commit 64d2b6376710ae769a554c34471e7709bea2f55b Author: jkooe <62494645+jkooe@users.noreply.github.com> Date: Mon Oct 12 17:58:30 2026 +0800 初始化 sdimport：检测视频 SD 卡自动导入工具（macOS/Windows 跨平台） diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..f236a13 --- /dev/null +++ b/.gitignore @@ -0,0 +1,16 @@ +# 运行时产物（日志/状态含真实卷名、楼栋号、用户名、时间戳，绝不入库） +runtime/ + +# 历史备份目录 +src.bak-*/ + +# WorkBuddy 项目记忆（私人工作记录） +.workbuddy/ + +# Python +__pycache__/ +*.py[cod] +*.egg-info/ + +# macOS +.DS_Store diff --git a/README.md b/README.md new file mode 100644 index 0000000..ad05794 --- /dev/null +++ b/README.md @@ -0,0 +1,154 @@ +# sdimport — 检测视频 SD 卡自动导入 + +插入记录仪 SD 卡 → 自动按拍摄日期归类 → 落盘 → 清空卡内副本 → 延时自动弹出。 + +macOS 与 Windows 共用同一套核心逻辑，系统差异全部收敛在 `src/sdimport/platforms/`。 + +``` +插卡 → 认卡 → 按日期分组 → 对账 → 倒计时确认 → 拷贝 → 校验 → 清卡 → 完成提示 → 延时弹卡 +``` + +| 环节 | 规则 | +|---|---| +| **认卡** | 不认卷名，认卡内有没有 `VID-YYYYMMDD-HHMMSS.AVI`。换卡、换记录仪、卡被改名都不受影响 | +| **分组** | 日期只取**文件名**里的八位数，不看卡上目录名（实测过目录名与拍摄日期错位） | +| **范围** | 只扫 `scan_roots` 指定的目录，**绝不进 `VIDEO/` 的子目录** —— 那里面是历史归档 | +| **对账** | 已落盘的跳过；同名但体积不符的另起名字，**绝不覆盖** | +| **校验** | 体积 + 头尾各 64KB 指纹一致才算成功 | +| **清卡** | 校验通过才删卡内源文件；单文件失败即整批中止，**不删任何源文件** | +| **弹卡** | 完成后弹提示，N 分钟后自动弹出；期间可点「暂不弹出」，或执行 `sdimport keep` 取消 | + +--- + +## 安装 + +### macOS（已实测） + +```bash +cd ~/WorkBuddy/检测视频自动导入 +./bin/sdimport selftest # 环境自检 +./bin/sdimport install # 装后台服务（launchd） +``` + +若 `install` 提示「未能在本次会话中加载」，双击项目根目录的 **安装后台服务.command**， +或在终端粘贴 `launchctl bootstrap gui/$(id -u) ~/Library/LaunchAgents/com.sdimport.agent.plist`。 +什么都不做也行 —— 注销或重启后 macOS 会自动加载。 + +触发方式：`WatchPaths` 监视 `/Volumes`（插卡即响应）+ 每 120 秒兜底检查一次。 + +### Windows（**尚未在真机实测**） + +1. 装 Python 3.9+（），安装时勾选 **Add python.exe to PATH**。 +2. 把本项目整个目录拷到那台 Windows 机器上，路径**不要含空格**最省事。 +3. 先自检，再演练，最后安装： + +```bat +bin\sdimport.cmd selftest +bin\sdimport.cmd scan +bin\sdimport.cmd once --no-delete +bin\sdimport.cmd install +``` + +或直接双击 **install-service.cmd**。 + +触发方式：常驻进程 `sdimport daemon` 订阅 WMI 卷插入事件（秒级响应）+ 每 30 秒轮询兜底， +由任务计划 `sdimport` 在登录时自动拉起。 + +> **可选提速**：`pip install wmi` 之后插卡是秒级响应；不装也能用，只是最坏延迟 30 秒。 + +--- + +## 命令 + +| 命令 | 作用 | +|---|---| +| `sdimport scan` | 演练：只显示会导入什么，**不做任何写入** | +| `sdimport once` | 立即执行一次导入 | +| `sdimport once --no-delete` | 导入但保留卡内文件 | +| `sdimport status` | 服务状态、待弹出的任务、最近导入记录 | +| `sdimport log` | 查看最近日志 | +| `sdimport keep` | 取消待执行的自动弹出（卡继续挂着） | +| `sdimport eject` | 立刻弹出 SD 卡 | +| `sdimport selftest` | 环境自检 | +| `sdimport install` / `uninstall` | 安装 / 卸载后台服务 | +| `sdimport daemon` | 常驻进程（Windows 用，一般无需手动跑） | + +macOS 用 `./bin/sdimport`，Windows 用 `bin\sdimport.cmd`，参数完全一致。 + +--- + +## 配置（`config.json`） + +改动前请先读 `.workbuddy/memory/MEMORY.md` 的「铁律」一节。 + +| 键 | 默认 | 说明 | +|---|---|---| +| `target_root` | `~/Downloads/检测视频` | 落盘根目录，按 `//` 组织 | +| `filename_regex` | `^VID-(\d{8})-(\d{6})\.(AVI\|MP4\|…)$` | **只碰匹配此模式的文件**，其余一律不动 | +| `scan_roots` | `["", "VIDEO", "DCIM:2", "MOVIE", "PRIVATE"]` | 扫描根。`""` 是卷根；`:N` 是该根下最多再进几层子目录 | +| `scan_max_depth` | `0` | 未写 `:N` 时的默认下潜层数 | +| `min_video_bytes` | `1024` | 只挡 0 字节 / 畸形残片 | +| `tiny_video_bytes` | `1048576` | 小于此值的只**标注**、照样导入，不丢数据 | +| `delete_source_after_verify` | `true` | 校验通过后是否清卡 | +| `notify_countdown_seconds` | `60` | 开始导入前的倒计时，期间可取消 | +| `eject_after_import` | `true` | 导入完成后是否自动弹出 | +| `eject_delay_seconds` | `300` | 延时多久弹出 | +| `completion_dialog_seconds` | `30` | 完成提示框停留秒数；设 `0` 则只发通知不弹框 | +| `eject_finder_fallback` | `false` | macOS：改走「访达推出」通道。**需先授予「自动化」权限**，否则会弹框阻塞 | +| `windows_only_removable` | `true` | Windows：只把可移动磁盘当候选卷（硬安全线，别轻易关） | + +调试用环境变量：`SDIMPORT_TARGET_ROOT`、`SDIMPORT_COUNTDOWN`、`SDIMPORT_EJECT_DELAY`、 +`SDIMPORT_COMPLETION_DIALOG`。 + +--- + +## 目录结构 + +``` +install-service.cmd Windows：双击安装后台服务 +uninstall-service.cmd Windows：双击卸载 +bin/sdimport macOS / Linux 入口 +bin/sdimport.cmd Windows 入口 +bin/on-volume-change.sh launchd 触发入口 +src/sdimport/ + planner.py 按日期分组与落盘规划 （跨平台） + transfer.py 拷贝 → 校验 → 删源 （跨平台） + pipeline.py 主流程编排 （跨平台） + scanner.py 卷内录像扫描规则 （跨平台） + runner.py auto 一轮流程 （跨平台） + daemon.py 常驻进程（Windows 触发用） （跨平台） + journal.py 日志、状态、单实例锁 （跨平台） + paths.py 路径与配置 （跨平台） + volumes.py / notify.py / ejector.py 薄门面，委托平台 + platforms/ + base.py 接口与 Volume 模型 + macos.py /Volumes + diskutil + osascript + launchd + windows.py 盘符 + kernel32 + PowerShell + 任务计划 + generic.py 兜底（Linux 等） +runtime/logs/ import-YYYYMMDD.log、launchd.err.log、eject.log +runtime/state/ state.json、pending-eject.json +``` + +--- + +## 故障排查 + +| 现象 | 处置 | +|---|---| +| 插卡没反应 | `sdimport status` 看服务在不在；`sdimport log` 看最近日志 | +| 说「判定为非记录仪卡」 | 卡上确实没有符合 `filename_regex` 的文件；或卡上文件都在子目录里（按设计不纳入） | +| 说「另有 N 个录像不纳入」 | 那些在扫描根的子目录里，属归档范围，**按设计不动**。真要处理就改 `scan_roots` | +| macOS 卡弹不出来 | 多半是访达窗口开着这张卡。关掉窗口，或在访达侧边栏点「推出」 | +| Windows 卡弹不出来 | 关掉资源管理器里该盘的窗口；或右键该盘选「弹出」 | +| 想临时保住卡不弹 | `sdimport keep` | + +**排查第一站永远是 `runtime/logs/` 下的日志**，每次导入的每个文件都留了痕。 + +--- + +## 边界声明 + +- 本程序只做「SD 卡 → `//` → 清卡」这一段。 + 电梯编号识别由 `elevator-video-classifier` skill 负责，那一步需要人工校对，两者不重叠。 +- **Windows 侧代码未在真机实测**（开发机是 macOS）。写法上已尽量保守 + （纯标准库优先、依赖缺失自动降级、无 GUI 会话也能工作），但仍建议先跑 `selftest` 与 `scan`。 diff --git a/bin/on-volume-change.sh b/bin/on-volume-change.sh new file mode 100755 index 0000000..10bf0e2 --- /dev/null +++ b/bin/on-volume-change.sh @@ -0,0 +1,32 @@ +#!/bin/bash +# 插卡触发入口（由 launchd 调用）。 +# +# launchd 在 /Volumes 变化或每 120 秒兜底触发本脚本；真正的逻辑与并发控制在 Python 侧。 +# 这里只负责：等待挂载落定 -> 交给 sdimport auto。 +# 路径按脚本自身位置推导，项目整个搬走也不会失效。 + +set -u + +BIN_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +PROJECT_ROOT="$(dirname "$BIN_DIR")" + +if [ -x /opt/homebrew/bin/python3 ]; then + PYTHON_BIN=/opt/homebrew/bin/python3 +elif command -v python3 >/dev/null 2>&1; then + PYTHON_BIN="$(command -v python3)" +elif [ -x /usr/bin/python3 ]; then + PYTHON_BIN=/usr/bin/python3 +else + echo "[sdimport] 找不到可用的 python3" >&2 + exit 127 +fi + +mkdir -p "$PROJECT_ROOT/runtime/logs" + +# 等文件系统挂载稳定（launchd 触发时卷可能刚出现） +sleep 4 + +export PYTHONPATH="$PROJECT_ROOT/src" +export PYTHONDONTWRITEBYTECODE=1 + +exec "$PYTHON_BIN" -m sdimport.cli auto "$@" diff --git a/bin/sdimport b/bin/sdimport new file mode 100755 index 0000000..c5377c9 --- /dev/null +++ b/bin/sdimport @@ -0,0 +1,31 @@ +#!/bin/bash +# sdimport 命令行包装（macOS / Linux）。 +# +# sdimport scan 演练，看看会导入什么 +# sdimport once 立即导入一次 +# sdimport status 查看服务状态与历史 +# sdimport keep 取消待执行的自动弹出 +# sdimport install 安装插卡自动触发 +# +# 路径全部按脚本自身位置推导，项目整个搬走也不会失效。 + +set -u + +BIN_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +PROJECT_ROOT="$(dirname "$BIN_DIR")" + +export PYTHONPATH="$PROJECT_ROOT/src" +export PYTHONDONTWRITEBYTECODE=1 + +if [ -x /opt/homebrew/bin/python3 ]; then + PYTHON_BIN=/opt/homebrew/bin/python3 +elif command -v python3 >/dev/null 2>&1; then + PYTHON_BIN="$(command -v python3)" +elif [ -x /usr/bin/python3 ]; then + PYTHON_BIN=/usr/bin/python3 +else + echo "找不到 python3，请先安装 Python 3.9 及以上。" >&2 + exit 1 +fi + +exec "$PYTHON_BIN" -m sdimport.cli "$@" diff --git a/bin/sdimport.cmd b/bin/sdimport.cmd new file mode 100644 index 0000000..f6d09a2 --- /dev/null +++ b/bin/sdimport.cmd @@ -0,0 +1,16 @@ +@echo off +rem sdimport launcher for Windows. Keep this file ASCII-only (encoding safety). +setlocal +chcp 65001 >nul +set "ROOT=%~dp0.." +set "PYEXE=" +for /f "delims=" %%i in ('where python.exe 2^>nul') do if not defined PYEXE set "PYEXE=%%i" +if not defined PYEXE for /f "delims=" %%i in ('where py.exe 2^>nul') do if not defined PYEXE set "PYEXE=%%i" +if not defined PYEXE ( + echo [ERROR] Python not found. Install Python 3.9+ and tick "Add python.exe to PATH". + exit /b 1 +) +set "PYTHONPATH=%ROOT%\src" +set "PYTHONDONTWRITEBYTECODE=1" +"%PYEXE%" -m sdimport.cli %* +exit /b %ERRORLEVEL% diff --git a/config.json b/config.json new file mode 100644 index 0000000..6209735 --- /dev/null +++ b/config.json @@ -0,0 +1,47 @@ +{ + "target_root": "~/Downloads/检测视频", + "filename_regex": "^VID-(\\d{8})-(\\d{6})\\.(?:AVI|MP4|MOV|MTS|MPG|MPEG)$", + "case_insensitive": true, + "scan_roots": [ + "", + "VIDEO", + "DCIM:2", + "MOVIE", + "PRIVATE" + ], + "scan_max_depth": 0, + "deny_dir_names": [ + "System Volume Information", + "$RECYCLE.BIN", + "System", + "FOUND.000" + ], + "report_skipped_archives": true, + "min_video_bytes": 1024, + "tiny_video_bytes": 1048576, + "copy_chunk_mb": 8, + "verify_sample_kb": 64, + "min_free_space_ratio": 1.1, + "delete_source_after_verify": true, + "delete_batch_size": 10, + "notify_countdown_seconds": 60, + "eject_after_import": true, + "eject_delay_seconds": 300, + "completion_dialog_seconds": 30, + "eject_retry_attempts": 5, + "eject_retry_interval": 15, + "eject_finder_fallback": false, + "deny_volumes": [ + "Macintosh HD", + "Recovery", + "TimeMachine", + "VM", + "Preboot", + "Update" + ], + "allow_readonly_copy": true, + "windows_only_removable": true, + "settle_seconds": 4, + "stability_checks": 3, + "stability_interval": 0.8 +} diff --git a/install-service.cmd b/install-service.cmd new file mode 100644 index 0000000..ffa3887 --- /dev/null +++ b/install-service.cmd @@ -0,0 +1,10 @@ +@echo off +chcp 65001 >nul +cd /d "%~dp0" +echo ================================================================ +echo sdimport - install background service (Windows) +echo ================================================================ +echo. +call "%~dp0bin\sdimport.cmd" install +echo. +pause diff --git a/src/sdimport/__init__.py b/src/sdimport/__init__.py new file mode 100644 index 0000000..394deea --- /dev/null +++ b/src/sdimport/__init__.py @@ -0,0 +1,3 @@ +"""检测视频 SD 卡自动导入。""" + +__version__ = "1.0.0" diff --git a/src/sdimport/cli.py b/src/sdimport/cli.py new file mode 100644 index 0000000..c7aba1e --- /dev/null +++ b/src/sdimport/cli.py @@ -0,0 +1,428 @@ +"""sdimport - 检测视频 SD 卡自动导入（macOS / Windows 通用）。 + +子命令： + auto 后台触发入口：macOS 由 launchd 调用，Windows 由常驻进程调用 + daemon 常驻进程（Windows：订阅卷插入事件 + 轮询兜底） + scan 演练：只显示会导入什么，不做任何写入 + once 立即手动执行一次导入 + keep 取消待执行的自动弹出 + eject 立即弹出 SD 卡 + status 查看服务状态与最近记录 + log 查看最近日志 + install 安装后台服务（插卡自动触发） + uninstall 卸载后台服务 + selftest 环境自检 +""" + +from __future__ import annotations + +import argparse +import json +import os +import subprocess +import sys +import time +from datetime import datetime +from pathlib import Path + +from . import notify, pipeline, transfer +from .journal import ( + Journal, + SingleInstanceLock, + human_duration, + human_size, + load_state, + note_run, + note_volume_seen, + save_state, +) +from .paths import ( + CONFIG_PATH, + LOCK_PATH, + LOG_DIR, + PROJECT_ROOT, + RUNTIME_DIR, + STATE_PATH, + ensure_dirs, + load_config, + write_default_config, +) +from .platforms import current as _platform +from .volumes import list_candidate_volumes, probe, volume_names + +#: 服务标识（Windows 的任务名 / macOS 的 launchd label） +SERVICE_ID = "com.sdimport.agent" + +SAMPLES = [ + "VID-20260912-090430.AVI", + "VID-20260911-131707.AVI", + "vid-20260910-083000.avi", + "IMG_0001.AVI", + "20260912_090430.AVI", +] + + +# -------------------------------------------------------------------------- +# 工具 +# -------------------------------------------------------------------------- + + +def volume_snapshot(cfg: dict) -> list[str]: + """当前候选卷的名字快照（轻量，不问系统要卷详情）。""" + return volume_names(cfg) + + + +# -------------------------------------------------------------------------- +# 子命令 +# -------------------------------------------------------------------------- + + +def cmd_auto(args: argparse.Namespace) -> int: + """后台触发入口。 + + macOS 由 launchd 调用（WatchPaths + StartInterval）； + Windows 由常驻进程 daemon 调用。真正的流程在 runner.auto_cycle，两端共用。 + """ + from .runner import auto_cycle + + return auto_cycle( + assume_yes=bool(getattr(args, "yes", False)), + lock_stale=float(getattr(args, "lock_stale", 3600.0)), + ) + + +def _run_interactive(args: argparse.Namespace, dry_run: bool) -> int: + cfg = load_config() + ensure_dirs() + journal = Journal("scan" if dry_run else "once") + + target = Path(args.volume) if args.volume else None + if target is not None and not target.exists(): + journal.error(f"路径不存在：{target}") + return 2 + + opts = pipeline.RunOptions( + dry_run=dry_run, + assume_yes=bool(getattr(args, "yes", False)) or dry_run, + no_delete=bool(getattr(args, "no_delete", False)), + target_volume=target, + quiet=bool(getattr(args, "quiet", False)), + ) + result = pipeline.run(cfg, journal, opts) + if dry_run: + print() + print(f"演练结束：将导入 {len(result.cards)} 张卡的待处理文件（未做任何写入）") + return 0 if not result.aborted else 1 + + +def cmd_status(args: argparse.Namespace) -> int: + cfg = load_config() + state = load_state() + print("=" * 68) + print("sdimport 状态") + print("=" * 68) + print(f"项目目录 : {PROJECT_ROOT}") + print(f"配置文件 : {CONFIG_PATH}") + print(f"落盘根目录 : {cfg['target_root']}") + print(f"倒计时 : {cfg.get('notify_countdown_seconds')} 秒") + print(f"导入后清卡 : {'是' if cfg.get('delete_source_after_verify') else '否'}") + + svc = _platform.service_status() + if svc.installed: + print(f"后台服务 : {svc.detail}") + else: + print(f"后台服务 : {svc.detail}（执行 sdimport install 安装）") + + from . import ejector + + pending = ejector.read_pending() + if pending: + print( + f"自动弹出 : 「{pending.get('name')}」计划 {pending.get('due_at_text')} 弹出" + f"（要取消就执行 sdimport keep）" + ) + else: + print("自动弹出 : 无待执行任务") + + print() + print("当前卷：") + vols = list_candidate_volumes(cfg) + if not vols: + print(" （无候选卷）") + for vol in vols: + print( + f" - {vol.describe()} 容量 {human_size(vol.total_bytes)}" + f" 可用 {human_size(vol.free_bytes)} uuid={vol.uuid or '未知'}" + ) + + hist = state.get("history", []) + print() + print(f"最近导入记录（共 {len(hist)} 条）：") + if not hist: + print(" （暂无）") + for rec in reversed(hist[-8:]): + flag = "中断" if rec.get("aborted") else "完成" + reason = f"（{rec.get('abort_reason')}）" if rec.get("aborted") else "" + print( + f" {rec.get('at')} {flag}{reason} 导入 {rec.get('imported', 0)} 个 / " + f"{human_size(rec.get('bytes', 0))} / 耗时 {human_duration(rec.get('seconds', 0))}" + ) + print() + print(f"状态文件 : {STATE_PATH}") + return 0 + + +def cmd_log(args: argparse.Namespace) -> int: + ensure_dirs() + files = sorted(LOG_DIR.glob("import-*.log")) + if not files: + print("暂无日志") + return 0 + path = files[-1] + print(f"# {path}") + lines = path.read_text(encoding="utf-8", errors="replace").splitlines() + for line in lines[-int(args.lines) :]: + print(line) + return 0 + + +def cmd_install(args: argparse.Namespace) -> int: + ensure_dirs() + write_default_config() + status = _platform.install_service(PROJECT_ROOT) + print(status.detail) + if status.ok and status.installed: + print(" 停用 : sdimport uninstall") + if _platform.needs_daemon: + print(" 说明 : 本平台靠常驻进程触发，登录后会随任务计划自动启动") + return 0 + return 1 + + +def cmd_uninstall(args: argparse.Namespace) -> int: + status = _platform.uninstall_service(PROJECT_ROOT) + print(status.detail) + return 0 if status.ok else 1 + + +def cmd_daemon(args: argparse.Namespace) -> int: + """常驻进程入口（Windows）。 + + macOS 用 launchd 的 WatchPaths + StartInterval，不需要常驻； + Windows 没有等价物，由常驻进程订阅卷插入事件做秒级响应， + 并辅以定时轮询兜底（两者任一可用即可工作）。 + """ + from .daemon import run_forever + + return run_forever(interval=float(args.interval)) + + +def cmd_selftest(args: argparse.Namespace) -> int: + ok = True + print("=" * 68) + print("sdimport 环境自检") + print("=" * 68) + + print(f"[1] Python : {sys.version.split()[0]} ({sys.executable})") + if sys.version_info < (3, 9): + print(" !! 需要 Python 3.9 及以上") + ok = False + + ensure_dirs() + for d in (RUNTIME_DIR, LOG_DIR): + print(f"[2] 运行目录 : {d} {'可写' if os.access(d, os.W_OK) else '不可写'}") + if not os.access(d, os.W_OK): + ok = False + + cfg = load_config() + root = Path(cfg["target_root"]) + root.mkdir(parents=True, exist_ok=True) + writable = os.access(root, os.W_OK) + import shutil + + free = shutil.disk_usage(root if root.exists() else Path.home()).free + print(f"[3] 落盘根目录 : {root}") + print(f" {'可写' if writable else '不可写'}，可用 {human_size(free)}") + if not writable: + ok = False + + print(f"[4] 平台 : {_platform.name}") + diag = getattr(_platform, "diagnose", None) + if callable(diag): + try: + for label, good, note in diag(): + suffix = f" {note}" if note else "" + print(f" - {label:<14} : {'可用' if good else '异常'}{suffix}") + if not good: + ok = False + except Exception as exc: # noqa: BLE001 - 自检本身出错不该带崩 + print(f" !! 平台诊断自身出错：{exc!r}") + ok = False + + notified = notify.notify("sdimport 自检", "通知通道正常") + print(f"[5] 通知通道 : {'可用' if notified else '不可用'}") + if not notified: + ok = False + + print("[6] 文件名正则自检：") + from .planner import parse_date + + for name in SAMPLES: + d = parse_date(name, cfg) + expect = "应识别" if name.lower().startswith("vid") else "应忽略" + got = d or "忽略" + mark = "OK " if (bool(d) == (expect == "应识别")) else "!! " + print(f" {mark}{name:<28} -> {got}") + if mark.strip() == "!!": + ok = False + + svc = _platform.service_status() + tail = "" if svc.installed else "（执行 sdimport install 安装）" + print(f"[7] 后台服务 : {svc.detail}{tail}") + print() + print("自检结果：" + ("全部通过" if ok else "存在问题，见上方 !! 标记")) + return 0 if ok else 1 + + +def cmd_keep(args: argparse.Namespace) -> int: + """取消待执行的「延时自动弹出 SD 卡」。""" + from . import ejector + + if ejector.cancel("用户手动执行 sdimport keep"): + print("已取消自动弹出 SD 卡，卡保持挂载。") + return 0 + print("当前没有待执行的自动弹出任务。") + return 0 + + +def cmd_eject(args: argparse.Namespace) -> int: + """立即弹出 SD 卡。""" + from . import ejector + + cfg = load_config() + target = args.volume + if not target: + vols = list_candidate_volumes(cfg) + if not vols: + print("未发现可弹出的卷。") + return 1 + target = str(vols[0].mount_point) + mount = Path(target) + if not mount.exists(): + print(f"挂载点不存在：{mount}") + return 2 + + ejector.cancel("改为立即弹出") + print(f"正在推出 {mount.name} …") + ok, why = ejector.eject_volume( + mount, + attempts=2, + interval=3.0, + use_finder=bool(cfg.get("eject_finder_fallback", False)), + ) + if ok: + print(f"{mount.name} {why}。可以拔卡了。") + return 0 + print(f"{mount.name} 弹出失败：{why}") + print("多半是卷被占用（访达窗口开着、或桌面显示着这张卡）。") + print("处置：关掉这张卡的访达窗口，或在访达侧边栏点该卷右侧的「推出」按钮。") + return 1 + + +def cmd_delayed_eject(args: argparse.Namespace) -> int: + """内部命令：延时弹出任务的实际执行者（由导入完成时以脱离进程方式拉起）。""" + from . import ejector + + return ejector.run_pending() + + +# -------------------------------------------------------------------------- +# 参数解析 +# -------------------------------------------------------------------------- + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="sdimport", + description="检测视频 SD 卡自动导入：插卡 -> 按日期分类 -> 落盘 -> 清卡", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=( + "示例：\n" + " sdimport scan 演练，看看会导入什么\n" + " sdimport once 立即执行一次\n" + " sdimport once --no-delete 执行但保留卡内文件\n" + " sdimport keep 取消自动弹出，卡继续挂着\n" + " sdimport eject 立刻弹出 SD 卡\n" + " sdimport status 查看状态与历史\n" + " sdimport install 安装插卡自动触发\n" + ), + ) + sub = parser.add_subparsers(dest="cmd", required=True) + + p_auto = sub.add_parser("auto", help="由 launchd 触发（插卡自动执行）") + p_auto.add_argument("--yes", action="store_true", help="跳过倒计时确认直接执行") + p_auto.add_argument("--lock-stale", type=float, default=3600.0, help="锁的超时秒数") + p_auto.set_defaults(func=cmd_auto) + + def add_common(p: argparse.ArgumentParser) -> None: + p.add_argument("--volume", help="只处理指定卷（默认扫描全部）") + p.add_argument("--yes", action="store_true", help="跳过倒计时确认") + p.add_argument("--quiet", action="store_true", help="不发送通知") + + p_scan = sub.add_parser("scan", help="演练：只显示会做什么") + add_common(p_scan) + p_scan.set_defaults(func=lambda a: _run_interactive(a, dry_run=True)) + + p_once = sub.add_parser("once", help="立即执行一次导入") + add_common(p_once) + p_once.add_argument("--no-delete", action="store_true", help="保留卡内源文件") + p_once.set_defaults(func=lambda a: _run_interactive(a, dry_run=False)) + + p_status = sub.add_parser("status", help="查看状态与历史") + p_status.set_defaults(func=cmd_status) + + p_log = sub.add_parser("log", help="查看最近日志") + p_log.add_argument("-n", "--lines", type=int, default=40, help="显示行数") + p_log.set_defaults(func=cmd_log) + + p_install = sub.add_parser("install", help="安装 launchd 后台服务") + p_install.set_defaults(func=cmd_install) + + p_uninstall = sub.add_parser("uninstall", help="卸载 launchd 后台服务") + p_uninstall.set_defaults(func=cmd_uninstall) + + p_test = sub.add_parser("selftest", help="环境自检") + p_test.set_defaults(func=cmd_selftest) + + p_keep = sub.add_parser("keep", help="取消待执行的自动弹出 SD 卡") + p_keep.set_defaults(func=cmd_keep) + + p_eject = sub.add_parser("eject", help="立即弹出 SD 卡") + p_eject.add_argument("--volume", help="指定挂载点（默认取唯一的候选卷）") + p_eject.set_defaults(func=cmd_eject) + + p_daemon = sub.add_parser( + "daemon", help="常驻进程：订阅卷插入事件 + 轮询兜底（Windows 靠它触发）" + ) + p_daemon.add_argument( + "--interval", type=float, default=30.0, help="轮询兜底间隔秒数（默认 30）" + ) + p_daemon.set_defaults(func=cmd_daemon) + + return parser + + +def main(argv: list[str] | None = None) -> int: + args_list = list(sys.argv[1:] if argv is None else argv) + # 内部命令：延时弹出任务的实际执行者。 + # 由导入完成时以脱离进程方式拉起，刻意不进帮助列表。 + if args_list[:1] == ["_delayed-eject"]: + return cmd_delayed_eject(argparse.Namespace()) + parser = build_parser() + args = parser.parse_args(args_list) + return int(args.func(args)) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/sdimport/daemon.py b/src/sdimport/daemon.py new file mode 100644 index 0000000..f03a01b --- /dev/null +++ b/src/sdimport/daemon.py @@ -0,0 +1,109 @@ +"""常驻进程：订阅卷插入事件 + 轮询兜底。 + +macOS 用 launchd 的 WatchPaths，不需要这层； +Windows 没有等价机制，这里做两条腿，任一可用即可工作： + + 1. **WMI 事件订阅**（`Win32_VolumeChangeEvent`）—— 秒级响应 + 2. **定时轮询** —— 兜底。WMI 并非万能：读卡器一直插着只换卡、 + WMI 服务异常、没装 wmi 包等情况下事件不会来，靠它保证不漏。 + +两条腿都指向同一个动作 `runner.auto_cycle`，与 macOS 侧行为一致。 +单轮之间用一把非阻塞锁互斥，避免事件与轮询同时触发跑两遍。 +""" + +from __future__ import annotations + +import threading +import time + +from .journal import Journal +from .platforms import current as _platform + + +def _trigger(source: str, journal: Journal, busy: threading.Lock) -> None: + if not busy.acquire(blocking=False): + journal.log(f"[{source}] 上一轮尚未结束，本次跳过") + return + try: + journal.log(f"[{source}] 触发一轮检查") + from .runner import auto_cycle + + auto_cycle(journal=Journal("auto"), lock_stale=3600.0) + except Exception as exc: # noqa: BLE001 - 常驻进程绝不能因单次异常退出 + journal.warn(f"[{source}] 本轮出错：{exc!r}") + finally: + busy.release() + + +def _poll_loop( + interval: float, journal: Journal, busy: threading.Lock, stop: threading.Event +) -> None: + journal.log(f"轮询兜底已启动，间隔 {interval:.0f} 秒") + while not stop.wait(interval): + _trigger("poll", journal, busy) + + +def _wmi_loop(journal: Journal, busy: threading.Lock, stop: threading.Event) -> None: + """订阅 WMI 卷变更事件（仅 Windows）。缺 wmi 包就安静退化成轮询。""" + try: + import wmi # type: ignore[import-untyped] + except ImportError: + journal.warn("未安装 wmi 包 → 插卡秒级响应不可用，只靠轮询兜底") + journal.warn(" 想开：pip install wmi") + return + + try: + client = wmi.WMI() + watcher = client.Win32_VolumeChangeEvent.watch_for() + except Exception as exc: # noqa: BLE001 + journal.warn(f"WMI 事件订阅失败（{exc!r}）→ 只靠轮询兜底") + return + + journal.log("WMI 卷变更事件订阅已就绪") + while not stop.is_set(): + try: + event = watcher(timeout_ms=5000) + except Exception as exc: # noqa: BLE001 - 含 wmi 的超时异常，属正常节拍 + name = type(exc).__name__ + if "timeout" in name.lower(): + continue + journal.warn(f"WMI 监听异常：{exc!r}") + time.sleep(5) + continue + if event is None: + continue + drive = getattr(event, "DriveName", "?") + kind = getattr(event, "EventType", "?") + journal.log(f"卷变更事件：DriveName={drive} EventType={kind}") + _trigger("wmi", journal, busy) + + +def run_forever(interval: float = 30.0) -> int: + journal = Journal("daemon") + journal.log(f"常驻进程启动（平台 {_platform.name}，轮询间隔 {interval:.0f} 秒）") + + stop = threading.Event() + busy = threading.Lock() + + threading.Thread( + target=_poll_loop, + args=(max(5.0, float(interval)), journal, busy, stop), + daemon=True, + name="sdimport-poll", + ).start() + + if _platform.needs_daemon: + threading.Thread( + target=_wmi_loop, args=(journal, busy, stop), daemon=True, name="sdimport-wmi" + ).start() + else: + journal.log("当前平台由系统服务负责触发，这里仅保留轮询兜底") + + try: + while not stop.wait(3600): + pass + except KeyboardInterrupt: + journal.log("收到中断，常驻进程退出") + finally: + stop.set() + return 0 diff --git a/src/sdimport/ejector.py b/src/sdimport/ejector.py new file mode 100644 index 0000000..9b5abbf --- /dev/null +++ b/src/sdimport/ejector.py @@ -0,0 +1,323 @@ +"""导入完成后的延时自动弹出 SD 卡。 + +用户要的是：导完给个明确交代，然后 5 分钟后自动把卡弹出来，省得手动去推。 + +安全设计（弹出涉及可移动介质，必须保守）： + * **绝不打断在跑的导入**：到点前先查 import.lock，有导入在进行就往后顺延。 + * **可能已被手动处理**：卷不在了就直接收工，不做任何多余动作。 + * **只弹出，不删除**：全程只调系统的弹出/卸载动作，不碰卡上任何文件。 + * **允许反悔**：完成提示框里可点「暂不弹出」；事后也可用 `sdimport keep` 取消。 + * **弹不出来要说清楚**：重试若干次，仍失败就通知用户手动推出，绝不静默失败。 + +实现要点：延时等待交给一个脱离父进程的独立进程， +这样触发链路可以立刻收工，不必为等 5 分钟而占着单实例锁。 +任务状态落在 runtime/state/pending-eject.json，进程被系统清掉也能由下次触发补做。 + +具体的弹出动作由平台实现（macOS：diskutil + 访达；Windows：Shell.Application + mountvol）。 +""" + +from __future__ import annotations + +import json +import os +import subprocess +import sys +import time +from datetime import datetime +from pathlib import Path + +from . import notify +from .journal import Journal, human_duration, human_size +from .paths import LOCK_PATH, LOG_DIR, PROJECT_ROOT, STATE_DIR, ensure_dirs +from .platforms import current as _platform + +PENDING_PATH = STATE_DIR / "pending-eject.json" +EJECT_LOG = LOG_DIR / "eject.log" + +# 到点后若仍有导入在跑，最多顺延这么久 +MAX_DEFER_SECONDS = 900 +DEFER_STEP = 10 +CHECK_STEP = 5.0 + + +# -------------------------------------------------------------------------- +# 任务状态 +# -------------------------------------------------------------------------- + + +def _write_pending(payload: dict) -> None: + ensure_dirs() + tmp = PENDING_PATH.with_suffix(".json.tmp") + tmp.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + os.replace(tmp, PENDING_PATH) + + +def read_pending() -> dict | None: + if not PENDING_PATH.exists(): + return None + try: + data = json.loads(PENDING_PATH.read_text(encoding="utf-8")) + return data if isinstance(data, dict) else None + except (json.JSONDecodeError, OSError): + return None + + +def clear_pending() -> None: + try: + PENDING_PATH.unlink(missing_ok=True) + except OSError: + pass + + +def cancel(reason: str = "手动取消") -> bool: + """取消待执行的弹出任务。返回是否确实取消了一个任务。""" + existed = PENDING_PATH.exists() + clear_pending() + if existed: + _append_eject_log(f"[{_now()}] 弹出任务已取消（{reason}）\n") + return existed + + +def _now() -> str: + return datetime.now().strftime("%Y-%m-%d %H:%M:%S") + + +def _append_eject_log(line: str) -> None: + try: + ensure_dirs() + with EJECT_LOG.open("a", encoding="utf-8") as fh: + fh.write(line) + except OSError: + pass + + +# -------------------------------------------------------------------------- +# 完成提示 +# -------------------------------------------------------------------------- + + +def delay_text(seconds: int) -> str: + """把延时说成人话（含短延时的调试值，避免出现「0 分钟后」）。""" + seconds = max(0, int(seconds)) + if seconds >= 120: + return f"{seconds // 60} 分钟后" + if seconds >= 60: + return "1 分钟后" + return f"{seconds} 秒后" + + +def completion_message(result, delay_seconds: int) -> str: + dates = "、".join(f"{d}({n})" for d, n in sorted(result.dates.items())) + lines = [ + f"{result.imported} 个视频 / {human_size(result.bytes_moved)} / " + f"耗时 {human_duration(result.seconds)}", + ] + if dates: + lines.append(f"归入日期：{dates}") + if result.deleted: + lines.append(f"卡内 {result.deleted} 个文件已清空。") + elif result.delete_failed: + lines.append(f"有 {result.delete_failed} 个卡内文件未能删除，仍有残留。") + else: + lines.append("卡内源文件已保留。") + lines.append("") + lines.append(f"{delay_text(delay_seconds)}自动弹出 SD 卡。") + lines.append("如需继续使用这张卡，请点「暂不弹出」。") + return "\n".join(lines) + + +def offer_and_schedule( + cfg: dict, + journal: Journal, + result, + mount_points: list[Path], +) -> None: + """弹完成提示；按用户选择安排延时弹出。""" + if not cfg.get("eject_after_import", True): + journal.log("已配置为不自动弹出 SD 卡（eject_after_import=false）") + return + if not mount_points: + return + + delay = max(0, int(cfg.get("eject_delay_seconds", 300))) + dialog_seconds = int(cfg.get("completion_dialog_seconds", 30)) + + verdict = "ok" + if dialog_seconds > 0: + verdict = notify.ask_completion( + "检测视频：导出完成", + completion_message(result, delay) if delay else "导出完成。", + timeout=dialog_seconds, + keep_label="暂不弹出", + ok_label="知道了", + ) + if verdict == "keep": + cancel("用户在完成提示中选了「暂不弹出」") + journal.log("用户选择暂不弹出，本次不安排自动弹出") + return + if verdict == "error": + # 弹框不可用（无图形会话等）：退回通知中心，仍按计划弹出 + notify.notify( + "检测视频：导出完成", + f"{result.imported} 个视频 / {human_size(result.bytes_moved)}，" + f"{delay_text(delay)}自动弹出 SD 卡。", + ) + + for mount in mount_points: + ok = schedule(mount, mount.name, cfg, journal) + if not ok: + break + + +# -------------------------------------------------------------------------- +# 安排与执行 +# -------------------------------------------------------------------------- + + +def _detach_kwargs() -> dict: + """让子进程脱离父进程。POSIX 与 Windows 的姿势不同。""" + if sys.platform.startswith("win"): + return { + "creationflags": getattr(subprocess, "DETACHED_PROCESS", 0x00000008) + | getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0x00000200) + } + return {"start_new_session": True} + + +def schedule(mount_point: Path, name: str, cfg: dict, journal: Journal) -> bool: + # 下限 10 秒：够用户看见提示并反悔；上限不设，配置写多少就是多少 + delay = max(10, int(cfg.get("eject_delay_seconds", 300))) + payload = { + "volume": str(mount_point), + "name": name, + "delay": delay, + "due_at": time.time() + delay, + "due_at_text": datetime.fromtimestamp(time.time() + delay).strftime("%H:%M:%S"), + "attempts": int(cfg.get("eject_retry_attempts", 5)), + "interval": float(cfg.get("eject_retry_interval", 15)), + "use_finder": bool(cfg.get("eject_finder_fallback", False)), + "created": _now(), + } + try: + _write_pending(payload) + except OSError as exc: + journal.warn(f"无法写入弹出任务：{exc}") + return False + + _append_eject_log( + f"[{_now()}] 已安排 {name} 于 {payload['due_at_text']} 自动弹出（{delay} 秒后）\n" + ) + + src_dir = Path(__file__).resolve().parents[1] + env = dict(os.environ) + env["PYTHONPATH"] = str(src_dir) + os.pathsep + env.get("PYTHONPATH", "") + env["PYTHONDONTWRITEBYTECODE"] = "1" + + argv = [sys.executable, "-m", "sdimport.cli", "_delayed-eject"] + try: + ensure_dirs() + with EJECT_LOG.open("ab") as fh: + subprocess.Popen( + argv, + env=env, + cwd=str(PROJECT_ROOT), + stdin=subprocess.DEVNULL, + stdout=fh, + stderr=subprocess.STDOUT, + **_detach_kwargs(), + ) + except OSError as exc: + journal.warn(f"无法启动延时弹出进程：{exc}（可执行 sdimport eject 手动弹出）") + return False + + journal.log(f"已安排 {name} 于 {payload['due_at_text']} 自动弹出") + return True + + +def eject_volume( + mount_point: Path, + attempts: int, + interval: float, + use_finder: bool = False, +) -> tuple[bool, str]: + """弹出卷（委托平台实现）。 + + macOS：先 diskutil，必要时改走访达双通道； + Windows：Shell.Application 的 Eject 动词，降级到 mountvol /P。 + 只弹出，不删除任何文件。 + """ + return _platform.eject( + Path(mount_point), + attempts=attempts, + interval=interval, + use_fallback=use_finder, + ) + + +def run_pending() -> int: + """独立进程入口：等到点，然后弹出。""" + journal = Journal("eject") + payload = read_pending() + if not payload: + journal.log("无待执行的弹出任务") + return 0 + + mount = Path(str(payload.get("volume", ""))) + name = str(payload.get("name") or mount.name) + due_at = float(payload.get("due_at", 0)) + + while True: + remain = due_at - time.time() + if remain <= 0: + break + time.sleep(min(CHECK_STEP, remain)) + if read_pending() is None: + journal.log("弹出任务已被取消") + return 0 + + if read_pending() is None: + journal.log("弹出任务已被取消") + return 0 + + # 有导入在跑就顺延，绝不打断拷贝 + deferred = 0.0 + while LOCK_PATH.exists() and deferred < MAX_DEFER_SECONDS: + journal.log("检测到导入仍在进行，顺延后再弹") + time.sleep(DEFER_STEP) + deferred += DEFER_STEP + if read_pending() is None: + journal.log("弹出任务已被取消") + return 0 + if LOCK_PATH.exists(): + journal.warn("导入长时间未结束，本次放弃自动弹出") + clear_pending() + notify.notify("检测视频：未自动弹出", f"{name} 因导入未结束而保留挂载，可稍后手动推出。") + return 1 + + if not mount.exists(): + journal.log(f"{name} 已不在挂载点，无需弹出") + clear_pending() + return 0 + + attempts = int(payload.get("attempts", 5)) + interval = float(payload.get("interval", 15)) + use_finder = bool(payload.get("use_finder", False)) + ok, why = eject_volume(mount, attempts, interval, use_finder=use_finder) + clear_pending() + + if ok: + journal.log(f"{name} {why}") + _append_eject_log(f"[{_now()}] {name} {why}\n") + notify.notify( + "检测视频：SD 卡已弹出", + f"「{name}」已安全弹出，可以拔卡了。", + ) + return 0 + + journal.warn(f"{name} 自动弹出失败：{why}") + _append_eject_log(f"[{_now()}] {name} 自动弹出失败：{why}\n") + notify.notify( + "检测视频：弹出失败", + f"「{name}」自动弹出失败（{why}）。数据已导入完毕，可手动推出该卡。", + ) + return 1 diff --git a/src/sdimport/journal.py b/src/sdimport/journal.py new file mode 100644 index 0000000..770746c --- /dev/null +++ b/src/sdimport/journal.py @@ -0,0 +1,178 @@ +"""日志与运行状态。 + +日志：runtime/logs/import-YYYYMMDD.log（追加） +状态：runtime/state/state.json（卷指纹 / 导入历史） +""" + +from __future__ import annotations + +import json +import os +import sys +import time +from datetime import datetime +from pathlib import Path + +from .paths import LOG_DIR, STATE_DIR, STATE_PATH, ensure_dirs + + +def _ts() -> str: + return datetime.now().strftime("%Y-%m-%d %H:%M:%S") + + +def human_size(n: int) -> str: + units = ["B", "KB", "MB", "GB", "TB"] + v = float(n) + for u in units: + if v < 1024 or u == units[-1]: + return f"{v:.1f}{u}" if u != "B" else f"{int(v)}B" + v /= 1024 + return f"{n}B" + + +def human_duration(sec: float) -> str: + sec = max(0, int(sec)) + if sec < 60: + return f"{sec}秒" + if sec < 3600: + return f"{sec // 60}分{sec % 60}秒" + return f"{sec // 3600}小时{(sec % 3600) // 60}分" + + +class Journal: + """一条运行一份，同时写文件、stderr 与内存副本。""" + + def __init__(self, tag: str = "auto") -> None: + ensure_dirs() + self.tag = tag + self.lines: list[str] = [] + self.log_path = LOG_DIR / f"import-{datetime.now().strftime('%Y%m%d')}.log" + self.t0 = time.time() + + def log(self, msg: str, level: str = "INFO") -> None: + line = f"[{_ts()}] [{self.tag}] [{level}] {msg}" + self.lines.append(line) + try: + with self.log_path.open("a", encoding="utf-8") as fh: + fh.write(line + "\n") + except OSError: + pass + print(line, file=sys.stderr, flush=True) + + def step(self, msg: str) -> None: + self.log(msg, "STEP") + + def warn(self, msg: str) -> None: + self.log(msg, "WARN") + + def error(self, msg: str) -> None: + self.log(msg, "ERROR") + + @property + def elapsed(self) -> float: + return time.time() - self.t0 + + def tail(self, n: int = 20) -> list[str]: + return self.lines[-n:] + + +# -------------------------------------------------------------------------- +# 状态：卷去重 + 导入历史 +# -------------------------------------------------------------------------- + + +def load_state() -> dict: + ensure_dirs() + if not STATE_PATH.exists(): + return {"volumes": {}, "history": []} + try: + data = json.loads(STATE_PATH.read_text(encoding="utf-8")) + if isinstance(data, dict): + data.setdefault("volumes", {}) + data.setdefault("history", []) + return data + except (json.JSONDecodeError, OSError): + pass + return {"volumes": {}, "history": []} + + +def save_state(state: dict) -> None: + ensure_dirs() + tmp = STATE_PATH.with_suffix(".json.tmp") + try: + tmp.write_text(json.dumps(state, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + os.replace(tmp, STATE_PATH) + except OSError: + pass + + +def note_volume_seen(state: dict, vol_id: str, name: str) -> None: + entry = state["volumes"].setdefault(vol_id, {"name": name, "first_seen": _ts()}) + entry["name"] = name + entry["last_seen"] = _ts() + + +def note_run(state: dict, record: dict) -> None: + state["history"].append(record) + # 只留最近 200 条 + if len(state["history"]) > 200: + del state["history"][:-200] + + +# -------------------------------------------------------------------------- +# 单实例锁 +# -------------------------------------------------------------------------- + + +class SingleInstanceLock: + """基于 O_EXCL 的跨进程锁；持有者崩溃超时后自动接管。""" + + def __init__(self, path: Path, stale_after: float = 900.0) -> None: + self.path = path + self.stale_after = stale_after + self.acquired = False + + def acquire(self) -> bool: + ensure_dirs() + if self.path.exists(): + try: + age = time.time() - self.path.stat().st_mtime + except OSError: + age = 0.0 + if age < self.stale_after: + return False + try: + self.path.unlink() + except OSError: + return False + try: + fd = os.open(self.path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644) + os.write(fd, f"{os.getpid()} {_ts()}\n".encode()) + os.close(fd) + self.acquired = True + return True + except FileExistsError: + return False + except OSError: + return False + + def refresh(self) -> None: + if self.acquired: + try: + os.utime(self.path, None) + except OSError: + pass + + def release(self) -> None: + if self.acquired: + try: + self.path.unlink() + except OSError: + pass + self.acquired = False + + def __enter__(self) -> "SingleInstanceLock": + return self + + def __exit__(self, *exc) -> None: + self.release() diff --git a/src/sdimport/notify.py b/src/sdimport/notify.py new file mode 100644 index 0000000..24e67c5 --- /dev/null +++ b/src/sdimport/notify.py @@ -0,0 +1,54 @@ +"""系统通知与确认框（跨平台门面）。 + +实际实现按运行平台走 platforms.current： + * macOS —— osascript 的 `display notification` / `display dialog` + * Windows —— PowerShell 的 toast 通知 + tkinter 自绘的带倒计时对话框 +调用方不必关心这些差异，签名与语义两端一致。 +""" + +from __future__ import annotations + +from .platforms import current as _platform + + +def notify(title: str, message: str, subtitle: str = "", sound: str = "") -> bool: + """发一条系统通知。返回是否发出（失败不抛异常，调用方照常继续）。""" + return _platform.notify(title, message, subtitle=subtitle, sound=sound) + + +def ask_with_countdown( + title: str, + message: str, + timeout: int = 60, + ok_label: str = "立即开始", + cancel_label: str = "取消导入", +) -> str: + """开始导入前的确认框。 + + 返回 'ok'（点了确认，或超时未操作 —— 用户要的就是「不响应即执行」）/ + 'cancel'（明确取消）/ 'error'（弹不出来，此时已改为通知 + 静默等待）。 + """ + return _platform.ask_countdown( + title, message, timeout=timeout, ok_label=ok_label, cancel_label=cancel_label + ) + + +def ask_completion( + title: str, + message: str, + timeout: int = 30, + keep_label: str = "暂不弹出", + ok_label: str = "知道了", +) -> str: + """导出完成提示框。返回 'ok'（继续弹出计划）/'keep'（明确暂不弹出）/'error'。""" + return _platform.ask_completion( + title, message, timeout=timeout, keep_label=keep_label, ok_label=ok_label + ) + + +def banner(title: str, message: str, timeout: int = 60) -> bool: + """纯提示框（无选择语义），超时自动消失。平台没实现就退化成通知。""" + impl = getattr(_platform, "banner", None) + if callable(impl): + return bool(impl(title, message, timeout=timeout)) + return notify(title, message) diff --git a/src/sdimport/paths.py b/src/sdimport/paths.py new file mode 100644 index 0000000..063a9af --- /dev/null +++ b/src/sdimport/paths.py @@ -0,0 +1,131 @@ +"""路径解析与配置加载。 + +配置优先级：config.json（用户可改） > DEFAULTS（代码内置） +""" + +from __future__ import annotations + +import json +import os +from pathlib import Path + +PROJECT_ROOT = Path(__file__).resolve().parents[2] +CONFIG_PATH = PROJECT_ROOT / "config.json" +RUNTIME_DIR = PROJECT_ROOT / "runtime" +LOG_DIR = RUNTIME_DIR / "logs" +STATE_DIR = RUNTIME_DIR / "state" +STATE_PATH = STATE_DIR / "state.json" +LOCK_PATH = RUNTIME_DIR / "import.lock" + +DEFAULTS: dict = { + # 视频落盘根目录，按 // 组织 + "target_root": "~/Downloads/检测视频", + # 只认这个文件名模式的文件，其余一律不碰（绝不误删卡上其它素材） + "filename_regex": r"^VID-(\d{8})-(\d{6})\.(?:AVI|MP4|MOV|MTS|MPG|MPEG)$", + "case_insensitive": True, + # ---- 扫描范围（2026-09-15 事故后重定，改动前务必读懂）---- + # 只在下列「扫描根」下找录像。path 相对卷根（"" 即卷根本身）， + # ":N" 表示在该根下最多再进 N 层子目录，不写则用 scan_max_depth。 + # + # 为什么写死层数、不再靠「本层有没有录像」推断边界 —— + # 上一版规则（本层有录像就停止下潜）会自毁：第一次导入把 VIDEO/ 本层清空后， + # 第二次扫描发现本层没录像，就下潜进了归档子目录， + # 把用户明确说过「不碰」的 409 个 / 33.5GB 历史归档全导走并清空了卡 + # （2026-09-15 18:45 真实发生）。VIDEO 用 0 层，永不进子目录。 + "scan_roots": ["", "VIDEO", "DCIM:2", "MOVIE", "PRIVATE"], + "scan_max_depth": 0, + # 完全不进入的目录名（大小写不敏感），用于躲开卡上无关目录 + "deny_dir_names": ["System Volume Information", "$RECYCLE.BIN", "System", "FOUND.000"], + # 停止下潜时，是否统计并告知被跳过了多少归档录像 + "report_skipped_archives": True, + # 小于此体积的匹配文件视为碎片不导入。只用来挡 0 字节 / 畸形残片， + # 不要用来挡「短录制」——2026-09-14 真卡上有个 192KB 的有效短录制差点被静默丢弃。 + "min_video_bytes": 1024, + # 小于此体积的录像会在报告里标注（疑似误触 / 中断的短段）， + # 但**照样导入**，绝不静默丢数据 + "tiny_video_bytes": 1048576, + # 拷贝块大小 / 校验采样大小 + "copy_chunk_mb": 8, + "verify_sample_kb": 64, + # 目标盘剩余空间至少是待导入总量的多少倍 + "min_free_space_ratio": 1.10, + # 校验通过后是否删除卡内源文件 + "delete_source_after_verify": True, + # 删源批量上限（每批之间停顿，避免卡 IO 过载） + "delete_batch_size": 10, + # 开始导入前的倒计时（秒），期间可取消 + "notify_countdown_seconds": 60, + # ---- 导出完成提示与自动弹出 SD 卡 ---- + # 导入完成后是否自动弹出 SD 卡（用户要的「5 分钟后弹卡」） + "eject_after_import": True, + # 导入完成后延迟多少秒弹出，给用户留反悔的余地（完成提示框里可点「暂不弹出」） + "eject_delay_seconds": 300, + # 完成提示框停留秒数；超时按「知道了」计，保证人不在也能照常弹卡。0 = 不弹框只用通知中心 + "completion_dialog_seconds": 30, + # 弹出失败的重试次数与间隔（卡被 Finder / 桌面引用时，diskutil 会被 loginwindow 拒绝， + # 此时改走访达推出通道，再失败才重试） + "eject_retry_attempts": 5, + "eject_retry_interval": 15, + # 弹出失败时是否改走「访达推出」通道。 + # 对付「卷被访达窗口 / 桌面引用、diskutil 卸载被 loginwindow 拒绝」很有效， + # 但 osascript 控制访达需要「自动化」权限，未授权时会弹授权框并**阻塞**，所以默认关闭。 + # 想开：先在终端手动执行一次 + # osascript -e 'tell application "Finder" to eject disk "卷名"' + # 在授权框点「允许」，再把这里改成 true。 + "eject_finder_fallback": False, + # 不处理的卷名（系统盘等） + "deny_volumes": ["Macintosh HD", "Recovery", "TimeMachine", "VM", "Preboot", "Update"], + # ---- Windows 专属 ---- + # 只把「可移动磁盘」当候选卷。这是硬安全线：绝不让程序碰到系统盘 / 数据盘。 + # 极少数读卡器会被 Windows 识别为固定磁盘，那种机器上可改成 false。 + "windows_only_removable": True, + # 卡为只读（NTFS 等）时是否降级为「只复制、不删源」 + "allow_readonly_copy": True, + # 挂载稳定等待 + "settle_seconds": 4, + "stability_checks": 3, + "stability_interval": 0.8, +} + + +def load_config() -> dict: + cfg = dict(DEFAULTS) + if CONFIG_PATH.exists(): + try: + user = json.loads(CONFIG_PATH.read_text(encoding="utf-8")) + if isinstance(user, dict): + cfg.update(user) + except (json.JSONDecodeError, OSError): + pass + overrides = { + "SDIMPORT_TARGET_ROOT": "target_root", + "SDIMPORT_COUNTDOWN": "notify_countdown_seconds", + "SDIMPORT_EJECT_DELAY": "eject_delay_seconds", + "SDIMPORT_COMPLETION_DIALOG": "completion_dialog_seconds", + } + int_keys = { + "notify_countdown_seconds", + "eject_delay_seconds", + "completion_dialog_seconds", + } + for env_key, cfg_key in overrides.items(): + raw = os.environ.get(env_key) + if raw: + cfg[cfg_key] = int(raw) if cfg_key in int_keys else raw + cfg["target_root"] = os.path.expanduser(str(cfg["target_root"])) + return cfg + + +def ensure_dirs() -> None: + for d in (RUNTIME_DIR, LOG_DIR, STATE_DIR): + d.mkdir(parents=True, exist_ok=True) + + +def write_default_config() -> Path: + """写出带注释性键名的默认配置（仅当文件不存在时）。""" + if CONFIG_PATH.exists(): + return CONFIG_PATH + CONFIG_PATH.write_text( + json.dumps(DEFAULTS, ensure_ascii=False, indent=2) + "\n", encoding="utf-8" + ) + return CONFIG_PATH diff --git a/src/sdimport/pipeline.py b/src/sdimport/pipeline.py new file mode 100644 index 0000000..5d0fbb2 --- /dev/null +++ b/src/sdimport/pipeline.py @@ -0,0 +1,468 @@ +"""主流程编排：找卡 -> 汇总 -> 倒计时确认 -> 拷贝校验 -> 删源 -> 汇报。""" + +from __future__ import annotations + +import time +from dataclasses import dataclass, field +from pathlib import Path + +from . import ejector, notify, planner, transfer +from .journal import Journal, human_duration, human_size +from .paths import ensure_dirs +from .volumes import ( + Volume, + check_writable, + find_videos, + list_candidate_volumes, + probe, + wait_until_stable, +) + +PROGRESS_MIN_INTERVAL = 5.0 + + +def _sweep_existing( + job: CardJob, + skipped: list[planner.PlanItem], + cfg: dict, + journal: Journal, + result: RunResult, + sample_kb: int, + batch: int, +) -> None: + """清理「此前已落盘、本次无需再拷」的卡内副本。 + + 场景：上一轮以「保留卡内文件」模式导入过，或导入中途卡被拔掉。 + 此时目标文件已存在且一致，按「移动」语义应当把卡内副本清掉，否则卡永远腾不空。 + 前提：必须通过完整校验（体积 + 头尾指纹），任何不一致一律不动。 + """ + if not skipped: + return + journal.step(f"{job.volume.name}：卡内有 {len(skipped)} 个文件此前已落盘，开始核对并清理") + cleaned = 0 + for item in skipped: + ok, why = transfer.verify(item.src, item.dst, sample_kb) + if not ok: + journal.warn(f" {item.src.name}：与本地副本不一致（{why}），保留卡内文件") + continue + journal.log(f" 删除清单：{item.src}") + ok, why = transfer.delete_source(item.src, item.dst, sample_kb) + if ok: + cleaned += 1 + result.deleted += 1 + else: + result.delete_failed += 1 + result.errors.append(f"{item.src.name}: 清卡失败 {why}") + journal.warn(f" {item.src.name}：清卡失败（本地副本完好）：{why}") + if cleaned and cleaned % batch == 0: + time.sleep(0.2) + journal.log(f" 核对完成：清理 {cleaned} 个，保留 {len(skipped) - cleaned} 个") + + +@dataclass +class RunOptions: + dry_run: bool = False + assume_yes: bool = False + no_delete: bool = False + target_volume: Path | None = None + quiet: bool = False + + +@dataclass +class CardJob: + volume: Volume + plan: planner.Plan + writable: bool = True + + +@dataclass +class RunResult: + imported: int = 0 + skipped: int = 0 + failed: int = 0 + deleted: int = 0 + delete_failed: int = 0 + bytes_moved: int = 0 + seconds: float = 0.0 + aborted: bool = False + abort_reason: str = "" + cards: list[str] = field(default_factory=list) + dates: dict[str, int] = field(default_factory=dict) + errors: list[str] = field(default_factory=list) + # 本次真正动过的卡挂载点（用于导入完成后的延时自动弹出） + volumes: list[str] = field(default_factory=list) + + +# -------------------------------------------------------------------------- +# 第一段：探测与规划 +# -------------------------------------------------------------------------- + + +def _report_tiny(files: list[Path], cfg: dict, journal: Journal) -> None: + """把体积极小的录像单独点出来，交用户自行判断。 + + 这些多半是误触或录制中断的短段，但**照样导入**。 + 2026-09-14 真卡上有个 192KB / 约 50 秒的短段，一度被体积阈值静默滤掉 —— + 静默丢数据不可接受，宁可多导一个废片，也不能漏一个有效段。 + """ + limit = int(cfg.get("tiny_video_bytes", 0)) + if limit <= 0: + return + tiny: list[tuple[Path, int]] = [] + for f in files: + try: + size = f.stat().st_size + except OSError: + continue + if size < limit: + tiny.append((f, size)) + if not tiny: + return + journal.warn( + f" 其中 {len(tiny)} 个体积极小（<{human_size(limit)}），疑似误触或中断的短录制，仍会一并导入：" + ) + for f, size in tiny: + journal.warn(f" {f.name} {human_size(size)}") + + +def discover_cards(cfg: dict, journal: Journal, opts: RunOptions) -> list[CardJob]: + """找出本次插着的记录仪卡，并生成导入计划。""" + if opts.target_volume is not None: + vol = probe(Path(opts.target_volume)) + vols = [vol] if vol else [] + if not vols: + journal.error(f"指定路径不是有效卷：{opts.target_volume}") + else: + vols = list_candidate_volumes(cfg) + + journal.log(f"扫描到 {len(vols)} 个候选卷：" + (", ".join(v.describe() for v in vols) or "无")) + + jobs: list[CardJob] = [] + for vol in vols: + if not wait_until_stable( + vol.mount_point, + checks=int(cfg.get("stability_checks", 3)), + interval=float(cfg.get("stability_interval", 0.8)), + ): + journal.warn(f"{vol.name}：挂载未稳定，本次跳过") + continue + files = find_videos(vol, cfg, journal) + if not files: + journal.log(f"{vol.name}：无符合命名规范的录像，判定为非记录仪卡，跳过") + continue + writable = check_writable(vol.mount_point) + if not writable: + journal.warn(f"{vol.name}：卷为只读（{vol.fs}），本次只能复制、无法清卡") + if not cfg.get("allow_readonly_copy", True): + continue + plan = planner.build_plan(vol, files, cfg, journal) + job = CardJob(volume=vol, plan=plan, writable=writable) + jobs.append(job) + journal.step( + f"{vol.name}：识别到录像 {len(files)} 个 / {human_size(plan.total_bytes)}，" + f"其中新增 {len(plan.to_copy)} 个 / {human_size(plan.pending_bytes)}，" + f"已导入过 {len(plan.skipped)} 个" + ) + for date, items in plan.group_by_date().items(): + journal.log( + f" -> {date}：{len(items)} 个 / {human_size(sum(i.size for i in items))}" + ) + _report_tiny(files, cfg, journal) + return jobs + + +def _confirm_message(jobs: list[CardJob], cfg: dict, timeout: int) -> tuple[str, str]: + total_files = sum(len(j.plan.to_copy) for j in jobs) + total_bytes = sum(j.plan.pending_bytes for j in jobs) + dates: dict[str, int] = {} + for job in jobs: + for date, items in job.plan.group_by_date().items(): + dates[date] = dates.get(date, 0) + len(items) + date_txt = "、".join(f"{d}({n})" for d, n in sorted(dates.items())) + + readonly_note = "" + if any(not j.writable for j in jobs): + readonly_note = "\n卡为只读，本次只复制、不清卡。" + + title = "检测视频自动导入" + message = ( + f"识别到 {len(jobs)} 张记录仪卡，共 {total_files} 个新视频 / {human_size(total_bytes)}。\n" + f"目标日期：{date_txt}\n" + f"落盘至：{cfg['target_root']}\n" + f"{timeout} 秒内未操作则自动开始导入，期间可点「取消导入」。{readonly_note}" + ) + return title, message + + +# -------------------------------------------------------------------------- +# 第二段：执行 +# -------------------------------------------------------------------------- + + +def _abort_item( + result: RunResult, + journal: Journal, + item: planner.PlanItem, + label: str, + kind: str, + detail: str, +) -> None: + """单文件失败即中止整批：已拷成功的文件安全落盘，未拷的源文件一律不动，重插卡可续传。""" + result.failed += 1 + result.aborted = True + result.abort_reason = f"{item.src.name} {kind}" + result.errors.append(f"{item.src.name}: {kind} {detail}") + journal.error(f" {label} -> {kind}：{detail}；为保数据安全，本次中止且不删除任何源文件") + + +def _run_one_card( + job: CardJob, + cfg: dict, + journal: Journal, + opts: RunOptions, + result: RunResult, + lock=None, +) -> None: + plan = job.plan + chunk = max(64 * 1024, int(cfg.get("copy_chunk_mb", 8)) * 1024 * 1024) + sample_kb = int(cfg.get("verify_sample_kb", 64)) + do_delete = ( + bool(cfg.get("delete_source_after_verify", True)) + and not opts.no_delete + and job.writable + ) + batch = max(1, int(cfg.get("delete_batch_size", 10))) + + items = plan.to_copy + result.skipped += len(plan.skipped) + if (items or (do_delete and plan.skipped)) and str(job.volume.mount_point) not in result.volumes: + result.volumes.append(str(job.volume.mount_point)) + stale = transfer.cleanup_partials(plan.items) + if stale: + journal.log(f" 清理历史残留临时文件 {stale} 个（.part）") + if items: + journal.step( + f"{job.volume.name}：开始导入 {len(items)} 个文件（{'清卡' if do_delete else '保留卡内文件'}）" + ) + else: + journal.step(f"{job.volume.name}：本次无新增视频") + + if do_delete: + _sweep_existing(job, plan.skipped, cfg, journal, result, sample_kb, batch) + + done_since_pause = 0 + for idx, item in enumerate(items, 1): + if lock is not None: + lock.refresh() + label = f"[{idx}/{len(items)}] {item.src.name} ({human_size(item.size)})" + + if not job.volume.mount_point.exists(): + result.aborted = True + result.abort_reason = "卡已被拔出" + journal.error(f"{label} -> 卡已拔出，中止导入") + return + + last_log = [0.0] + + def on_progress(written: int, total: int, _label=label, _last=last_log) -> None: + now = time.time() + if now - _last[0] >= PROGRESS_MIN_INTERVAL: + _last[0] = now + pct = (written / total * 100) if total else 0 + journal.log(f" {_label} 拷贝中 {pct:.0f}%（{human_size(written)}）") + + try: + res = transfer.copy_file(item.src, item.dst, chunk, on_progress) + if not res.ok: + _abort_item(result, journal, item, label, "拷贝失败", res.error) + return + + ok, why = transfer.verify(item.src, item.dst, sample_kb) + if not ok: + _abort_item(result, journal, item, label, "校验失败", why) + return + + result.imported += 1 + result.bytes_moved += res.size + result.dates[item.date] = result.dates.get(item.date, 0) + 1 + journal.log( + f" {label} -> 完成，{human_size(res.size)}，" + f"耗时 {human_duration(res.seconds)}，均速 {human_size(int(res.speed))}/s，校验通过" + ) + + if do_delete: + journal.log(f" 删除清单：{item.src}") + ok2, why2 = transfer.delete_source(item.src, item.dst, sample_kb) + if ok2: + result.deleted += 1 + journal.log(" 已删除卡内源文件") + else: + result.delete_failed += 1 + result.errors.append(f"{item.src.name}: 删源失败 {why2}") + journal.warn(f" 删源失败（文件已安全落盘，卡内残留）：{why2}") + done_since_pause += 1 + if done_since_pause >= batch: + done_since_pause = 0 + time.sleep(0.2) + except Exception as exc: # noqa: BLE001 - 兜底：未预期异常绝不清卡、绝不带崩后台服务 + _abort_item(result, journal, item, label, "处理异常", repr(exc)) + return + + return + + +def run( + cfg: dict, + journal: Journal, + opts: RunOptions | None = None, + lock=None, +) -> RunResult: + opts = opts or RunOptions() + ensure_dirs() + result = RunResult() + t0 = time.time() + + jobs = discover_cards(cfg, journal, opts) + result.cards = [j.volume.name for j in jobs] + + if not jobs: + journal.log("本次未发现需要处理的记录仪卡") + result.seconds = time.time() - t0 + return result + + do_delete_global = ( + bool(cfg.get("delete_source_after_verify", True)) and not opts.no_delete + ) + + pending_jobs = [j for j in jobs if j.plan.to_copy] + # 无新增、但卡内还留着「此前已落盘」的副本 => 只差清卡这一步 + sweep_jobs = [ + j + for j in jobs + if not j.plan.to_copy and do_delete_global and j.writable and j.plan.skipped + ] + + if not pending_jobs and not sweep_jobs: + journal.log("卡已识别，所有录像此前均已导入且卡内无待清理副本，无动作") + result.skipped = sum(len(j.plan.skipped) for j in jobs) + result.seconds = time.time() - t0 + if not opts.quiet and not opts.dry_run: + notify.notify( + "检测视频：无新增", + f"{'、'.join(result.cards)} 内有 {result.skipped} 个录像，此前均已导入。", + ) + return result + + work_jobs = pending_jobs + sweep_jobs + sweep_only = not pending_jobs and bool(sweep_jobs) + + # 空间预检（仅新拷贝需要） + for job in pending_jobs: + ok, msg = planner.check_space(job.plan, cfg, journal) + journal.log(f"空间预检（{job.volume.name}）：{msg}") + if not ok: + result.aborted = True + result.abort_reason = "目标盘空间不足" + result.errors.append(msg) + journal.error(f"中止：{msg}") + notify.notify("检测视频：空间不足", msg) + result.seconds = time.time() - t0 + return result + + total_files = sum(len(j.plan.to_copy) for j in pending_jobs) + total_bytes = sum(j.plan.pending_bytes for j in pending_jobs) + + if opts.dry_run: + if pending_jobs: + journal.step( + f"[演练模式] 将导入 {total_files} 个文件 / {human_size(total_bytes)}，不做任何写入" + ) + for job in pending_jobs: + for item in job.plan.to_copy: + journal.log(f" {item.src} -> {item.dst}") + for job in sweep_jobs: + journal.step( + f"[演练模式] {job.volume.name}：将核对并清理 {len(job.plan.skipped)} 个卡内副本" + ) + result.seconds = time.time() - t0 + return result + + # 倒计时确认：只有「有新视频要导入」才打断用户；纯清卡静默执行 + if pending_jobs: + if not opts.assume_yes: + timeout = int(cfg.get("notify_countdown_seconds", 60)) + title, message = _confirm_message(pending_jobs, cfg, timeout) + verdict = notify.ask_with_countdown(title, message, timeout=timeout) + if verdict == "cancel": + journal.log("用户取消，本次不执行") + result.aborted = True + result.abort_reason = "用户取消" + result.seconds = time.time() - t0 + return result + journal.log(f"确认结果：{verdict}（ok=点击确认或超时自动执行）") + else: + journal.step("已指定 --yes，跳过倒计时确认") + + journal.step( + f"开始导入：{total_files} 个文件 / {human_size(total_bytes)}，" + f"目标 {cfg['target_root']}" + ) + notify.notify( + "检测视频：开始导入", + f"{total_files} 个视频 / {human_size(total_bytes)}，正在拷贝，期间可继续用电脑。", + ) + elif sweep_only: + journal.step("卡内无新视频，仅执行清卡核对（不打断用户）") + + for job in work_jobs: + _run_one_card(job, cfg, journal, opts, result, lock=lock) + if result.aborted: + break + + result.seconds = time.time() - t0 + + # 结果汇报 + if result.aborted: + notify.notify( + "检测视频：导入中断", + f"已成功 {result.imported} 个 / {human_size(result.bytes_moved)}，" + f"原因：{result.abort_reason}。卡内未成功导入的文件已保留，重新插卡即可续传。", + ) + journal.error(f"导入中断：{result.abort_reason}") + elif result.imported or (result.deleted and not sweep_only): + date_txt = "、".join(f"{d}({n})" for d, n in sorted(result.dates.items())) + if result.delete_failed: + tail = f"有 {result.delete_failed} 个卡内文件未能删除，仍有残留。" + elif result.deleted: + tail = f"卡内 {result.deleted} 个文件已清空。" + else: + tail = "卡内源文件已保留。" + eject_note = "" + if cfg.get("eject_after_import", True) and result.volumes and not opts.dry_run: + delay = max(10, int(cfg.get("eject_delay_seconds", 300))) + eject_note = f" {ejector.delay_text(delay)}自动弹出 SD 卡。" + notify.notify( + "检测视频：导出完成", + f"{result.imported} 个视频 / {human_size(result.bytes_moved)} / " + f"耗时 {human_duration(result.seconds)}。{tail}{eject_note}", + subtitle=date_txt, + ) + elif sweep_only and not opts.quiet: + notify.notify( + "检测视频：卡内已清空", + f"卡内 {result.deleted} 个已导入文件核对通过并清理完毕。", + ) + + if result.imported or result.deleted: + journal.step( + f"完成：导入 {result.imported} 个 / {human_size(result.bytes_moved)} / " + f"耗时 {human_duration(result.seconds)}；清卡 {result.deleted} 个，" + f"清卡失败 {result.delete_failed}" + ) + + # 导出完成 -> 明确交代结果 -> 延时自动弹出卡。 + # 中止时不安排弹出：卡里还有没导完的东西，用户多半要重插重试。 + if not result.aborted and not opts.dry_run and (result.imported or result.deleted): + ejector.offer_and_schedule(cfg, journal, result, [Path(v) for v in result.volumes]) + + return result diff --git a/src/sdimport/planner.py b/src/sdimport/planner.py new file mode 100644 index 0000000..c65b1cb --- /dev/null +++ b/src/sdimport/planner.py @@ -0,0 +1,158 @@ +"""导入规划：按文件名日期分组、幂等判定、空间预检。 + +纪律（来自实战教训）： + 1. 日期一律取文件名里的 YYYYMMDD，不看卡上的目录名（导入时目录名常与拍摄日期错位）。 + 2. 目标已存在且体积一致 => 视为已导入，跳过，不覆盖、不改名。 + 3. 体积不一致 => 判为冲突，改名保存并告警，绝不覆盖既有文件。 +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass, field +from pathlib import Path + +from .journal import human_size +from .volumes import Volume + +COPY = "copy" +SKIP = "skip" + + +@dataclass +class PlanItem: + src: Path + dst: Path + date: str + size: int + action: str + reason: str = "" + + +@dataclass +class Plan: + volume: Volume + items: list[PlanItem] = field(default_factory=list) + conflicts: list[PlanItem] = field(default_factory=list) + + @property + def to_copy(self) -> list[PlanItem]: + return [i for i in self.items if i.action == COPY] + + @property + def skipped(self) -> list[PlanItem]: + return [i for i in self.items if i.action == SKIP] + + @property + def pending_bytes(self) -> int: + return sum(i.size for i in self.to_copy) + + @property + def total_bytes(self) -> int: + return sum(i.size for i in self.items) + + def dates(self) -> list[str]: + return sorted({i.date for i in self.to_copy}) + + def group_by_date(self) -> dict[str, list[PlanItem]]: + out: dict[str, list[PlanItem]] = {} + for item in self.to_copy: + out.setdefault(item.date, []).append(item) + return {k: sorted(v, key=lambda x: x.src.name) for k, v in sorted(out.items())} + + +def parse_date(filename: str, cfg: dict) -> str | None: + """从文件名提取 YYYYMMDD。""" + flags = re.IGNORECASE if cfg.get("case_insensitive", True) else 0 + try: + pattern = re.compile(str(cfg["filename_regex"]), flags) + except re.error: + return None + m = pattern.match(filename) + if not m or not m.groups(): + return None + raw = m.group(1) + if len(raw) != 8 or not raw.isdigit(): + return None + y, mo, d = int(raw[:4]), int(raw[4:6]), int(raw[6:8]) + if not (1990 <= y <= 2200 and 1 <= mo <= 12 and 1 <= d <= 31): + return None + return raw + + +def _unique_dest(dest: Path) -> Path: + stem, suffix = dest.stem, dest.suffix + for n in range(2, 1000): + cand = dest.with_name(f"{stem} ({n}){suffix}") + if not cand.exists(): + return cand + return dest.with_name(f"{stem} ({dest.stat().st_mtime_ns}){suffix}") + + +def build_plan(volume: Volume, files: list[Path], cfg: dict, journal=None) -> Plan: + root = Path(cfg["target_root"]) + plan = Plan(volume=volume) + + for src in files: + date = parse_date(src.name, cfg) + if date is None: + if journal: + journal.warn(f"文件名无法解析日期，跳过：{src.name}") + continue + try: + size = src.stat().st_size + except OSError as exc: + if journal: + journal.warn(f"读取失败，跳过：{src}（{exc}）") + continue + + dest_dir = root / date + dest = dest_dir / src.name + if dest.exists(): + try: + same = dest.stat().st_size == size + except OSError: + same = False + if same: + plan.items.append( + PlanItem(src, dest, date, size, SKIP, "目标已存在且体积一致") + ) + continue + new_dest = _unique_dest(dest) + item = PlanItem(src, new_dest, date, size, COPY, "同名但体积不同，改用新名") + plan.items.append(item) + plan.conflicts.append(item) + if journal: + journal.warn( + f"冲突：{dest.name} 已存在且体积不符" + f"（本地 {human_size(dest.stat().st_size)} vs 卡上 {human_size(size)}），" + f"本次落为 {new_dest.name}" + ) + continue + + plan.items.append(PlanItem(src, dest, date, size, COPY, "新增")) + + return plan + + +def check_space(plan: Plan, cfg: dict, journal=None) -> tuple[bool, str]: + """目标盘剩余空间是否够用。""" + root = Path(cfg["target_root"]) + probe = root + while not probe.exists() and probe.parent != probe: + probe = probe.parent + import shutil + + try: + free = shutil.disk_usage(probe).free + except OSError as exc: + return False, f"无法读取目标盘剩余空间：{exc}" + + need = int(plan.pending_bytes * float(cfg.get("min_free_space_ratio", 1.10))) + if free < need: + return ( + False, + f"空间不足：待导入 {human_size(plan.pending_bytes)}，" + f"需预留 {human_size(need)}，目标盘可用 {human_size(free)}", + ) + return True, f"空间充足：待导入 {human_size(plan.pending_bytes)}，可用 {human_size(free)}" diff --git a/src/sdimport/platforms/__init__.py b/src/sdimport/platforms/__init__.py new file mode 100644 index 0000000..dc52e5a --- /dev/null +++ b/src/sdimport/platforms/__init__.py @@ -0,0 +1,22 @@ +"""按运行平台自动选择实现。 + +用法：`from .platforms import current, Volume` +""" + +from __future__ import annotations + +import sys + +from .base import Platform, ServiceStatus, Volume + +if sys.platform == "darwin": + from .macos import MacPlatform as _Impl +elif sys.platform.startswith("win"): + from .windows import WindowsPlatform as _Impl +else: + from .generic import GenericPlatform as _Impl + +#: 当前平台的实现实例 +current: Platform = _Impl() + +__all__ = ["Platform", "ServiceStatus", "Volume", "current"] diff --git a/src/sdimport/platforms/base.py b/src/sdimport/platforms/base.py new file mode 100644 index 0000000..5cc0400 --- /dev/null +++ b/src/sdimport/platforms/base.py @@ -0,0 +1,207 @@ +"""平台适配层：接口定义与跨平台数据结构。 + +核心逻辑（按日期分组、拷贝、校验、清卡、编排）完全不关心操作系统。 +所有系统差异都收敛到这里定义的四组动作： + + 1. **卷** —— 枚举候选卷、读卷信息、判断可写、等挂载稳定 + 2. **通知** —— 发通知、两种带倒计时的确认框 + 3. **弹出** —— 把可移动卷弹出来 + 4. **服务** —— 安装 / 卸载 / 查询后台服务 + +各平台实现见 macos.py / windows.py；运行时由 platforms.current 自动选。 +接口里能跨平台实现的（写测试、等稳定）直接在基类给出，子类只实现真正有差异的部分。 +""" + +from __future__ import annotations + +import os +import shutil +import time +from dataclasses import dataclass +from pathlib import Path + +# 等待挂载稳定时的默认节拍 +DEFAULT_STABILITY_CHECKS = 3 +DEFAULT_STABILITY_INTERVAL = 0.8 + + +@dataclass +class Volume: + """一个候选卷。macOS 上是 /Volumes/<卷名>，Windows 上是 E:\\。""" + + name: str + mount_point: Path + uuid: str = "" + fs: str = "" + readonly: bool = False + removable: bool = False + bus: str = "" + total_bytes: int = 0 + free_bytes: int = 0 + + @property + def identity(self) -> str: + """跨插拔稳定的标识，用于状态去重。""" + return self.uuid or f"{self.name}::{self.mount_point}" + + def describe(self) -> str: + kind = self.bus or self.fs or "?" + ro = "只读" if self.readonly else "可写" + return f"{self.name}({kind}, {ro})" + + +@dataclass +class ServiceStatus: + """后台服务状态。 + + installed：服务**当前**是否处于已安装状态（卸载成功后应为 False） + ok ：**本次操作**本身是否成功（用于让调用方判断退出码） + """ + + installed: bool + detail: str = "" + ok: bool = True + + +class Platform: + """平台能力接口。子类只需覆盖真正有差异的方法。""" + + name = "generic" + # 该平台是否靠「常驻进程」实现插卡触发。 + # macOS 用 launchd 的 WatchPaths + StartInterval，无需常驻；Windows 需要常驻。 + needs_daemon = False + + # ------------------------------------------------------------------ + # 1. 卷 + # ------------------------------------------------------------------ + + def list_volumes(self, cfg: dict) -> list[Volume]: + raise NotImplementedError + + def probe_volume(self, mount_point: Path) -> Volume | None: + raise NotImplementedError + + def default_mount_root(self) -> Path | None: + """候选卷所在的位置（macOS 是 /Volumes，Windows 无固定根，返回 None）。""" + return None + + def volume_names(self, cfg: dict) -> list[str]: + """当前候选卷的**轻量**名字快照，用于判断「是不是新插了一张卡」。 + + 必须轻量：兜底触发每 30~120 秒就会调一次，不能每次都去问系统要卷详情。 + 子类应覆写成不调慢命令的版本；默认实现退化到完整枚举。 + """ + return sorted(v.name for v in self.list_volumes(cfg)) + + def check_writable(self, mount_point: Path) -> bool: + """真实写测试，比读只读标记可靠。""" + probe_file = Path(mount_point) / ".sdimport_write_test" + try: + probe_file.write_bytes(b"ok") + probe_file.unlink() + return True + except OSError: + return False + + def wait_until_stable( + self, + mount_point: Path, + checks: int = DEFAULT_STABILITY_CHECKS, + interval: float = DEFAULT_STABILITY_INTERVAL, + ) -> bool: + """等到卷挂载稳定。 + + 判据只用**卷总容量**：它在挂载完成后固定。 + 不能用可用空间 / 文件数 —— 系统后台写入会让这些值持续变动， + 永远等不到相等（macOS 上实测踩过：整个等待窗口跑满仍未稳定）。 + shutil.disk_usage 两端都可用，故放在基类。 + """ + mount_point = Path(mount_point) + last_total = None + stable = 0 + deadline = time.time() + checks * interval * 6 + 5 + while time.time() < deadline: + try: + usage = shutil.disk_usage(mount_point) + if usage.total <= 0: + raise OSError("卷容量为 0，可能仍在挂载") + os.listdir(mount_point) + except OSError: + last_total, stable = None, 0 + time.sleep(interval) + continue + if usage.total == last_total: + stable += 1 + if stable >= checks: + return True + else: + last_total, stable = usage.total, 0 + time.sleep(interval) + return False + + # ------------------------------------------------------------------ + # 2. 通知 + # ------------------------------------------------------------------ + + def notify(self, title: str, message: str, subtitle: str = "", sound: str = "") -> bool: + raise NotImplementedError + + def ask_countdown( + self, + title: str, + message: str, + timeout: int, + ok_label: str = "立即开始", + cancel_label: str = "取消导入", + ) -> str: + """开始导入前的确认框。 + + 返回 'ok'（点了确认，或超时未操作）/'cancel'（明确取消）/'error'（弹不出来）。 + 超时必须按「同意」算 —— 用户要的是「不响应就执行」。 + """ + raise NotImplementedError + + def ask_completion( + self, + title: str, + message: str, + timeout: int = 30, + keep_label: str = "暂不弹出", + ok_label: str = "知道了", + ) -> str: + """导出完成提示框。 + + 返回 'ok'（继续弹出计划）/'keep'（明确选了暂不弹出）/'error'（弹不出来）。 + """ + raise NotImplementedError + + # ------------------------------------------------------------------ + # 3. 弹出 + # ------------------------------------------------------------------ + + def eject( + self, + mount_point: Path, + attempts: int, + interval: float, + use_fallback: bool = False, + ) -> tuple[bool, str]: + """弹出可移动卷。只弹出，绝不删除任何文件。""" + raise NotImplementedError + + # ------------------------------------------------------------------ + # 4. 后台服务 + # ------------------------------------------------------------------ + + def install_service(self, project_root: Path) -> ServiceStatus: + raise NotImplementedError + + def uninstall_service(self, project_root: Path) -> ServiceStatus: + raise NotImplementedError + + def service_status(self) -> ServiceStatus: + raise NotImplementedError + + def daemon_argv(self, project_root: Path) -> list[str]: + """常驻进程的启动命令（needs_daemon 为 True 的平台才有意义）。""" + raise NotImplementedError diff --git a/src/sdimport/platforms/generic.py b/src/sdimport/platforms/generic.py new file mode 100644 index 0000000..06196e8 --- /dev/null +++ b/src/sdimport/platforms/generic.py @@ -0,0 +1,136 @@ +"""兜底平台实现（非 macOS、非 Windows）。 + +只为保证程序能启动并给出可读提示，不假装支持。 +Linux 上顺手支持常见挂载点与 notify-send / umount。 +""" + +from __future__ import annotations + +import os +import shutil +import subprocess +import time +from pathlib import Path + +from .base import Platform, ServiceStatus, Volume + +LINUX_MOUNT_ROOTS = ("/media", "/run/media", "/mnt") + + +def _iter_mount_dirs(): + for root in LINUX_MOUNT_ROOTS: + base = Path(root) + if not base.is_dir(): + continue + try: + entries = sorted(base.iterdir()) + except OSError: + continue + for entry in entries: + if not entry.is_dir() or entry.is_symlink(): + continue + yield entry + # /run/media//