43 lines
2.2 KiB
Python
43 lines
2.2 KiB
Python
# -*- coding:utf-8 -*-
|
||
"""P2-③ 建表：opp_demand_atoms / opp_std_demands / opp_demand_domains（幂等 IF NOT EXISTS）。"""
|
||
import asyncio, os, re, sys
|
||
W = "/d/pipeline/pipeline-app"; os.chdir(W); sys.path.insert(0, W)
|
||
from appPublic.folderUtils import ProgramPath
|
||
from appPublic.jsonConfig import getConfig
|
||
from appPublic.event_dispatcher import EventDispatcher
|
||
from sqlor.dbpools import DBPools
|
||
from ahserver.serverenv import ServerEnv
|
||
p = ProgramPath(); c = getConfig(W, {"workdir": W, "ProgramPath": p}); DBPools(c.databases)
|
||
se = ServerEnv(); se.event_dispatcher = EventDispatcher(); se.get_module_dbname = lambda m: "pipeline"
|
||
|
||
NEW = ["opp_demand_atoms", "opp_std_demands", "opp_demand_domains"]
|
||
|
||
async def main():
|
||
ddl = open("/tmp/p2_tables.sql", encoding="utf-8").read()
|
||
async with DBPools().sqlorContext("pipeline") as sor:
|
||
for stmt in [s.strip() for s in ddl.split(";") if s.strip()]:
|
||
m = re.match(r"(?is)CREATE\s+TABLE\s+(?:IF\s+NOT\s+EXISTS\s+)?`?(\w+)`?", stmt)
|
||
if not m or m.group(1) not in NEW:
|
||
continue
|
||
await sor.sqlExe(stmt, {})
|
||
await sor.sqlExe("COMMIT", {})
|
||
print("table OK:", m.group(1))
|
||
for t in NEW:
|
||
r = await sor.sqlExe(
|
||
"SELECT COUNT(*) c FROM information_schema.TABLES WHERE table_schema=DATABASE() AND table_name=${t}$", {"t": t})
|
||
await sor.sqlExe("COMMIT", {})
|
||
print("verify", t, "exists:", (r[0].c if r else 0))
|
||
# 索引
|
||
for name, tbl, cols in [("idx_opp_atoms_cluster", "opp_demand_atoms", "cluster_id"),
|
||
("idx_opp_atoms_snap", "opp_demand_atoms", "snap_id"),
|
||
("idx_opp_std_cluster", "opp_std_demands", "cluster_id, tier"),
|
||
("idx_opp_domains_cluster", "opp_demand_domains", "cluster_id")]:
|
||
try:
|
||
await sor.sqlExe("CREATE INDEX IF NOT EXISTS %s ON %s (%s)" % (name, tbl, cols), {})
|
||
await sor.sqlExe("COMMIT", {})
|
||
print("index OK:", name)
|
||
except Exception as e:
|
||
print("index skip:", name, str(e)[:60])
|
||
|
||
asyncio.run(main())
|