102 lines
3.4 KiB
Python
102 lines
3.4 KiB
Python
"""一次性脚本：从「翁福 数据远传变量.xlsx」解析点位，建立并灌入 mpoint_name 映射表。
|
||
|
||
- 位号(mpoint_id) 由 mqttc.code_from_name 切出，与线上入库逻辑同一套规则，杜绝漂移。
|
||
- point_name = 全名去掉尾部位号；full_name 保留 Excel 原文便于溯源。
|
||
- 幂等：ON CONFLICT(mpoint_id) DO UPDATE，可反复执行。
|
||
|
||
用法:
|
||
python load_mpoint_name.py [Excel路径] [数据库URL]
|
||
默认读取桌面上的《数据远传变量.xlsx》，写入 conf.DATABASE_URL 指向的库；
|
||
可传第二个参数指定目标库（如线上 10.0.11.52）。
|
||
"""
|
||
import os
|
||
import sys
|
||
|
||
import openpyxl
|
||
from sqlalchemy import create_engine, text
|
||
|
||
CUR_DIR = os.path.dirname(os.path.abspath(__file__))
|
||
sys.path.insert(0, CUR_DIR)
|
||
import conf
|
||
from mqttc import code_from_name
|
||
|
||
DEFAULT_XLSX = r"C:\Users\11825\Desktop\01-文档资料\资料\翁福采集\数据远传变量.xlsx"
|
||
TABLE = "wengfu_mpoint_name"
|
||
|
||
DDL = f"""
|
||
CREATE TABLE IF NOT EXISTS {TABLE} (
|
||
mpoint_id varchar PRIMARY KEY,
|
||
point_name text,
|
||
full_name text,
|
||
dtype varchar
|
||
)
|
||
"""
|
||
|
||
UPSERT = f"""
|
||
INSERT INTO {TABLE} (mpoint_id, point_name, full_name, dtype)
|
||
VALUES (:mpoint_id, :point_name, :full_name, :dtype)
|
||
ON CONFLICT (mpoint_id) DO UPDATE
|
||
SET point_name = EXCLUDED.point_name,
|
||
full_name = EXCLUDED.full_name,
|
||
dtype = EXCLUDED.dtype
|
||
"""
|
||
|
||
|
||
def read_items(path):
|
||
"""Excel 为左右两张并排的表：列 A/B 和 D/E 各是 (变量名, 类型)。"""
|
||
wb = openpyxl.load_workbook(path, read_only=True, data_only=True)
|
||
ws = wb["Sheet1"]
|
||
items = []
|
||
for r in list(ws.iter_rows(values_only=True))[1:]: # 跳过表头
|
||
r = list(r)
|
||
if r and r[0]:
|
||
items.append((str(r[0]).strip(), str(r[1]).strip() if len(r) > 1 and r[1] else ''))
|
||
if len(r) > 3 and r[3]:
|
||
items.append((str(r[3]).strip(), str(r[4]).strip() if len(r) > 4 and r[4] else ''))
|
||
return items
|
||
|
||
|
||
def split_row(full_name, dtype):
|
||
code = code_from_name(full_name)
|
||
point = full_name[:len(full_name) - len(code)].rstrip('_') if code and full_name.endswith(code) else full_name
|
||
return {
|
||
"mpoint_id": code,
|
||
"point_name": point or None,
|
||
"full_name": full_name,
|
||
"dtype": dtype or None,
|
||
}
|
||
|
||
|
||
def main():
|
||
path = sys.argv[1] if len(sys.argv) > 1 else DEFAULT_XLSX
|
||
db_url = sys.argv[2] if len(sys.argv) > 2 else conf.DATABASE_URL
|
||
items = read_items(path)
|
||
rows = [split_row(name, typ) for name, typ in items]
|
||
|
||
# 位号唯一性自检
|
||
seen = {}
|
||
collisions = []
|
||
for row in rows:
|
||
mid = row["mpoint_id"]
|
||
if mid in seen and seen[mid] != row["full_name"]:
|
||
collisions.append((mid, seen[mid], row["full_name"]))
|
||
seen[mid] = row["full_name"]
|
||
if collisions:
|
||
print(f"!! 位号冲突 {len(collisions)} 处（不同全名切出同一位号），请先核对：")
|
||
for mid, a, b in collisions[:20]:
|
||
print(f" {mid}: {a} <> {b}")
|
||
sys.exit(1)
|
||
|
||
engine = create_engine(db_url, connect_args={'connect_timeout': 10})
|
||
with engine.begin() as conn:
|
||
conn.execute(text(DDL))
|
||
for row in rows:
|
||
conn.execute(text(UPSERT), row)
|
||
total = conn.execute(text(f"SELECT COUNT(*) FROM {TABLE}")).scalar()
|
||
|
||
print(f"目标库写入完成：读取 {len(items)} 条，去重后位号 {len(seen)} 个，{TABLE} 表内现有 {total} 行。")
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main()
|