qiaoqichaoandClaude Code
d56c699ce6
按 docs/v2/05-implementation-plan.md §1 候选迁移表与 claude_use/T00 白名单， 从 55e2cdb 抽取 224 个文件（2.48 MB）到独立目录 F:/mini-OpenClaw-v2。 - 排除：凭据(.env/server-deploy.conf/旧DELIVERY.md)、运行数据(xingbang_data 等 4 个 SQLite)、依赖(venv/node_modules 符号链接)、归档(_backup*/test-evidence/ ori_file)、Office 临时文件 - 目标目录独立 git 仓库，初始恢复点；Gitea 远程与分支待 OQ-06 Q2 定案 - 未部署、未启动服务、未调用模型 Co-Authored-By: Claude Code <noreply@anthropic.com>
87 lines
4.2 KiB
Python
87 lines
4.2 KiB
Python
"""Read-only migration audit by default; apply creates a consistent backup first.
|
|
|
|
Use backend/venv/Scripts/python.exe scripts/assistant_migrate.py --dry-run.
|
|
Restore produces a separate recovered database, never overwrites post-release writes.
|
|
"""
|
|
import argparse
|
|
import json
|
|
import sqlite3
|
|
import sys
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT / "backend"))
|
|
|
|
|
|
def audit(path):
|
|
if not path.is_file():
|
|
return {"database": str(path), "exists": False, "action": "initialize-demo-instance", "dangling_references": []}
|
|
with sqlite3.connect(path.as_uri() + "?mode=ro", uri=True) as db:
|
|
tables = {r[0] for r in db.execute("SELECT name FROM sqlite_master WHERE type='table'")}
|
|
counts = {name: db.execute('SELECT count(*) FROM "' + name + '"').fetchone()[0] for name in tables if name in ("projects", "runs", "snapshots", "reports", "experiences", "assistant_objects", "assistant_jobs")}
|
|
dangling = []
|
|
if "assistant_objects" in tables:
|
|
objects = {(r[0], r[1]): json.loads(r[2]) for r in db.execute("SELECT kind,id,body FROM assistant_objects")}
|
|
for (kind, key), body in objects.items():
|
|
for field, target in (("snapshot_id", "snapshot"), ("input_snapshot_id", "input_snapshot"), ("run_id", "run"), ("document_ids", "document")):
|
|
if kind not in ("run", "report", "report_final", "project", "experience_application"):
|
|
continue
|
|
values = body.get(field)
|
|
for value in (values if isinstance(values, list) else [values]):
|
|
if value and (target, value) not in objects:
|
|
dangling.append({"kind": kind, "id": key, "field": field, "missing": value})
|
|
return {"database": str(path), "exists": True, "counts": counts, "integrity": db.execute("PRAGMA integrity_check").fetchone()[0], "dangling_references": dangling}
|
|
|
|
|
|
def backup(source, target):
|
|
if target.exists() or source == target:
|
|
raise ValueError("备份目标已存在或与源相同，拒绝覆盖")
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
original = sqlite3.connect(source.as_uri() + "?mode=ro", uri=True)
|
|
output = sqlite3.connect(target)
|
|
try:
|
|
original.backup(output)
|
|
finally:
|
|
output.close()
|
|
original.close()
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--database", type=Path, default=ROOT / "backend/xingbang_data/project_assistant.sqlite3")
|
|
parser.add_argument("--dry-run", action="store_true")
|
|
parser.add_argument("--apply", action="store_true")
|
|
parser.add_argument("--restore-copy", type=Path, help="Output recovered DB alongside current DB; never replace live data")
|
|
args = parser.parse_args()
|
|
path = args.database.resolve()
|
|
if not path.is_relative_to(ROOT):
|
|
raise ValueError("迁移数据库须位于当前工作区")
|
|
before = audit(path)
|
|
if before.get("integrity", "ok") != "ok" or before["dangling_references"]:
|
|
print(json.dumps(before, ensure_ascii=False, indent=2))
|
|
return 1
|
|
if args.restore_copy:
|
|
target = args.restore_copy.resolve()
|
|
if not target.is_relative_to(ROOT) or target == path:
|
|
raise ValueError("恢复副本须为工作区内的新路径")
|
|
backup(path, target)
|
|
print(json.dumps({"restored_copy": str(target), "live_data_untouched": True, "audit": audit(target)}, ensure_ascii=False, indent=2))
|
|
return 0
|
|
if args.apply and not args.dry_run:
|
|
if path.exists():
|
|
target = ROOT / ("_backup_pre_assistant_migration_" + datetime.now().strftime("%Y%m%d_%H%M%S")) / path.name
|
|
backup(path, target)
|
|
before["backup"] = str(target)
|
|
from services.project_assistant import ProjectAssistantService
|
|
ProjectAssistantService(path)
|
|
result = {"before": before, "after": audit(path), "legacy_tables": "preserved-read-only"}
|
|
else:
|
|
result = {"mode": "dry-run", **before, "writes": 0}
|
|
print(json.dumps(result, ensure_ascii=False, indent=2))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|