ila
0fd661e061
- 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 改为章级口径
792 lines
41 KiB
Python
792 lines
41 KiB
Python
"""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())
|