from __future__ import annotations
|
|
import argparse
|
import csv
|
import getpass
|
import hashlib
|
import json
|
import os
|
from datetime import date, datetime, time
|
from pathlib import Path
|
from typing import Any
|
|
import pymysql
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
DEFAULT_OUT = ROOT / "data" / "mysql_seed_20260624"
|
TABLES = [
|
"dl_source_document",
|
"dl_case_import_batch",
|
"dl_case_record",
|
"dl_darkline_hypothesis",
|
"dl_case_reasoning_step",
|
"dl_expected_line",
|
"dl_event_node",
|
"dl_evidence",
|
"dl_evidence_node_link",
|
"dl_impact_target",
|
"dl_manifestation_bridge",
|
"dl_alternative_explanation",
|
"dl_chain_state",
|
]
|
|
|
def sha256_file(path: Path) -> str:
|
h = hashlib.sha256()
|
with path.open("rb") as f:
|
for chunk in iter(lambda: f.read(1024 * 1024), b""):
|
h.update(chunk)
|
return h.hexdigest()
|
|
|
def json_default(value: Any) -> str:
|
if isinstance(value, datetime):
|
return value.isoformat(sep=" ")
|
if isinstance(value, (date, time)):
|
return value.isoformat()
|
raise TypeError(f"Unsupported type: {type(value)!r}")
|
|
|
def get_password(args: argparse.Namespace) -> str:
|
if args.password:
|
return args.password
|
if os.environ.get("DARKLINE_MYSQL_PASSWORD"):
|
return os.environ["DARKLINE_MYSQL_PASSWORD"]
|
if os.environ.get("TIANXIA_MYSQL_PASSWORD"):
|
return os.environ["TIANXIA_MYSQL_PASSWORD"]
|
return getpass.getpass("MySQL password: ")
|
|
|
def connect(args: argparse.Namespace):
|
return pymysql.connect(
|
host=args.host,
|
port=args.port,
|
user=args.user,
|
password=get_password(args),
|
database=args.database,
|
charset="utf8mb4",
|
autocommit=True,
|
cursorclass=pymysql.cursors.DictCursor,
|
)
|
|
|
def write_manifest(out_dir: Path, files: list[Path]) -> None:
|
rows = []
|
for path in sorted(files):
|
rows.append(
|
{
|
"relative_path": str(path.relative_to(out_dir)).replace("\\", "/"),
|
"size_bytes": path.stat().st_size,
|
"sha256": sha256_file(path),
|
"artifact_status": "READY",
|
}
|
)
|
with (out_dir / "manifest.csv").open("w", newline="", encoding="utf-8-sig") as f:
|
writer = csv.DictWriter(
|
f,
|
fieldnames=["relative_path", "size_bytes", "sha256", "artifact_status"],
|
)
|
writer.writeheader()
|
writer.writerows(rows)
|
|
|
def main() -> None:
|
parser = argparse.ArgumentParser(description="Export darkline dl_ tables as portable seed data.")
|
parser.add_argument("--host", default=os.environ.get("DARKLINE_MYSQL_HOST", "127.0.0.1"))
|
parser.add_argument("--port", type=int, default=int(os.environ.get("DARKLINE_MYSQL_PORT", "3306")))
|
parser.add_argument("--user", default=os.environ.get("DARKLINE_MYSQL_USER", "root"))
|
parser.add_argument("--password", default=None)
|
parser.add_argument("--database", default=os.environ.get("DARKLINE_MYSQL_DATABASE", "tianxia"))
|
parser.add_argument("--out-dir", default=str(DEFAULT_OUT))
|
args = parser.parse_args()
|
|
out_dir = Path(args.out_dir)
|
tables_dir = out_dir / "tables"
|
tables_dir.mkdir(parents=True, exist_ok=True)
|
|
conn = connect(args)
|
schema_parts: list[str] = []
|
summary: dict[str, Any] = {
|
"seed_id": "darkline_mysql_seed_20260624",
|
"database": args.database,
|
"tables": {},
|
"exported_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
}
|
generated_files: list[Path] = []
|
|
with conn.cursor() as cur:
|
for table in TABLES:
|
cur.execute(f"SHOW CREATE TABLE `{table}`")
|
create_sql = cur.fetchone()["Create Table"]
|
schema_parts.append(f"DROP TABLE IF EXISTS `{table}`;\n{create_sql};")
|
|
cur.execute(f"SELECT * FROM `{table}`")
|
rows = cur.fetchall()
|
out_path = tables_dir / f"{table}.jsonl"
|
with out_path.open("w", encoding="utf-8") as f:
|
for row in rows:
|
f.write(json.dumps(row, ensure_ascii=False, default=json_default) + "\n")
|
generated_files.append(out_path)
|
summary["tables"][table] = {
|
"row_count": len(rows),
|
"jsonl": str(out_path.relative_to(out_dir)).replace("\\", "/"),
|
"sha256": sha256_file(out_path),
|
}
|
|
conn.close()
|
|
schema_path = out_dir / "schema.sql"
|
schema_path.write_text("\n\n".join(schema_parts) + "\n", encoding="utf-8")
|
generated_files.append(schema_path)
|
|
order_path = out_dir / "table_order.json"
|
order_path.write_text(json.dumps(TABLES, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
generated_files.append(order_path)
|
|
summary_path = out_dir / "seed_summary.json"
|
summary_path.write_text(json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
generated_files.append(summary_path)
|
|
readme_path = out_dir / "README.md"
|
readme_path.write_text(
|
"# darkline MySQL seed data\n\n"
|
"本目录用于在另一台机器复原 darkline 案例数据。\n\n"
|
"导入方式见项目根目录 `darkline安装文档.md`。\n\n"
|
"包含表：\n\n"
|
+ "\n".join(f"- `{table}`" for table in TABLES)
|
+ "\n",
|
encoding="utf-8",
|
)
|
generated_files.append(readme_path)
|
|
write_manifest(out_dir, generated_files)
|
print(json.dumps(summary, ensure_ascii=False, indent=2))
|
|
|
if __name__ == "__main__":
|
main()
|
