wolfydw/xsk-education-management
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

初始化新时空数据应用

Browse Source
This commit is contained in:
Codex
2026-06-12 02:11:47 +08:00
commit a18ef4fc42
18 changed files with 3492 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/scripts/deploy_to_vps.py
Executable
+148
View File
@@ -0,0 +1,148 @@
#!/usr/bin/env python3
from __future__ import annotations
import argparse
import os
from pathlib import Path
import shlex
import subprocess
import sys
import tempfile
PROJECT_DIR = Path(__file__).resolve().parents[1]
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"
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="部署新时空课程记录查询网页到 VPS")
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("--app-port", default=os.getenv("XSK_APP_PORT", "18080"))
parser.add_argument(
"--python-image",
default=os.getenv(
"XSK_PYTHON_IMAGE",
"swr.cn-north-4.myhuaweicloud.com/ddn-k8s/docker.io/library/python:3.12-slim",
),
help="Docker 基础镜像；VPS 无法访问 Docker Hub 时可使用镜像站",
)
parser.add_argument("--auth-user", default=os.getenv("XSK_WEB_USER", "wolfydw"))
parser.add_argument("--auth-password", default=os.getenv("XSK_WEB_PASSWORD"))
parser.add_argument("--use-sshpass", action="store_true", default=os.getenv("XSK_USE_SSHPASS", "") == "1")
return parser.parse_args()
def run(command: list[str], cwd: Path | None = None) -> None:
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, cwd=cwd, text=True, capture_output=True, check=False, env=env)
if result.returncode != 0:
redacted = ["***" if part == os.getenv("XSK_SSH_PASSWORD") else part for part in command]
raise RuntimeError(
f"命令失败({result.returncode}): {' '.join(shlex.quote(part) for part in redacted)}\n"
f"STDOUT: {result.stdout.strip()}\nSTDERR: {result.stderr.strip()}"
)
if result.stdout.strip():
print(result.stdout.strip())
def base_remote(args: argparse.Namespace) -> str:
return f"{args.remote_user}@{args.remote_host}"
def ssh_options(args: argparse.Namespace) -> list[str]:
options = ["-p", str(args.remote_port), "-o", "StrictHostKeyChecking=accept-new"]
if args.identity_file:
options.extend(["-i", str(Path(args.identity_file).expanduser()), "-o", "IdentitiesOnly=yes"])
return options
def ssh_command(args: argparse.Namespace) -> list[str]:
command: list[str] = []
if args.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(args), base_remote(args)])
return command
def rsync_command(args: argparse.Namespace) -> list[str]:
command: list[str] = []
if args.use_sshpass:
password = os.getenv("XSK_SSH_PASSWORD")
if not password:
raise RuntimeError("启用 --use-sshpass 时必须设置 XSK_SSH_PASSWORD")
command.extend(["sshpass", "-e"])
ssh_command_text = " ".join(shlex.quote(part) for part in ["ssh", *ssh_options(args)])
command.extend(
[
"rsync",
"-az",
"--delete",
"--exclude",
".env",
"--exclude",
"*.log",
"-e",
ssh_command_text,
]
)
return command
def write_env(args: argparse.Namespace) -> Path:
if not args.auth_password:
raise RuntimeError("必须通过 XSK_WEB_PASSWORD 或 --auth-password 设置网页访问密码")
env_text = "\n".join(
[
f"APP_PORT={args.app_port}",
"TZ=Asia/Shanghai",
f"PYTHON_IMAGE={args.python_image}",
f"BASIC_AUTH_USERNAME={args.auth_user}",
f"BASIC_AUTH_PASSWORD={args.auth_password}",
"CLASSNOTES_PATH=/data/classnotes.txt",
"ACCOUNTS_PATH=/data/学生课时账户.md",
"",
]
)
handle = tempfile.NamedTemporaryFile("w", encoding="utf-8", delete=False)
try:
handle.write(env_text)
return Path(handle.name)
finally:
handle.close()
def main() -> int:
args = parse_args()
app_dir = f"{args.remote_dir.rstrip('/')}/app"
data_dir = f"{args.remote_dir.rstrip('/')}/data"
env_file = write_env(args)
try:
run(ssh_command(args) + [f"mkdir -p {shlex.quote(app_dir)} {shlex.quote(data_dir)}"])
run(rsync_command(args) + [f"{PROJECT_DIR}/", f"{base_remote(args)}:{app_dir}/"])
run(rsync_command(args) + [str(env_file), f"{base_remote(args)}:{app_dir}/.env"])
run(ssh_command(args) + [f"cd {shlex.quote(app_dir)} && docker compose up -d --build"])
print(f"部署完成: http://{args.remote_host}:{args.app_port}/")
finally:
env_file.unlink(missing_ok=True)
return 0
if __name__ == "__main__":
raise SystemExit(main())
app/scripts/install_launch_agent.py
Executable
+60
View File
@@ -0,0 +1,60 @@
#!/usr/bin/env python3
from __future__ import annotations
import argparse
import os
from pathlib import Path
import plistlib
import subprocess
PROJECT_DIR = Path(__file__).resolve().parents[1]
LABEL = "com.xsk.records.sync"
PLIST_PATH = Path.home() / "Library" / "LaunchAgents" / f"{LABEL}.plist"
LOG_DIR = Path.home() / "Library" / "Logs" / "xsk-records-web"
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="安装新时空课程记录同步 LaunchAgent")
parser.add_argument("--use-sshpass", action="store_true", default=os.getenv("XSK_USE_SSHPASS", "") == "1")
parser.add_argument("--ssh-password", default=os.getenv("XSK_SSH_PASSWORD"))
return parser.parse_args()
def main() -> int:
args = parse_args()
LOG_DIR.mkdir(parents=True, exist_ok=True)
PLIST_PATH.parent.mkdir(parents=True, exist_ok=True)
program_arguments = [
"/usr/bin/python3",
str(PROJECT_DIR / "scripts" / "sync_to_vps.py"),
]
if args.use_sshpass:
program_arguments.append("--use-sshpass")
environment = {
"PATH": "/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin:/usr/sbin:/sbin",
}
if args.use_sshpass and args.ssh_password:
environment["XSK_SSH_PASSWORD"] = args.ssh_password
plist = {
"Label": LABEL,
"ProgramArguments": program_arguments,
"RunAtLoad": True,
"KeepAlive": True,
"StandardOutPath": str(LOG_DIR / "sync.log"),
"StandardErrorPath": str(LOG_DIR / "sync.err.log"),
"EnvironmentVariables": environment,
}
with PLIST_PATH.open("wb") as handle:
plistlib.dump(plist, handle)
subprocess.run(["launchctl", "unload", str(PLIST_PATH)], check=False, capture_output=True)
subprocess.run(["launchctl", "load", str(PLIST_PATH)], check=True)
print(f"已安装并启动 LaunchAgent: {PLIST_PATH}")
return 0
if __name__ == "__main__":
raise SystemExit(main())
app/scripts/sync_to_vps.py
Executable
+204
View File
@@ -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())