Files
lexgo/scripts/ops.py
T
ila 0fd661e061 feat: 音频与插图改为章节级,封面维持书级 (#37)
- schema v9:lexgo_chapter_attachments(chapter_id+kind 唯一,MEDIUMBLOB)与
  lexgo_chapter_playback_positions(owner_id+chapter_id),并用幂等语句清空书级音频历史行
- 书级音频与播放位置接口下线(路由不再注册);封面仍是书级且行为不变
- 章节级接口:POST/DELETE/GET /chapters/:id/{audio,illustration}、PUT /chapters/:id/playback;
  章节列表与阅读器响应带 illustrationVersion/audioVersion/playbackSeconds(始终存在)
- 复用 #21 的类型嗅探、尺寸校验、ServeContent Range/ETag 包装与“先校验后写入”流程
- 学习端:封面区块收窄、章节行新增缩略图与「附件」入口、独立的章节附件对话框、
  阅读页正文上方插图与章级播放器(audioChapterId 防止复用上一章音频)、切换章节上报位置
- 测试:Go 87 项(含 v8→v9 迁移与回退、书级接口已下线、章级附件集成)、学习端 157 单测与
  26 项 E2E、真实 API 52 项、#15 恢复演练第三次 22 项
- 文档:Architecture / Business-Rules / Local-Development / Requirements / Home /
  Deployment 改为章级口径
2026-09-15 21:08:43 +08:00

792 lines
41 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""LexGo operations: dependency check, backup, restore and post-restore verification.
Everything a self-hosted instance needs to be installed, backed up and restored is in MySQL:
accounts, books and chapter text, personal terms, review schedules and answers, the imported
WordNet archive and the audit logs. This script therefore never invents a second storage
location: a dump plus the environment file is a complete backup.
Safety rules, because a restore can overwrite a working instance:
* a restore always needs ``--confirm``;
* the target database name must contain ``lexgo`` and must not be a MySQL system schema;
* a database that already holds LexGo data is refused unless ``--force`` is given;
* credentials are read from the environment or ``.env.local`` and are never written into a
manifest, a log line or a dump name.
Only the standard library is used, so the script runs on the deployment host without installing
anything beyond the MySQL client.
"""
from __future__ import annotations
import argparse
import gzip
import hashlib
import json
import os
import re
import shutil
import subprocess
import tempfile
import time
import sys
from datetime import datetime, timezone
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
ENV_FILE = ROOT / ".env.local"
SYSTEM_SCHEMAS = {"mysql", "information_schema", "performance_schema", "sys"}
# Columns that must never appear in the audit tables: they exist to prove that no credential,
# request body or private learning content is stored there.
AUDIT_BANNED_COLUMNS = {"password", "token", "body", "content", "definition", "text"}
TABLES = [
"sys_user", "lexgo_spaces", "lexgo_sessions", "lexgo_login_logs", "lexgo_operation_logs",
"lexgo_books", "lexgo_chapters", "lexgo_ingest_jobs", "lexgo_dictionaries",
"lexgo_terms", "lexgo_term_reviews", "lexgo_review_answers", "lexgo_chapter_progress",
"lexgo_book_attachments", "lexgo_playback_positions",
"lexgo_chapter_attachments", "lexgo_chapter_playback_positions",
]
SCHEMA_VERSION = 7
class OpsError(SystemExit):
"""A failure the operator has to read and act on."""
def load_env(path=ENV_FILE):
"""Reads .env.local. Values are never printed; only key names are."""
values = {}
if path.exists():
for line in path.read_text(encoding="utf-8").splitlines():
line = line.strip()
if not line or line.startswith("#") or "=" not in line:
continue
key, value = line.split("=", 1)
values[key.strip()] = value.strip()
for key, value in values.items():
os.environ.setdefault(key, value)
return values
def mysql_binary(name):
"""Finds mysqldump/mysql via LEXGO_MYSQL_BIN, then PATH."""
configured = os.environ.get("LEXGO_MYSQL_BIN")
if configured:
candidate = Path(configured) / (name + (".exe" if os.name == "nt" else ""))
if candidate.exists():
return str(candidate)
found = shutil.which(name)
if not found:
raise OpsError(
"找不到 " + name + ":请安装 MySQL 客户端,或用 LEXGO_MYSQL_BIN 指向客户端目录。"
)
return found
def connection_args(database=None):
"""Builds client arguments. The password travels through the environment, not argv."""
args = [
"--host=" + os.environ.get("LEXGO_DB_HOST", "127.0.0.1"),
"--port=" + os.environ.get("LEXGO_DB_PORT", "3306"),
"--user=" + os.environ.get("LEXGO_DB_USER", ""),
]
if database:
args.append(database)
return args
def client_env():
env = dict(os.environ)
password = os.environ.get("LEXGO_DB_PASSWORD")
if password:
env["MYSQL_PWD"] = password
return env
def run(command, env=None, capture=True):
result = subprocess.run(command, env=env, capture_output=capture, text=True, encoding="utf-8", errors="replace")
if result.returncode != 0:
message = (result.stderr or "").strip().splitlines()
# The last line of a client error is the one an operator needs; never echo arguments,
# which is why the command line itself is not part of the message.
raise OpsError("命令失败:" + (message[-1] if message else "未知错误"))
return result.stdout or ""
def query(database, sql):
# SQL travels through stdin: a long INSERT would exceed the Windows command line limit.
result = subprocess.run([mysql_binary("mysql"), *connection_args(database), "--batch", "--skip-column-names"],
input=sql.encode("utf-8"), env=client_env(), capture_output=True)
if result.returncode != 0:
lines = result.stderr.decode("utf-8", "replace").strip().splitlines()
raise OpsError("查询失败:" + (lines[-1] if lines else "未知错误"))
output = result.stdout.decode("utf-8", "replace")
return [line.split("\t") for line in output.strip().splitlines() if line]
def scalar(database, sql, default=None):
rows = query(database, sql)
if not rows or not rows[0] or rows[0][0] == "NULL":
return default
return rows[0][0]
def database_exists(database):
rows = query("", "SELECT SCHEMA_NAME FROM information_schema.SCHEMATA WHERE SCHEMA_NAME='%s'" % database)
return bool(rows)
def sha256_of(path, chunk=1024 * 1024):
digest = hashlib.sha256()
with open(path, "rb") as handle:
while True:
block = handle.read(chunk)
if not block:
break
digest.update(block)
return digest.hexdigest()
def git_commit():
try:
return subprocess.run(["git", "rev-parse", "HEAD"], cwd=str(ROOT), capture_output=True, text=True,
encoding="utf-8", errors="replace").stdout.strip()
except OSError:
return ""
def version_of(text):
"""Pulls the version out of a client or server banner.
A client banner reads "Ver 14.14 Distrib 5.7.38", where the first number is the protocol
version, so the version after "Distrib" is the one to compare.
"""
banner = text or ""
match = re.search(r"Distrib\s+(\d+)\.(\d+)(?:\.(\d+))?", banner)
if not match:
match = re.search(r"(\d+)\.(\d+)(?:\.(\d+))?", banner)
if not match:
return None
return tuple(int(part) for part in match.groups(default="0"))
def validate_database_name(name, action):
if not name or name in SYSTEM_SCHEMAS or "lexgo" not in name:
raise OpsError("拒绝在 %s 上执行:库名必须包含 lexgo 且不能是系统库。" % action)
return name
def row_counts(database):
counts = {}
present = {row[0] for row in query("information_schema", "SELECT TABLE_NAME FROM TABLES WHERE TABLE_SCHEMA='%s'" % database)}
for table in TABLES:
if table in present:
counts[table] = int(scalar(database, "SELECT COUNT(*) FROM `%s`" % table, "0"))
return counts
# ---------------------------------------------------------------- install-check
def command_install_check(_args):
load_env()
checks = []
problems = []
def add(name, ok, detail):
checks.append((name, ok, detail))
if not ok:
problems.append(name)
# The client has to be at least as new as the server: an older mysqldump produces a dump the
# server cannot be restored from, which is worth catching before an incident rather than after.
try:
client = run([mysql_binary("mysql"), "--version"]).strip()
except OpsError as error:
client = str(error)
client_version = version_of(client)
add("MySQL 客户端存在", bool(client_version), client)
server = ""
database = os.environ.get("LEXGO_DB_NAME", "")
if database:
try:
server = scalar("", "SELECT VERSION()", "") or ""
except OpsError as error:
problems.append("数据库连接")
checks.append(("数据库连接", False, str(error)))
server_version = version_of(server)
if server_version:
add("MySQL 服务端为 8.x", server_version[0] == 8, server)
if client_version and server_version:
add("客户端版本不低于服务端", client_version >= server_version,
"客户端 " + ".".join(str(part) for part in client_version) + " / 服务端 " + ".".join(str(part) for part in server_version))
add("环境配置已加载", bool(database), "LEXGO_DB_NAME=" + (database or "(未设置)"))
for key in ("LEXGO_DB_USER", "LEXGO_DB_PASSWORD", "LEXGO_BOOTSTRAP_USERNAME"):
add("配置项 " + key + " 存在", bool(os.environ.get(key)), "已设置" if os.environ.get(key) else "未设置")
if not os.environ.get("LEXGO_LISTEN"):
print(" [info] LEXGO_LISTEN 未设置,服务将使用内置默认监听地址。")
# The dictionary resource is pinned by digest and imported explicitly, never downloaded at
# runtime; the pin file has to be present for an install to be reproducible.
resource = ROOT / "server" / "wordnet-resource.json"
if resource.exists():
pin = json.loads(resource.read_text(encoding="utf-8"))
add("WordNet 资源 pin", len(pin.get("sha256", "")) == 64 and bool(pin.get("source")), "sha256=" + pin.get("sha256", "")[:12] + "…")
else:
add("WordNet 资源 pin", False, "缺少 server/wordnet-resource.json")
for binary, label in (("go", "Go 工具链"), ("node", "Node"), ("pnpm", "pnpm")):
found = shutil.which(binary)
add(label + "(仅构建需要)", bool(found), found or "未安装:使用预构建产物时可不安装")
if os.environ.get("LEXGO_LISTEN", "").startswith("0.0.0.0"):
add("服务监听地址", False, "LEXGO_LISTEN 对外监听:" + os.environ["LEXGO_LISTEN"] + ",应由反向代理转发而不是直接暴露")
for name, ok, detail in checks:
print((" [ok] " if ok else " [fail] ") + name + " — " + detail)
if problems:
print("\n检查未通过:" + "、".join(problems))
print("按 wiki 的 Deployment-and-Operations 页面补齐后再安装。")
raise SystemExit(1)
print("\n依赖检查通过。")
if database:
try:
version = scalar(database, "SELECT version FROM lexgo_schema WHERE id=1", None)
print("当前 schema 版本:" + (version or "未知(尚未迁移)"))
except OpsError:
print("当前 schema 版本:无法读取(数据库可能尚未初始化)")
return 0
# ---------------------------------------------------------------- init-database
def command_init_database(args):
"""Creates the empty schema an installation starts from.
The server cannot connect to a database that does not exist yet, so this is the first step of
a fresh install. Grants are printed instead of applied: which account may reach the schema is
a decision for the operator's database policy, not for this tool.
"""
load_env()
database = validate_database_name(args.database or os.environ.get("LEXGO_DB_NAME", ""), "建库")
if database_exists(database):
tables = int(scalar(database, "SELECT COUNT(*) FROM information_schema.TABLES WHERE TABLE_SCHEMA='%s'" % database, "0") or 0)
print("库已存在:" + database + "(" + str(tables) + " 张表)")
if tables:
print("这是已有实例;需要升级时下一步是 python scripts/server.py migrate。")
return 0
else:
run([mysql_binary("mysql"), *connection_args(), "-e", "CREATE DATABASE `%s` CHARACTER SET utf8mb4" % database], env=client_env())
print("已创建库:" + database)
print(
"\n请为应用账号授予该库的最小权限(只限此库,不要使用管理员账号):\n"
" GRANT SELECT, INSERT, UPDATE, DELETE, CREATE, ALTER, INDEX, DROP, REFERENCES\n"
" ON `" + database + "`.* TO '<应用账号>'@'<来源主机>';\n"
" FLUSH PRIVILEGES;\n"
"\n下一步:写入 .env.local(凭据取自运维密码库)后执行\n"
" python scripts/server.py migrate\n"
" python scripts/server.py bootstrap"
)
return 0
# ---------------------------------------------------------------- backup
def command_backup(args):
load_env()
database = validate_database_name(args.database or os.environ.get("LEXGO_DB_NAME", ""), "备份")
out_dir = Path(args.out) if args.out else ROOT / ".local" / "backups" / datetime.now().strftime("%Y%m%d-%H%M%S")
if out_dir.exists() and any(out_dir.iterdir()) and not args.force:
raise OpsError("输出目录非空:" + str(out_dir) + "(加 --force 覆盖)")
out_dir.mkdir(parents=True, exist_ok=True)
stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
sql_path = out_dir / ("lexgo-" + stamp + ".sql")
gz_path = Path(str(sql_path) + ".gz")
command = [
mysql_binary("mysqldump"), *connection_args(),
"--single-transaction", "--routines", "--triggers", "--hex-blob", "--no-tablespaces",
"--default-character-set=utf8mb4",
# Deliberately without --databases: that option writes CREATE DATABASE and USE into the
# dump, which would send a restore straight back into the source schema instead of the
# target one. The client is given the target database instead.
database,
]
print("正在备份 " + database + " …")
# The dump is captured and written as bytes: nothing about it is decoded or re-encoded on the
# way to disk, which keeps a binary-safe dump byte-exact.
dump = subprocess.run(command, env=client_env(), capture_output=True)
if dump.returncode != 0:
message = dump.stderr.decode("utf-8", "replace").strip().splitlines()
raise OpsError("mysqldump 失败:" + (message[-1] if message else "未知错误"))
with open(sql_path, "wb") as handle:
handle.write(dump.stdout)
with open(sql_path, "rb") as source, gzip.open(gz_path, "wb", compresslevel=6) as target:
shutil.copyfileobj(source, target)
sql_path.unlink()
manifest = {
"created_at": stamp,
"database": database,
"product": "lexgo",
"schema_version": int(scalar(database, "SELECT version FROM lexgo_schema WHERE id=1", "0") or 0),
"git_commit": git_commit(),
"dump_file": gz_path.name,
"dump_sha256": sha256_of(gz_path),
"dump_bytes": gz_path.stat().st_size,
"row_counts": row_counts(database),
"mysql_client": run([mysql_binary("mysqldump"), "--version"]).strip(),
"mysql_server": scalar("", "SELECT VERSION()", ""),
"contents": "accounts, spaces, sessions, audit logs, books, chapters, ingest jobs, "
"dictionary archive, terms, review schedules, review answers, reading progress",
"excludes": "credentials and the environment file; back those up separately from the operations vault",
}
(out_dir / "manifest.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8")
print("备份完成:" + str(gz_path))
print(" schema 版本 " + str(manifest["schema_version"]) + ",提交 " + (manifest["git_commit"][:8] or "未知"))
print(" 表行数 " + json.dumps(manifest["row_counts"], ensure_ascii=False))
print(" sha256 " + manifest["dump_sha256"][:16] + "…(完整值见 manifest.json)")
print(" 凭据与 .env.local 不在其中,请用运维密码库单独保存。")
return 0
# ---------------------------------------------------------------- restore
def ensure_empty_target(database, force):
if not database_exists(database):
run([mysql_binary("mysql"), *connection_args(), "-e", "CREATE DATABASE `%s` CHARACTER SET utf8mb4" % database], env=client_env())
print("已创建空库 " + database)
return
existing = scalar(database, "SELECT COUNT(*) FROM information_schema.TABLES WHERE TABLE_SCHEMA='%s'" % database, "0")
if int(existing or 0) == 0:
return
if not force:
raise OpsError(
"目标库 " + database + " 已有 " + str(existing) + " 张表。恢复默认只写空库;"
"确认要覆盖时再加 --force,并先对现有库做一次备份。"
)
print("警告:目标库已有数据,按 --force 覆盖。")
run([mysql_binary("mysql"), *connection_args(), "-e",
"DROP DATABASE `%s`; CREATE DATABASE `%s` CHARACTER SET utf8mb4" % (database, database)], env=client_env())
def dump_targets_instead_of_source(dump):
"""Reads the head of a dump and refuses one that would switch databases.
A dump taken with ``mysqldump --databases`` carries CREATE DATABASE and USE statements, so a
restore would write into the schema named inside the file rather than the requested target.
An older backup is rejected instead of being trusted.
"""
with gzip.open(dump, "rb") as handle:
head = handle.read(256 * 1024).decode("utf-8", "replace")
return "CREATE DATABASE" in head.upper() or bool(re.search(r"^USE `", head, re.M))
def dump_checksums(database):
"""Per-table content checksums, used to prove a restore equals its source."""
checksums = {}
for table in TABLES:
rows = query(database, "CHECKSUM TABLE `%s`" % table)
if rows and len(rows[0]) > 1:
checksums[table] = rows[0][1]
return checksums
def command_restore(args):
load_env()
if not args.confirm:
raise OpsError(
"恢复会写入数据库,需要显式确认:\n"
" python scripts/ops.py restore --dump <文件.sql.gz> --database <库名> --confirm\n"
"默认只恢复到空库;覆盖已有库还要加 --force。"
)
dump = Path(args.dump)
if not dump.exists():
raise OpsError("找不到备份文件:" + str(dump))
database = validate_database_name(args.database, "恢复")
manifest = {}
manifest_path = dump.parent / "manifest.json"
if manifest_path.exists():
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
expected = manifest.get("dump_sha256")
if expected and expected != sha256_of(dump):
raise OpsError("备份文件与 manifest 的 sha256 不一致,拒绝恢复。")
print("校验通过:备份文件与 manifest 的 sha256 一致。")
elif not args.skip_manifest_check:
raise OpsError("找不到 manifest.json,无法校验备份完整性(确认要继续时加 --skip-manifest-check)。")
if dump_targets_instead_of_source(dump):
raise OpsError(
"备份文件里带有 CREATE DATABASE / USE,恢复会写进文件里指定的库而不是目标库,拒绝执行。\n"
"请用本工具重新生成备份(它导出的是只含表数据的 dump)。"
)
# The source database must not change: its checksums are compared again afterwards.
source_database = manifest.get("database", "")
source_before = dump_checksums(source_database) if source_database and source_database != database and database_exists(source_database) else {}
ensure_empty_target(database, args.force)
print("正在恢复 " + dump.name + " 到 " + database + " …")
# The dump is decompressed to a temporary file and that file is handed to the client. Feeding
# a pipe from Python is unreliable on Windows for a stream this size, and a real file
# descriptor also means the client sees the plain SQL it expects.
work = Path(tempfile.mkdtemp(prefix="lexgo-restore-"))
plain = work / "dump.sql"
try:
with gzip.open(dump, "rb") as compressed, open(plain, "wb") as target:
shutil.copyfileobj(compressed, target)
with open(plain, "rb") as handle:
load = subprocess.run([mysql_binary("mysql"), *connection_args(database), "--default-character-set=utf8mb4"],
stdin=handle, env=client_env(), capture_output=True)
if load.returncode != 0:
message = load.stderr.decode("utf-8", "replace").strip().splitlines()
raise OpsError("恢复失败:" + (message[-1] if message else "未知错误"))
finally:
shutil.rmtree(work, ignore_errors=True)
print("恢复完成,开始校验。")
if source_before:
source_after = dump_checksums(source_database)
if source_after != source_before:
raise OpsError("源库 " + source_database + " 在校验过程中发生变化,已停止;请人工比对后再继续。")
print("源库 " + source_database + " 的内容校验和未变化。")
verify_args = argparse.Namespace(database=database, manifest=str(manifest_path) if manifest_path.exists() else "",
expect_schema=manifest.get("schema_version"), api="", user="", password_env="",
password_env_b="", skip_api=True)
return command_verify(verify_args)
# ---------------------------------------------------------------- verify
def integrity_checks(database):
checks = []
def add(name, ok, detail=""):
checks.append((name, ok, detail))
add("schema 表存在", int(scalar(database, "SELECT COUNT(*) FROM information_schema.TABLES WHERE TABLE_SCHEMA='%s' AND TABLE_NAME='lexgo_schema'" % database, "0") or 0) == 1)
add("产品标识为 lexgo", scalar(database, "SELECT product FROM lexgo_schema WHERE id=1", "") == "lexgo")
add("没有孤立章节(章节属于其书)", int(scalar(database, "SELECT COUNT(*) FROM lexgo_chapters c LEFT JOIN lexgo_books b ON b.id=c.book_id WHERE b.id IS NULL", "0") or 0) == 0)
add("章节归属与书归属一致", int(scalar(database, "SELECT COUNT(*) FROM lexgo_chapters c JOIN lexgo_books b ON b.id=c.book_id WHERE c.owner_id<>b.owner_id", "0") or 0) == 0)
add("每个词条都有排期行", int(scalar(database, "SELECT COUNT(*) FROM lexgo_terms t LEFT JOIN lexgo_term_reviews r ON r.term_id=t.id WHERE r.term_id IS NULL", "0") or 0) == 0)
add("排期行都指向存在的词条", int(scalar(database, "SELECT COUNT(*) FROM lexgo_term_reviews r LEFT JOIN lexgo_terms t ON t.id=r.term_id WHERE t.id IS NULL", "0") or 0) == 0)
add("复习记录引用有效词条", int(scalar(database, "SELECT COUNT(*) FROM lexgo_review_answers a LEFT JOIN lexgo_terms t ON t.id=a.term_id WHERE t.id IS NULL", "0") or 0) == 0)
add("等级只出现在学习中词条", int(scalar(database, "SELECT COUNT(*) FROM lexgo_terms WHERE (status='learning' AND (level<1 OR level>7)) OR (status<>'learning' AND level<>0)", "0") or 0) == 0)
add("完成记录都指向存在的章节", int(scalar(database, "SELECT COUNT(*) FROM lexgo_chapter_progress p LEFT JOIN lexgo_chapters c ON c.id=p.chapter_id WHERE c.id IS NULL", "0") or 0) == 0)
add("词条语言与所属空间一致", int(scalar(database, "SELECT COUNT(*) FROM lexgo_terms t JOIN lexgo_spaces s ON s.owner_id=t.owner_id WHERE t.language<>s.language", "0") or 0) == 0)
add("附件都指向存在的书", int(scalar(database, "SELECT COUNT(*) FROM lexgo_book_attachments a LEFT JOIN lexgo_books b ON b.id=a.book_id WHERE b.id IS NULL", "0") or 0) == 0)
add("附件归属与书归属一致", int(scalar(database, "SELECT COUNT(*) FROM lexgo_book_attachments a JOIN lexgo_books b ON b.id=a.book_id WHERE a.owner_id<>b.owner_id", "0") or 0) == 0)
add("播放位置都指向存在的书", int(scalar(database, "SELECT COUNT(*) FROM lexgo_playback_positions p LEFT JOIN lexgo_books b ON b.id=p.book_id WHERE b.id IS NULL", "0") or 0) == 0)
add("章节附件都指向存在的章节", int(scalar(database, "SELECT COUNT(*) FROM lexgo_chapter_attachments a LEFT JOIN lexgo_chapters c ON c.id=a.chapter_id WHERE c.id IS NULL", "0") or 0) == 0)
add("章节附件归属与章节归属一致", int(scalar(database, "SELECT COUNT(*) FROM lexgo_chapter_attachments a JOIN lexgo_chapters c ON c.id=a.chapter_id WHERE a.owner_id<>c.owner_id", "0") or 0) == 0)
add("章节播放位置都指向存在的章节", int(scalar(database, "SELECT COUNT(*) FROM lexgo_chapter_playback_positions p LEFT JOIN lexgo_chapters c ON c.id=p.chapter_id WHERE c.id IS NULL", "0") or 0) == 0)
# The audit tables must not gain a column that could hold a credential or private content.
for table in ("lexgo_login_logs", "lexgo_operation_logs"):
columns = {row[0].lower() for row in query("information_schema", "SELECT COLUMN_NAME FROM COLUMNS WHERE TABLE_SCHEMA='%s' AND TABLE_NAME='%s'" % (database, table))}
banned = sorted(columns & AUDIT_BANNED_COLUMNS)
add("审计表 " + table + " 不含敏感列", not banned, "命中:" + ",".join(banned) if banned else "列名白名单通过")
return checks
class Api:
"""A very small API client: the operations tool must not depend on the test harness."""
def __init__(self, base):
self.base = base.rstrip("/")
def call(self, method, path, token=None, body=None):
import urllib.error
import urllib.request
request = urllib.request.Request(self.base + "/api/v1/" + path, method=method)
if token:
request.add_header("Authorization", "Bearer " + token)
payload = None
if body is not None:
payload = json.dumps(body).encode("utf-8")
request.add_header("Content-Type", "application/json")
try:
with urllib.request.urlopen(request, payload, timeout=60) as response:
return response.status, json.loads(response.read().decode("utf-8"))
except urllib.error.HTTPError as error:
try:
return error.code, json.loads(error.read().decode("utf-8"))
except Exception:
return error.code, {}
def login(self, account, password):
status, payload = self.call("POST", "login", body={"username": account, "password": password})
return status, payload.get("data", {}).get("token", "")
def report(checks):
"""Prints one line per check and fails the command when any of them failed."""
failures = [name for name, ok, _ in checks if not ok]
for name, ok, detail in checks:
print((" [ok] " if ok else " [fail] ") + name + ("— " + detail if detail else ""))
if failures:
print("\n检查未通过 " + str(len(failures)) + " 项:" + "、".join(failures))
raise SystemExit(1)
print("\n检查通过:" + str(len(checks)) + " 项全部成功。")
return 0
def api_walk(api, user, password_a, password_b=None):
"""Two accounts walk the learning loop on the restored instance and one must not see the other."""
checks = []
client = Api(api)
def add(name, ok, detail=""):
checks.append((name, ok, detail))
status, token_a = client.login(user + "_a", password_a)
add("账号 " + user + "_a 可以登录恢复实例", status == 200 and bool(token_a))
status, token_b = client.login(user + "_b", password_b or password_a)
add("账号 " + user + "_b 可以登录恢复实例", status == 200 and bool(token_b))
if not token_a or not token_b:
return checks
status, books_a = client.call("GET", "books", token_a)
items_a = books_a.get("data", {}).get("items", [])
add("账号 A 能看到自己的书库", status == 200 and len(items_a) > 0, str(len(items_a)) + " 本书")
status, progress_a = client.call("GET", "progress", token_a)
add("账号 A 能读到自己的进度", status == 200, json.dumps(progress_a.get("data", {}), ensure_ascii=False)[:120])
# Isolation: ids that belong to A must not resolve for B.
if items_a:
book_id = items_a[0]["id"]
status, _ = client.call("GET", "books/%d" % book_id, token_b)
add("B 读取 A 的书籍被拒绝", status == 404, "HTTP " + str(status))
status, detail = client.call("GET", "books/%d" % book_id, token_a)
chapters = [chapter for chapter in detail.get("data", {}).get("chapters", []) if chapter["status"] == "ready"]
if chapters:
chapter_id = chapters[0]["id"]
status, _ = client.call("GET", "chapters/%d" % chapter_id, token_b)
add("B 读取 A 的章节被拒绝", status == 404, "HTTP " + str(status))
# A completes the chapter twice: the mark must stay idempotent after a restore.
status, _ = client.call("POST", "chapters/%d/complete" % chapter_id, token_a)
status2, second = client.call("POST", "chapters/%d/complete" % chapter_id, token_a)
duplicate = second.get("data", {}).get("progress", {}).get("duplicate")
add("恢复后完成章节仍然幂等", status == 200 and status2 == 200 and bool(duplicate), "duplicate=" + str(duplicate))
status, _ = client.call("POST", "chapters/%d/complete" % chapter_id, token_b)
add("B 不能完成 A 的章节", status == 404, "HTTP " + str(status))
# The review loop works on restored schedules.
status, queue = client.call("GET", "reviews/queue", token_a)
items = queue.get("data", {}).get("items", [])
add("账号 A 的到期队列可读", status == 200, "到期 " + str(queue.get("data", {}).get("total")) + " 条")
if items:
item = items[0]
answer = {"answerId": "ops-verify-restore-0001", "grade": "correct", "expectedDueAt": item["dueAt"]}
status, result = client.call("POST", "reviews/%d/answers" % item["id"], token_a, answer)
add("恢复后答题成功并更新排期", status in (200, 201) and result.get("data", {}).get("result") == "applied",
"level " + str(result.get("data", {}).get("levelAfter")))
status, replay = client.call("POST", "reviews/%d/answers" % item["id"], token_a, answer)
add("重复提交同一答案不重复记账", status in (200, 201) and bool(replay.get("data", {}).get("duplicate")),
"duplicate=" + str(replay.get("data", {}).get("duplicate")))
status, _ = client.call("POST", "reviews/%d/answers" % item["id"], token_b,
{"answerId": "ops-verify-restore-0002", "grade": "correct", "expectedDueAt": item["dueAt"]})
add("B 不能给 A 的词条答题", status == 404, "HTTP " + str(status))
return checks
def command_smoke(args):
"""Walks the whole learning loop on a freshly installed instance with two trial accounts."""
load_env()
client = Api(args.api)
password = os.environ.get(args.password_env, "")
if not password:
raise OpsError("环境变量 " + args.password_env + " 为空:请先为演练账号设置密码。")
checks = []
def add(name, ok, detail=""):
checks.append((name, ok, detail))
# 1. The first administrator exists and can log in: the installation is usable.
status, admin_token = client.login(args.admin_user, os.environ.get(args.admin_password_env, ""))
add("初始管理员可以登录", status == 200 and bool(admin_token))
if not admin_token:
return report(checks)
# 2. Two trial learners, created by the administrator. There is no self-registration.
for suffix in ("a", "b"):
account = args.user + "_" + suffix
status, payload = client.call("POST", "accounts", admin_token, {"username": account, "password": password})
if status == 201:
add("管理员创建演练账号 " + account, True, "HTTP 201")
else:
# An existing account is fine on a rerun; the password was set the first time.
existing = client.login(account, password)[0] == 200
add("演练账号 " + account + " 已存在且可登录", existing, payload.get("msg", "HTTP " + str(status)))
status, token_a = client.login(args.user + "_a", password)
status2, token_b = client.login(args.user + "_b", password)
add("两个演练账号都能登录", status == 200 and status2 == 200 and bool(token_a) and bool(token_b))
if not token_a or not token_b:
return report(checks)
add("新账号的学习空间是空的", client.call("GET", "books", token_a)[1].get("data", {}).get("items", []) == [])
add("新账号的进度从零开始", client.call("GET", "progress", token_a)[1].get("data", {}).get("totalChapters") == 0)
# 3. The learning loop on the new instance: paste, read, complete, save a word, review it.
text = "Curiosity opens the first door.\nThe second door stays closed.\n" * 6
status, pasted = client.call("POST", "books", token_a, {
"requestId": args.request_id, "title": "Fictional trial chapter", "text": text, "language": "en"})
add("演练账号可以粘贴章节", status == 201, "HTTP " + str(status))
if status != 201:
return report(checks)
book_id = pasted["data"]["book"]["id"]
chapter_id = pasted["data"]["chapter"]["id"]
ready = {}
for _ in range(40):
status, detail = client.call("GET", "chapters/%d" % chapter_id, token_a)
ready = detail.get("data", {}).get("chapter", {})
if ready.get("status") == "ready":
break
time.sleep(0.5)
add("章节在干净实例上处理完成", ready.get("status") == "ready", str(ready.get("status")))
if ready.get("status") != "ready":
return report(checks)
status, tokens = client.call("GET", "chapters/%d/tokens" % chapter_id, token_a)
words = [token for token in tokens.get("data", {}).get("tokens", []) if token.get("kind") == "word"]
add("章节可以分词并返回可点选的词", status == 200 and len(words) > 3, str(len(words)) + " 个词")
status, _ = client.call("GET", "terms/lookup?query=curiosity", token_a)
add("词典查询返回明确状态(含资源缺失时的降级)", status in (200, 404, 503), "lookup HTTP " + str(status))
if words:
word = words[0]
status, saved = client.call("POST", "terms", token_a, {
"chapterId": chapter_id, "start": word["start"], "end": word["end"],
"definition": "虚构释义", "status": "new"})
add("演练账号可以保存自己的词义", status in (200, 201), "HTTP " + str(status))
term_id = saved.get("data", {}).get("term", {}).get("id")
if term_id:
other_terms = client.call("GET", "terms?query=curiosity", token_b)[1].get("data", {}).get("total")
add("另一个账号看不到该词条", other_terms == 0, "B 查到 " + str(other_terms) + " 条")
queue = client.call("GET", "reviews/queue", token_a)[1]
items = queue.get("data", {}).get("items", [])
due = [entry for entry in items if entry["id"] == term_id]
add("新保存的词条立刻到期", bool(due), "到期 " + str(queue.get("data", {}).get("total")))
if due:
answer = {"answerId": args.answer_id, "grade": "correct", "expectedDueAt": due[0]["dueAt"]}
status, result = client.call("POST", "reviews/%d/answers" % term_id, token_a, answer)
add("到期复习可以作答", status in (200, 201) and result.get("data", {}).get("result") == "applied",
"level " + str(result.get("data", {}).get("levelAfter")))
replay = client.call("POST", "reviews/%d/answers" % term_id, token_a, answer)[1]
add("重复作答不重复记账", replay.get("data", {}).get("duplicate") is True)
status, completed = client.call("POST", "chapters/%d/complete" % chapter_id, token_a)
add("完成章节只记已读", status == 200 and completed.get("data", {}).get("progress", {}).get("read") is True)
data = client.call("GET", "progress", token_a)[1].get("data", {})
add("进度反映新实例上的活动", data.get("readChapters", 0) >= 1 and data.get("savedTerms", 0) >= 1,
json.dumps({key: data.get(key) for key in ("readChapters", "totalChapters", "knownTerms", "learningTerms", "newTerms", "dueNow")}, ensure_ascii=False))
add("B 无法读取 A 的书籍", client.call("GET", "books/%d" % book_id, token_b)[0] == 404)
add("B 的书库仍然为空", client.call("GET", "books", token_b)[1].get("data", {}).get("items", []) == [])
return report(checks)
def command_verify(args):
load_env()
database = validate_database_name(args.database or os.environ.get("LEXGO_DB_NAME", ""), "校验")
failures = []
print("校验数据库 " + database)
if not database_exists(database):
raise OpsError("数据库不存在:" + database)
version = int(scalar(database, "SELECT version FROM lexgo_schema WHERE id=1", "0") or 0)
if version != SCHEMA_VERSION:
print(" [warn] schema 版本 " + str(version) + ",本工具期望 " + str(SCHEMA_VERSION) + "(升级或回退阶段属正常)")
checks = integrity_checks(database)
if args.manifest:
manifest = json.loads(Path(args.manifest).read_text(encoding="utf-8"))
actual = row_counts(database)
for table, expected in manifest.get("row_counts", {}).items():
got = actual.get(table, 0)
checks.append(("表 " + table + " 行数与备份一致", got == expected, "备份 " + str(expected) + " / 现在 " + str(got)))
for name, ok, detail in checks:
print((" [ok] " if ok else " [fail] ") + name + ("— " + detail if detail else ""))
if not ok:
failures.append(name)
if args.api:
print("校验接口 " + args.api + "(账号前缀 " + args.user + ")")
password = os.environ.get(args.password_env, "")
if not password:
raise OpsError("环境变量 " + args.password_env + " 为空,无法登录演练账号。")
password_b = os.environ.get(args.password_env_b, "") or password
for name, ok, detail in api_walk(args.api, args.user, password, password_b):
print((" [ok] " if ok else " [fail] ") + name + ("— " + detail if detail else ""))
if not ok:
failures.append(name)
if failures:
print("\n校验未通过 " + str(len(failures)) + " 项:" + "、".join(failures))
raise SystemExit(1)
print("\n校验通过。")
return 0
def build_parser():
parser = argparse.ArgumentParser(description="LexGo 运维工具:依赖检查、备份、恢复与校验")
sub = parser.add_subparsers(dest="command", required=True)
install = sub.add_parser("install-check", help="只读检查依赖与资源版本")
install.set_defaults(func=command_install_check)
init = sub.add_parser("init-database", help="为全新安装创建空库并提示所需权限")
init.add_argument("--database", default="", help="要创建的库名,默认取 LEXGO_DB_NAME")
init.set_defaults(func=command_init_database)
backup = sub.add_parser("backup", help="导出全库并写出 manifest")
backup.add_argument("--out", default="", help="输出目录,默认 .local/backups/<时间戳>")
backup.add_argument("--database", default="", help="要备份的库,默认取 LEXGO_DB_NAME")
backup.add_argument("--force", action="store_true", help="输出目录非空时覆盖")
backup.set_defaults(func=command_backup)
restore = sub.add_parser("restore", help="把备份恢复到空库(覆盖已有库要 --force)")
restore.add_argument("--dump", required=True, help="backup 生成的 .sql.gz")
restore.add_argument("--database", required=True, help="目标库名,必须包含 lexgo")
restore.add_argument("--confirm", action="store_true", help="确认执行写入")
restore.add_argument("--force", action="store_true", help="目标库已有数据时覆盖")
restore.add_argument("--skip-manifest-check", action="store_true", help="没有 manifest 时跳过完整性校验")
restore.set_defaults(func=command_restore)
verify = sub.add_parser("verify", help="校验恢复结果的完整性与两账号隔离")
verify.add_argument("--database", default="", help="要校验的库,默认取 LEXGO_DB_NAME")
verify.add_argument("--manifest", default="", help="与备份的行数逐表比对")
verify.add_argument("--expect-schema", type=int, default=None)
verify.add_argument("--api", default="", help="例如 http://127.0.0.1:8010 ,启用两账号闭环校验")
verify.add_argument("--user", default="", help="演练账号前缀,实际账号为 <前缀>_a 与 <前缀>_b")
verify.add_argument("--password-env", default="LEXGO_TRIAL_PASSWORD", help="存放演练账号 _a 密码的环境变量名")
verify.add_argument("--password-env-b", default="", help="存放演练账号 _b 密码的环境变量名,默认与 _a 相同")
verify.add_argument("--skip-api", action="store_true")
verify.set_defaults(func=command_verify)
smoke = sub.add_parser("smoke", help="在干净实例上创建两个演练账号并走通学习闭环")
smoke.add_argument("--api", required=True, help="例如 http://127.0.0.1:8010")
smoke.add_argument("--admin-user", required=True, help="初始管理员账号")
smoke.add_argument("--admin-password-env", default="LEXGO_BOOTSTRAP_PASSWORD", help="初始管理员密码所在的环境变量名")
smoke.add_argument("--user", required=True, help="演练账号前缀,实际账号为 <前缀>_a 与 <前缀>_b")
smoke.add_argument("--password-env", default="LEXGO_TRIAL_PASSWORD", help="演练账号密码所在的环境变量名")
smoke.add_argument("--request-id", default="ops-smoke-0001", help="粘贴章节的幂等请求号")
smoke.add_argument("--answer-id", default="ops-smoke-answer-0001", help="复习作答的幂等编号")
smoke.set_defaults(func=command_smoke)
return parser
def main(argv=None):
args = build_parser().parse_args(argv)
try:
return args.func(args)
except OpsError as error:
print("错误:" + str(error), file=sys.stderr)
return 2
if __name__ == "__main__":
sys.exit(main())