79 lines
3.0 KiB
Python
79 lines
3.0 KiB
Python
"""
|
||
数据库列修复脚本
|
||
自动检测 ORM 模型与实际数据库表的列差异，仅添加缺失的列。
|
||
用法: python fix_missing_columns.py
|
||
"""
|
||
import pymysql
|
||
|
||
# 数据库连接配置（从 .env 读取）
|
||
DB_CONFIG = {
|
||
"host": "fn.taisan.online",
|
||
"port": 33306,
|
||
"user": "root",
|
||
"password": "taiyi1224",
|
||
"database": "order_flow",
|
||
"charset": "utf8mb4",
|
||
}
|
||
|
||
# 需要检查并修复的表和列定义
|
||
# 格式: { "表名": [ (列名, 列定义SQL), ... ] }
|
||
TABLE_FIXES = {
|
||
"product": [
|
||
("pricing_type", "VARCHAR(32) NULL DEFAULT NULL COMMENT '计价方式：area/weight'"),
|
||
("pricing_unit", "VARCHAR(16) NULL DEFAULT '㎡' COMMENT '计价单位'"),
|
||
("thickness", "VARCHAR(32) NULL DEFAULT NULL COMMENT '厚度'"),
|
||
("weight_gsm", "INT NULL DEFAULT NULL COMMENT '克重(gsm)'"),
|
||
("default_width_m", "DECIMAL(10,4) NULL DEFAULT NULL COMMENT '默认宽度(米)'"),
|
||
("is_default", "INT NOT NULL DEFAULT 0 COMMENT '是否默认规格：0否 1是'"),
|
||
],
|
||
"sales_order_item": [
|
||
("pricing_type", "VARCHAR(32) NULL DEFAULT NULL COMMENT '计价方式：area/weight'"),
|
||
("length_m", "DECIMAL(10,4) NULL DEFAULT NULL COMMENT '长度(米)'"),
|
||
("width_m", "DECIMAL(10,4) NULL DEFAULT NULL COMMENT '宽度(米)'"),
|
||
("area_sqm", "DECIMAL(18,4) NULL DEFAULT NULL COMMENT '面积(平方米)'"),
|
||
("surcharge_detail", "TEXT NULL DEFAULT NULL COMMENT '附加费用明细'"),
|
||
("processing_detail", "TEXT NULL DEFAULT NULL COMMENT '加工费用明细'"),
|
||
("supplier_id", "INT NULL DEFAULT NULL COMMENT '供应商ID'"),
|
||
("supplier_model", "VARCHAR(64) NULL DEFAULT NULL COMMENT '供应商型号'"),
|
||
("price_tier", "VARCHAR(32) NULL DEFAULT NULL COMMENT '价格层级'"),
|
||
],
|
||
}
|
||
|
||
|
||
def get_existing_columns(cursor, table_name: str) -> set[str]:
|
||
"""获取表中已存在的列名集合。"""
|
||
cursor.execute(f"SHOW COLUMNS FROM `{table_name}`")
|
||
return {row[0] for row in cursor.fetchall()}
|
||
|
||
|
||
def main():
|
||
conn = pymysql.connect(**DB_CONFIG)
|
||
cursor = conn.cursor()
|
||
total_added = 0
|
||
|
||
for table_name, columns in TABLE_FIXES.items():
|
||
existing = get_existing_columns(cursor, table_name)
|
||
print(f"\n[{table_name}] 已有 {len(existing)} 列: {', '.join(sorted(existing))}")
|
||
|
||
for col_name, col_def in columns:
|
||
if col_name in existing:
|
||
print(f" ✓ {col_name} 已存在，跳过")
|
||
else:
|
||
sql = f"ALTER TABLE `{table_name}` ADD COLUMN `{col_name}` {col_def}"
|
||
try:
|
||
cursor.execute(sql)
|
||
conn.commit()
|
||
print(f" + {col_name} 已添加")
|
||
total_added += 1
|
||
except Exception as e:
|
||
conn.rollback()
|
||
print(f" ✗ {col_name} 添加失败: {e}")
|
||
|
||
print(f"\n完成！共添加 {total_added} 个缺失列。")
|
||
cursor.close()
|
||
conn.close()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|