fix(compute): 图标中心 URL 统一 resolve 补全 + 生产库运维脚本
- icon_to_dict/options/create/update 用 resolve_url 补全相对路径 - 新增生产库结构对比/索引修复/建表脚本
This commit is contained in:
Pine
@@ -0,0 +1,170 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""对比生产数据库和开发数据库的表结构一致性。"""
|
||||
import sys
|
||||
|
||||
# 数据库连接配置
|
||||
PROD_HOST = "124.221.143.140"
|
||||
DEV_HOST = "192.168.1.3"
|
||||
PORT = 8091
|
||||
USER = "opc"
|
||||
PASSWORD = "jjjsgysyujkwjwgb"
|
||||
DB_NAME = "opc"
|
||||
|
||||
PROD_URL = f"mysql+pymysql://{USER}:{PASSWORD}@{PROD_HOST}:{PORT}/{DB_NAME}?charset=utf8mb4"
|
||||
DEV_URL = f"mysql+pymysql://{USER}:{PASSWORD}@{DEV_HOST}:{PORT}/{DB_NAME}?charset=utf8mb4"
|
||||
|
||||
print(f"生产数据库: {PROD_HOST}:{PORT}/{DB_NAME}")
|
||||
print(f"开发数据库: {DEV_HOST}:{PORT}/{DB_NAME}")
|
||||
print()
|
||||
|
||||
from sqlalchemy import create_engine, inspect
|
||||
|
||||
def get_db_structure(url):
|
||||
"""获取数据库的完整结构信息。"""
|
||||
engine = create_engine(url, pool_pre_ping=True, connect_args={'connect_timeout': 10})
|
||||
inspector = inspect(engine)
|
||||
|
||||
structure = {}
|
||||
|
||||
# 获取所有表
|
||||
tables = inspector.get_table_names()
|
||||
structure['tables'] = sorted(tables)
|
||||
|
||||
# 获取每个表的详细结构
|
||||
table_details = {}
|
||||
for table in tables:
|
||||
cols = inspector.get_columns(table)
|
||||
col_info = {}
|
||||
for col in cols:
|
||||
col_info[col['name']] = {
|
||||
'type': str(col['type']),
|
||||
'nullable': col.get('nullable', True),
|
||||
'default': str(col.get('default', '')),
|
||||
}
|
||||
# 获取主键
|
||||
try:
|
||||
pk_constraint = inspector.get_pk_constraint(table)
|
||||
if isinstance(pk_constraint, dict):
|
||||
primary_key = pk_constraint.get('constrained_columns', [])
|
||||
else:
|
||||
primary_key = list(pk_constraint) if pk_constraint else []
|
||||
except Exception:
|
||||
primary_key = []
|
||||
|
||||
table_details[table] = {
|
||||
'columns': col_info,
|
||||
'primary_key': primary_key,
|
||||
'indexes': [(idx['name'], tuple(idx['column_names']), idx.get('unique', False)) for idx in inspector.get_indexes(table)],
|
||||
}
|
||||
|
||||
structure['table_details'] = table_details
|
||||
engine.dispose()
|
||||
return structure
|
||||
|
||||
def compare_structures(prod, dev):
|
||||
"""对比两个数据库结构，返回差异列表。"""
|
||||
diffs = []
|
||||
|
||||
# 1. 对比表列表
|
||||
prod_tables = set(prod['tables'])
|
||||
dev_tables = set(dev['tables'])
|
||||
|
||||
only_in_prod = prod_tables - dev_tables
|
||||
only_in_dev = dev_tables - prod_tables
|
||||
|
||||
if only_in_prod:
|
||||
diffs.append(f"⚠️ 仅在生产数据库存在的表 ({len(only_in_prod)}): {sorted(only_in_prod)}")
|
||||
if only_in_dev:
|
||||
diffs.append(f"⚠️ 仅在开发数据库存在的表 ({len(only_in_dev)}): {sorted(only_in_dev)}")
|
||||
|
||||
# 2. 对比共有表的结构
|
||||
common_tables = prod_tables & dev_tables
|
||||
for table in sorted(common_tables):
|
||||
prod_detail = prod['table_details'][table]
|
||||
dev_detail = dev['table_details'][table]
|
||||
|
||||
# 对比列
|
||||
prod_cols = set(prod_detail['columns'].keys())
|
||||
dev_cols = set(dev_detail['columns'].keys())
|
||||
|
||||
cols_only_in_prod = prod_cols - dev_cols
|
||||
cols_only_in_dev = dev_cols - prod_cols
|
||||
|
||||
if cols_only_in_prod:
|
||||
diffs.append(f" 📋 表 [{table}]: 仅在生产存在的列: {sorted(cols_only_in_prod)}")
|
||||
if cols_only_in_dev:
|
||||
diffs.append(f" 📋 表 [{table}]: 仅在开发存在的列: {sorted(cols_only_in_dev)}")
|
||||
|
||||
# 对比共有列的属性
|
||||
common_cols = prod_cols & dev_cols
|
||||
for col in sorted(common_cols):
|
||||
prod_col = prod_detail['columns'][col]
|
||||
dev_col = dev_detail['columns'][col]
|
||||
|
||||
if prod_col['type'] != dev_col['type']:
|
||||
diffs.append(f" 🔧 表 [{table}].列 [{col}]: 类型不一致 - 生产={prod_col['type']}, 开发={dev_col['type']}")
|
||||
if prod_col['nullable'] != dev_col['nullable']:
|
||||
diffs.append(f" 🔧 表 [{table}].列 [{col}]: 可空性不一致 - 生产={prod_col['nullable']}, 开发={dev_col['nullable']}")
|
||||
|
||||
# 对比主键
|
||||
if prod_detail['primary_key'] != dev_detail['primary_key']:
|
||||
diffs.append(f" 🔑 表 [{table}]: 主键不一致 - 生产={prod_detail['primary_key']}, 开发={dev_detail['primary_key']}")
|
||||
|
||||
# 对比索引
|
||||
prod_indexes = set(prod_detail['indexes'])
|
||||
dev_indexes = set(dev_detail['indexes'])
|
||||
idx_only_in_prod = prod_indexes - dev_indexes
|
||||
idx_only_in_dev = dev_indexes - prod_indexes
|
||||
if idx_only_in_prod:
|
||||
diffs.append(f" 📊 表 [{table}]: 仅在生产存在的索引: {sorted(idx_only_in_prod)}")
|
||||
if idx_only_in_dev:
|
||||
diffs.append(f" 📊 表 [{table}]: 仅在开发存在的索引: {sorted(idx_only_in_dev)}")
|
||||
|
||||
return diffs
|
||||
|
||||
try:
|
||||
print("正在连接生产数据库...")
|
||||
prod_structure = get_db_structure(PROD_URL)
|
||||
print(f"✓ 生产数据库: {len(prod_structure['tables'])} 张表")
|
||||
|
||||
print("正在连接开发数据库...")
|
||||
dev_structure = get_db_structure(DEV_URL)
|
||||
print(f"✓ 开发数据库: {len(dev_structure['tables'])} 张表")
|
||||
print()
|
||||
|
||||
diffs = compare_structures(prod_structure, dev_structure)
|
||||
|
||||
if not diffs:
|
||||
print("✅ 生产数据库和开发数据库结构完全一致！")
|
||||
else:
|
||||
print(f"❌ 发现 {len(diffs)} 处差异：")
|
||||
print()
|
||||
for diff in diffs:
|
||||
print(diff)
|
||||
|
||||
# 输出表数量统计
|
||||
print()
|
||||
print("=" * 60)
|
||||
print(f"表总数: 生产={len(prod_structure['tables'])}, 开发={len(dev_structure['tables'])}")
|
||||
print(f"共有表: {len(set(prod_structure['tables']) & set(dev_structure['tables']))}")
|
||||
|
||||
# 重点检查新增的 compute_series 和 compute_icons 表
|
||||
print()
|
||||
print("=" * 60)
|
||||
print("重点检查新增表：")
|
||||
for new_table in ['compute_series', 'compute_icons']:
|
||||
in_prod = new_table in prod_structure['tables']
|
||||
in_dev = new_table in dev_structure['tables']
|
||||
print(f" {new_table}: 生产={'✓' if in_prod else '✗'}, 开发={'✓' if in_dev else '✗'}")
|
||||
if in_prod and in_dev:
|
||||
prod_cols = list(prod_structure['table_details'][new_table]['columns'].keys())
|
||||
dev_cols = list(dev_structure['table_details'][new_table]['columns'].keys())
|
||||
print(f" 生产列 ({len(prod_cols)}): {prod_cols}")
|
||||
print(f" 开发列 ({len(dev_cols)}): {dev_cols}")
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ 错误: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
sys.exit(1)
|
||||
Reference in New Issue
Block a user