196 lines
7.4 KiB
Python
196 lines
7.4 KiB
Python
#!/usr/bin/env python3
|
||
"""下载并导入 2026-06-12 库存明细 CSV 文件"""
|
||
import sys, os, csv, json, subprocess, pymysql, re
|
||
from datetime import datetime
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
import importlib.util
|
||
spec = importlib.util.spec_from_file_location("config", "config.py")
|
||
config = importlib.util.module_from_spec(spec)
|
||
spec.loader.exec_module(config)
|
||
DB_CONFIG = config.DB_CONFIG
|
||
|
||
TMP_DIR = "tmp_csv"
|
||
os.makedirs(TMP_DIR, exist_ok=True)
|
||
|
||
DATA_DATE = "2026-06-12"
|
||
SOURCE = "wdt"
|
||
SPACE_ID = "28814565790"
|
||
|
||
# File list from 钉盘: (name, fileId)
|
||
FILES = [
|
||
("20260612-wdt-kcmx-01", "ndMj49yWjXkyODyehwNYo0L0J3pmz5aA"),
|
||
("20260612-wdt-kcmx-02", "G53mjyd80pGevAeNtgN6Epjx86zbX04v"),
|
||
("20260612-wdt-kcmx-03", "4lgGw3P8vRrBDmBeUp5kDkQR85daZ90D"),
|
||
("20260612-wdt-kcmx-04", "1R7q3QmWeeDRbPROCZyQO423WxkXOEP2"),
|
||
("20260612-wdt-kcmx-05", "MyQA2dXW7eKDkAD2C5mG2gD2JzlwrZgb"),
|
||
("20260612-wdt-kcmx-06", "GZLxjv9VGqm0xA04u12yODm686EDybno"),
|
||
("20260612-wdt-kcmx-07", "1DKw2zgV2PdeDAe0HPb2on2O8B5r9YAn"),
|
||
("20260612-wdt-kcmx-08", "wva2dxOW4Y14M642H0wpqOlwVbkz3BRL"),
|
||
("20260612-wdt-kcmx-09", "y20BglGWO2ywEgwkh04KEQ2g8A7depqY"),
|
||
("20260612-wdt-kcmx-10", "QBnd5ExVEvZ73d7xC2mDEdAdJyeZqMmz"),
|
||
("20260612-wdt-kcmx-11", "NkDwLng8ZLGrXmrnFx4y0A1rVKMEvZBY"),
|
||
("20260612-wdt-kcmx-12", "4lgGw3P8vRrBDmBeUp5kDePa85daZ90D"),
|
||
("20260612-wdt-kcmx-13", "gpG2NdyVX3EMNZMACGz4OQqzWMwvDqPk"),
|
||
("20260612-wdt-kcmx-15", "7QG4Yx2JpLaqBXqeFgN9opxdJ9dEq3XD"),
|
||
("20260612-wdt-kcmx-16", "vNG4YZ7JnPzwqNwLH9dan7QkW2LD0oRE"),
|
||
("20260612-wdt-kcmx-20", "NDoBb60VLQbxqvxMtBELN3o2JlemrZQ3"),
|
||
("20260612-wdt-kcmx-22", "NDoBb60VLQbxqvxMtBELN9kDJlemrZQ3"),
|
||
("20260612-wdt-kcmx-23", "NkDwLng8ZLGrXmrnFx4yYyMlVKMEvZBY"),
|
||
("20260612-wdt-kcmx-24", "QBnd5ExVEvZ73d7xC2mDj2y7JyeZqMmz"),
|
||
]
|
||
|
||
def parse_int(val):
|
||
if val is None or val == '' or str(val).upper() in ('NA', '#N/A', 'N/A'):
|
||
return 0
|
||
try:
|
||
return int(float(val))
|
||
except (ValueError, TypeError):
|
||
return 0
|
||
|
||
def parse_float(val):
|
||
if val is None or val == '' or str(val).upper() in ('NA', '#N/A', 'N/A'):
|
||
return 0.0
|
||
try:
|
||
return float(val)
|
||
except (ValueError, TypeError):
|
||
return 0.0
|
||
|
||
def parse_datetime(val):
|
||
if not val or val == '' or str(val).upper() in ('NA', '#N/A', 'N/A'):
|
||
return None
|
||
try:
|
||
return datetime.strptime(val.strip(), '%Y-%m-%d %H:%M:%S')
|
||
except ValueError:
|
||
try:
|
||
return datetime.strptime(val.strip(), '%Y-%m-%d')
|
||
except ValueError:
|
||
return None
|
||
|
||
def extract_warehouse_code(name):
|
||
"""从文件名提取仓库代码。如 20260612-wdt-kcmx-01 → 01"""
|
||
m = re.search(r'kcmx-(\d+)$', name)
|
||
return m.group(1) if m else '00'
|
||
|
||
def download_file(name, file_id):
|
||
"""用 dws drive download 下载 CSV"""
|
||
local_path = os.path.join(TMP_DIR, f"{name}.csv")
|
||
result = subprocess.run(
|
||
["dws", "drive", "download", "--node", file_id,
|
||
"--space-id", SPACE_ID, "--output", local_path, "--format", "json"],
|
||
capture_output=True, text=True, timeout=60
|
||
)
|
||
if result.returncode != 0:
|
||
print(f" ❌ 下载失败: {result.stderr[:300]}")
|
||
return None
|
||
if os.path.exists(local_path) and os.path.getsize(local_path) > 50:
|
||
print(f" ✅ 下载完成: {name}.csv ({os.path.getsize(local_path)} bytes)")
|
||
return local_path
|
||
print(f" ❌ 下载后文件过小或不存在")
|
||
return None
|
||
|
||
def count_existing(conn, wh_code):
|
||
"""查询该仓库已有记录数"""
|
||
with conn.cursor() as cur:
|
||
cur.execute("SELECT COUNT(*) as cnt FROM inventory_detail WHERE data_date=%s AND warehouse_code=%s",
|
||
(DATA_DATE, wh_code))
|
||
return cur.fetchone()['cnt']
|
||
|
||
print(f"📋 导入 {DATA_DATE} 库存明细数据（来源: {SOURCE}）")
|
||
print(f"共 {len(FILES)} 个仓库文件\n")
|
||
|
||
# 第一步：删除该日期已有数据
|
||
print("🧹 清理已有数据...")
|
||
conn = pymysql.connect(**DB_CONFIG, cursorclass=pymysql.cursors.DictCursor)
|
||
with conn.cursor() as cur:
|
||
cur.execute("DELETE FROM inventory_detail WHERE data_date=%s", (DATA_DATE,))
|
||
deleted = cur.rowcount
|
||
conn.commit()
|
||
conn.close()
|
||
print(f" 已删除 {deleted} 条记录\n")
|
||
|
||
# 第二步：逐个下载并导入
|
||
total_imported = 0
|
||
total_files = 0
|
||
errors = []
|
||
|
||
for name, file_id in FILES:
|
||
wh_code = extract_warehouse_code(name)
|
||
total_files += 1
|
||
print(f"[{total_files}/{len(FILES)}] {name} (仓库:{wh_code})")
|
||
|
||
# 下载
|
||
local_path = download_file(name, file_id)
|
||
if not local_path:
|
||
errors.append((name, "下载失败"))
|
||
continue
|
||
|
||
# 检查编码
|
||
file_result = subprocess.run(["file", local_path], capture_output=True, text=True, timeout=5)
|
||
if 'ISO' in file_result.stdout or '8859' in file_result.stdout:
|
||
utf8_path = local_path.replace('.csv', '_utf8.csv')
|
||
subprocess.run(["iconv", "-f", "GBK", "-t", "UTF-8", local_path, "-o", utf8_path],
|
||
capture_output=True, timeout=10)
|
||
os.replace(utf8_path, local_path)
|
||
|
||
# 导入
|
||
conn = pymysql.connect(**DB_CONFIG, cursorclass=pymysql.cursors.DictCursor)
|
||
inserted = 0
|
||
try:
|
||
with open(local_path, 'r', encoding='utf-8-sig') as f:
|
||
reader = csv.DictReader(f)
|
||
rows = list(reader)
|
||
|
||
columns = reader.fieldnames
|
||
print(f" 读取到 {len(rows)} 条记录, 列: {columns[:6]}...")
|
||
|
||
with conn.cursor() as cur:
|
||
for row in rows:
|
||
try:
|
||
cur.execute("""
|
||
INSERT INTO inventory_detail
|
||
(data_date, source, warehouse_code, goods_code, product_code, product_name,
|
||
brand, cost_price, available_qty, production_in_transit, purchase_in_transit,
|
||
category1, category2, category3, created_time)
|
||
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
|
||
""", (
|
||
DATA_DATE, SOURCE, wh_code,
|
||
row.get('商家编码', '').strip(),
|
||
row.get('货品编号', '').strip(),
|
||
row.get('货品名称', '').strip(),
|
||
row.get('品牌', '').strip(),
|
||
parse_float(row.get('成本价', 0)),
|
||
parse_int(row.get('可用库存', 0)),
|
||
parse_int(row.get('生产在途量', 0)),
|
||
parse_int(row.get('采购在途', 0)),
|
||
row.get('一级类目', '').strip(),
|
||
row.get('二级类目', '').strip(),
|
||
row.get('三级类目', '').strip(),
|
||
parse_datetime(row.get('单品创建时间', ''))
|
||
))
|
||
inserted += 1
|
||
except Exception as e:
|
||
if inserted < 3:
|
||
print(f" 插入异常: {e}")
|
||
|
||
conn.commit()
|
||
print(f" ✅ 导入 {inserted} 条记录")
|
||
total_imported += inserted
|
||
|
||
except Exception as e:
|
||
conn.rollback()
|
||
print(f" ❌ 导入失败: {e}")
|
||
errors.append((name, str(e)))
|
||
finally:
|
||
conn.close()
|
||
|
||
# 汇总
|
||
print(f"\n{'='*50}")
|
||
print(f"库存导入完成!")
|
||
print(f"处理 {total_files} 个文件, 导入 {total_imported} 条记录")
|
||
if errors:
|
||
print(f"\n⚠️ 失败:")
|
||
for name, reason in errors:
|
||
print(f" - {name}: {reason}")
|
||
print(f"{'='*50}")
|