388 lines
13 KiB
Python
388 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Export the museum guide/explain static data package from a Navicat SQL dump.
|
|
|
|
The parser streams the dump statement by statement and only buffers CREATE TABLE
|
|
or INSERT statements for the guide static-data tables. It does not connect to a
|
|
database and never records credentials in generated metadata.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import datetime as dt
|
|
import decimal
|
|
import importlib.util
|
|
import json
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Any, Iterable
|
|
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
|
DEFAULT_OUTPUT = PROJECT_ROOT / "static" / "guide-data"
|
|
DEFAULT_NAV_ASSETS = (
|
|
PROJECT_ROOT
|
|
/ "static"
|
|
/ "nav-assets"
|
|
/ "codex_nav_20260621_175310_museum_default_centerline_finalxy"
|
|
)
|
|
SOURCE_SCRIPT = Path(__file__).with_name("export-guide-static-data.py")
|
|
|
|
|
|
def load_source_module():
|
|
spec = importlib.util.spec_from_file_location("guide_static_export", SOURCE_SCRIPT)
|
|
if spec is None or spec.loader is None:
|
|
raise RuntimeError(f"failed to load source export script: {SOURCE_SCRIPT}")
|
|
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
guide_export = load_source_module()
|
|
SCHEMA_VERSION = guide_export.SCHEMA_VERSION
|
|
TABLES: dict[str, str] = guide_export.TABLES
|
|
TARGET_TABLES = set(TABLES)
|
|
|
|
CREATE_RE = re.compile(r'CREATE\s+TABLE\s+"NATURE_SGS"\."(?P<table>[A-Z0-9_]+)"', re.I)
|
|
INSERT_RE = re.compile(r'INSERT\s+INTO\s+"NATURE_SGS"\."(?P<table>[A-Z0-9_]+)"', re.I | re.S)
|
|
INSERT_VALUES_RE = re.compile(
|
|
r'^\s*INSERT\s+INTO\s+"NATURE_SGS"\."(?P<table>[A-Z0-9_]+)"\s*'
|
|
r'(?P<columns>\((?:\s*"[^"]+"\s*,?)+\))?\s*VALUES\s*(?P<values>.*?);?\s*$',
|
|
re.I | re.S,
|
|
)
|
|
HEADER_FIELD_RE = re.compile(r"^\s*(Source Host|Source Schema|Date)\s*:\s*(.*?)\s*$", re.I)
|
|
|
|
|
|
def statement_complete(sql: str) -> bool:
|
|
in_string = False
|
|
i = 0
|
|
while i < len(sql):
|
|
char = sql[i]
|
|
if char == "'":
|
|
if in_string and i + 1 < len(sql) and sql[i + 1] == "'":
|
|
i += 2
|
|
continue
|
|
in_string = not in_string
|
|
elif char == ";" and not in_string:
|
|
return True
|
|
i += 1
|
|
return False
|
|
|
|
|
|
def iter_target_statements(sql_dump_path: Path) -> Iterable[tuple[str, str]]:
|
|
buffer: list[str] = []
|
|
table: str | None = None
|
|
|
|
with sql_dump_path.open("r", encoding="utf-8", errors="replace", newline="") as handle:
|
|
for line in handle:
|
|
if table is None:
|
|
create_match = CREATE_RE.search(line)
|
|
insert_match = INSERT_RE.search(line)
|
|
match = create_match or insert_match
|
|
if not match or match.group("table").upper() not in TARGET_TABLES:
|
|
continue
|
|
|
|
table = match.group("table").upper()
|
|
buffer = [line]
|
|
else:
|
|
buffer.append(line)
|
|
|
|
current = "".join(buffer)
|
|
if statement_complete(current):
|
|
yield table, current
|
|
buffer = []
|
|
table = None
|
|
|
|
if table is not None:
|
|
raise RuntimeError(f"unterminated SQL statement for {table}")
|
|
|
|
|
|
def parse_header(sql_dump_path: Path) -> dict[str, str]:
|
|
source: dict[str, str] = {}
|
|
with sql_dump_path.open("r", encoding="utf-8", errors="replace") as handle:
|
|
for line_number, line in enumerate(handle, 1):
|
|
match = HEADER_FIELD_RE.match(line)
|
|
if match:
|
|
key = match.group(1).lower().replace(" ", "")
|
|
value = match.group(2).strip()
|
|
if key == "sourcehost":
|
|
source["sourceHost"] = value
|
|
elif key == "sourceschema":
|
|
source["sourceSchema"] = value
|
|
elif key == "date":
|
|
source["sourceDate"] = value
|
|
|
|
if line_number > 80 or line.strip() == "*/":
|
|
break
|
|
|
|
return source
|
|
|
|
|
|
def extract_parenthesized(text: str, start_index: int = 0) -> tuple[str, int]:
|
|
while start_index < len(text) and text[start_index].isspace():
|
|
start_index += 1
|
|
if start_index >= len(text) or text[start_index] != "(":
|
|
raise ValueError("expected parenthesized expression")
|
|
|
|
in_string = False
|
|
depth = 0
|
|
start = start_index + 1
|
|
i = start_index
|
|
while i < len(text):
|
|
char = text[i]
|
|
if char == "'":
|
|
if in_string and i + 1 < len(text) and text[i + 1] == "'":
|
|
i += 2
|
|
continue
|
|
in_string = not in_string
|
|
elif not in_string:
|
|
if char == "(":
|
|
depth += 1
|
|
elif char == ")":
|
|
depth -= 1
|
|
if depth == 0:
|
|
return text[start:i], i + 1
|
|
i += 1
|
|
|
|
raise ValueError("unterminated parenthesized expression")
|
|
|
|
|
|
def split_top_level_csv(text: str) -> list[str]:
|
|
values: list[str] = []
|
|
start = 0
|
|
depth = 0
|
|
in_string = False
|
|
i = 0
|
|
while i < len(text):
|
|
char = text[i]
|
|
if char == "'":
|
|
if in_string and i + 1 < len(text) and text[i + 1] == "'":
|
|
i += 2
|
|
continue
|
|
in_string = not in_string
|
|
elif not in_string:
|
|
if char == "(":
|
|
depth += 1
|
|
elif char == ")":
|
|
depth -= 1
|
|
elif char == "," and depth == 0:
|
|
values.append(text[start:i].strip())
|
|
start = i + 1
|
|
i += 1
|
|
|
|
values.append(text[start:].strip())
|
|
return values
|
|
|
|
|
|
def parse_sql_string(token: str) -> str:
|
|
token = token.strip()
|
|
if token[:2].upper() == "N'":
|
|
token = token[1:].lstrip()
|
|
if not token.startswith("'"):
|
|
raise ValueError(f"not a SQL string literal: {token[:80]}")
|
|
|
|
result: list[str] = []
|
|
i = 1
|
|
while i < len(token):
|
|
char = token[i]
|
|
if char == "'":
|
|
if i + 1 < len(token) and token[i + 1] == "'":
|
|
result.append("'")
|
|
i += 2
|
|
continue
|
|
return "".join(result)
|
|
result.append(char)
|
|
i += 1
|
|
|
|
raise ValueError("unterminated SQL string literal")
|
|
|
|
|
|
def timestamp_to_iso(value: str) -> str:
|
|
value = value.strip()
|
|
if re.match(r"^[+-]?\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2}", value):
|
|
return value.replace(" ", "T", 1)
|
|
return value
|
|
|
|
|
|
def parse_function_value(token: str) -> Any:
|
|
name_match = re.match(r"^\s*([A-Z_][A-Z0-9_]*)\s*\(", token, re.I)
|
|
if not name_match:
|
|
return token
|
|
|
|
function_name = name_match.group(1).upper()
|
|
args_text, _ = extract_parenthesized(token, name_match.end() - 1)
|
|
args = split_top_level_csv(args_text)
|
|
if function_name in {"TO_TIMESTAMP", "TO_DATE", "TIMESTAMP"} and args:
|
|
return timestamp_to_iso(str(parse_sql_value(args[0])))
|
|
if function_name in {"TO_CLOB", "TO_NCLOB"} and args:
|
|
return parse_sql_value(args[0])
|
|
return token
|
|
|
|
|
|
def parse_sql_value(token: str) -> Any:
|
|
value = token.strip()
|
|
if not value:
|
|
return None
|
|
if value.upper() == "NULL":
|
|
return None
|
|
if value.startswith("'") or value[:2].upper() == "N'":
|
|
return parse_sql_string(value)
|
|
if re.match(r"^[A-Z_][A-Z0-9_]*\s*\(", value, re.I):
|
|
return parse_function_value(value)
|
|
if re.match(r"^[+-]?\d+$", value):
|
|
return decimal.Decimal(value)
|
|
if re.match(r"^[+-]?(?:\d+\.\d*|\.\d+)(?:[Ee][+-]?\d+)?$", value) or re.match(r"^[+-]?\d+[Ee][+-]?\d+$", value):
|
|
return decimal.Decimal(value)
|
|
return value
|
|
|
|
|
|
def parse_create_columns(statement: str) -> list[str]:
|
|
body, _ = extract_parenthesized(statement, statement.index("("))
|
|
columns: list[str] = []
|
|
for raw_line in body.splitlines():
|
|
match = re.match(r'\s*"([^"]+)"\s+', raw_line)
|
|
if match:
|
|
columns.append(match.group(1))
|
|
if not columns:
|
|
raise RuntimeError("no columns found in CREATE TABLE statement")
|
|
return columns
|
|
|
|
|
|
def parse_insert_columns(columns_text: str | None) -> list[str] | None:
|
|
if not columns_text:
|
|
return None
|
|
return re.findall(r'"([^"]+)"', columns_text)
|
|
|
|
|
|
def parse_insert_rows(statement: str, create_columns: dict[str, list[str]]) -> tuple[str, list[dict[str, Any]]]:
|
|
match = INSERT_VALUES_RE.match(statement)
|
|
if not match:
|
|
raise RuntimeError(f"unsupported INSERT syntax: {statement[:240]}")
|
|
|
|
table = match.group("table").upper()
|
|
columns = parse_insert_columns(match.group("columns")) or create_columns.get(table)
|
|
if not columns:
|
|
raise RuntimeError(f"CREATE TABLE column order not found before INSERT for {table}")
|
|
|
|
rows: list[dict[str, Any]] = []
|
|
values_text = match.group("values").strip()
|
|
cursor = 0
|
|
while cursor < len(values_text):
|
|
while cursor < len(values_text) and values_text[cursor] in " \r\n\t,":
|
|
cursor += 1
|
|
if cursor >= len(values_text) or values_text[cursor] == ";":
|
|
break
|
|
|
|
row_text, cursor = extract_parenthesized(values_text, cursor)
|
|
raw_values = split_top_level_csv(row_text)
|
|
if len(raw_values) != len(columns):
|
|
raise RuntimeError(
|
|
f"{table} INSERT value count mismatch: {len(raw_values)} values for {len(columns)} columns"
|
|
)
|
|
|
|
parsed_values = [parse_sql_value(value) for value in raw_values]
|
|
rows.append(guide_export.normalize_record(columns, parsed_values))
|
|
|
|
return table, rows
|
|
|
|
|
|
def load_data_from_sql_dump(sql_dump_path: Path) -> tuple[dict[str, list[dict[str, Any]]], dict[str, str]]:
|
|
create_columns: dict[str, list[str]] = {}
|
|
data: dict[str, list[dict[str, Any]]] = {file_stem: [] for file_stem in TABLES.values()}
|
|
|
|
for table, statement in iter_target_statements(sql_dump_path):
|
|
if CREATE_RE.search(statement):
|
|
create_columns[table] = parse_create_columns(statement)
|
|
continue
|
|
|
|
if INSERT_RE.search(statement):
|
|
insert_table, rows = parse_insert_rows(statement, create_columns)
|
|
data[TABLES[insert_table]].extend(rows)
|
|
|
|
missing_create = sorted(table for table in TARGET_TABLES if table not in create_columns)
|
|
if missing_create:
|
|
raise RuntimeError(f"CREATE TABLE statements not found: {', '.join(missing_create)}")
|
|
|
|
return data, parse_header(sql_dump_path)
|
|
|
|
|
|
def write_package(
|
|
data: dict[str, list[dict[str, Any]]],
|
|
output_dir: Path,
|
|
nav_assets_dir: Path,
|
|
source: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
for table, file_stem in TABLES.items():
|
|
rows = data[file_stem]
|
|
guide_export.write_json(
|
|
output_dir / f"{file_stem}.json",
|
|
{
|
|
"schemaVersion": SCHEMA_VERSION,
|
|
"sourceTable": table,
|
|
"rowCount": len(rows),
|
|
"rows": rows,
|
|
},
|
|
)
|
|
|
|
indexes = guide_export.build_indexes(data)
|
|
guide_export.write_json(output_dir / "indexes.json", indexes)
|
|
poi_bridge = guide_export.build_poi_bridge(data, nav_assets_dir)
|
|
guide_export.write_json(output_dir / "poi-bridge.json", poi_bridge)
|
|
|
|
manifest = {
|
|
"schemaVersion": SCHEMA_VERSION,
|
|
"generatedAt": dt.datetime.now(dt.timezone.utc).isoformat(),
|
|
"source": source,
|
|
"files": {
|
|
"halls": "halls.json",
|
|
"outlines": "outlines.json",
|
|
"guideStops": "guide-stops.json",
|
|
"exhibits": "exhibits.json",
|
|
"guideContents": "guide-contents.json",
|
|
"pois": "pois.json",
|
|
"indexes": "indexes.json",
|
|
"poiBridge": "poi-bridge.json",
|
|
},
|
|
"counts": {key: len(value) for key, value in data.items()},
|
|
"bridgeStats": poi_bridge["stats"],
|
|
}
|
|
guide_export.write_json(output_dir / "manifest.json", manifest)
|
|
return manifest
|
|
|
|
|
|
def export_package_from_sql_dump(sql_dump_path: Path, output_dir: Path, nav_assets_dir: Path) -> dict[str, Any]:
|
|
data, header = load_data_from_sql_dump(sql_dump_path)
|
|
source = {
|
|
"type": "sql-dump",
|
|
"sqlDumpPath": str(sql_dump_path),
|
|
"sourceHost": header.get("sourceHost"),
|
|
"sourceSchema": header.get("sourceSchema", "NATURE_SGS"),
|
|
"sourceDate": header.get("sourceDate"),
|
|
}
|
|
return write_package(data, output_dir, nav_assets_dir, source)
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description="Export frontend guide static data from a Navicat SQL dump.")
|
|
parser.add_argument("--sql-dump", type=Path, required=True)
|
|
parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT)
|
|
parser.add_argument("--nav-assets", type=Path, default=DEFAULT_NAV_ASSETS)
|
|
args = parser.parse_args()
|
|
|
|
manifest = export_package_from_sql_dump(
|
|
args.sql_dump.resolve(),
|
|
args.output.resolve(),
|
|
args.nav_assets.resolve(),
|
|
)
|
|
print(f"Exported {manifest['schemaVersion']} to {args.output.resolve()}")
|
|
print("Counts:", json.dumps(manifest["counts"], ensure_ascii=False, sort_keys=True))
|
|
print("Bridge:", json.dumps(manifest["bridgeStats"], ensure_ascii=False, sort_keys=True))
|
|
print("Source:", json.dumps(manifest["source"], ensure_ascii=False, sort_keys=True))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|