75 lines
2.2 KiB
Python
75 lines
2.2 KiB
Python
|
"""补齐 breeding 全部建表（幂等，逐句隔离）。
|
|||
|
|
|
|||
|
|
用正则按顶层 SQL 关键字（CREATE TABLE / CREATE INDEX / COMMENT ON / ALTER TABLE ...）
|
|||
|
|
切分，避免部分 CREATE TABLE 不以分号结尾导致多表被合并成一条语句的问题。
|
|||
|
|
每张 CREATE TABLE 成功/失败均打印表名与真实异常（repr 保证 ascii 安全）。
|
|||
|
|
"""
|
|||
|
|
import re
|
|||
|
|
from pathlib import Path
|
|||
|
|
|
|||
|
|
from sqlalchemy import text
|
|||
|
|
|
|||
|
|
from app.core.database import engine
|
|||
|
|
|
|||
|
|
SQL_PATH = Path(r"d:\dpb\dpb\backend\sql\bre_tables.sql")
|
|||
|
|
|
|||
|
|
CHECK_TABLES = [
|
|||
|
|
"bre_trait", "bre_tree", "bre_germplasm",
|
|||
|
|
"bre_prediction", "bre_prediction_value",
|
|||
|
|
"bre_combining_ability", "bre_statistics_job",
|
|||
|
|
]
|
|||
|
|
|
|||
|
|
|
|||
|
|
def classify(s: str) -> str:
|
|||
|
|
u = s.upper()
|
|||
|
|
if u.startswith("CREATE TABLE"):
|
|||
|
|
return "CREATE"
|
|||
|
|
if "CREATE INDEX" in u:
|
|||
|
|
return "IDX"
|
|||
|
|
if u.startswith("COMMENT"):
|
|||
|
|
return "CMT"
|
|||
|
|
return "OTHER"
|
|||
|
|
|
|||
|
|
|
|||
|
|
def main() -> None:
|
|||
|
|
sql = SQL_PATH.read_text(encoding="utf-8")
|
|||
|
|
parts = re.split(
|
|||
|
|
r"(?=(?:CREATE TABLE|CREATE INDEX|COMMENT ON|ALTER TABLE|DROP TABLE|DROP INDEX))",
|
|||
|
|
sql,
|
|||
|
|
flags=re.IGNORECASE,
|
|||
|
|
)
|
|||
|
|
stmts: list[str] = []
|
|||
|
|
for p in parts:
|
|||
|
|
s = p.strip().rstrip(";").strip()
|
|||
|
|
if s and not s.startswith("--"):
|
|||
|
|
stmts.append(s)
|
|||
|
|
|
|||
|
|
ok = fail = 0
|
|||
|
|
with engine.connect() as conn:
|
|||
|
|
for s in stmts:
|
|||
|
|
kind = classify(s)
|
|||
|
|
try:
|
|||
|
|
conn.execute(text(s))
|
|||
|
|
conn.commit()
|
|||
|
|
ok += 1
|
|||
|
|
if kind == "CREATE":
|
|||
|
|
m = re.search(r"breed_\w+", s)
|
|||
|
|
print("CREATE OK", m.group(0) if m else s[:30])
|
|||
|
|
except Exception as e: # noqa: BLE001
|
|||
|
|
conn.rollback()
|
|||
|
|
fail += 1
|
|||
|
|
if kind in ("CREATE", "IDX", "CMT"):
|
|||
|
|
print(kind, "FAIL", s[:38].replace("\n", " "), "|", repr(e)[:140])
|
|||
|
|
with engine.connect() as conn:
|
|||
|
|
for t in CHECK_TABLES:
|
|||
|
|
try:
|
|||
|
|
conn.execute(text(f"select 1 from {t} limit 1"))
|
|||
|
|
print("EXISTS", t)
|
|||
|
|
except Exception: # noqa: BLE001
|
|||
|
|
print("MISSING", t)
|
|||
|
|
print(f"DDL done: ok={ok} fail={fail}")
|
|||
|
|
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
main()
|