leefer/xiaobaifupan
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

migration: close candidate maintenance audit

Browse Source
This commit is contained in:
leefer
2026-07-31 21:24:37 +08:00
parent faac60b1a6
commit 406118bba6
17 changed files with 844 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/tools/README.md
+36
View File
@@ -0,0 +1,36 @@
# Candidate tools
Run these commands from `webapp/app/`. The tools are separated by responsibility so a
maintenance command cannot be mistaken for a historical migration rewrite.
## Normal maintenance
- `python tools/verify_baseline.py`: candidate unit tests, registry checks, every frontend
JavaScript syntax check, Git whitespace check, and read-only SQLite integrity check.
- `python tools/verify_baseline.py --e2e`: the same checks plus Playwright. The verifier owns
the local static-server lifecycle so the command exits cleanly on Windows.
- `python tools/build_api_registry.py [--check]`: generate or verify
`config/api.config.json` from `backend/application.py`.
- `python tools/build_architecture_inventory.py [--check]`: generate or verify
`config/architecture-inventory.json` from the candidate source tree.
## Acceptance and differential checks
- `run_preservation_runtime.py`: start an isolated original or candidate runtime with an
explicitly selected data directory and port.
- `compare_preservation_apis.py`: compare authenticated responses from two isolated runtimes.
- `compare_preservation_databases.py`: compare schema and selected table contents from two
SQLite copies.
These tools require explicit paths and do not select the production database automatically.
## Migration-only tools
- `build_preservation_manifest.py`: builds an exact-copy manifest for a specified source and
target. `--output` is mandatory so committed historical evidence is not overwritten.
- `move_class_methods.py`: mechanically moves named class methods between explicit files.
- `split_frontend_runtime.py`: reproduces the one-time Slice 10 split. It refuses to write
unless `--apply` is supplied and is not a normal maintenance command.
The migration-only tools are retained for audit and reproducibility. They are not part of
application startup, normal testing, or future feature development.
app/tools/build_architecture_inventory.py
+45 -23
View File
@@ -9,7 +9,7 @@ from typing import Any
ROOT = Path(__file__).resolve().parents[1]
OUTPUT = ROOT / "docs" / "governance" / "architecture-inventory.json"
OUTPUT = ROOT / "config" / "architecture-inventory.json"
def relative(path: Path) -> str:
@@ -51,8 +51,15 @@ def api_inventory(server: str) -> dict[str, list[str]]:
return {"exact": exact, "prefixes": prefixes, "patterns": patterns}
def database_inventory(database: str) -> list[str]:
return re.findall(r"CREATE TABLE IF NOT EXISTS\s+([a-zA-Z0-9_]+)", database)
def database_inventory(sources: list[str]) -> list[str]:
tables: list[str] = []
for database in sources:
tables.extend(
re.findall(
r"CREATE TABLE IF NOT EXISTS\s+([a-zA-Z0-9_]+)", database
)
)
return list(dict.fromkeys(tables))
def python_functions(path: str, prefixes: tuple[str, ...]) -> list[str]:
@@ -70,11 +77,13 @@ def css_layers(html: str) -> list[str]:
def code_hotspots() -> list[dict[str, Any]]:
candidates = [
"server.py",
"backend/application.py",
"database.py",
"screener.py",
"market_insights.py",
"tushare_client.py",
"backend/features/screener/engine.py",
"backend/features/market/insights.py",
"backend/data/providers/tushare_client.py",
"backend/features/heaven/service.py",
"backend/features/heaven/engine.py",
"frontend/index.html",
"frontend/app.js",
"frontend/styles/styles.css",
@@ -88,6 +97,8 @@ def code_hotspots() -> list[dict[str, Any]]:
rows = []
for name in candidates:
path = ROOT / name
if not path.is_file():
continue
rows.append(
{
"path": name,
@@ -100,14 +111,21 @@ def code_hotspots() -> list[dict[str, Any]]:
def build() -> dict[str, Any]:
html = source("frontend/index.html")
server = source("server.py")
database = source("database.py")
server = source("backend/application.py")
database_sources = [source("database.py")]
database_sources.extend(
path.read_text(encoding="utf-8")
for path in sorted((ROOT / "backend" / "database" / "migrations").glob("m*.py"))
)
database_sources.append(
source("backend/database/migrations/runner.py")
)
pages = page_inventory(html)
api = api_inventory(server)
tables = database_inventory(database)
tables = database_inventory(database_sources)
return {
"schema_version": 1,
"captured_from": "governed source tree",
"captured_from": "app modular preservation candidate",
"runtime": {
"http_server": "http.server.ThreadingHTTPServer",
"application_processes": 1,
@@ -126,21 +144,22 @@ def build() -> dict[str, Any]:
"api": api,
"database_tables": tables,
"background_job_methods": python_functions(
"server.py", ("_background", "_run_background", "_schedule_", "run_automatic")
"backend/application.py",
("_background", "_run_background", "_schedule_", "run_automatic"),
),
"external_data_adapters": [
{"provider": "tushare", "path": "tushare_client.py", "runtime_role": "primary deterministic market data"},
{"provider": "ifind", "path": "ifind_client.py", "runtime_role": "realtime, charts, snapshots, enrichment"},
{"provider": "eastmoney", "path": "chart_data_provider.py", "runtime_role": "display chart fallback"},
{"provider": "eastmoney", "path": "realtime_aggregator.py", "runtime_role": "isolated realtime observation"},
{"provider": "tencent", "path": "realtime_aggregator.py", "runtime_role": "index observation fallback"},
{"provider": "tushare", "path": "backend/data/providers/tushare_client.py", "runtime_role": "primary deterministic market data"},
{"provider": "ifind", "path": "backend/data/providers/ifind_client.py", "runtime_role": "realtime, charts, snapshots, enrichment"},
{"provider": "eastmoney", "path": "backend/features/market/charts.py", "runtime_role": "display chart fallback"},
{"provider": "eastmoney", "path": "backend/data/realtime.py", "runtime_role": "isolated realtime observation"},
{"provider": "tencent", "path": "backend/data/realtime.py", "runtime_role": "index observation fallback"},
],
"llm_entrypoints": [
{"function": "stream_with_mentor", "path": "mentor_agent.py"},
{"function": "interpret_heaven", "path": "heaven_agent.py"},
{"function": "stream_review_assistant", "path": "assistant_agent.py"},
{"function": "compile_strategy_with_llm", "path": "llm_strategy.py"},
{"function": "test_llm_connection", "path": "llm_strategy.py"},
{"function": "stream_with_mentor", "path": "backend/features/mentor/agent.py"},
{"function": "interpret_heaven", "path": "backend/features/heaven/agent.py"},
{"function": "stream_review_assistant", "path": "backend/features/review/agent.py"},
{"function": "compile_strategy_with_llm", "path": "backend/features/screener/compiler.py"},
{"function": "test_llm_connection", "path": "backend/features/screener/compiler.py"},
],
"css_layers": css_layers(html),
"code_hotspots": code_hotspots(),
@@ -154,7 +173,10 @@ def main() -> int:
rendered = json.dumps(build(), ensure_ascii=False, indent=2) + "\n"
if args.check:
if not OUTPUT.exists() or OUTPUT.read_text(encoding="utf-8") != rendered:
raise SystemExit("architecture inventory is stale; run tools/build_architecture_inventory.py")
raise SystemExit(
"architecture inventory is stale; run "
"tools/build_architecture_inventory.py"
)
print("Architecture inventory is current.")
return 0
OUTPUT.parent.mkdir(parents=True, exist_ok=True)
app/tools/build_preservation_manifest.py
+64 -21
View File
@@ -1,14 +1,13 @@
from __future__ import annotations
import argparse
import hashlib
import json
from datetime import datetime
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
TARGET = ROOT / "app"
OUTPUT = ROOT / "docs" / "migration" / "原版资产清单.json"
REPOSITORY_ROOT = Path(__file__).resolve().parents[2]
DIRECTORIES = (
"backend",
@@ -69,33 +68,57 @@ def digest(path: Path) -> str:
return checksum.hexdigest()
def source_files() -> list[Path]:
files = [ROOT / name for name in ROOT_FILES]
def display_path(path: Path, base: Path) -> str:
try:
return path.relative_to(base).as_posix() or "."
except ValueError:
return path.as_posix()
def source_files(source_root: Path) -> list[Path]:
files = [source_root / name for name in ROOT_FILES]
for directory in DIRECTORIES:
root = source_root / directory
if not root.is_dir():
continue
files.extend(
path
for path in (ROOT / directory).rglob("*")
for path in root.rglob("*")
if path.is_file() and "__pycache__" not in path.parts
)
return sorted(set(files), key=lambda path: path.relative_to(ROOT).as_posix())
return sorted(
set(files), key=lambda path: path.relative_to(source_root).as_posix()
)
def build_manifest() -> dict[str, object]:
def build_manifest(
source_root: Path,
target_root: Path,
source_commit: str,
) -> dict[str, object]:
assets = []
mismatches = []
for source in source_files():
relative = source.relative_to(ROOT)
target = TARGET / relative
source_hash = digest(source)
for source in source_files(source_root):
relative = source.relative_to(source_root)
target = target_root / relative
source_exists = source.is_file()
source_hash = digest(source) if source_exists else ""
target_hash = digest(target) if target.is_file() else ""
status = "identical" if source_hash == target_hash else "mismatch"
if not source_exists:
status = "missing_source"
elif not target.is_file():
status = "missing_target"
elif source_hash == target_hash:
status = "identical"
else:
status = "mismatch"
if status != "identical":
mismatches.append(relative.as_posix())
assets.append(
{
"source": relative.as_posix(),
"target": f"app/{relative.as_posix()}",
"bytes": source.stat().st_size,
"target": display_path(target, REPOSITORY_ROOT),
"bytes": source.stat().st_size if source_exists else 0,
"sha256": source_hash,
"disposition": "original_copy_pending_move",
"status": status,
@@ -104,9 +127,9 @@ def build_manifest() -> dict[str, object]:
return {
"schema_version": 1,
"generated_at": datetime.now().astimezone().isoformat(timespec="seconds"),
"source_commit": "41329943c4878fc09ed82ec376eb93ab151e4092",
"source_root": ".",
"target_root": "app",
"source_commit": source_commit,
"source_root": display_path(source_root, REPOSITORY_ROOT),
"target_root": display_path(target_root, REPOSITORY_ROOT),
"excluded": [
"next/",
"data/review.db and SQLite sidecars",
@@ -123,14 +146,34 @@ def build_manifest() -> dict[str, object]:
def main() -> int:
manifest = build_manifest()
OUTPUT.write_text(
parser = argparse.ArgumentParser(
description=(
"Build an exact-copy manifest for a preservation migration stage. "
"This is a migration-only tool; it does not describe the final moved layout."
)
)
parser.add_argument("--source-root", type=Path, default=REPOSITORY_ROOT)
parser.add_argument("--target-root", type=Path, default=REPOSITORY_ROOT / "app")
parser.add_argument("--source-commit", default="")
parser.add_argument(
"--output",
type=Path,
required=True,
help="write to a new audit path; do not overwrite committed slice evidence",
)
args = parser.parse_args()
source_root = args.source_root.resolve()
target_root = args.target_root.resolve()
output = args.output.resolve()
manifest = build_manifest(source_root, target_root, args.source_commit)
output.parent.mkdir(parents=True, exist_ok=True)
output.write_text(
json.dumps(manifest, ensure_ascii=False, indent=2) + "\n",
encoding="utf-8",
)
print(
f"assets={manifest['asset_count']} mismatches={manifest['mismatch_count']} "
f"output={OUTPUT}"
f"output={output}"
)
return 1 if manifest["mismatch_count"] else 0
app/tools/split_frontend_runtime.py
+15
View File
@@ -1,5 +1,6 @@
from __future__ import annotations
import argparse
import hashlib
import json
import re
@@ -91,6 +92,20 @@ def extract_written_chunks(paths: list[Path]) -> dict[tuple[int, int], str]:
def main() -> None:
parser = argparse.ArgumentParser(
description="Reproduce the one-time Slice 10 frontend source split"
)
parser.add_argument(
"--apply",
action="store_true",
help="perform the historical split; this rewrites frontend runtime files",
)
args = parser.parse_args()
if not args.apply:
parser.error(
"this migration-only tool rewrites files; pass --apply only when "
"reproducing Slice 10 from its documented pre-split checkpoint"
)
source_path = ORIGINAL_STATIC / "app.js"
target_path = FRONTEND_ROOT / "app.js"
source_bytes = source_path.read_bytes()
app/tools/verify_baseline.py
+85 -7
View File
@@ -5,10 +5,15 @@ import shutil
import sqlite3
import subprocess
import sys
import time
import urllib.error
import urllib.request
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
FRONTEND_ROOT = ROOT / "frontend"
E2E_URL = "http://127.0.0.1:8876/index.html"
def run(label: str, command: list[str]) -> None:
@@ -28,8 +33,63 @@ def verify_database() -> None:
print(f"\n[database] integrity_check=ok size={database.stat().st_size}")
def url_reachable(url: str) -> bool:
try:
with urllib.request.urlopen(url, timeout=1) as response:
return response.status == 200
except (OSError, urllib.error.URLError):
return False
def start_e2e_server() -> subprocess.Popen[bytes] | None:
if url_reachable(E2E_URL):
print(f"\n[e2e-server] reusing {E2E_URL}")
return None
creation_flags = subprocess.CREATE_NO_WINDOW if sys.platform == "win32" else 0
process = subprocess.Popen(
[
sys.executable,
"-m",
"http.server",
"8876",
"--bind",
"127.0.0.1",
"--directory",
"frontend",
],
cwd=ROOT,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
creationflags=creation_flags,
)
deadline = time.monotonic() + 15
while time.monotonic() < deadline:
if process.poll() is not None:
raise RuntimeError("Playwright static server exited before becoming ready")
if url_reachable(E2E_URL):
print(f"\n[e2e-server] started {E2E_URL}")
return process
time.sleep(0.1)
stop_e2e_server(process)
raise RuntimeError("Playwright static server did not become ready within 15 seconds")
def stop_e2e_server(process: subprocess.Popen[bytes] | None) -> None:
if process is None:
return
process.terminate()
try:
process.wait(timeout=5)
except subprocess.TimeoutExpired:
process.kill()
process.wait(timeout=5)
print("\n[e2e-server] stopped")
def main() -> int:
parser = argparse.ArgumentParser(description="Verify the governance regression baseline")
parser = argparse.ArgumentParser(
description="Verify the modular preservation candidate"
)
parser.add_argument(
"--e2e",
action="store_true",
@@ -38,19 +98,37 @@ def main() -> int:
args = parser.parse_args()
run("python", [sys.executable, "-m", "unittest", "discover", "-s", "tests"])
run(
"api-registry",
[sys.executable, "tools/build_api_registry.py", "--check"],
)
run(
"architecture-inventory",
[sys.executable, "tools/build_architecture_inventory.py", "--check"],
)
node = shutil.which("node")
if not node:
raise RuntimeError("node is required for JavaScript syntax checks")
for script in ("static/app.js", "static/heaven-loading-v2.js"):
run("javascript", [node, "--check", script])
scripts = sorted(FRONTEND_ROOT.rglob("*.js"))
if not scripts:
raise RuntimeError(f"no JavaScript files found under {FRONTEND_ROOT}")
for script in scripts:
run(
"javascript",
[node, "--check", script.relative_to(ROOT).as_posix()],
)
run("patch", ["git", "diff", "--check"])
verify_database()
if args.e2e:
npm = shutil.which("npm.cmd" if sys.platform == "win32" else "npm")
if not npm:
raise RuntimeError("npm is required for the Playwright suite")
run("playwright", [npm, "run", "test:e2e"])
npx = shutil.which("npx.cmd" if sys.platform == "win32" else "npx")
if not npx:
raise RuntimeError("npx is required for the Playwright suite")
server = start_e2e_server()
try:
run("playwright", [npx, "playwright", "test", "--reporter=dot"])
finally:
stop_e2e_server(server)
print("\nBaseline verification passed.")
return 0