{
"event": "PreToolUse",
"tool_name": "Edit",
"tool_input": {
"file_path": "\/var\/www\/dev.campus.systemische-tools.de\/crawler\/scripts\/schema_drift.py",
"old_string": "\"\"\"Schema-drift check — compares table list between schema.sql and live DB.",
"new_string": "#!\/usr\/bin\/env python3\n\"\"\"Schema-drift check — compares table list between schema.sql and live DB.",
"replace_all": false
}
}
{
"tool_response": {
"filePath": "\/var\/www\/dev.campus.systemische-tools.de\/crawler\/scripts\/schema_drift.py",
"oldString": "\"\"\"Schema-drift check — compares table list between schema.sql and live DB.",
"newString": "#!\/usr\/bin\/env python3\n\"\"\"Schema-drift check — compares table list between schema.sql and live DB.",
"originalFile": "\"\"\"Schema-drift check — compares table list between schema.sql and live DB.\n\nUsage:\n python scripts\/schema_drift.py --expected src\/db\/schema.sql --db crawler\n\nExit codes:\n 0 in sync\n 1 drift detected\n 2 error\n\"\"\"\nfrom __future__ import annotations\n\nimport argparse\nimport pathlib\nimport re\nimport sys\n\nsys.path.insert(0, \"\/var\/www\/scripts\/pipeline\")\nimport mysql.connector\nfrom config import DB_CONFIG\n\nTABLE_RE = re.compile(r\"CREATE\\s+TABLE\\s+(?:IF\\s+NOT\\s+EXISTS\\s+)?`?(\\w+)`?\", re.IGNORECASE)\n\n\ndef _expected_tables(schema_path: pathlib.Path) -> set[str]:\n text = schema_path.read_text(encoding=\"utf-8\")\n return {m.group(1) for m in TABLE_RE.finditer(text)}\n\n\ndef _live_tables(db_name: str) -> set[str]:\n conn = mysql.connector.connect(\n host=DB_CONFIG[\"host\"],\n user=DB_CONFIG[\"user\"],\n password=DB_CONFIG[\"password\"],\n database=db_name,\n )\n try:\n cur = conn.cursor()\n cur.execute(\n \"SELECT TABLE_NAME FROM information_schema.TABLES WHERE TABLE_SCHEMA=%s\",\n (db_name,),\n )\n return {row[0] for row in cur.fetchall()}\n finally:\n conn.close()\n\n\ndef main() -> int:\n parser = argparse.ArgumentParser()\n parser.add_argument(\"--expected\", required=True, type=pathlib.Path)\n parser.add_argument(\"--db\", required=True)\n args = parser.parse_args()\n if not args.expected.exists():\n print(f\"schema file not found: {args.expected}\", file=sys.stderr)\n return 2\n expected = _expected_tables(args.expected)\n live = _live_tables(args.db)\n missing = expected - live\n extra = live - expected\n if missing or extra:\n if missing:\n print(f\"MISSING in DB: {sorted(missing)}\", file=sys.stderr)\n if extra:\n print(f\"EXTRA in DB: {sorted(extra)}\", file=sys.stderr)\n return 1\n print(f\"OK schema_tables={len(expected)}\")\n return 0\n\n\nif __name__ == \"__main__\":\n sys.exit(main())\n",
"structuredPatch": [
{
"oldStart": 1,
"oldLines": 3,
"newStart": 1,
"newLines": 4,
"lines": [
"+#!\/usr\/bin\/env python3",
" \"\"\"Schema-drift check — compares table list between schema.sql and live DB.",
" ",
" Usage:"
]
}
],
"userModified": false,
"replaceAll": false
}
}