check 优化

alist_sync/__main__.py

@@ -7,6 +7,7 @@
 
 from alist_sync.models import AlistServer
 
+from alist_sync.base_sync import SyncBase
 from alist_sync.run_copy import CopyToTarget
 from alist_sync.run_mirror import Mirror
 from alist_sync.run_sync import Sync
@@ -44,8 +45,34 @@
     help="一个包含存储配置的JSON文件，可以是AList的备份文件"
 )
 
+@app.command(name='check')
+def check(
+        base_url: str = _base_url,
+        username: str = _username,
+        password: str = _password,
+        token: str = _token,
+        verify: bool = _verify,
+        storage_config: str = _stores_config,
+        target: list[str] = Option(..., "--target", "-t", help="Check Path"),
+):
+    """检查各个存储中的文件状态"""
+    alist_info = AlistServer(
+        base_url=base_url,
+        username=username,
+        password=password,
+        token=token,
+        verify=verify,
+        storage_config=storage_config,
+    )
+    echo(
+        f"Will Be Check {target} "
+        "on {alist_info.base_url} [{alist_info.username}]"
+    )
+    return SyncBase(alist_info=alist_info, sync_dirs=target).sync_task.checker.model_dump_table()
 
-@app.command()
+
+
+@app.command(name="copy")
 def copy(
         base_url: str = _base_url,
         username: str = _username,

alist_sync/base_sync.py

@@ -4,7 +4,7 @@
 
 from alist_sync.alist_client import AlistClient
 from alist_sync.config import cache_dir
-from alist_sync.models import SyncTask, AlistServer
+from alist_sync.models import SyncTask, AlistServer, Checker
 from alist_sync.scan_dir import scan_dir
 from alist_sync.common import sha1_6, is_task_all_success, timeout_input
 
@@ -103,6 +103,7 @@ async def async_run(self):
         if not self.sync_task.sync_dirs.values():
             await self.scans()
             self.save_to_cache()
+            self.sync_task.checker = Checker.checker(*self.sync_task.sync_dirs.values())
         else:
             logger.info(f"一件从缓存中找到 %d 个 SyncDir",
                         len(self.sync_task.sync_dirs))

alist_sync/models.py

@@ -1,16 +1,16 @@
 import datetime
 import json
 from pathlib import PurePosixPath, Path
-from typing import Optional, Literal
+from typing import  Optional, Literal
 
 from pydantic import BaseModel, computed_field, Field
 from alist_sdk import Item
 
 from alist_sync.alist_client import AlistClient
 
-__all__ = ["AlistServer", "SyncDir", "CopyTask", "RemoveTask", "SyncTask"]
+__all__ = ["AlistServer", "SyncDir", "CopyTask", "RemoveTask", "SyncTask", "Checker"]
 
-StatusModify = Literal["init", "created", ]
+CopyStatusModify = Literal["init", "created", "waitting","getting src object", "", "running", "seccess"]
 
 
 class AlistServer(BaseModel):
@@ -76,12 +76,6 @@ def in_items(self, path) -> bool:
                 self.base_path) for i in self.items]
         return path in self.items_relative
 
-    # @field_serializer('items_relative')
-    # def serializer_items_relative(self, value, info) -> list:
-    #     value: list[PurePosixPath] = []
-    #     info: object
-    #     return value
-
 
 class CopyTask(BaseModel):
     """复制任务"""
@@ -90,13 +84,8 @@ class CopyTask(BaseModel):
     copy_source: PurePosixPath  # 需要复制的源文件夹
     copy_target: PurePosixPath  # 需要复制到的目标文件夹
     # 任务状态 init: 初始化，created: 已创建，"getting src object": 运行中，"": 已完成，"failed": 失败
-    status: str = "init"
+    status: CopyStatusModify = "init"
     message: Optional[str] = ''
-
-    # @field_serializer('copy_source', 'copy_target')
-    # def serializer_path(self, value: PurePosixPath, info):
-    #     return value.as_posix()
-
     @computed_field()
     @property
     def name(self) -> str:
@@ -138,6 +127,7 @@ class SyncTask(BaseModel):
     """同步任务"""
     alist_info: AlistServer  # Alist Info
     sync_dirs: dict[str, SyncDir] = {}  # 同步目录
+    checker: Optional['Checker'] = None
     copy_tasks: dict[str, CopyTask] = {}  # 复制目录
     remove_tasks: dict[str, RemoveTask] = {}  # 删除目录
 
@@ -154,17 +144,49 @@ class Config(BaseModel):
     create_time: datetime.datetime
 
 
-if __name__ == '__main__':
-    print(
-        SyncTask(
-            alist_info=AlistServer(),
-            sync_dirs=[],
-            copy_tasks={
-                'aa': CopyTask(
-                    copy_name='aa',
-                    copy_source=PurePosixPath('/a'),
-                    copy_target=PurePosixPath('/b/aa')
+class Checker(BaseModel):
+    mixmatrix: 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
                 )
-            },
-        ).model_dump_json(indent=2)
-    )
+                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(mixmatrix=_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.mixmatrix.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.json').open())
+                        ])
+    checker.model_dump_table()

bootstarp.sh

@@ -0,0 +1,49 @@
+#!/bin/env bash
+
+cd "$(dirname "$0")" || exit 1
+
+all_clear() {
+    echo ".pytest_cache"
+    find -type d -name ".pytest_cache" -exec rm -rf {} \;
+    echo  "__pycache__"
+    find -type d -name "__pycache__" -exec rm -rf {} \;
+    echo ".cache"
+    rm -rf alist_sync/.cache alist_sync.egg-info
+}
+
+case $1 in
+install ) 
+    pip install -U pip
+    pip install -e .
+    ;;
+
+alist-init ) 
+    pkill alist
+    rm -rf tests/alist
+    tests/init_alist.sh
+    ;;
+
+alist-run )
+    cd tests/alist  || {
+        echo "未初始化 -  执行 alist-init"
+        exit 2
+    }
+    ./alist restart
+    ;;
+
+alist-stop ) 
+    ./tests/alist/alist stop || pkill alist
+    ;;
+
+clear )  
+    all_clear 
+    ;;
+
+test ) 
+    whereis pytest || pip install pytest
+    clear
+    pkill alist
+    all_clear
+    pytest -v
+    ;;
+esac

tests/common.py

@@ -25,10 +25,12 @@ def create_storage_local(client_, mount_name, local_path: Path):
     if f'/{mount_name}' not in [
         i['mount_path'] for i in client_.get("/api/admin/storage/list").json()['data']['content']
     ]:
-        assert client_.post(
+        data = client_.post(
             "/api/admin/storage/create",
-            json=local_storage
-        ).json().get('code') == 200, "创建Storage失败。"
+            json=local_storage).json()
+
+        assert data.get('code') == 200, data.get(
+            "message") + f", {mount_name = }"
     else:
         print("已经创建，跳过 ...")
 
@@ -42,4 +44,3 @@ def clear_dir(path: Path):
         else:
             item.unlink()
     path.rmdir()
-

tests/init_alist.sh

@@ -5,8 +5,11 @@ cd "$(dirname "$0")" || exit
 mkdir -p alist
 cd alist || exit
 
+VERSION=${ALIST_VERSION:-"v3.29.1"}
+
 if [ ! -f alist ]; then
-    wget -q https://github.com/alist-org/alist/releases/download/v3.29.1/alist-linux-amd64.tar.gz
+    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
 fi
 

tests/test_check.py

@@ -3,7 +3,7 @@
 
 import pytest
 
-from alist_sync.models import SyncDir
+from alist_sync.models import SyncDir, CheckModel
 from alist_sync.checker import Checker
 
 SUP_DIR = Path(__file__).parent.joinpath('resource')
@@ -16,3 +16,11 @@
 def test_checker(scanned_dirs):
     checker = Checker(*scanned_dirs)
     assert checker.result
+
+@pytest.mark.parametrize(
+    "scanned_dirs",
+    [[SyncDir(**s) for s in json.load(SUP_DIR.joinpath('SyncDirs.json').open())]]
+)
+def test_check(scanned_dirs):
+    checker = CheckModel.checker(*scanned_dirs)
+    assert checker.mixmatrix

tests/test_sync.py

@@ -22,7 +22,7 @@
 
 def setup_module():
     print("setup_module")
-    os.system(f'bash {WORKDIR}/init_alist.sh')
+    assert os.system(f'bash {WORKDIR}/init_alist.sh') == 0
     DATA_DIR.mkdir(parents=True, exist_ok=True)
     DATA_DIR_DST.mkdir(parents=True, exist_ok=True)
     DATA_DIR_DST2.mkdir(parents=True, exist_ok=True)