196 lines
6.6 KiB
Python
196 lines
6.6 KiB
Python
"""Step 1: 全量采集ERP表结构到 erp_schema
|
|
通过 erp-api-gateway 采集985张表的字段信息
|
|
运行: python3 scripts/collect_erp_schema.py
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
import json
|
|
import urllib.request
|
|
import urllib.error
|
|
import logging
|
|
import time
|
|
from datetime import datetime
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from app.database import get_session_local
|
|
from sqlalchemy import text
|
|
|
|
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
|
|
logger = logging.getLogger("erp_schema_collect")
|
|
|
|
ERP_API_BASE = "http://127.0.0.1:8300/api/v1"
|
|
ERP_API_KEY = os.getenv("ERP_API_KEY", "erp-gateway-key-bhwl-2026")
|
|
|
|
HEADERS = {
|
|
"X-API-Key": ERP_API_KEY,
|
|
"User-Agent": "CMA-ERP-SCHEMA/1.0",
|
|
"Content-Type": "application/json",
|
|
}
|
|
|
|
|
|
def api_get(path: str) -> dict:
|
|
"""调用ERP API"""
|
|
url = f"{ERP_API_BASE}{path}"
|
|
req = urllib.request.Request(url, headers=HEADERS)
|
|
with urllib.request.urlopen(req, timeout=30) as resp:
|
|
return json.loads(resp.read().decode())
|
|
|
|
|
|
def get_table_columns(table_name: str) -> list:
|
|
"""通过 INFORMATION_SCHEMA 查询表字段"""
|
|
sql = f"""
|
|
SELECT
|
|
COLUMN_NAME,
|
|
DATA_TYPE,
|
|
CHARACTER_MAXIMUM_LENGTH,
|
|
IS_NULLABLE,
|
|
COLUMN_DEFAULT
|
|
FROM INFORMATION_SCHEMA.COLUMNS
|
|
WHERE TABLE_NAME = '{table_name}'
|
|
ORDER BY ORDINAL_POSITION
|
|
"""
|
|
params = json.dumps({"sql": sql}).encode()
|
|
req = urllib.request.Request(
|
|
f"{ERP_API_BASE}/query",
|
|
data=params,
|
|
headers=HEADERS,
|
|
method="POST"
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=30) as resp:
|
|
data = json.loads(resp.read().decode())
|
|
return data.get("data", [])
|
|
except Exception as e:
|
|
logger.warning(f" ⚠️ {table_name}: 查询失败 - {e}")
|
|
return []
|
|
|
|
|
|
def get_row_count(table_name: str) -> int:
|
|
"""获取表行数"""
|
|
sql = f"SELECT COUNT(*) as cnt FROM [{table_name}]"
|
|
params = json.dumps({"sql": sql}).encode()
|
|
req = urllib.request.Request(
|
|
f"{ERP_API_BASE}/query",
|
|
data=params,
|
|
headers=HEADERS,
|
|
method="POST"
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=30) as resp:
|
|
data = json.loads(resp.read().decode())
|
|
rows = data.get("data", [])
|
|
return rows[0]["cnt"] if rows else 0
|
|
except:
|
|
return -1 # 未知
|
|
|
|
|
|
def main():
|
|
db = get_session_local()()
|
|
|
|
try:
|
|
# 1. 获取全部表名
|
|
logger.info("📡 获取ERP全量表名列表...")
|
|
tables_data = api_get("/tables")
|
|
all_tables = tables_data.get("tables", [])
|
|
logger.info(f" 共 {len(all_tables)} 张表")
|
|
|
|
# 2. 获取已采集的表名
|
|
existing = set()
|
|
try:
|
|
rows = db.execute(text("SELECT table_name FROM erp_schema")).fetchall()
|
|
existing = set(row[0] for row in rows)
|
|
except:
|
|
pass
|
|
logger.info(f" 已采集 {len(existing)} 张表，待采集 {len(all_tables) - len(existing)} 张")
|
|
|
|
# 3. 逐表采集
|
|
collected = 0
|
|
skipped = 0
|
|
errors = 0
|
|
|
|
for i, table_name in enumerate(all_tables):
|
|
if table_name in existing:
|
|
skipped += 1
|
|
continue
|
|
|
|
# 进度显示
|
|
if (i + 1) % 50 == 0:
|
|
logger.info(f" 进度: {i+1}/{len(all_tables)} (已采{collected}, 跳过{skipped}, 错误{errors})")
|
|
|
|
# 采集字段
|
|
columns = get_table_columns(table_name)
|
|
if not columns:
|
|
errors += 1
|
|
# 即使查不到字段也记录一个空记录，避免重复查
|
|
collect_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
fields_json = "[]"
|
|
db.execute(
|
|
text("""
|
|
INSERT INTO erp_schema (table_name, total_rows, field_count, fields_json, created_at, updated_at)
|
|
VALUES (:tn, :tr, :fc, :fj, NOW(), NOW())
|
|
ON DUPLICATE KEY UPDATE fields_json=:fj2, total_rows=:tr2, updated_at=NOW()
|
|
"""),
|
|
{"tn": table_name, "tr": -1, "fc": 0, "fj": fields_json, "fj2": fields_json, "tr2": -1}
|
|
)
|
|
db.commit()
|
|
continue
|
|
|
|
# 获取行数
|
|
row_count = get_row_count(table_name)
|
|
field_count = len(columns)
|
|
|
|
fields = []
|
|
for col in columns:
|
|
fields.append({
|
|
"name": col.get("COLUMN_NAME", ""),
|
|
"type": col.get("DATA_TYPE", ""),
|
|
"max_length": col.get("CHARACTER_MAXIMUM_LENGTH"),
|
|
"nullable": col.get("IS_NULLABLE", "YES"),
|
|
"default": col.get("COLUMN_DEFAULT"),
|
|
})
|
|
fields_json = json.dumps(fields, ensure_ascii=False)
|
|
|
|
# 写入 erp_schema
|
|
try:
|
|
collect_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
db.execute(
|
|
text("""
|
|
INSERT INTO erp_schema (table_name, total_rows, field_count, fields_json, created_at, updated_at)
|
|
VALUES (:tn, :tr, :fc, :fj, NOW(), NOW())
|
|
ON DUPLICATE KEY UPDATE total_rows=:tr2, field_count=:fc2, fields_json=:fj2, updated_at=NOW()
|
|
"""),
|
|
{"tn": table_name, "tr": row_count, "fc": field_count, "fj": fields_json,
|
|
"tr2": row_count, "fc2": field_count, "fj2": fields_json}
|
|
)
|
|
db.commit()
|
|
collected += 1
|
|
except Exception as e:
|
|
db.rollback()
|
|
logger.warning(f" ⚠️ {table_name}: 写入数据库失败 - {e}")
|
|
errors += 1
|
|
|
|
# 限流：不要打太快
|
|
if collected > 0 and collected % 10 == 0:
|
|
time.sleep(0.5)
|
|
|
|
# 4. 统计
|
|
total = db.execute(text("SELECT COUNT(*) FROM erp_schema")).scalar()
|
|
with_data = db.execute(text("SELECT COUNT(*) FROM erp_schema WHERE field_count > 0")).scalar()
|
|
logger.info(f"\n🎉 采集完成!")
|
|
logger.info(f" 总计: {total} 张表 (erp_schema)")
|
|
logger.info(f" 有字段信息: {with_data} 张")
|
|
logger.info(f" 本次新增: {collected} 张")
|
|
logger.info(f" 跳过(已存在): {skipped} 张")
|
|
logger.info(f" 错误: {errors} 张")
|
|
|
|
except Exception as e:
|
|
logger.error(f"采集失败: {e}", exc_info=True)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|