初始化 sdimport:检测视频 SD 卡自动导入工具(macOS/Windows 跨平台)

This commit is contained in:
jkooe committed 2026-10-12 17:58:30 +08:00
commit 64d2b63767
28 files changed
+4118

No files matched your search

+16
View File
@@ -0,0 +1,16 @@
# 运行时产物(日志/状态含真实卷名、楼栋号、用户名、时间戳,绝不入库)
runtime/
# 历史备份目录
src.bak-*/
# WorkBuddy 项目记忆(私人工作记录)
.workbuddy/
# Python
__pycache__/
*.py[cod]
*.egg-info/
# macOS
.DS_Store
+154
View File
@@ -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+(<https://www.python.org/downloads/>),安装时勾选 **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/检测视频` | 落盘根目录,按 `<root>/<YYYYMMDD>/` 组织 |
| `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 卡 → `<target_root>/<YYYYMMDD>/` → 清卡」这一段。
电梯编号识别由 `elevator-video-classifier` skill 负责,那一步需要人工校对,两者不重叠。
- **Windows 侧代码未在真机实测**(开发机是 macOS)。写法上已尽量保守
(纯标准库优先、依赖缺失自动降级、无 GUI 会话也能工作),但仍建议先跑 `selftest` 与 `scan`。
+32
View File
@@ -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 "$@"
Executable
+31
View File
@@ -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 "$@"
+16
View File
@@ -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%
+47
View File
@@ -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
}
+10
View File
@@ -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
+3
View File
@@ -0,0 +1,3 @@
"""检测视频 SD 卡自动导入。"""
__version__ = "1.0.0"
+428
View File
@@ -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())
+109
View File
@@ -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
+323
View File
@@ -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
+178
View File
@@ -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()
+54
View File
@@ -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)
+131
View File
@@ -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>/<YYYYMMDD>/ 组织
"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
+468
View File
@@ -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
+158
View File
@@ -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)}"
+22
View File
@@ -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"]
+207
View File
@@ -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
+136
View File
@@ -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/<user>/<label> 这种两层结构
if root == "/run/media":
try:
for sub in sorted(entry.iterdir()):
if sub.is_dir() and not sub.is_symlink():
yield sub
except OSError:
continue
class GenericPlatform(Platform):
name = "generic"
needs_daemon = False
def default_mount_root(self) -> Path | None:
return Path("/media") if Path("/media").is_dir() else None
def list_volumes(self, cfg: dict) -> list[Volume]:
deny = {str(x) for x in cfg.get("deny_volumes", [])}
out: list[Volume] = []
for entry in _iter_mount_dirs():
if entry.name in deny:
continue
vol = self.probe_volume(entry)
if vol is not None:
out.append(vol)
return out
def probe_volume(self, mount_point: Path) -> Volume | None:
mp = Path(mount_point)
if not mp.is_dir():
return None
try:
usage = shutil.disk_usage(mp)
os.listdir(mp)
except OSError:
return None
writable = os.access(mp, os.W_OK)
return Volume(
name=mp.name,
mount_point=mp,
readonly=not writable,
removable=True,
total_bytes=usage.total,
free_bytes=usage.free,
)
def notify(self, title: str, message: str, subtitle: str = "", sound: str = "") -> bool:
try:
proc = subprocess.run(
["notify-send", title, message], capture_output=True, timeout=10
)
return proc.returncode == 0
except (subprocess.SubprocessError, OSError):
return False
def ask_countdown(
self, title, message, timeout, ok_label="立即开始", cancel_label="取消导入"
) -> str:
# 非目标平台:给通知 + 静默等待,保持「不响应即执行」语义
self.notify(title, message)
time.sleep(min(max(int(timeout), 0), 30))
return "ok"
def ask_completion(
self, title, message, timeout=30, keep_label="暂不弹出", ok_label="知道了"
) -> str:
self.notify(title, message)
return "ok"
def eject(
self, mount_point: Path, attempts: int, interval: float, use_fallback: bool = False
) -> tuple[bool, str]:
mp = Path(mount_point)
last = ""
for i in range(max(1, attempts)):
if not mp.exists():
return True, "卷已不在"
try:
proc = subprocess.run(
["umount", str(mp)], capture_output=True, text=True, timeout=30
)
except (subprocess.SubprocessError, OSError) as exc:
return False, str(exc)
if proc.returncode == 0:
return True, "已卸载"
last = ((proc.stdout or "") + (proc.stderr or "")).strip().replace("\n", " ")
if i < attempts - 1:
time.sleep(max(1.0, interval))
return False, last or "未知错误"
def install_service(self, project_root: Path) -> ServiceStatus:
return ServiceStatus(
False, "当前平台不支持自动安装后台服务,请自行配置定时任务", ok=False
)
def uninstall_service(self, project_root: Path) -> ServiceStatus:
return ServiceStatus(False, "当前平台无需卸载")
def service_status(self) -> ServiceStatus:
return ServiceStatus(False, "当前平台不支持后台服务")
def daemon_argv(self, project_root: Path) -> list[str]:
return []
+452
View File
@@ -0,0 +1,452 @@
"""macOS 平台实现:/Volumes + diskutil + osascript + launchd。
这些代码原先散在 volumes.py / notify.py / ejector.py / cli.py 里。
2026-09-25 为适配 Windows 收拢到此处,**行为与之前保持一致**,
只把 os.statvfs 换成了跨平台的 shutil.disk_usage(等价)。
"""
from __future__ import annotations
import os
import plistlib
import shutil
import subprocess
import time
from pathlib import Path
from .base import Platform, ServiceStatus, Volume
VOLUMES_ROOT = Path("/Volumes")
# 网络文件系统,不参与扫描
NETWORK_FS = {"nfs", "smbfs", "afpfs", "webdav", "cddafs", "ftp", "cifs"}
LAUNCH_AGENT_LABEL = "com.sdimport.agent"
def _as_str(s: str) -> str:
"""转成 AppleScript 字符串字面量(含引号)。"""
body = str(s).replace("\\", "\\\\").replace('"', '\\"')
return f'"{body}"'
class MacPlatform(Platform):
name = "macos"
# launchd 的 WatchPaths + StartInterval 已够用,不需要常驻进程
needs_daemon = False
def __init__(self) -> None:
self._finder_blocked = False
# ------------------------------------------------------------------
# 卷
# ------------------------------------------------------------------
def default_mount_root(self) -> Path | None:
return VOLUMES_ROOT
def volume_names(self, cfg: dict) -> list[str]:
"""轻量快照:只列 /Volumes 目录名,不问 diskutil。"""
deny = {str(x) for x in cfg.get("deny_volumes", [])}
names: list[str] = []
try:
for entry in sorted(VOLUMES_ROOT.iterdir()):
if not entry.is_dir() or entry.is_symlink():
continue
if entry.name in deny or entry.name.startswith("."):
continue
if entry.name.startswith("com.apple.TimeMachine"):
continue
names.append(entry.name)
except OSError:
pass
return names
@staticmethod
def _diskutil_info(mount_point: Path) -> dict:
try:
out = subprocess.run(
["diskutil", "info", "-plist", str(mount_point)],
capture_output=True,
timeout=15,
)
if out.returncode != 0 or not out.stdout:
return {}
return plistlib.loads(out.stdout)
except (subprocess.SubprocessError, plistlib.InvalidFileException, OSError, ValueError):
return {}
@staticmethod
def _mount_flags(mount_point: Path) -> str:
"""返回该挂载点在 mount(8) 输出中的括号标志串,取不到返回空。"""
try:
out = subprocess.run(["mount"], capture_output=True, text=True, timeout=10).stdout
except (subprocess.SubprocessError, OSError):
return ""
target = str(mount_point)
escaped = target.replace(" ", "\\040")
for line in out.splitlines():
if f" on {target} " in line or f" on {escaped} " in line:
idx = line.rfind("(")
if idx != -1:
return line[idx + 1 :].rstrip(")").strip()
return ""
return ""
def probe_volume(self, mount_point: Path) -> Volume | None:
mp = Path(mount_point)
if not mp.is_dir():
return None
try:
if os.stat(mp).st_dev == os.stat(mp.parent).st_dev and mp.parent == VOLUMES_ROOT:
# /Volumes 下与父目录同设备 => 不是真实挂载
return None
except OSError:
return None
info = self._diskutil_info(mp)
flags = self._mount_flags(mp)
if flags and not flags.split(",")[0].strip():
flags = ""
if any(fs in flags for fs in NETWORK_FS):
return None
name = str(info.get("VolumeName") or mp.name)
fs_type = str(info.get("FilesystemType") or info.get("FileSystemPersonality") or "")
readonly = bool(info.get("ReadOnlyVolume")) or "read-only" in flags
removable = bool(
info.get("RemovableMedia")
or info.get("RemovableMediaOrExternalDevice")
or info.get("Ejectable")
)
bus = str(info.get("BusProtocol") or "")
try:
usage = shutil.disk_usage(mp)
total, free = usage.total, usage.free
except OSError:
total = free = 0
return Volume(
name=name,
mount_point=mp,
uuid=str(info.get("VolumeUUID") or ""),
fs=fs_type,
readonly=readonly,
removable=removable,
bus=bus,
total_bytes=total,
free_bytes=free,
)
def list_volumes(self, cfg: dict) -> list[Volume]:
"""列出 /Volumes 下值得检查的卷(已排除系统卷与网络卷)。"""
deny = {str(x) for x in cfg.get("deny_volumes", [])}
out: list[Volume] = []
try:
entries = sorted(VOLUMES_ROOT.iterdir())
except OSError:
return out
for entry in entries:
if not entry.is_dir() or entry.is_symlink():
continue
if entry.name in deny or entry.name.startswith("."):
continue
if entry.name.startswith("com.apple.TimeMachine"):
continue
vol = self.probe_volume(entry)
if vol is not None:
out.append(vol)
return out
# ------------------------------------------------------------------
# 通知
# ------------------------------------------------------------------
@staticmethod
def _osascript(script: str, timeout: float = 30.0) -> tuple[int, str]:
try:
proc = subprocess.run(
["osascript", "-e", script],
capture_output=True,
text=True,
timeout=timeout,
)
return proc.returncode, (proc.stdout or "") + (proc.stderr or "")
except subprocess.TimeoutExpired:
return 124, "timeout"
except (subprocess.SubprocessError, OSError) as exc:
return 1, str(exc)
def notify(self, title: str, message: str, subtitle: str = "", sound: str = "") -> bool:
parts = [
f"display notification {_as_str(message)}",
f"with title {_as_str(title)}",
]
if subtitle:
parts.append(f"subtitle {_as_str(subtitle)}")
if sound:
parts.append(f"sound name {_as_str(sound)}")
code, _ = self._osascript(" ".join(parts), timeout=15)
return code == 0
def ask_countdown(
self,
title: str,
message: str,
timeout: int,
ok_label: str = "立即开始",
cancel_label: str = "取消导入",
) -> str:
script = (
f"display dialog {_as_str(message)} with title {_as_str(title)} "
f"buttons {{{_as_str(cancel_label)}, {_as_str(ok_label)}}} "
f"default button {_as_str(ok_label)} with icon note giving up after {int(timeout)}"
)
code, out = self._osascript(script, timeout=timeout + 20)
if code != 0:
# 弹窗不可用(无 GUI session 等),降级为通知 + 静默倒计时
self.notify(title, message)
time.sleep(min(timeout, 60))
return "error"
if f"button returned:{cancel_label}" in out:
return "cancel"
return "ok"
def ask_completion(
self,
title: str,
message: str,
timeout: int = 30,
keep_label: str = "暂不弹出",
ok_label: str = "知道了",
) -> str:
script = (
f"display dialog {_as_str(message)} with title {_as_str(title)} "
f"buttons {{{_as_str(keep_label)}, {_as_str(ok_label)}}} "
f"default button {_as_str(ok_label)} with icon note giving up after {int(timeout)}"
)
code, out = self._osascript(script, timeout=timeout + 20)
if code != 0:
return "error"
if f"button returned:{keep_label}" in out:
return "keep"
return "ok"
def banner(self, title: str, message: str, timeout: int = 60) -> bool:
"""纯提示框(无选择语义),超时自动消失。"""
script = (
f"display dialog {_as_str(message)} with title {_as_str(title)} "
f'buttons {{"知道了"}} default button "知道了" with icon note '
f"giving up after {int(timeout)}"
)
code, _ = self._osascript(script, timeout=timeout + 20)
return code == 0
# ------------------------------------------------------------------
# 弹出
# ------------------------------------------------------------------
def _eject_via_diskutil(self, mount_point: Path) -> tuple[bool, str]:
try:
proc = subprocess.run(
["diskutil", "eject", str(mount_point)],
capture_output=True,
text=True,
timeout=30,
)
except subprocess.TimeoutExpired:
return False, "diskutil 超时(卷正被占用)"
except (subprocess.SubprocessError, OSError) as exc:
return False, str(exc)
if proc.returncode == 0:
return True, "已弹出"
return False, ((proc.stdout or "") + (proc.stderr or "")).strip().replace("\n", " ")
def _eject_via_finder(self, volume_name: str) -> tuple[bool, str]:
"""让访达推出(默认关闭,见配置 eject_finder_fallback)。
2026-09-15 实测:diskutil 报
`Unmount was dissented by PID 182 (.../loginwindow)`,
那是卷被图形会话(访达窗口 / 桌面)引用,diskutil 说不通。
但 osascript 控制访达需要「自动化」权限;未授权时会弹授权框并**阻塞**,
所以默认不开。一旦发现权限被拒,本次运行不再尝试,免得反复卡住。
"""
if self._finder_blocked:
return False, "访达通道已停用(缺自动化权限)"
script = f'tell application "Finder" to eject disk {_as_str(volume_name)}'
try:
proc = subprocess.run(
["osascript", "-e", script], capture_output=True, text=True, timeout=10
)
except subprocess.TimeoutExpired:
self._finder_blocked = True
return False, "访达通道超时(多半是缺自动化权限)"
except (subprocess.SubprocessError, OSError) as exc:
return False, str(exc)
if proc.returncode == 0:
return True, "已通过访达推出"
err = ((proc.stdout or "") + (proc.stderr or "")).strip().replace("\n", " ")
if "-10004" in err or "Not authorized" in err or "权限违例" in err:
self._finder_blocked = True
return False, err
def eject(
self,
mount_point: Path,
attempts: int,
interval: float,
use_fallback: bool = False,
) -> tuple[bool, str]:
"""弹出卷:先 diskutil,必要时改走访达,失败则重试。只弹出,不删除任何文件。"""
mount_point = Path(mount_point)
last = ""
for i in range(max(1, attempts)):
if not mount_point.exists():
return True, "卷已不在(可能已被手动拔出)"
ok, why = self._eject_via_diskutil(mount_point)
if ok:
return True, why
if use_fallback:
ok2, why2 = self._eject_via_finder(mount_point.name)
if ok2 or not mount_point.exists():
return True, why2 if ok2 else "已弹出"
if why2 and why2 not in why:
why = f"{why};访达通道:{why2}"
last = why
if i < attempts - 1:
time.sleep(max(1.0, interval))
return False, last or "未知错误"
# ------------------------------------------------------------------
# 后台服务(launchd)
# ------------------------------------------------------------------
@property
def plist_path(self) -> Path:
return Path.home() / "Library" / "LaunchAgents" / f"{LAUNCH_AGENT_LABEL}.plist"
@staticmethod
def _launchctl(*args: str) -> tuple[int, str]:
try:
proc = subprocess.run(
["launchctl", *args], capture_output=True, text=True, timeout=30
)
return proc.returncode, (proc.stdout or "") + (proc.stderr or "")
except (subprocess.SubprocessError, OSError) as exc:
return 1, str(exc)
def install_service(self, project_root: Path) -> ServiceStatus:
project_root = Path(project_root)
log_dir = project_root / "runtime" / "logs"
entry = project_root / "bin" / "on-volume-change.sh"
log_dir.mkdir(parents=True, exist_ok=True)
payload = {
"Label": LAUNCH_AGENT_LABEL,
"ProgramArguments": ["/bin/bash", str(entry)],
"WatchPaths": ["/Volumes"],
"StartInterval": 120,
"RunAtLoad": True,
"ThrottleInterval": 15,
"StandardOutPath": str(log_dir / "launchd.out.log"),
"StandardErrorPath": str(log_dir / "launchd.err.log"),
}
plist_path = self.plist_path
plist_path.parent.mkdir(parents=True, exist_ok=True)
try:
plist_path.write_bytes(plistlib.dumps(payload))
os.chmod(entry, 0o755)
except OSError as exc:
return ServiceStatus(False, f"写入 plist 失败:{exc}", ok=False)
uid = os.getuid()
self._launchctl("bootout", f"gui/{uid}/{LAUNCH_AGENT_LABEL}")
code, out = self._launchctl("bootstrap", f"gui/{uid}", str(plist_path))
if code != 0:
self._launchctl("load", "-w", str(plist_path))
code2, _ = self._launchctl("print", f"gui/{uid}/{LAUNCH_AGENT_LABEL}")
if code2 == 0:
return ServiceStatus(
True,
f"后台服务已安装并加载\n"
f" plist : {plist_path}\n"
f" 触发 : /Volumes 发生变化(插卡)或每 120 秒兜底检查一次",
)
first = (out.strip().splitlines() or ["未知"])[0]
return ServiceStatus(
False,
"plist 已就位,但未能在本次会话中加载。\n"
f" plist : {plist_path}\n"
f" 底层报错: {first}\n"
"\n"
" 原因:执行本命令的进程没有完整的用户图形会话权限,无法与 launchd 通信。\n"
" 任意一种方式即可生效:\n"
" 1) 双击项目根目录的「安装后台服务.command」\n"
f" 2) 在「终端」里粘贴:launchctl bootstrap gui/$(id -u) {plist_path}\n"
" 3) 什么都不做 —— 注销或重启后,macOS 会自动加载这个 plist",
ok=False,
)
def uninstall_service(self, project_root: Path) -> ServiceStatus:
uid = os.getuid()
self._launchctl("bootout", f"gui/{uid}/{LAUNCH_AGENT_LABEL}")
self._launchctl("unload", "-w", str(self.plist_path))
if self.plist_path.exists():
try:
self.plist_path.unlink()
except OSError as exc:
return ServiceStatus(False, f"卸载失败,plist 删不掉:{exc}", ok=False)
return ServiceStatus(False, "后台服务已卸载")
def service_status(self) -> ServiceStatus:
code, out = self._launchctl("print", f"gui/{os.getuid()}/{LAUNCH_AGENT_LABEL}")
if code != 0:
return ServiceStatus(False, "未安装")
# `launchctl print` 里 state 会出现多次(本级与各子域),只取第一条
state = ""
pid = ""
for line in out.splitlines():
text = line.strip()
if not state and text.startswith("state ="):
state = text.split("=", 1)[1].strip()
elif not pid and text.startswith("pid ="):
pid = text.split("=", 1)[1].strip()
bits = [f"state={state}"] if state else []
if pid:
bits.append(f"pid={pid}")
return ServiceStatus(True, "已安装 " + (", ".join(bits) or "(待触发)"))
def diagnose(self) -> list[tuple[str, bool, str]]:
"""自检用:各项外部依赖是否可用。"""
out: list[tuple[str, bool, str]] = []
try:
proc = subprocess.run(
["diskutil", "info", "-plist", "/"], capture_output=True, timeout=15
)
out.append(("diskutil", proc.returncode == 0, ""))
except (subprocess.SubprocessError, OSError) as exc:
out.append(("diskutil", False, str(exc)))
try:
proc = subprocess.run(
["osascript", "-e", "return 1"], capture_output=True, timeout=10
)
out.append(("osascript", proc.returncode == 0, ""))
except (subprocess.SubprocessError, OSError) as exc:
out.append(("osascript", False, str(exc)))
code, _ = self._launchctl("version")
out.append(("launchctl", code == 0, "" if code == 0 else "无法与 launchd 通信"))
return out
def daemon_argv(self, project_root: Path) -> list[str]:
return []
+577
View File
@@ -0,0 +1,577 @@
"""Windows 平台实现:盘符 + kernel32(ctypes) + PowerShell + 任务计划。
⚠ **本文件未在真实 Windows 机器上实测过**(开发机是 macOS)。
写法上刻意保守,尽量不引入失败点:
* 卷枚举/探测 —— ctypes 直接调 kernel32,不依赖任何第三方包
* 通知 —— PowerShell 的 Windows Runtime toast;失败降级为消息框
* 确认框 —— tkinter 自绘(Python 自带),支持倒计时自动关闭
* 弹出 —— 先试 Shell.Application 的 Eject 动词,再降级 mountvol /P
* 后台服务 —— schtasks 任务计划,登录自启,并立即拉起一次
首次在那台 Windows 上跑,请依次执行:
sdimport selftest 逐个依赖自检
sdimport scan 演练,看识别对不对
sdimport once 手动导一次
sdimport install 装后台服务
"""
from __future__ import annotations
import base64
import ctypes
import os
import shutil
import subprocess
import sys
import time
from pathlib import Path
from .base import Platform, ServiceStatus, Volume
TASK_NAME = "sdimport"
APP_ID = "sdimport"
# GetDriveTypeW 的返回值
DRIVE_REMOVABLE = 2
DRIVE_FIXED = 3
DRIVE_REMOTE = 4
DRIVE_CDROM = 5
DRIVE_RAMDISK = 6
_CREATE_NO_WINDOW = 0x08000000
# --------------------------------------------------------------------------
# 底层工具
# --------------------------------------------------------------------------
def _run(args: list[str], timeout: float = 30.0) -> tuple[int, str]:
try:
proc = subprocess.run(
args,
capture_output=True,
text=True,
timeout=timeout,
creationflags=_CREATE_NO_WINDOW,
)
return proc.returncode, (proc.stdout or "") + (proc.stderr or "")
except (subprocess.SubprocessError, OSError) as exc:
return 1, str(exc)
def _run_powershell(script: str, timeout: float = 30.0) -> tuple[int, str]:
exe = shutil.which("powershell") or shutil.which("pwsh") or "powershell"
return _run(
[exe, "-NoProfile", "-NonInteractive", "-ExecutionPolicy", "Bypass", "-Command", script],
timeout=timeout,
)
def _kernel32():
return ctypes.WinDLL("kernel32", use_last_error=True) # type: ignore[attr-defined]
def _logical_drives() -> list[str]:
"""当前存在的盘符,如 ['C:\\\\', 'E:\\\\']。"""
try:
mask = int(_kernel32().GetLogicalDrives())
except (OSError, AttributeError):
return []
return [f"{chr(65 + i)}:\\" for i in range(26) if mask & (1 << i)]
def _drive_type(root: str) -> int:
try:
return int(_kernel32().GetDriveTypeW(ctypes.c_wchar_p(root)))
except (OSError, AttributeError):
return 0
def _volume_info(root: str) -> tuple[str, str, str]:
"""返回 (卷标, 文件系统, 序列号)。取不到给空串,绝不抛。"""
try:
k32 = _kernel32()
except (OSError, AttributeError):
return "", "", ""
name_buf = ctypes.create_unicode_buffer(261)
fs_buf = ctypes.create_unicode_buffer(261)
serial = ctypes.c_ulong(0)
max_len = ctypes.c_ulong(0)
flags = ctypes.c_ulong(0)
try:
ok = k32.GetVolumeInformationW(
ctypes.c_wchar_p(root),
name_buf,
len(name_buf),
ctypes.byref(serial),
ctypes.byref(max_len),
ctypes.byref(flags),
fs_buf,
len(fs_buf),
)
except (OSError, AttributeError):
return "", "", ""
if not ok:
return "", "", ""
return name_buf.value, fs_buf.value, f"{serial.value:08X}"
def _xml_escape(text: str) -> str:
return (
str(text)
.replace("&", "&amp;")
.replace("<", "&lt;")
.replace(">", "&gt;")
)
class WindowsPlatform(Platform):
name = "windows"
# Windows 没有 launchd 那样的卷监视,靠常驻进程订阅事件
needs_daemon = True
# ------------------------------------------------------------------
# 卷
# ------------------------------------------------------------------
@staticmethod
def _deny_set(cfg: dict) -> set[str]:
return {str(x).upper().rstrip("\\") for x in cfg.get("deny_volumes", [])}
def _only_removable(self, cfg: dict) -> bool:
"""默认只看可移动盘 —— 绝不把系统盘当候选,这是硬安全线。"""
return bool(cfg.get("windows_only_removable", True))
def _is_candidate(self, root: str, cfg: dict) -> bool:
letter = root.rstrip("\\").upper()
if letter in self._deny_set(cfg):
return False
# 系统盘永远不碰 —— 双保险,即便 windows_only_removable 被关掉
if letter == os.environ.get("SystemDrive", "C:").upper():
return False
kind = _drive_type(root)
if kind in (DRIVE_REMOTE, DRIVE_CDROM, 0):
return False
if self._only_removable(cfg) and kind != DRIVE_REMOVABLE:
return False
return True
def list_volumes(self, cfg: dict) -> list[Volume]:
out: list[Volume] = []
for root in _logical_drives():
if not self._is_candidate(root, cfg):
continue
vol = self.probe_volume(Path(root))
if vol is not None:
out.append(vol)
return out
def volume_names(self, cfg: dict) -> list[str]:
"""轻量快照:只列候选盘符,不问卷标。"""
names: list[str] = []
for root in _logical_drives():
if self._is_candidate(root, cfg):
names.append(root.rstrip("\\").upper())
return sorted(names)
def probe_volume(self, mount_point: Path) -> Volume | None:
root = str(mount_point)
if not root.endswith("\\"):
root += "\\"
kind = _drive_type(root)
if kind == 0:
return None
try:
usage = shutil.disk_usage(root)
os.listdir(root)
except OSError:
return None
label, fs, serial = _volume_info(root)
removable = kind == DRIVE_REMOVABLE
return Volume(
name=label or root.rstrip("\\"),
mount_point=Path(root),
uuid=serial,
fs=fs,
readonly=not os.access(root, os.W_OK),
removable=removable,
bus="Removable" if removable else "Fixed",
total_bytes=usage.total,
free_bytes=usage.free,
)
# ------------------------------------------------------------------
# 通知
# ------------------------------------------------------------------
def notify(self, title: str, message: str, subtitle: str = "", sound: str = "") -> bool:
body = f"{subtitle}\n{message}" if subtitle else message
if self._toast(title, body):
return True
return self._message_box(title, body)
def _toast(self, title: str, body: str) -> bool:
xml = (
'<toast><visual><binding template="ToastGeneric">'
f"<text>{_xml_escape(title)}</text><text>{_xml_escape(body)}</text>"
"</binding></visual></toast>"
)
b64 = base64.b64encode(xml.encode("utf-8")).decode("ascii")
script = (
"$ErrorActionPreference='Stop';"
"[Windows.UI.Notifications.ToastNotificationManager, Windows.UI.Notifications,"
" ContentType=WindowsRuntime] > $null;"
"[Windows.Data.Xml.Dom.XmlDocument, Windows.Data.Xml.Dom.XmlDocument,"
" ContentType=WindowsRuntime] > $null;"
f"$raw=[Text.Encoding]::UTF8.GetString([Convert]::FromBase64String('{b64}'));"
"$x=New-Object Windows.Data.Xml.Dom.XmlDocument;$x.LoadXml($raw);"
"$n=New-Object Windows.UI.Notifications.ToastNotification $x;"
f"[Windows.UI.Notifications.ToastNotificationManager]::CreateToastNotifier('{APP_ID}')"
".Show($n);"
)
code, _ = _run_powershell(script, timeout=25)
return code == 0
def _message_box(self, title: str, body: str) -> bool:
"""toast 不可用时的降级:系统消息框(不阻塞流程,用 MB_SYSTEMMODAL 太吵,
这里用普通模态框并限时,失败也不影响导入本身)。"""
try:
MB_OK = 0x00000000
MB_SETFOREGROUND = 0x00010000
ctypes.windll.user32.MessageBoxW( # type: ignore[attr-defined]
None, body, title, MB_OK | MB_SETFOREGROUND
)
return True
except (OSError, AttributeError):
return False
# ------------------------------------------------------------------
# 确认框(tkinter 自绘,支持倒计时)
# ------------------------------------------------------------------
@staticmethod
def _ask_via_tk(
title: str,
message: str,
timeout: int,
buttons: list[tuple[str, str]],
default_key: str,
) -> str | None:
"""返回被点按钮的 key;超时返回 default_key;tkinter 不可用返回 None。"""
try:
import tkinter as tk
except ImportError:
return None
try:
root = tk.Tk()
except Exception: # noqa: BLE001 - 无图形会话时 TclError
return None
result = {"key": default_key}
try:
root.title(title)
root.attributes("-topmost", True)
root.resizable(False, False)
frame = tk.Frame(root, padx=18, pady=14)
frame.pack(fill="both", expand=True)
tk.Label(frame, text=message, justify="left", anchor="w").pack(fill="x")
hint = tk.StringVar(value="")
tk.Label(frame, textvariable=hint, fg="#606060").pack(anchor="w", pady=(10, 0))
row = tk.Frame(frame)
row.pack(fill="x", pady=(14, 0))
def choose(key: str) -> None:
result["key"] = key
try:
root.destroy()
except Exception: # noqa: BLE001
pass
for key, label in buttons:
btn = tk.Button(row, text=label, width=12, command=lambda k=key: choose(k))
btn.pack(side="right", padx=(8, 0))
if key == default_key:
btn.focus_set()
root.bind("<Return>", lambda _e, k=key: choose(k))
deadline = time.time() + max(0, int(timeout))
def tick() -> None:
remain = int(deadline - time.time())
if remain <= 0:
choose(default_key)
return
hint.set(f"{remain} 秒后自动继续")
root.after(1000, tick)
if timeout > 0:
tick()
else:
hint.set("请选择")
# 直接关窗按默认处理,与 macOS 侧「超时即同意」一致
root.protocol("WM_DELETE_WINDOW", lambda: choose(default_key))
root.update_idletasks()
root.mainloop()
except Exception: # noqa: BLE001
return None
return result["key"]
def ask_countdown(
self,
title: str,
message: str,
timeout: int,
ok_label: str = "立即开始",
cancel_label: str = "取消导入",
) -> str:
key = self._ask_via_tk(
title,
message,
timeout,
buttons=[(cancel_label, "cancel"), (ok_label, "ok")],
default_key="ok",
)
if key is None:
# 无图形会话:退回通知 + 静默等待,保持「不响应即执行」语义
self.notify(title, message)
time.sleep(min(max(int(timeout), 0), 60))
return "error"
return key
def ask_completion(
self,
title: str,
message: str,
timeout: int = 30,
keep_label: str = "暂不弹出",
ok_label: str = "知道了",
) -> str:
key = self._ask_via_tk(
title,
message,
timeout,
buttons=[(keep_label, "keep"), (ok_label, "ok")],
default_key="ok",
)
if key is None:
return "error"
return key
# ------------------------------------------------------------------
# 弹出
# ------------------------------------------------------------------
@staticmethod
def _eject_via_shell(letter: str) -> tuple[bool, str]:
"""用资源管理器的「弹出」动词。"""
script = (
"$ErrorActionPreference='Stop';"
"$s=New-Object -ComObject Shell.Application;"
f"$i=$s.Namespace(17).ParseName('{letter}\\');"
"if($i -eq $null){exit 3};"
"$i.InvokeVerb('Eject');"
)
code, out = _run_powershell(script, timeout=30)
if code == 0:
return True, "已通过资源管理器弹出"
return False, out.strip().replace("\n", " ")
@staticmethod
def _eject_via_mountvol(letter: str) -> tuple[bool, str]:
"""mountvol <盘符> /P:卸除并弹出可移动介质。"""
code, out = _run(["mountvol", letter + "\\", "/P"], timeout=30)
if code == 0:
return True, "已通过 mountvol 弹出"
return False, out.strip().replace("\n", " ")
def eject(
self,
mount_point: Path,
attempts: int,
interval: float,
use_fallback: bool = False,
) -> tuple[bool, str]:
mount = Path(mount_point)
letter = str(mount).rstrip("\\")
check = Path(letter + "\\")
last = ""
for i in range(max(1, attempts)):
if not check.exists():
return True, "卷已不在(可能已被手动拔出)"
ok, why = self._eject_via_shell(letter)
if ok or not check.exists():
return True, why or "已弹出"
ok2, why2 = self._eject_via_mountvol(letter)
if ok2 or not check.exists():
return True, why2 or "已弹出"
last = why or why2
if i < attempts - 1:
time.sleep(max(1.0, interval))
return False, last or "未知错误"
# ------------------------------------------------------------------
# 后台服务(任务计划 + 常驻进程)
# ------------------------------------------------------------------
def daemon_argv(self, project_root: Path) -> list[str]:
"""常驻进程命令。优先 pythonw.exe,避免弹出黑色控制台窗口。"""
exe = Path(sys.executable)
pyw = exe.with_name("pythonw.exe")
runner = pyw if pyw.exists() else exe
return [str(runner), "-m", "sdimport.cli", "daemon", "--interval", "30"]
def _write_launcher(self, project_root: Path) -> tuple[Path, str | None]:
"""写一个启动器 .cmd。
schtasks 无法直接设工作目录与环境变量,所以中间垫一层批处理。
"""
project_root = Path(project_root)
launcher = project_root / "run-daemon.cmd"
exe = Path(sys.executable)
pyw = exe.with_name("pythonw.exe")
interp = pyw if pyw.exists() else exe
content = (
"@echo off\r\n"
"rem 由 sdimport install 生成:启动常驻进程(插卡自动导入)\r\n"
f'cd /d "{project_root}"\r\n'
f'set "PYTHONPATH={project_root / "src"}"\r\n'
f'"{interp}" -m sdimport.cli daemon --interval 30\r\n'
)
try:
launcher.write_text(content, encoding="utf-8")
except OSError as exc:
return launcher, str(exc)
return launcher, None
def install_service(self, project_root: Path) -> ServiceStatus:
launcher, err = self._write_launcher(project_root)
if err:
return ServiceStatus(False, f"写入启动器失败:{err}", ok=False)
code, out = _run(
[
"schtasks",
"/Create",
"/TN",
TASK_NAME,
"/TR",
f'"{launcher}"',
"/SC",
"ONLOGON",
"/RL",
"LIMITED",
"/F",
],
timeout=60,
)
if code != 0:
return ServiceStatus(
False,
"任务计划创建失败。\n"
f" 启动器: {launcher}\n"
f" 报错 : {out.strip() or '(无输出)'}\n"
" 可手动兜底:任务计划程序 → 创建任务 → 触发器「登录时」"
f"→ 操作「启动程序」指向 {launcher}",
ok=False,
)
_run(["schtasks", "/Run", "/TN", TASK_NAME], timeout=30)
return ServiceStatus(
True,
"后台服务已安装(任务计划)。\n"
f" 任务名: {TASK_NAME}\n"
f" 启动器: {launcher}\n"
" 触发 : 登录时自动启动;本次已立即拉起\n"
" 停用 : sdimport uninstall",
)
def uninstall_service(self, project_root: Path) -> ServiceStatus:
code, out = _run(["schtasks", "/Delete", "/TN", TASK_NAME, "/F"], timeout=60)
self._kill_daemons()
if code != 0:
low = out.lower()
if "cannot find" in low or "找不到" in out or "不存在" in out:
return ServiceStatus(False, "后台服务未安装,无需卸载")
return ServiceStatus(False, f"卸载失败:{out.strip()}", ok=False)
return ServiceStatus(False, "后台服务已卸载")
@staticmethod
def _kill_daemons() -> None:
"""结束可能还在跑的常驻进程(按命令行精确匹配,不误伤别的 Python 程序)。"""
script = (
"Get-CimInstance Win32_Process -Filter \"Name like 'python%'\" | "
"Where-Object { $_.CommandLine -like '*sdimport.cli daemon*' } | "
"ForEach-Object { Stop-Process -Id $_.ProcessId -Force }"
)
_run_powershell(script, timeout=30)
def service_status(self) -> ServiceStatus:
code, _ = _run(["schtasks", "/Query", "/TN", TASK_NAME], timeout=30)
if code == 0:
running = self._daemon_running()
return ServiceStatus(
True, "已安装(任务计划)" + (",常驻进程在跑" if running else ",常驻进程未在跑")
)
return ServiceStatus(False, "未安装")
@staticmethod
def _daemon_running() -> bool:
script = (
"(Get-CimInstance Win32_Process -Filter \"Name like 'python%'\" | "
"Where-Object { $_.CommandLine -like '*sdimport.cli daemon*' } | "
"Measure-Object).Count"
)
code, out = _run_powershell(script, timeout=30)
if code != 0:
return False
try:
return int((out.strip().splitlines() or ["0"])[-1]) > 0
except ValueError:
return False
def diagnose(self) -> list[tuple[str, bool, str]]:
out: list[tuple[str, bool, str]] = []
code, _ = _run_powershell("exit 0", timeout=25)
out.append(("powershell", code == 0, "" if code == 0 else "通知将降级为消息框"))
drives = _logical_drives()
out.append(
("盘符枚举", bool(drives), ", ".join(d.rstrip("\\") for d in drives) or "未发现任何盘符")
)
try:
import tkinter # noqa: F401
out.append(("tkinter", True, "确认框可用"))
except ImportError as exc:
out.append(("tkinter", False, f"确认框不可用:{exc}"))
has_schtasks = shutil.which("schtasks") is not None
out.append(("schtasks", has_schtasks, "" if has_schtasks else "无法安装后台服务"))
has_wmi = True
try:
import wmi # type: ignore[import-untyped] # noqa: F401
except ImportError:
has_wmi = False
out.append(
(
"wmi 包",
True, # 缺了也能靠轮询工作,故不算异常
"已装,插卡秒级响应" if has_wmi else "未装,靠 30 秒轮询兜底(pip install wmi 可提速)",
)
)
return out
+89
View File
@@ -0,0 +1,89 @@
"""auto 流程:一次完整的「找卡 → 判定是否新插入 → 导入 → 记历史」。
macOS 由 launchd 触发(`sdimport auto`),Windows 由常驻进程触发(`sdimport daemon`),
两条路都走这里,保证两端行为一致。
"""
from __future__ import annotations
from datetime import datetime
from . import pipeline
from .journal import (
Journal,
SingleInstanceLock,
load_state,
note_run,
note_volume_seen,
save_state,
)
from .paths import LOCK_PATH, ensure_dirs, load_config
from .volumes import list_candidate_volumes, volume_names
def auto_cycle(
journal: Journal | None = None,
assume_yes: bool = False,
lock_stale: float = 3600.0,
) -> int:
"""跑一轮。
返回 0 = 正常(含「无卡」与「已有实例在跑」等跳过情形);1 = 本轮导入被中止。
"""
cfg = load_config()
ensure_dirs()
journal = journal or Journal("auto")
# 单实例锁:launchd 的 WatchPaths 与 StartInterval 可能同时触发,
# Windows 侧的事件与轮询也可能撞上,靠它保证同一时刻只有一条导入链路。
lock = SingleInstanceLock(LOCK_PATH, stale_after=lock_stale)
if not lock.acquire():
journal.log("已有导入实例在运行,本次触发忽略")
return 0
try:
state = load_state()
snap = volume_names(cfg)
prev = state.get("last_snapshot", [])
new_insert = snap != prev
if not snap:
journal.log("当前无候选卷,退出")
state["last_snapshot"] = snap
save_state(state)
return 0
journal.log(f"卷快照:{snap}(上次 {prev},{'新插入' if new_insert else '无变化'})")
for vol in list_candidate_volumes(cfg):
note_volume_seen(state, vol.identity, vol.name)
state["last_snapshot"] = snap
save_state(state)
# 不是新插入的卡就别反复播报「无新增」,免得烦人
opts = pipeline.RunOptions(quiet=not new_insert, assume_yes=assume_yes)
result = pipeline.run(cfg, journal, opts, lock=lock)
state = load_state()
note_run(
state,
{
"at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"trigger": journal.tag,
"new_insert": new_insert,
"cards": result.cards,
"imported": result.imported,
"deleted": result.deleted,
"failed": result.failed,
"bytes": result.bytes_moved,
"seconds": round(result.seconds, 1),
"aborted": result.aborted,
"abort_reason": result.abort_reason,
"errors": result.errors[:10],
},
)
state["last_snapshot"] = volume_names(cfg)
save_state(state)
return 0 if not result.aborted else 1
finally:
lock.release()
+143
View File
@@ -0,0 +1,143 @@
"""卷内录像扫描(纯跨平台逻辑,不调用任何系统专有 API)。
这里承载项目最要紧的一条规则,改动前请先读 .workbuddy/memory/MEMORY.md 的铁律。
规则(2026-09-15 事故后重定):只在**显式配置的扫描根**下找录像,下潜层数受限,
**绝不靠「本层有没有录像」来推断边界**。
推翻上一版「本层有录像就停止下潜」的原因 —— 那条规则会自毁:
第一次导入把 VIDEO/ 本层清空后,第二次扫描发现本层没录像,就继续下潜进了
归档子目录,把用户明确说过「不碰」的 409 个 / 33.5GB 历史归档全导走、
还把卡内副本清空了。2026-09-15 18:45 真实发生。
现在 VIDEO 的层数写死 0 层:只认 `VIDEO/*.AVI`,永不进 `VIDEO/<归档目录>/` 这类归档子目录。
"""
from __future__ import annotations
import os
import re
from pathlib import Path
from .platforms import Volume
def parse_scan_root(spec: str, default_depth: int) -> tuple[str, int]:
"""解析扫描根配置。
"VIDEO" -> ("VIDEO", default_depth)
"DCIM:2" -> ("DCIM", 2)
"" -> ("", default_depth) 卷根本身
"""
text = str(spec).strip().strip("/")
if ":" in text:
path, _, raw = text.partition(":")
try:
return path.strip("/"), max(0, int(raw))
except ValueError:
return path.strip("/"), default_depth
return text, default_depth
def _count_nested_videos(
dirs: list[Path], pattern: re.Pattern, min_bytes: int, max_depth: int
) -> int:
"""只统计更深处的录像数量,用于告知用户「卡上还有多少没被纳入」。"""
total = 0
stack: list[tuple[Path, int]] = [(d, 0) for d in dirs]
while stack:
cur, depth = stack.pop()
try:
with os.scandir(cur) as it:
for entry in it:
if entry.name.startswith(".") or entry.name.startswith("$"):
continue
try:
if entry.is_dir(follow_symlinks=False):
if depth < max_depth:
stack.append((Path(entry.path), depth + 1))
elif entry.is_file(follow_symlinks=False):
if pattern.match(entry.name) and entry.stat().st_size >= min_bytes:
total += 1
except OSError:
continue
except (PermissionError, OSError):
continue
return total
def find_videos(vol: Volume, cfg: dict, journal=None) -> list[Path]:
"""收集本次要导入的录像。"""
flags = re.IGNORECASE if cfg.get("case_insensitive", True) else 0
try:
pattern = re.compile(str(cfg["filename_regex"]), flags)
except re.error as exc:
if journal:
journal.error(f"filename_regex 非法:{exc}")
return []
min_bytes = int(cfg.get("min_video_bytes", 0))
default_depth = int(cfg.get("scan_max_depth", 0))
deny_dirs = {str(x).lower() for x in cfg.get("deny_dir_names", [])}
report = bool(cfg.get("report_skipped_archives", True))
found: list[Path] = []
seen: set[Path] = set()
for spec in cfg.get("scan_roots") or [""]:
rel, depth_limit = parse_scan_root(spec, default_depth)
base = (Path(vol.mount_point) / rel) if rel else Path(vol.mount_point)
if not base.is_dir():
continue
subdirs_at_base: list[Path] = []
stack: list[tuple[Path, int]] = [(base, 0)]
while stack:
cur, depth = stack.pop()
try:
with os.scandir(cur) as it:
for entry in it:
if entry.name.startswith(".") or entry.name.startswith("$"):
continue
try:
if entry.is_dir(follow_symlinks=False):
if entry.name.lower() in deny_dirs:
continue
if cur == base:
subdirs_at_base.append(Path(entry.path))
if depth < depth_limit:
stack.append((Path(entry.path), depth + 1))
elif entry.is_file(follow_symlinks=False):
if not pattern.match(entry.name):
continue
if entry.stat().st_size < min_bytes:
continue
path = Path(entry.path)
if path not in seen:
seen.add(path)
found.append(path)
except OSError:
continue
except (PermissionError, OSError):
continue
# 告知用户「卡上还有多少没被纳入」,避免以为程序漏了。
# 统计时排除掉本身就是扫描根的子目录,否则会重复计数
# (卷根的统计会把 VIDEO/ 本层已纳入的录像也算成「未纳入」)。
if report and subdirs_at_base and journal:
other_roots: set[Path] = set()
for other in cfg.get("scan_roots") or [""]:
other_rel, _ = parse_scan_root(other, default_depth)
if other_rel and other_rel != rel:
other_roots.add(Path(vol.mount_point) / other_rel)
candidates = [d for d in subdirs_at_base if d not in other_roots]
nested = (
_count_nested_videos(candidates, pattern, min_bytes, 3) if candidates else 0
)
if nested:
journal.log(
f" {base}/ 的子目录中另有 {nested} 个录像,"
f"按配置(不进子目录)不纳入本次导入"
)
return sorted(found, key=lambda p: (p.parent.as_posix(), p.name))
+137
View File
@@ -0,0 +1,137 @@
"""传输层:拷贝 -> 校验 -> 删源。
安全设计:
* 先写 .part 临时文件,校验通过后才原子改名,绝不出现半截文件冒充成品。
* 校验 = 体积一致 + 头尾采样指纹一致(比单纯比体积可靠,成本远低于全量哈希)。
* 删源前二次确认目标文件存在且体积一致;只删除本次成功导入的文件,不动卡上任何其它内容。
* 单个文件失败即中止整批,不删任何源文件。
"""
from __future__ import annotations
import hashlib
import os
import time
from dataclasses import dataclass
from pathlib import Path
from .journal import human_size
@dataclass
class CopyResult:
ok: bool
src: Path
dst: Path
size: int
seconds: float
error: str = ""
@property
def speed(self) -> float:
return self.size / self.seconds if self.seconds > 0 else 0.0
def _sample_digest(path: Path, sample_bytes: int) -> tuple[int, bytes, bytes] | None:
"""返回 (体积, 头部样本摘要, 尾部样本摘要);失败返回 None。"""
try:
size = path.stat().st_size
with path.open("rb") as fh:
head = fh.read(sample_bytes)
if size > sample_bytes:
fh.seek(max(0, size - sample_bytes))
tail = fh.read(sample_bytes)
else:
tail = head
except OSError:
return None
return size, hashlib.md5(head).digest(), hashlib.md5(tail).digest()
def copy_file(src: Path, dst: Path, chunk_bytes: int, progress=None) -> CopyResult:
t0 = time.time()
part = dst.with_name(dst.name + ".part")
written = 0
try:
st = src.stat()
total = st.st_size
dst.parent.mkdir(parents=True, exist_ok=True)
with src.open("rb") as fin, part.open("wb") as fout:
while True:
buf = fin.read(chunk_bytes)
if not buf:
break
fout.write(buf)
written += len(buf)
if progress:
progress(written, total)
fout.flush()
os.fsync(fout.fileno())
os.utime(part, (st.st_atime, st.st_mtime))
os.replace(part, dst)
return CopyResult(True, src, dst, written, time.time() - t0)
except OSError as exc:
try:
part.unlink(missing_ok=True)
except OSError:
pass
return CopyResult(False, src, dst, written, time.time() - t0, str(exc))
def verify(src: Path, dst: Path, sample_kb: int = 64) -> tuple[bool, str]:
sample = max(4096, sample_kb * 1024)
a = _sample_digest(src, sample)
b = _sample_digest(dst, sample)
if a is None or b is None:
return False, "无法读取比对样本"
if a[0] != b[0]:
return False, f"体积不符(源 {human_size(a[0])} / 目标 {human_size(b[0])})"
if a[1] != b[1]:
return False, "头部指纹不符"
if a[2] != b[2]:
return False, "尾部指纹不符"
return True, "校验通过"
def delete_source(src: Path, dst: Path, sample_kb: int = 64) -> tuple[bool, str]:
"""删除卡内源文件;删前再次做完整校验(体积 + 头尾指纹)。
只有确认目标文件与源文件内容一致才删除,杜绝「目标不全却清卡」。
"""
try:
if not dst.is_file():
return False, "目标文件不存在,拒绝删除源文件"
if not src.is_file():
return False, "源文件已不存在"
except OSError as exc:
return False, f"复核失败:{exc}"
ok, why = verify(src, dst, sample_kb)
if not ok:
return False, f"删前校验未通过({why}),拒绝删除源文件"
try:
src.unlink()
except OSError as exc:
return False, f"删除失败:{exc}"
if src.exists():
return False, "删除后文件仍存在"
return True, "已删除"
def cleanup_partials(plan_items) -> int:
"""清理历史遗留的 .part 残留。"""
removed = 0
seen: set[Path] = set()
for item in plan_items:
part = item.dst.with_name(item.dst.name + ".part")
if part in seen:
continue
seen.add(part)
try:
if part.exists():
part.unlink()
removed += 1
except OSError:
pass
return removed
+56
View File
@@ -0,0 +1,56 @@
"""卷相关门面。
职责划分:
* 跨平台的**扫描规则** -> scanner.py
* 平台相关的**卷操作** -> platforms.current(枚举、探测、可写、稳定等待)
保留这一层,是为了让 pipeline / cli 等调用方不必关心平台差异。
"""
from __future__ import annotations
from pathlib import Path
from .platforms import Volume
from .platforms import current as _platform
from .scanner import find_videos
__all__ = [
"Volume",
"find_videos",
"probe",
"list_candidate_volumes",
"volume_names",
"check_writable",
"wait_until_stable",
"default_mount_root",
]
def probe(mount_point: Path) -> Volume | None:
"""读取卷信息;不是有效挂载点则返回 None。"""
return _platform.probe_volume(Path(mount_point))
def list_candidate_volumes(cfg: dict) -> list[Volume]:
"""列出值得检查的候选卷(已排除系统卷与网络卷)。"""
return _platform.list_volumes(cfg)
def volume_names(cfg: dict) -> list[str]:
"""轻量的卷名快照,用于判断是否新插入了卡。"""
return _platform.volume_names(cfg)
def check_writable(mount_point: Path) -> bool:
"""真实写测试,避免只读标记误判。"""
return _platform.check_writable(Path(mount_point))
def wait_until_stable(mount_point: Path, checks: int = 3, interval: float = 0.8) -> bool:
"""等卷挂载稳定(判据是卷总容量,理由见 platforms/base.py)。"""
return _platform.wait_until_stable(Path(mount_point), checks=checks, interval=interval)
def default_mount_root() -> Path | None:
return _platform.default_mount_root()
+96
View File
@@ -0,0 +1,96 @@
#!/usr/bin/env python3
"""造一张假的记录仪 SD 卡,用于在无实体卡的情况下验证全流程。
python3 tests/make_fake_card.py <目标目录> [--mb 8] [--clean]
结构按一张真实记录仪卡复刻(卷名与楼栋名均已隐去,仅保留目录层级语义):
<root>/VIDEO/VID-20260912-090430.AVI 新录:平铺在 VIDEO 下
<root>/VIDEO/VID-20260914-070000.AVI 新录
<root>/VIDEO/archive-01/VID-20250731-103204.AVI 早年的归档 —— 一个都不许碰
<root>/VIDEO/project-2025/unit-east/... 更深的归档 —— 同样不许碰
<root>/PHOTO/IMG_0001.JPG 干扰项
<root>/LOG.TXT <root>/VIDEO.LST 记录仪自己的文件
要点:VIDEO/ 这一层既有新录像、又有归档子目录,
正是用来验证「收下本层录像即停止下潜」这条关键规则。
"""
from __future__ import annotations
import argparse
import random
import shutil
import sys
from pathlib import Path
NEW_VIDEOS = [
("VIDEO", "VID-20260912-090430.AVI"),
("VIDEO", "VID-20260912-093634.AVI"),
("VIDEO", "VID-20260913-080000.AVI"),
("VIDEO", "VID-20260913-083000.AVI"),
("VIDEO", "VID-20260913-090000.AVI"),
("VIDEO", "VID-20260914-070000.AVI"),
]
ARCHIVE_VIDEOS = [
("VIDEO/archive-01", "VID-20250731-103204.AVI"),
("VIDEO/archive-01", "VID-20250731-103312.AVI"),
("VIDEO/archive-02", "VID-20260618-111306.AVI"),
("VIDEO/project-2025/unit-east", "VID-20250627-135316.AVI"),
]
DECOYS = [
("PHOTO", "IMG_0001.JPG"),
("VIDEO", "VID-20260912-090430.AVI.thumb"),
("", "LOG.TXT"),
("", "VIDEO.LST"),
("", "Param.ini"),
]
def make(path: Path, size_mb: int) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("wb") as fh:
for _ in range(size_mb):
fh.write(random.randbytes(1024 * 1024))
def main(argv: list[str] | None = None) -> int:
ap = argparse.ArgumentParser(description="造一张假的记录仪 SD 卡")
ap.add_argument("root", help="假卡根目录")
ap.add_argument("--mb", type=int, default=8, help="每个视频的体积(MB)")
ap.add_argument("--clean", action="store_true", help="先清空目标目录")
args = ap.parse_args(argv)
root = Path(args.root).expanduser().resolve()
if args.clean and root.exists():
shutil.rmtree(root)
root.mkdir(parents=True, exist_ok=True)
print(f"生成假卡:{root}")
print(" [新录] 应当被导入:")
for sub, name in NEW_VIDEOS:
make(root / sub / name, args.mb)
print(f" + {sub}/{name}")
print(" [归档] 一个都不许碰:")
for sub, name in ARCHIVE_VIDEOS:
make(root / sub / name, args.mb)
print(f" + {sub}/{name}")
print(" [干扰] 同样不许碰:")
for sub, name in DECOYS:
target = root / sub / name
target.parent.mkdir(parents=True, exist_ok=True)
target.write_bytes(b"decoy" * 16)
print(f" + {sub}/{name}")
total = sum(p.stat().st_size for p in root.rglob("*") if p.is_file())
print(
f"完成:新录 {len(NEW_VIDEOS)} 个 / 归档 {len(ARCHIVE_VIDEOS)} 个 / "
f"干扰 {len(DECOYS)} 个,共 {total / 1048576:.1f}MB"
)
return 0
if __name__ == "__main__":
sys.exit(main())
+10
View File
@@ -0,0 +1,10 @@
@echo off
chcp 65001 >nul
cd /d "%~dp0"
echo ================================================================
echo sdimport - uninstall background service (Windows)
echo ================================================================
echo.
call "%~dp0bin\sdimport.cmd" uninstall
echo.
pause
+35
View File
@@ -0,0 +1,35 @@
#!/bin/bash
# 双击本文件即可安装「插卡自动导入」后台服务。
# 安装后:插入记录仪 SD 卡 -> 自动识别 -> 弹倒计时确认 -> 按日期落盘 -> 清空卡内录像。
cd "$(dirname "$0")" || exit 1
clear
echo "================================================================"
echo " 检测视频 SD 卡自动导入 - 安装后台服务"
echo "================================================================"
echo
./bin/sdimport install
STATUS=$?
echo
if [ "$STATUS" -eq 0 ]; then
echo "----------------------------------------------------------------"
echo " 安装完成。现在可以拔掉这张窗口,正常使用了。"
echo
echo " 常用命令(在终端里执行):"
echo " cd \"$(pwd)\""
echo " ./bin/sdimport status 查看状态与导入历史"
echo " ./bin/sdimport scan 演练:看看会导入什么(不动文件)"
echo " ./bin/sdimport log 看最近日志"
echo " ./bin/sdimport uninstall 卸载后台服务"
echo "----------------------------------------------------------------"
else
echo "----------------------------------------------------------------"
echo " 安装未成功,请把上面的报错发给开发者。"
echo "----------------------------------------------------------------"
fi
echo
read -n 1 -s -r -p "按任意键关闭本窗口..."
echo