194 lines
6.6 KiB
Python
194 lines
6.6 KiB
Python
#!/usr/bin/env python3
|
||
"""导入国内下单表到数据库"""
|
||
import os, sys, re
|
||
import openpyxl
|
||
import pymysql
|
||
from decimal import Decimal
|
||
from config import DB_CONFIG
|
||
|
||
EXCEL_PATH = "/home/gxd/.openclaw/workspace-dingtalk-assistant/media/inbound/国内下单表（跟单）6-17-1781682618512.xlsx"
|
||
|
||
# 59列字段定义
|
||
COLUMNS = [
|
||
("日期", "DATE"),
|
||
("计划下单时间", "DATETIME"),
|
||
("新旧品", "VARCHAR(50)"),
|
||
("货品编号", "VARCHAR(100)"),
|
||
("货品名称", "VARCHAR(500)"),
|
||
("下单备注", "TEXT"),
|
||
("单价", "DECIMAL(12,4)"),
|
||
("是否审核价格", "VARCHAR(50)"),
|
||
("数量", "INT"),
|
||
("金额", "DECIMAL(14,2)"),
|
||
("供货商", "VARCHAR(200)"),
|
||
("仓库", "VARCHAR(200)"),
|
||
("金蝶采购单", "VARCHAR(50)"),
|
||
("采购员", "VARCHAR(500)"),
|
||
("下单人", "VARCHAR(100)"),
|
||
("进度", "VARCHAR(100)"),
|
||
("旺店通生产单号", "VARCHAR(100)"),
|
||
("是否已录入旺店通", "VARCHAR(50)"),
|
||
("采购开发", "VARCHAR(200)"),
|
||
("实际到货日期", "VARCHAR(50)"),
|
||
("实际到货数量辅助", "INT"),
|
||
("辅助1", "VARCHAR(100)"),
|
||
("实际到货数量", "INT"),
|
||
("未到货数", "INT"),
|
||
("预计到货日期", "VARCHAR(50)"),
|
||
("金蝶入库", "VARCHAR(50)"),
|
||
("状态", "VARCHAR(50)"),
|
||
]
|
||
|
||
# 1-31 到货日明细
|
||
for i in range(1, 32):
|
||
COLUMNS.append((str(i), "INT"))
|
||
|
||
# 父记录
|
||
COLUMNS.append(("父记录", "VARCHAR(200)"))
|
||
|
||
CREATE_SQL = """
|
||
CREATE TABLE IF NOT EXISTS domestic_orders (
|
||
id INT AUTO_INCREMENT PRIMARY KEY,
|
||
{fields},
|
||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
|
||
"""
|
||
|
||
def main():
|
||
conn = pymysql.connect(**DB_CONFIG)
|
||
cur = conn.cursor()
|
||
|
||
# 1. 建表
|
||
field_defs = ", ".join([f"`{name}` {dtype} DEFAULT NULL" for name, dtype in COLUMNS])
|
||
sql = CREATE_SQL.format(fields=field_defs)
|
||
cur.execute(sql)
|
||
print("✅ 表 domestic_orders 已就绪")
|
||
|
||
# 2. 先清空（增量导入用 TRUNCATE 避免重复）
|
||
cur.execute("TRUNCATE TABLE domestic_orders")
|
||
print("✅ 已清空旧数据")
|
||
|
||
# 3. 读 Excel
|
||
wb = openpyxl.load_workbook(EXCEL_PATH, data_only=True)
|
||
ws = wb.active
|
||
|
||
# 验证表头
|
||
headers = []
|
||
for col in range(1, ws.max_column + 1):
|
||
val = ws.cell(row=1, column=col).value
|
||
headers.append(str(val).strip() if val else "")
|
||
|
||
expected = [c[0] for c in COLUMNS]
|
||
mismatch = []
|
||
for i, (h, e) in enumerate(zip(headers, expected)):
|
||
if h != e:
|
||
mismatch.append(f" 列{i+1}: 期望'{e}' 实际'{h}'")
|
||
if mismatch:
|
||
print("⚠️ 表头不匹配，需检查：")
|
||
for m in mismatch:
|
||
print(m)
|
||
return
|
||
|
||
# 4. 导入数据
|
||
placeholders = ", ".join(["%s"] * len(COLUMNS))
|
||
col_names = ", ".join([f"`{c[0]}`" for c in COLUMNS])
|
||
insert_sql = f"INSERT INTO domestic_orders ({col_names}) VALUES ({placeholders})"
|
||
|
||
# 各列数据类型（用于清洗）
|
||
col_types = [c[1] for c in COLUMNS] # 0-based
|
||
|
||
def clean_val(val, dtype):
|
||
"""根据目标类型清洗单元格值"""
|
||
if val is None:
|
||
return None
|
||
s = str(val).strip()
|
||
if not s:
|
||
return None
|
||
if dtype.startswith('INT'):
|
||
# 转整数，去除非数字字符
|
||
nums = re.sub(r'[^0-9\-]', '', s)
|
||
if nums and nums != '-':
|
||
try:
|
||
return int(nums)
|
||
except:
|
||
return None
|
||
# 检查是否日期格式的数值
|
||
clean = re.sub(r'[^0-9]', '', s)
|
||
if clean:
|
||
try: return int(clean)
|
||
except: return None
|
||
return None
|
||
if dtype.startswith('DECIMAL'):
|
||
clean = re.sub(r'[^0-9\.\-]', '', s)
|
||
if clean and clean != '.':
|
||
try:
|
||
return float(clean)
|
||
except:
|
||
return None
|
||
return None
|
||
if dtype == 'DATE':
|
||
# 尝试解析日期
|
||
date_match = re.match(r'(\d{4})[\-\/](\d{1,2})[\-\/](\d{1,2})', s)
|
||
if date_match:
|
||
return f"{date_match.group(1)}-{int(date_match.group(2)):02d}-{int(date_match.group(3)):02d}"
|
||
return None if not s else s
|
||
if dtype == 'DATETIME':
|
||
dt_match = re.match(r'(\d{4})[\-\/](\d{1,2})[\-\/](\d{1,2})\s+(\d{1,2}):(\d{1,2})(:(\d{1,2}))?', s)
|
||
if dt_match:
|
||
return f"{dt_match.group(1)}-{int(dt_match.group(2)):02d}-{int(dt_match.group(3)):02d} {int(dt_match.group(4)):02d}:{dt_match.group(5)}:00"
|
||
return None if not s else s
|
||
return s if s else None
|
||
|
||
total = 0
|
||
batch = []
|
||
field_index = {name: idx for idx, (name, _) in enumerate(COLUMNS)}
|
||
for row_idx in range(2, ws.max_row + 1):
|
||
row_data = []
|
||
has_data = False
|
||
for col_idx in range(1, len(COLUMNS) + 1):
|
||
val = ws.cell(row=row_idx, column=col_idx).value
|
||
# 清洗
|
||
cleaned = clean_val(val, col_types[col_idx - 1])
|
||
row_data.append(cleaned)
|
||
if cleaned is not None:
|
||
has_data = True
|
||
if not has_data:
|
||
continue
|
||
# 导入口径与订单接口一致：金蝶入库默认“否”，未到货数由订单数量和实际到货数量计算。
|
||
if row_data[field_index['金蝶入库']] in (None, ''):
|
||
row_data[field_index['金蝶入库']] = '否'
|
||
if row_data[field_index['未到货数']] in (None, ''):
|
||
order_qty = row_data[field_index['数量']] or 0
|
||
arrived_qty = row_data[field_index['实际到货数量']] or 0
|
||
row_data[field_index['未到货数']] = order_qty - arrived_qty
|
||
batch.append(tuple(row_data))
|
||
total += 1
|
||
|
||
if len(batch) >= 500:
|
||
cur.executemany(insert_sql, batch)
|
||
conn.commit()
|
||
print(f" 已导入 {total} 行...")
|
||
batch = []
|
||
|
||
if batch:
|
||
cur.executemany(insert_sql, batch)
|
||
conn.commit()
|
||
|
||
print(f"✅ 导入完成，共 {total} 行")
|
||
|
||
# 5. 验证
|
||
cur.execute("SELECT COUNT(*) FROM domestic_orders")
|
||
cnt = cur.fetchone()[0]
|
||
print(f"📊 数据库记录数: {cnt}")
|
||
|
||
# 显示最新几条
|
||
cur.execute("SELECT `日期`,`货品编号`,`货品名称`,`数量`,`供货商`,`状态` FROM domestic_orders WHERE `日期` IS NOT NULL ORDER BY `日期` DESC LIMIT 5")
|
||
print("\n最新记录：")
|
||
for r in cur.fetchall():
|
||
print(f" {r[0]} | {r[1]} | {r[2][:20]}... | {r[3]}件 | {r[4]} | {r[5]}")
|
||
|
||
conn.close()
|
||
|
||
if __name__ == "__main__":
|
||
main()
|