49 lines
2.1 KiB
Python
49 lines
2.1 KiB
Python
# -*- coding:utf-8 -*-
|
||
"""P1 建表：opp_mining_batches / opp_demand_snap / opp_clusters（幂等 IF NOT EXISTS + 增量索引）。"""
|
||
import asyncio, os, sys
|
||
WORKDIR = "/d/pipeline/pipeline-app"
|
||
os.chdir(WORKDIR); sys.path.insert(0, WORKDIR)
|
||
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()
|
||
config = getConfig(WORKDIR, NS={'workdir': WORKDIR, 'ProgramPath': p})
|
||
DBPools(config.databases)
|
||
se = ServerEnv(); se.event_dispatcher = EventDispatcher(); se.get_module_dbname = lambda m: 'pipeline'
|
||
|
||
DDL_FILE = "/tmp/p1_tables.sql"
|
||
INDEXES = [
|
||
("idx_opp_snap_batch_src", "opp_demand_snap", "batch_id, src_id"),
|
||
("idx_opp_snap_cluster", "opp_demand_snap", "cluster_id"),
|
||
("idx_opp_clusters_batch", "opp_clusters", "batch_id, org_id"),
|
||
("idx_opp_batches_org", "opp_mining_batches", "org_id, status"),
|
||
]
|
||
|
||
|
||
async def main():
|
||
sql = open(DDL_FILE, encoding="utf-8").read()
|
||
async with DBPools().sqlorContext("pipeline") as sor:
|
||
for stmt in [s.strip() for s in sql.split(";") if s.strip()]:
|
||
await sor.sqlExe(stmt, {})
|
||
await sor.sqlExe("COMMIT", {})
|
||
print("table OK:", stmt.split("(")[0].strip()[:50])
|
||
# 增量索引（IF NOT EXISTS 语法 MariaDB 支持）
|
||
for name, table, cols in INDEXES:
|
||
try:
|
||
await sor.sqlExe("CREATE INDEX IF NOT EXISTS %s ON %s (%s)" % (name, table, cols), {})
|
||
await sor.sqlExe("COMMIT", {})
|
||
print("index OK:", name)
|
||
except Exception as e:
|
||
print("index SKIP:", name, str(e)[:80])
|
||
# 验证三表存在
|
||
for t in ("opp_mining_batches", "opp_demand_snap", "opp_clusters"):
|
||
r = await sor.sqlExe(
|
||
"SELECT COUNT(*) c FROM information_schema.TABLES WHERE table_schema=DATABASE() AND table_name=${t}$", {"t": t})
|
||
print("verify", t, "exists:", (r[0].c if r else 0))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|