#!/usr/bin/env python3 """批量下载并导入2026-05-29库存明细数据""" import sys import os import csv import json import subprocess import pymysql from datetime import datetime sys.path.insert(0, os.path.dirname(__file__)) from config import DB_CONFIG TMP_DIR = os.path.join(os.path.dirname(__file__), "tmp_csv") os.makedirs(TMP_DIR, exist_ok=True) def parse_int(val): if val is None or val == '' or str(val).upper() == 'NA': 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() == 'NA': 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() == 'NA': 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 download_file(node_id, local_path): """下载钉盘文件 - 两步：获取凭证 → curl下载""" # 1. 获取下载凭证（获取resourceUrl） result = subprocess.run( ["dws", "doc", "download", "--node", node_id, "--output", "/dev/stdout", "--format", "json"], capture_output=True, text=True, timeout=60 ) try: data = json.loads(result.stdout) url = data.get('resourceUrl') if not url: print(f" ❌ 无法获取下载链接") return False except: print(f" ❌ 解析响应失败: {result.stdout[:200]}") return False # 2. 用curl下载文件 dl_result = subprocess.run( ["curl", "-sL", url, "-o", local_path], capture_output=True, timeout=60 ) if os.path.exists(local_path) and os.path.getsize(local_path) > 50: return True else: print(f" ❌ curl下载失败: size={os.path.getsize(local_path) if os.path.exists(local_path) else 0}") return False def import_file(filepath, data_date, warehouse_code, source='wdt'): """导入单个文件到数据库""" conn = pymysql.connect( host=DB_CONFIG["host"], port=DB_CONFIG["port"], user=DB_CONFIG["user"], password=DB_CONFIG["password"], database=DB_CONFIG["database"], charset="utf8mb4", cursorclass=pymysql.cursors.DictCursor ) try: # 检测编码 with open(filepath, 'rb') as f: raw = f.read(10000) encodings = ['utf-8-sig', 'gbk', 'gb2312', 'utf-8'] content = None for enc in encodings: try: content = raw.decode(enc) + open(filepath, 'r', encoding=enc).read()[10000:] break except: continue if content is None: print(f" 无法解码文件") return 0 # 重新以正确编码读取 for enc in encodings: try: with open(filepath, 'r', encoding=enc) as f: reader = csv.DictReader(f) rows = list(reader) break except: continue else: print(f" 无法解码文件") return 0 print(f" 读取到 {len(rows)} 条记录") # 先删除该仓库该日期的旧数据 with conn.cursor() as cur: cur.execute( "DELETE FROM inventory_detail WHERE data_date = %s AND warehouse_code = %s", (data_date, warehouse_code) ) deleted = cur.rowcount if deleted > 0: print(f" 已清除 {deleted} 条旧记录") conn.commit() inserted = 0 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, warehouse_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: print(f" 插入失败: {e}") conn.commit() return inserted finally: conn.close() # 2026-05-29 库存文件列表 (从钉盘获取) FILES = [ ("20260529-wdt-kcmx-01", "01", "G53mjyd80pGevAeNtgeNyYz186zbX04v"), ("20260529-wdt-kcmx-02", "02", "7dx2rn0JbY4m7rmQH2Np9px3VMGjLRb3"), ("20260529-wdt-kcmx-03", "03", "QPGYqjpJYrAjkajLCZKPARp98akx1Z5N"), ("20260529-wdt-kcmx-04", "04", "7QG4Yx2JpLaqBXqeFgqNElOZJ9dEq3XD"), ("20260529-wdt-kcmx-05", "05", "nYMoO1rWxa3BM7BKSz9Xnpn6V47Z3je9"), ("20260529-wdt-kcmx-06", "06", "ndMj49yWjXkyODyehwbNejAjJ3pmz5aA"), ("20260529-wdt-kcmx-07", "07", "NDoBb60VLQbxqvxMtBaEOo0QJlemrZQ3"), ("20260529-wdt-kcmx-08", "08", "NkDwLng8ZLGrXmrnFx34E0aQVKMEvZBY"), ("20260529-wdt-kcmx-09", "09", "NkDwLng8ZLGrXmrnFx34EyogVKMEvZBY"), ("20260529-wdt-kcmx-10", "10", "3NwLYZXWynyBjRB4tZGMb9z1VkyEqBQm"), ("20260529-wdt-kcmx-11", "11", "y20BglGWO2ywEgwkh0v4NmEz8A7depqY"), ("20260529-wdt-kcmx-12", "12", "QBnd5ExVEvZ73d7xC26mqRZ9JyeZqMmz"), ("20260529-wdt-kcmx-13", "13", "NkDwLng8ZLGrXmrnFx34EqqPVKMEvZBY"), ("20260529-wdt-kcmx-15", "15", "1R7q3QmWeeDRbPROCZ6ypvOnWxkXOEP2"), ("20260529-wdt-kcmx-16", "16", "P0MALyR8klXxPgxAtKDXxyXKW3bzYmDO"), ("20260529-wdt-kcmx-22", "22", "yQod3RxJKG0y54ygF4lX0YOLJkb4Mw9r"), ("20260529-wdt-kcmx-23", "23", "QPGYqjpJYrAjkajLCZKPy70M8akx1Z5N"), ("20260529-wdt-kcmx-24", "24", "OG9lyrgJPzx5KM5ACzl5AjE6WzN67Mw4"), ] DATA_DATE = "2026-05-29" total_imported = 0 total_files = len(FILES) success_count = 0 for i, (name, wh_code, node_id) in enumerate(FILES, 1): csv_name = f"{name}.csv" local_path = os.path.join(TMP_DIR, csv_name) # 如果文件已存在则跳过下载 if os.path.exists(local_path) and os.path.getsize(local_path) > 50: print(f"[{i}/{total_files}] {csv_name} (仓库:{wh_code}) - 文件已存在，直接导入") else: print(f"[{i}/{total_files}] {csv_name} (仓库:{wh_code}) - 下载中...") if not download_file(node_id, local_path): continue # 检查编码 result = subprocess.run(["file", local_path], capture_output=True, text=True, timeout=5) if 'ISO' in result.stdout or '8859' in 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) print(f" 编码已转换: GBK → UTF-8") # 导入 count = import_file(local_path, DATA_DATE, wh_code) if count > 0: print(f" ✅ 仓库{wh_code}: 导入 {count} 条") total_imported += count success_count += 1 else: print(f" ⚠️ 仓库{wh_code}: 导入 0 条") print(f"\n{'='*50}") print(f"5月29日库存数据导入完成!") print(f"成功仓库: {success_count}/{total_files}") print(f"总计导入: {total_imported} 条记录") print(f"{'='*50}")