安装alist-sdk现在从github强制安装

重构checker
init_alist支持多平台
完成CopyJob创建

alist_sync/checker.py

@@ -1,54 +1,63 @@
+from pathlib import PurePosixPath
+from pydantic import BaseModel
+
 from alist_sdk import Item
-from alist_sync.models import SyncDir
 
 
-class Checker:
-    """检查结果"""
+class Checker(BaseModel):
+    # matrix: 相对文件路径 -> {同步目录: Item}
+    matrix: dict[PurePosixPath, dict[PurePosixPath, Item]]
+    # cols: 同步目录 - base_path
+    cols: list[PurePosixPath]
 
-    def __init__(self, *scanned_dirs: SyncDir):
-        self._result: dict[str, dict[str, Item]] = dict()
-        self.table_title = [t.base_path for t in scanned_dirs]
+    @classmethod
+    def checker(cls, *scanned_dirs):
+        _result = {}
         for scanned_dir in scanned_dirs:
-            self.check(scanned_dir)
-
-    @property
-    def result(self):
-        return self._result
-
-    def check(self, scanned_dir: SyncDir):
-        for item in scanned_dir.items:
-            r_path = item.full_name.relative_to(
-                scanned_dir.base_path
-            )
-            try:
-                self._result[str(r_path)].setdefault(scanned_dir.base_path, item)
-            except KeyError:
-                self._result[str(r_path)] = {scanned_dir.base_path: item}
-
-    def rich_table(self):
-        """Rich的报表  """
+            for item in scanned_dir.items:
+                r_path = item.full_name.relative_to(
+                    scanned_dir.base_path
+                )
+                try:
+                    _result[PurePosixPath(r_path)].setdefault(
+                        PurePosixPath(scanned_dir.base_path),
+                        item
+                    )
+                except KeyError:
+                    _result[PurePosixPath(r_path)] = {
+                        PurePosixPath(scanned_dir.base_path): item
+                    }
+
+        return cls(
+            matrix=_result,
+            cols=[PurePosixPath(t.base_path) for t in scanned_dirs]
+        )
+
+    def model_dump_table(self):
+        """"""
         from rich.console import Console
         from rich.table import Table
 
         console = Console()
         table = Table(show_header=True, header_style="bold magenta")
-        table.add_column('r_path', style="dim red", width=120)
-        for col in self.table_title:
-            table.add_column(col, justify="center", vertical='middle')
+        table.add_column('r_path', style="dim red", )
+        for col in self.cols:
+            table.add_column(str(col), justify="center", vertical='middle')
 
-        for r_path, raw in self._result.items():
+        for r_path, raw in self.matrix.items():
             table.add_row(
-                r_path,
-                *[True if raw.get(tt) else "False" for tt in self.table_title]
+                str(r_path),
+                *["True" if raw.get(tt) else "False" for tt in self.cols]
             )
         console.print(table)
 
 
 if __name__ == '__main__':
     import json
     from pathlib import Path
+    from alist_sync.models import SyncDir
 
-    checker = Checker(*[SyncDir(**s) for s in json.load(
-        Path(__file__).parent.parent.joinpath('tests/resource/SyncDirs.json').open())
-                        ])
-    checker.rich_table()
+    checker = Checker.checker(*[SyncDir(**s) for s in json.load(
+        Path(__file__).parent.parent.joinpath('tests/resource/SyncDirs-m.json').open())
+                                ])
+    checker.model_dump_table()

alist_sync/job_copy.py

@@ -1,16 +1,19 @@
-
 from pathlib import Path, PurePosixPath
 from typing import Literal, Optional
-from pydantic import BaseModel, computed_field, Field
-
-from alist_sdk import Item
+from pydantic import BaseModel, computed_field
 
 from alist_sync.alist_client import AlistClient
-from alist_sync.models import Checker
+from alist_sync.checker import Checker
+from alist_sync.config import cache_dir
 
 CopyStatusModify = Literal[
-    "init", "created", "waiting",
-    "getting src object", "", "running", "success"
+    "init",
+    "created",
+    "waiting",
+    "getting src object",
+    "",
+    "running",
+    "success"
 ]
 
 
@@ -57,30 +60,59 @@ async def check_status(self, client: AlistClient, ):
 
 class CopyJob(BaseModel):
     """复制工作 - 从Checker中找出需要被复制的任务并创建"""
-
-    # client: AlistClient = Field(exclude=True)
-    # checker: Checker = Field(exclude=True)
+    # tasks: task_name -> task
     tasks: dict[str, CopyTask]
 
-    def dump(self):
-        pass
+    @classmethod
+    def from_json_file(cls, file: Path):
+        return cls.model_validate_json(
+            Path(file).read_text(encoding='utf-8')
+        )
 
     @classmethod
-    def load_from_file(file: Path):
-        pass
+    def from_cache(cls):
+        file = cache_dir.joinpath('copy_job.json')
+        return cls.from_json_file(file)
+
+    def save_to_cache(self):
+        file = cache_dir.joinpath('copy_job.json')
+        file.write_text(self.model_dump_json(indent=2), encoding='utf-8')
 
     @classmethod
-    def from_checker(source, target, checker: Checker, ):
+    def from_checker(cls, source, target, checker: Checker, ):
         """从Checker中创建Task"""
-        _tasks: dict[str, CopyTask] = {}
-        for r_path, pp in checker.matrix.items():
-            _s: Item | None = pp.get(source)
-            _t: Item | None = pp.get(target)
-            if _s is not None and _t is None:
-                _task = CopyTask(
-                    copy_name=_s.name,  # 需要复制到文件名
-                    copy_source=PurePosixPath(source.base_path).joinpath(
-                        copy_path.parent),  # 需要复制的源文件夹
-                    copy_target=PurePosixPath(target.base_path).joinpath(
-                        copy_path.parent),  # 需要复制到的目标文件夹
-                )
+        _tasks = {}
+
+        def _1_1(sp: PurePosixPath, tp: PurePosixPath):
+            sp, tp = PurePosixPath(sp), PurePosixPath(tp)
+            if sp not in checker.cols and tp not in checker.cols:
+                raise ValueError(f"Source: {sp} or Target: {tp} not in Checker")
+            for r_path, pp in checker.matrix.items():
+                if pp.get(sp) is not None and pp.get(tp) is None:
+                    __t = CopyTask(
+                        copy_name=pp.get(sp).name,
+                        copy_source=PurePosixPath(sp).joinpath(r_path.parent),
+                        copy_target=PurePosixPath(tp).joinpath(r_path.parent),
+                    )
+                    _tasks[__t.name] = __t
+
+        def _1_n(sp, tps):
+            sp = PurePosixPath(sp)
+            _tps = [PurePosixPath(tp) for tp in tps]
+            [_1_1(sp, tp) for tp in _tps if sp != tp]
+
+        def _n_n(sps, tps):
+            [_1_n(sp, tps) for sp in sps]
+
+        if isinstance(source, str | PurePosixPath) and isinstance(target, str | PurePosixPath):
+            _1_1(source, target)
+        elif isinstance(source, str | PurePosixPath) and isinstance(target, list | tuple):
+            _1_n(source, target)
+        elif isinstance(source, list | tuple) and isinstance(target, list | tuple):
+            _n_n(source, target)
+        else:
+            raise ValueError(f"source: {source} or target: {target} not support")
+
+        self = cls(tasks=_tasks)
+        self.save_to_cache()
+        return cls(tasks=_tasks)

alist_sync/models.py

@@ -8,8 +8,7 @@
 
 from alist_sync.alist_client import AlistClient
 
-__all__ = ["AlistServer", "SyncDir", "CopyTask",
-           "RemoveTask", "SyncJob", "Checker"]
+__all__ = ["AlistServer", "SyncDir", "RemoveTask", "SyncJob", ]
 
 
 class AlistServer(BaseModel):
@@ -100,59 +99,3 @@ class Config(BaseModel):
 
     update_time: datetime.datetime
     create_time: datetime.datetime
-
-
-class Checker(BaseModel):
-    matrix: dict[PurePosixPath, dict[PurePosixPath, Item]]
-    cols: list[PurePosixPath]
-
-    @classmethod
-    def checker(cls, *scanned_dirs):
-        _result = {}
-        for scanned_dir in scanned_dirs:
-            for item in scanned_dir.items:
-                r_path = item.full_name.relative_to(
-                    scanned_dir.base_path
-                )
-                try:
-                    _result[PurePosixPath(r_path)].setdefault(
-                        PurePosixPath(scanned_dir.base_path),
-                        item
-                    )
-                except KeyError:
-                    _result[PurePosixPath(r_path)] = {
-                        PurePosixPath(scanned_dir.base_path): item
-                    }
-
-        return cls(
-            matrix=_result,
-            cols=[PurePosixPath(t.base_path) for t in scanned_dirs]
-        )
-
-    def model_dump_table(self):
-        """"""
-        from rich.console import Console
-        from rich.table import Table
-
-        console = Console()
-        table = Table(show_header=True, header_style="bold magenta")
-        table.add_column('r_path', style="dim red", )
-        for col in self.cols:
-            table.add_column(str(col), justify="center", vertical='middle')
-
-        for r_path, raw in self.matrix.items():
-            table.add_row(
-                str(r_path),
-                *["True" if raw.get(tt) else "False" for tt in self.cols]
-            )
-        console.print(table)
-
-
-if __name__ == '__main__':
-    import json
-    from pathlib import Path
-
-    checker = Checker.checker(*[SyncDir(**s) for s in json.load(
-        Path(__file__).parent.parent.joinpath('tests/resource/SyncDirs-m.json').open())
-    ])
-    checker.model_dump_table()

bootstarp.sh

@@ -1,12 +1,12 @@
-#!/bin/env bash
+#!/bin/bash
 
 cd "$(dirname "$0")" || exit 1
 
 all_clear() {
     echo ".pytest_cache"
-    find -type d -name ".pytest_cache" -exec rm -rf {} \;
+    find . -type d -name ".pytest_cache" -exec rm -rf {} \; 2>/dev/null
     echo  "__pycache__"
-    find -type d -name "__pycache__" -exec rm -rf {} \;
+    find . -type d -name "__pycache__" -exec rm -rf {} \; 2>/dev/null
     echo ".cache"
     rm -rf alist_sync/.cache alist_sync.egg-info
 }
@@ -15,6 +15,7 @@ case $1 in
 install ) 
     pip install -U pip
     pip install -e .
+    pip install git+https://github.com/lee-cq/alist-sdk --no-cache-dir
     ;;
 
 alist-init ) 
@@ -23,6 +24,14 @@ alist-init )
     tests/init_alist.sh
     ;;
 
+alist-version )
+    cd tests/alist  || {
+        echo "未初始化 -  执行 alist-init"
+        exit 2
+    }
+    ./alist version
+    ;;
+
 alist-run )
     cd tests/alist  || {
         echo "未初始化 -  执行 alist-init"
@@ -46,4 +55,8 @@ test )
     all_clear
     pytest -v
     ;;
+* )
+    echo "Usage: $0 {install|alist-init|alist-version|alist-run|alist-stop|clear|test}"
+    exit 1
+    ;;
 esac

tests/init_alist.sh

@@ -7,10 +7,27 @@ cd alist || exit
 
 VERSION=${ALIST_VERSION:-"v3.29.1"}
 
+
+platform=$(uname -s | tr '[:upper:]' '[:lower:]')
+case $platform in
+  linux | darwin)    fix=".tar.gz"    ;;
+  windows*)          fix=".zip"       ;;
+esac
+
+case $(uname -m) in
+  x86_64)                 cpu="amd64"        ;;
+  i386 | i686)            cpu="386"          ;;
+  aarch64 | arm64)        cpu="arm64"        ;;
+esac
+filename="alist-${platform}-${cpu}${fix}"
+export download_url="https://github.com/alist-org/alist/releases/download/${VERSION}/${filename}"
+
 if [ ! -f alist ]; then
-    echo Will Install ${VERSION}
-    wget -q https://github.com/alist-org/alist/releases/download/${VERSION}/alist-linux-amd64.tar.gz
-    tar xzvf alist-linux-amd64.tar.gz
+  set -e
+  echo "Will Install ${download_url}"
+  wget -q "$download_url"
+  tar xzvf "$filename"
+  set +e
 fi
 
 rm -rf data/ test_dir/