101 lines
3.6 KiB
Python
101 lines
3.6 KiB
Python
#!/usr/bin/env python3
|
||
"""统一所有日期列格式为 YYYY-MM-DD HH:mm:00"""
|
||
import re, pymysql
|
||
from config import DB_CONFIG
|
||
|
||
# 各表需要转换的列
|
||
UPDATE_PLANS = {
|
||
'domestic_orders': {
|
||
'alter_date_cols': ['日期'], # DATE→VARCHAR
|
||
'reformat_cols': ['实际到货日期', '预计到货日期'], # 已有VARCHAR值重新格式化
|
||
},
|
||
'overseas_orders': {
|
||
'alter_date_cols': ['日期'],
|
||
'reformat_cols': ['实际到货日期', '预计到货日期'],
|
||
},
|
||
'packaging_orders': {
|
||
'alter_date_cols': ['日期'],
|
||
'reformat_cols': ['到货日期'],
|
||
},
|
||
}
|
||
|
||
def normalize_date(val):
|
||
"""将各种日期格式转为 YYYY-MM-DD 00:00:00"""
|
||
if val is None:
|
||
return None
|
||
s = str(val).strip()
|
||
if not s:
|
||
return None
|
||
|
||
# 匹配 YYYY-MM-DD HH:MM:SS 已有格式
|
||
m = re.match(r'(\d{4})-(\d{1,2})-(\d{1,2})\s+(\d{1,2}):(\d{1,2})(:\d{2})?', s)
|
||
if m:
|
||
return f"{m.group(1)}-{int(m.group(2)):02d}-{int(m.group(3)):02d} {int(m.group(4)):02d}:{int(m.group(5)):02d}:00"
|
||
|
||
# 匹配 YYYY/MM/DD 或 YYYY-MM-DD
|
||
m = re.match(r'(\d{4})[-/](\d{1,2})[-/](\d{1,2})', s)
|
||
if m:
|
||
return f"{m.group(1)}-{int(m.group(2)):02d}-{int(m.group(3)):02d} 00:00:00"
|
||
|
||
return s # 无法识别则保持原样
|
||
|
||
def main():
|
||
conn = pymysql.connect(**DB_CONFIG)
|
||
cur = conn.cursor()
|
||
|
||
for tbl, plan in UPDATE_PLANS.items():
|
||
print(f"\n=== {tbl} ===")
|
||
|
||
# 1. 先处理 VARCHAR 列的格式重写
|
||
for col in plan['reformat_cols']:
|
||
# 检查列是否存在
|
||
cur.execute(f"SHOW COLUMNS FROM {tbl} WHERE Field=%s", (col,))
|
||
if not cur.fetchone():
|
||
print(f" 列 {col} 不存在，跳过")
|
||
continue
|
||
|
||
# 获取所有值，逐行更新
|
||
cur.execute(f"SELECT id, `{col}` FROM {tbl} WHERE `{col}` IS NOT NULL AND `{col}` != ''")
|
||
rows = cur.fetchall()
|
||
updated = 0
|
||
for row_id, old_val in rows:
|
||
new_val = normalize_date(old_val)
|
||
if new_val and new_val != old_val:
|
||
cur.execute(f"UPDATE {tbl} SET `{col}`=%s WHERE id=%s", (new_val, row_id))
|
||
updated += 1
|
||
conn.commit()
|
||
print(f" {col}: {updated}/{len(rows)} 行已格式化")
|
||
|
||
# 2. 将 DATE 类型列转 VARCHAR 并格式化
|
||
for col in plan['alter_date_cols']:
|
||
cur.execute(f"SHOW COLUMNS FROM {tbl} WHERE Field=%s", (col,))
|
||
col_info = cur.fetchone()
|
||
if not col_info:
|
||
print(f" 列 {col} 不存在，跳过")
|
||
continue
|
||
|
||
current_type = col_info[1].upper()
|
||
if current_type.startswith('VARCHAR'):
|
||
print(f" {col}: 已是 VARCHAR，跳过")
|
||
continue
|
||
|
||
# 先获取所有值格式化
|
||
cur.execute(f"SELECT id, `{col}` FROM {tbl} WHERE `{col}` IS NOT NULL")
|
||
rows = cur.fetchall()
|
||
for row_id, old_val in rows:
|
||
new_val = normalize_date(str(old_val))
|
||
if new_val:
|
||
cur.execute(f"UPDATE {tbl} SET `{col}`=%s WHERE id=%s", (new_val, row_id))
|
||
conn.commit()
|
||
|
||
# 改字段类型为 VARCHAR(20)
|
||
cur.execute(f"ALTER TABLE {tbl} MODIFY `{col}` VARCHAR(20) DEFAULT NULL")
|
||
conn.commit()
|
||
print(f" {col}: DATE→VARCHAR(20)，{len(rows)} 行已格式化")
|
||
|
||
conn.close()
|
||
print("\n✅ 全部日期格式转换完成")
|
||
|
||
if __name__ == "__main__":
|
||
main()
|