from __future__ import annotations
|
|
import argparse
|
import getpass
|
import json
|
import os
|
from pathlib import Path
|
|
import pymysql
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
DEFAULT_SEED_SUMMARY = ROOT / "data" / "mysql_seed_20260624" / "seed_summary.json"
|
|
|
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 expected_counts(seed_summary_path: Path) -> dict[str, int]:
|
summary = json.loads(seed_summary_path.read_text(encoding="utf-8"))
|
return {
|
table: int(meta["row_count"])
|
for table, meta in summary.get("tables", {}).items()
|
}
|
|
|
def main() -> None:
|
parser = argparse.ArgumentParser(description="Validate local darkline data and skill installation.")
|
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("--seed-summary", default=str(DEFAULT_SEED_SUMMARY))
|
parser.add_argument(
|
"--skill-dir",
|
default=str(Path.home() / ".codex2" / "skills" / "darkline"),
|
)
|
args = parser.parse_args()
|
|
expected = expected_counts(Path(args.seed_summary))
|
password = get_password(args)
|
conn = pymysql.connect(
|
host=args.host,
|
port=args.port,
|
user=args.user,
|
password=password,
|
database=args.database,
|
charset="utf8mb4",
|
)
|
db_counts = {}
|
with conn.cursor() as cur:
|
for table in expected:
|
cur.execute(f"SELECT COUNT(*) FROM `{table}`")
|
db_counts[table] = cur.fetchone()[0]
|
conn.close()
|
|
skill_dir = Path(args.skill_dir)
|
skill_file = skill_dir / "SKILL.md"
|
result = {
|
"database": args.database,
|
"expected_counts": expected,
|
"db_counts": db_counts,
|
"db_count_match": db_counts == expected,
|
"skill_dir": str(skill_dir),
|
"skill_file_exists": skill_file.exists(),
|
"skill_name_darkline": False,
|
"overall_status": "FAIL",
|
}
|
if skill_file.exists():
|
text = skill_file.read_text(encoding="utf-8", errors="ignore")
|
result["skill_name_darkline"] = "name: darkline" in text
|
|
if result["db_count_match"] and result["skill_file_exists"] and result["skill_name_darkline"]:
|
result["overall_status"] = "PASS"
|
|
print(json.dumps(result, ensure_ascii=False, indent=2))
|
|
|
if __name__ == "__main__":
|
main()
|