feat: 完成三角色工作台及金租对公Skills全流程演示

backend/services/business_demo_database.py

@@ -0,0 +1,42 @@
+"""Build the sanitized multi-table leasing database used by the local demo."""
+import os
+import sqlite3
+from contextlib import closing
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[1]
+SCHEMA_PATH = ROOT / "fixtures" / "xingbang_demo" / "business_database.sql"
+DEFAULT_PATH = ROOT / "xingbang_data" / "xingbang_business_demo.sqlite3"
+
+
+def build_business_demo_database(target=DEFAULT_PATH, force=False):
+    target = Path(target).resolve()
+    if target.exists() and not force:
+        return target
+    target.parent.mkdir(parents=True, exist_ok=True)
+    temporary = target.with_name(target.name + ".building")
+    if temporary.exists():
+        temporary.unlink()
+    try:
+        with closing(sqlite3.connect(temporary)) as db:
+            db.executescript(SCHEMA_PATH.read_text(encoding="utf-8"))
+            result = db.execute("PRAGMA integrity_check").fetchone()[0]
+            if result != "ok":
+                raise RuntimeError("演示业务库完整性检查失败")
+            db.commit()
+        os.replace(temporary, target)
+    finally:
+        if temporary.exists():
+            temporary.unlink()
+    return target
+
+
+def database_summary(target=DEFAULT_PATH):
+    target = Path(target).resolve()
+    with sqlite3.connect(target) as db:
+        tables = [row[0] for row in db.execute(
+            "SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' ORDER BY name"
+        )]
+        counts = {table: db.execute(f'SELECT COUNT(*) FROM "{table}"').fetchone()[0] for table in tables}
+    return {"database": str(target), "tables": counts}