Merge pull request #4 from ehdgua01/develop

.pre-commit-config.yaml

@@ -22,8 +22,15 @@ repos:
     rev: v5.8.0
     hooks:
       - id: isort
+        args: [--profile=black, --line-length=120]
 
   - repo: https://github.com/pre-commit/mirrors-mypy
     rev: v0.812
     hooks:
       - id: mypy
+
+  - repo: https://github.com/pre-commit/pygrep-hooks
+    rev: v1.8.0
+    hooks:
+      - id: python-check-blanket-noqa
+      - id: python-check-mock-methods

README.md

@@ -1,21 +1,20 @@
 # blocksync
 
-[![Build](https://travis-ci.com/ehdgua01/blocksync.svg?branch=master)](https://travis-ci.com/github/ehdgua01/blocksync)
-[![Coverage](https://codecov.io/gh/ehdgua01/blocksync/branch/master/graph/badge.svg)](https://app.codecov.io/gh/ehdgua01/blocksync)
-[![PyPi](https://badge.fury.io/py/blocksync.svg)](https://pypi.org/project/blocksync/)
-[![PyVersions](https://img.shields.io/pypi/pyversions/blocksync)](https://pypi.org/project/blocksync/)
-
 Blocksync Python package allows [blocksync script](https://github.com/theraser/blocksync) to be used as Python packages,
 and supports more convenient and various functions than blocksync script.
 
+[![Build](https://img.shields.io/travis/ehdgua01/blocksync/master.svg?style=for-the-badge&logo=travis)](https://travis-ci.com/github/ehdgua01/blocksync)
+[![PyPi](https://img.shields.io/pypi/v/blocksync?logo=pypi&style=for-the-badge)](https://pypi.org/project/blocksync/)
+[![PyVersions](https://img.shields.io/pypi/pyversions/blocksync?logo=python&style=for-the-badge)](https://pypi.org/project/blocksync/)
+
 # Prerequisites
 
 - Python 3.8 or later
 
 # Features
 
 - Synchronize the destination (remote or local) files using an incremental algorithm.
-- Supports all synchronization directions. (local-local, local-remote, remote-local, remote-remote)
+- Supports all synchronization directions. (local-local, local-remote, remote-local)
 - Support for callbacks that can run before(run once or per workers), after(run once or per workers), and during synchronization of files
 - Support for synchronization suspend/resume, cancel.
 - Most methods support method chaining.
@@ -31,23 +30,22 @@ pip install blocksync
 
 # Quick start
 
-When using SFTP files, you can check the SSH connection options in [paramiko docs](http://docs.paramiko.org/en/stable/api/client.html#paramiko.client.SSHClient).
+When sync from/to remote, you can check the SSH connection options in [paramiko docs](http://docs.paramiko.org/en/stable/api/client.html#paramiko.client.SSHClient).
 
 ```python
-from blocksync import LocalFile, SFTPFile, Syncer
-
-
-syncer = Syncer(
-    src=SFTPFile(
-        path="src.file",
-        hostname="hostname",
-        username="username",
-        password="password",
-        key_filename="key_filepath",
-    ),
-    dest=LocalFile(path="dest.file"),
-)
-syncer.start_sync(workers=2, create=True, wait=True)
+from blocksync import local_to_local
+
+
+manager, status = local_to_local("src.txt", "dest.txt", workers=4)
+manager.wait_sync()
+print(status)
+
+# Output
+[Worker 1]: Start sync(src.txt -> dest.txt) 1 blocks
+[Worker 2]: Start sync(src.txt -> dest.txt) 1 blocks
+[Worker 3]: Start sync(src.txt -> dest.txt) 1 blocks
+[Worker 4]: Start sync(src.txt -> dest.txt) 1 blocks
+{'workers': 4, 'chunk_size': 250, 'block_size': 250, 'src_size': 1000, 'dest_size': 1000, 'blocks': {'same': 4, 'diff': 0, 'done': 4}}
 ```
 
 

blocksync/__init__.py

@@ -1,4 +1,5 @@
-from blocksync.files import File, LocalFile, SFTPFile
-from blocksync.syncer import Syncer
+from blocksync._status import Status
+from blocksync._sync_manager import SyncManager
+from blocksync.sync import local_to_local, local_to_remote, remote_to_local
 
-__all__ = ["File", "LocalFile", "SFTPFile", "Syncer"]
+__all__ = ["local_to_local", "local_to_remote", "remote_to_local", "Status", "SyncManager"]

blocksync/_consts.py

@@ -0,0 +1,37 @@
+import re
+from pathlib import Path
+
+__all__ = ["BASE_DIR", "ByteSizes", "SAME", "SKIP", "DIFF"]
+
+BASE_DIR = Path(__file__).parent
+SAME: str = "0"
+SKIP: str = "1"
+DIFF: str = "2"
+
+
+class ByteSizes:
+    BLOCK_SIZE_PATTERN = re.compile("([0-9]+)(B|KB|MB|GB|KiB|K|MiB|M|GiB|G)")
+
+    B: int = 1
+    KB: int = 1000
+    MB: int = 1000 ** 2
+    GB: int = 1000 ** 3
+    KiB: int = 1 << 10
+    K = KiB
+    MiB: int = 1 << 20
+    M = MiB
+    GiB: int = 1 << 30
+    G = GiB
+
+    @classmethod
+    def parse_readable_byte_size(cls, size: str) -> int:
+        """
+        Examples
+        1MB -> 1000000
+        1M, 1MiB -> 10478576
+        """
+        if not size.isdigit():
+            if matched := cls.BLOCK_SIZE_PATTERN.match(size):
+                size, unit = matched.group(1), matched.group(2).strip()
+                return int(size) * getattr(ByteSizes, unit.upper())
+        return int(size)

blocksync/_hooks.py

@@ -0,0 +1,35 @@
+from typing import Any, Callable, Optional
+
+from blocksync._status import Status
+
+__all__ = ["Hooks"]
+
+
+class Hooks:
+    def __init__(
+        self,
+        on_before: Optional[Callable[..., Any]],
+        on_after: Optional[Callable[[Status], Any]],
+        monitor: Optional[Callable[[Status], Any]],
+        on_error: Optional[Callable[[Exception, Status], Any]],
+    ):
+        self.before: Optional[Callable[..., Any]] = on_before
+        self.after: Optional[Callable[[Status], Any]] = on_after
+        self.monitor: Optional[Callable[[Status], Any]] = monitor
+        self.on_error: Optional[Callable[[Exception, Status], Any]] = on_error
+
+    def _run(self, hook: Optional[Callable], *args, **kwargs):
+        if hook:
+            hook(*args, **kwargs)
+
+    def run_before(self):
+        self._run(self.before)
+
+    def run_after(self, status: Status):
+        self._run(self.after, status)
+
+    def run_monitor(self, status: Status):
+        self._run(self.monitor, status)
+
+    def run_on_error(self, exc: Exception, status: Status):
+        self._run(self.on_error, exc, status)

blocksync/_read_server.py

@@ -0,0 +1,29 @@
+import hashlib
+import io
+import sys
+from typing import Callable
+
+DIFF = b"2"
+COMPLEN = len(DIFF)
+path: bytes = sys.stdin.buffer.readline().strip()
+stdout = sys.stdout.buffer
+stdin = sys.stdin.buffer
+
+fileobj = open(path, "rb")
+fileobj.seek(io.SEEK_SET, io.SEEK_END)
+print(fileobj.tell(), flush=True)
+
+block_size: int = int(stdin.readline())
+hash_: Callable = getattr(hashlib, stdin.readline().strip().decode())
+startpos: int = int(stdin.readline())
+maxblock: int = int(stdin.readline())
+
+with fileobj:
+    fileobj.seek(startpos)
+    for _ in range(maxblock):
+        block = fileobj.read(block_size)
+        stdout.write(hash_(block).digest())
+        stdout.flush()
+        if stdin.read(COMPLEN) == DIFF:
+            stdout.write(block)
+            stdout.flush()

blocksync/_status.py

@@ -0,0 +1,41 @@
+import threading
+from typing import Literal, TypedDict
+
+
+class Blocks(TypedDict):
+    same: int
+    diff: int
+    done: int
+
+
+class Status:
+    def __init__(
+        self,
+        workers: int,
+        block_size: int,
+        src_size: int,
+        dest_size: int = 0,
+    ):
+        self._lock = threading.Lock()
+        self.workers: int = workers
+        self.chunk_size: int = src_size // workers
+        self.block_size: int = block_size
+        self.src_size: int = src_size
+        self.dest_size: int = dest_size
+        self.blocks: Blocks = Blocks(same=0, diff=0, done=0)
+
+    def __repr__(self):
+        return str({k: v for k, v in self.__dict__.items() if k != "_lock"})
+
+    def add_block(self, block_type: Literal["same", "diff"]):
+        with self._lock:
+            self.blocks[block_type] += 1
+            self.blocks["done"] = self.blocks["same"] + self.blocks["diff"]
+
+    @property
+    def rate(self) -> float:
+        return (
+            min(100.00, (self.blocks["done"] / (self.src_size // self.block_size)) * 100)
+            if self.blocks["done"] > 1
+            else 0.00
+        )

blocksync/_sync_manager.py

@@ -0,0 +1,41 @@
+import threading
+from typing import List
+
+
+class SyncManager:
+    def __init__(self):
+        self.workers: List[threading.Thread] = []
+        self._suspend: threading.Event = threading.Event()
+        self._suspend.set()
+        self._cancel: bool = False
+
+    def cancel_sync(self):
+        self._cancel = True
+
+    def wait_sync(self):
+        for worker in self.workers:
+            worker.join()
+
+    def suspend(self):
+        self._suspend.clear()
+
+    def resume(self):
+        self._suspend.set()
+
+    def _wait_resuming(self):
+        self._suspend.wait()
+
+    @property
+    def canceled(self) -> bool:
+        return self._cancel
+
+    @property
+    def suspended(self) -> bool:
+        return not self._suspend.is_set()
+
+    @property
+    def finished(self) -> bool:
+        for worker in self.workers:
+            if worker.is_alive():
+                return False
+        return True

blocksync/_write_server.py

@@ -0,0 +1,25 @@
+import io
+import sys
+
+DIFF = b"2"
+COMPLEN = len(DIFF)
+stdin = sys.stdin.buffer
+
+path = stdin.readline().strip()
+
+size = int(stdin.readline())
+if size > 0:
+    with open(path, "a+") as fileobj:
+        fileobj.truncate(size)
+
+block_size = int(stdin.readline())
+startpos = int(stdin.readline())
+maxblock = int(stdin.readline())
+
+with open(path, mode="rb+") as f:
+    f.seek(startpos)
+    for _ in range(maxblock):
+        if stdin.read(COMPLEN) == DIFF:
+            f.write(stdin.read(block_size))
+        else:
+            f.seek(block_size, io.SEEK_CUR)