#!/usr/bin/env python3 """批量下载并导入所有库存明细数据""" import sys import os import csv import json import subprocess import pymysql from datetime import datetime import re # 从 config.py 导入配置 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 # 特殊成本价规则：命中的商品不使用CSV/库存表成本价 from services.inventory_calc import SPECIAL_COST_PRICES TMP_DIR = "tmp_csv" 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 is_invalid_product_code(code): """过滤报表汇总行、占位编码和空编码。""" normalized = str(code or '').strip().upper().rstrip(':') return not normalized or normalized in ('NA', 'N/A', '合计') def is_invalid_product_name(name): """过滤货品名称为单横线或双横线的占位数据。""" return str(name or '').strip() in ('-', '--') def download_file(name, node_id): """下载钉盘文件，返回本地路径""" local_path = os.path.join(TMP_DIR, f"{name}.csv") # 获取下载 URL result = subprocess.run( ["dws", "doc", "download", "--node", node_id, "--output", "/dev/stdout", "--format", "json"], capture_output=True, text=True, timeout=30 ) try: data = json.loads(result.stdout) url = data.get('resourceUrl') if not url: print(f" ❌ 无法获取下载链接: {name}") return None except: print(f" ❌ 解析下载响应失败: {name}") return None # 下载文件 subprocess.run( ["curl", "-sL", url, "-o", local_path], capture_output=True, timeout=30 ) 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 else: print(f" ❌ 下载失败或文件过小: {name}") return None 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, 'r', encoding='utf-8-sig') as f: reader = csv.DictReader(f) rows = list(reader) print(f" 读取到 {len(rows)} 条记录") inserted = 0 with conn.cursor() as cur: for row in rows: try: product_code = row.get('货品编号', '').strip() if is_invalid_product_code(product_code): continue if is_invalid_product_name(row.get('货品名称')): continue cost_price = parse_float(row.get('成本价', 0)) # 特殊成本价规则：命中的商品不使用CSV/库存表成本价 cost_price = SPECIAL_COST_PRICES.get(product_code, cost_price) 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(), product_code, row.get('货品名称', '').strip(), row.get('品牌', '').strip(), cost_price, 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() # 所有需要导入的文件列表 FILES = [ # 2026-05-27 (除了已导入的01) ("20260527-wdt-kcmx-04", "2026-05-27", "04", "3NwLYZXWynyBjRB4tZZ1gM2oVkyEqBQm"), ("20260527-wdt-kcmx-05", "2026-05-27", "05", "y20BglGWO2ywEgwkh00oxDN98A7depqY"), ("20260527-wdt-kcmx-06", "2026-05-27", "06", "14dA3GK8gjMb34bZIKKRXEQ0J9ekBD76"), ("20260527-wdt-kcmx-07", "2026-05-27", "07", "7dx2rn0JbY4m7rmQH22aYgwdVMGjLRb3"), ("20260527-wdt-kcmx-10", "2026-05-27", "10", "y20BglGWO2ywEgwkh00oxOzP8A7depqY"), ("20260527-wdt-kcmx-11", "2026-05-27", "11", "np9zOoBVBYpa0AaxHLL7Ke3DW1DK0g6l"), ("20260527-wdt-kcmx-12", "2026-05-27", "12", "P0MALyR8klXxPgxAtKKAv379W3bzYmDO"), ("20260527-wdt-kcmx-13", "2026-05-27", "13", "MyQA2dXW7eKDkAD2C55OaxvjJzlwrZgb"), ("20260527-wdt-kcmx-15", "2026-05-27", "15", "1OQX0akWmx2P5jPZFvvzyvKg8GlDd3mE"), ("20260527-wdt-kcmx-16", "2026-05-27", "16", "4lgGw3P8vRrBDmBeUppbwDLd85daZ90D"), ("20260527-wdt-kcmx-22", "2026-05-27", "22", "OG9lyrgJPzx5KM5ACzz0qrdZWzN67Mw4"), ("20260527-wdt-kcmx-23", "2026-05-27", "23", "nYMoO1rWxa3BM7BKSzzADrneV47Z3je9"), ("20260527-wdt-kcmx-24", "2026-05-27", "24", "DnRL6jAJMGYOl9OjFXX0rQyOWyMoPYe1"), # 2026-05-28 (全部18个仓库) ("20260528-wdt-kcmx-01", "2026-05-28", "01", "Amq4vjg890vMpAMPCPmdyejGJ3kdP0wQ"), ("20260528-wdt-kcmx-02", "2026-05-28", "02", "ndMj49yWjXkyODyehwR0Obn0J3pmz5aA"), ("20260528-wdt-kcmx-03", "2026-05-28", "03", "4lgGw3P8vRrBDmBeUpgdKP5385daZ90D"), ("20260528-wdt-kcmx-04", "2026-05-28", "04", "vNG4YZ7JnPzwqNwLH9NyxoqkW2LD0oRE"), ("20260528-wdt-kcmx-05", "2026-05-28", "05", "LeBq413JAwXZRAZxtr3naNPKWDOnGvpb"), ("20260528-wdt-kcmx-06", "2026-05-28", "06", "bva6QBXJwaYBgnBwS2Ml5wAQWn4qY5Pr"), ("20260528-wdt-kcmx-07", "2026-05-28", "07", "Amq4vjg890vMpAMPCPmd79jKJ3kdP0wQ"), ("20260528-wdt-kcmx-08", "2026-05-28", "08", "1zknDm0WRaPKvNKASx09oDEg8BQEx5rG"), ("20260528-wdt-kcmx-09", "2026-05-28", "09", "DnRL6jAJMGYOl9OjFXqDeaY9WyMoPYe1"), ("20260528-wdt-kcmx-10", "2026-05-28", "10", "GZLxjv9VGqm0xA04u167zyOq86EDybno"), ("20260528-wdt-kcmx-11", "2026-05-28", "11", "gpG2NdyVX3EMNZMACGP3pzPXWMwvDqPk"), ("20260528-wdt-kcmx-12", "2026-05-28", "12", "0eMKjyp813eL9ALkC4edLLXjVxAZB1Gv"), ("20260528-wdt-kcmx-13", "2026-05-28", "13", "R4GpnMqJzGxBN0BvFakK9L7w8Ke0xjE3"), ("20260528-wdt-kcmx-15", "2026-05-28", "15", "Exel2BLV5z6vjAv2CmPEA9gPJgk9rpMq"), ("20260528-wdt-kcmx-16", "2026-05-28", "16", "QPGYqjpJYrAjkajLCZBdmd1v8akx1Z5N"), ("20260528-wdt-kcmx-22", "2026-05-28", "22", "yQod3RxJKG0y54ygF4oPGOeeJkb4Mw9r"), ("20260528-wdt-kcmx-23", "2026-05-28", "23", "dxXB52LJqnmBErB3tMKnGK7w8qjMp697"), ("20260528-wdt-kcmx-24", "2026-05-28", "24", "9E05BDRVQ2Oq4jqAhyDmrqRqJ63zgkYA"), ] os.makedirs(TMP_DIR, exist_ok=True) total_imported = 0 total_files = len(FILES) success_count = 0 for i, (name, data_date, wh_code, node_id) in enumerate(FILES, 1): print(f"\n[{i}/{total_files}] {name} (日期:{data_date}, 仓库:{wh_code})") # 下载 local_path = download_file(name, node_id) if not local_path: continue # 检查文件编码和内容 result = subprocess.run( ["file", local_path], capture_output=True, text=True, timeout=5 ) # 如果是 GBK 编码，转为 UTF-8 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) # 导入 count = import_file(local_path, data_date, wh_code) if count > 0: print(f" ✅ 导入 {count} 条记录") total_imported += count success_count += 1 else: print(f" ⚠️ 导入 0 条记录") print(f"\n{'='*50}") print(f"批量导入完成!") print(f"成功: {success_count}/{total_files} 个文件") print(f"总计导入: {total_imported} 条记录") print(f"{'='*50}")