初始化新时空数据应用

2026-06-12 02:11:47 +08:00
commit aab9da4f31
18 changed files with 3492 additions and 0 deletions
@@ -0,0 +1,204 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+from dataclasses import dataclass
+from datetime import datetime
+import os
+from pathlib import Path
+import shlex
+import subprocess
+import sys
+import time
+
+
+DEFAULT_LOCAL_DIR = Path("/Users/yangdawei/Desktop/新时空业务源数据/新时空课程记录与课时账户")
+DEFAULT_REMOTE_HOST = "121.199.172.246"
+DEFAULT_REMOTE_USER = "root"
+DEFAULT_REMOTE_PORT = 22222
+DEFAULT_REMOTE_DIR = "/root/新时空数据"
+DEFAULT_IDENTITY_FILE = Path.home() / ".ssh" / "xsk_records_vps_ed25519"
+SYNC_FILES = ("classnotes.txt", "学生课时账户.md")
+
+
+@dataclass(frozen=True)
+class Config:
+    local_dir: Path
+    remote_host: str
+    remote_user: str
+    remote_port: int
+    remote_dir: str
+    identity_file: Path | None
+    interval: float
+    use_sshpass: bool
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="同步新时空课程记录数据到 VPS")
+    parser.add_argument("--once", action="store_true", help="只同步一次后退出")
+    parser.add_argument("--interval", type=float, default=float(os.getenv("XSK_SYNC_INTERVAL", "3")))
+    parser.add_argument("--local-dir", default=os.getenv("XSK_LOCAL_DIR", str(DEFAULT_LOCAL_DIR)))
+    parser.add_argument("--remote-host", default=os.getenv("XSK_REMOTE_HOST", DEFAULT_REMOTE_HOST))
+    parser.add_argument("--remote-user", default=os.getenv("XSK_REMOTE_USER", DEFAULT_REMOTE_USER))
+    parser.add_argument("--remote-port", type=int, default=int(os.getenv("XSK_REMOTE_PORT", str(DEFAULT_REMOTE_PORT))))
+    parser.add_argument("--remote-dir", default=os.getenv("XSK_REMOTE_DIR", DEFAULT_REMOTE_DIR))
+    parser.add_argument(
+        "--identity-file",
+        default=os.getenv("XSK_IDENTITY_FILE", str(DEFAULT_IDENTITY_FILE) if DEFAULT_IDENTITY_FILE.exists() else ""),
+        help="SSH 私钥路径；默认使用 ~/.ssh/xsk_records_vps_ed25519（如果存在）",
+    )
+    parser.add_argument(
+        "--use-sshpass",
+        action="store_true",
+        default=os.getenv("XSK_USE_SSHPASS", "") == "1",
+        help="从 XSK_SSH_PASSWORD 读取密码并通过 sshpass 连接；推荐改用 SSH key",
+    )
+    return parser.parse_args()
+
+
+def log(message: str) -> None:
+    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+    print(f"[{timestamp}] {message}", flush=True)
+
+
+def run_command(command: list[str]) -> subprocess.CompletedProcess[str]:
+    env = os.environ.copy()
+    if env.get("XSK_SSH_PASSWORD") and not env.get("SSHPASS"):
+        env["SSHPASS"] = env["XSK_SSH_PASSWORD"]
+    result = subprocess.run(command, text=True, capture_output=True, check=False, env=env)
+    if result.returncode != 0:
+        safe_command = " ".join(shlex.quote(part) for part in command if part != os.getenv("XSK_SSH_PASSWORD", ""))
+        raise RuntimeError(
+            f"命令失败({result.returncode}): {safe_command}\n"
+            f"STDOUT: {result.stdout.strip()}\nSTDERR: {result.stderr.strip()}"
+        )
+    return result
+
+
+def base_remote(config: Config) -> str:
+    return f"{config.remote_user}@{config.remote_host}"
+
+
+def ssh_options(config: Config) -> list[str]:
+    options = [
+        "-p",
+        str(config.remote_port),
+        "-o",
+        "StrictHostKeyChecking=accept-new",
+    ]
+    if config.identity_file:
+        options.extend(["-i", str(config.identity_file), "-o", "IdentitiesOnly=yes"])
+    return options
+
+
+def ssh_prefix(config: Config) -> list[str]:
+    command: list[str] = []
+    if config.use_sshpass:
+        password = os.getenv("XSK_SSH_PASSWORD")
+        if not password:
+            raise RuntimeError("启用 --use-sshpass 时必须设置 XSK_SSH_PASSWORD")
+        command.extend(["sshpass", "-e"])
+    command.extend(["ssh", *ssh_options(config), base_remote(config)])
+    return command
+
+
+def rsync_prefix(config: Config) -> list[str]:
+    command: list[str] = []
+    if config.use_sshpass:
+        password = os.getenv("XSK_SSH_PASSWORD")
+        if not password:
+            raise RuntimeError("启用 --use-sshpass 时必须设置 XSK_SSH_PASSWORD")
+        command.extend(["sshpass", "-e"])
+    ssh_command = " ".join(shlex.quote(part) for part in ["ssh", *ssh_options(config)])
+    command.extend(
+        [
+            "rsync",
+            "-az",
+            "-e",
+            ssh_command,
+        ]
+    )
+    return command
+
+
+def remote_shell_quote(value: str) -> str:
+    return shlex.quote(value)
+
+
+def ensure_remote_dirs(config: Config) -> None:
+    data_dir = f"{config.remote_dir.rstrip('/')}/data"
+    run_command(ssh_prefix(config) + [f"mkdir -p {remote_shell_quote(data_dir)}"])
+
+
+def sync_once(config: Config) -> None:
+    ensure_remote_dirs(config)
+    data_dir = f"{config.remote_dir.rstrip('/')}/data"
+    for filename in SYNC_FILES:
+        source = config.local_dir / filename
+        if not source.exists():
+            raise FileNotFoundError(f"本地文件不存在: {source}")
+        temp_name = f".{filename}.tmp"
+        remote_temp = f"{base_remote(config)}:{data_dir}/{temp_name}"
+        run_command(rsync_prefix(config) + [str(source), remote_temp])
+        run_command(
+            ssh_prefix(config)
+            + [
+                "mv "
+                f"{remote_shell_quote(data_dir + '/' + temp_name)} "
+                f"{remote_shell_quote(data_dir + '/' + filename)}"
+            ]
+        )
+        log(f"已同步 {source.name}")
+
+
+def file_signature(path: Path) -> tuple[int, int]:
+    stat = path.stat()
+    return stat.st_mtime_ns, stat.st_size
+
+
+def current_signatures(local_dir: Path) -> dict[str, tuple[int, int]]:
+    return {filename: file_signature(local_dir / filename) for filename in SYNC_FILES}
+
+
+def watch(config: Config) -> None:
+    log("启动课程记录同步监听")
+    signatures: dict[str, tuple[int, int]] = {}
+    while True:
+        try:
+            next_signatures = current_signatures(config.local_dir)
+            if next_signatures != signatures:
+                time.sleep(0.4)
+                sync_once(config)
+                signatures = current_signatures(config.local_dir)
+        except Exception as exc:
+            log(f"同步失败: {exc}")
+        time.sleep(config.interval)
+
+
+def main() -> int:
+    args = parse_args()
+    config = Config(
+        local_dir=Path(args.local_dir).expanduser(),
+        remote_host=args.remote_host,
+        remote_user=args.remote_user,
+        remote_port=args.remote_port,
+        remote_dir=args.remote_dir,
+        identity_file=Path(args.identity_file).expanduser() if args.identity_file else None,
+        interval=args.interval,
+        use_sshpass=args.use_sshpass,
+    )
+    try:
+        if args.once:
+            sync_once(config)
+        else:
+            watch(config)
+    except KeyboardInterrupt:
+        return 130
+    except Exception as exc:
+        log(f"退出: {exc}")
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())