299 lines
11 KiB
Python
299 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
"""动态导入新的库存数据（加固版）
|
||
|
||
改进：
|
||
1. 编码探测（utf-8-sig/gbk/gb18030），去除 iconv 风险路径
|
||
2. 清洗表头字段名中的 BOM 和不可见字符
|
||
3. 导入后校验货品编号空置率
|
||
4. 记录失败原因，不静默吞掉错误
|
||
"""
|
||
import sys, os, csv, json, subprocess, pymysql, re
|
||
from datetime import datetime
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
from config import DB_CONFIG
|
||
from services.inventory_calc import SPECIAL_COST_PRICES
|
||
|
||
TMP_DIR = "tmp_csv"
|
||
os.makedirs(TMP_DIR, exist_ok=True)
|
||
|
||
DB = {**DB_CONFIG, 'cursorclass': pymysql.cursors.DictCursor}
|
||
|
||
EMPTY_CODE_THRESHOLD = 5 # 货品编号空置率超过此百分比则回滚
|
||
INVENTORY_COLUMNS = [
|
||
'商家编码', '货品编号', '货品名称', '品牌', '成本价',
|
||
'库存', '生产在途量', '采购在途',
|
||
'一级类目', '二级类目', '三级类目', '单品创建时间',
|
||
]
|
||
|
||
|
||
def clean_header(val):
|
||
"""清洗表头字段名 — 去 BOM、引号、等号、首尾空白"""
|
||
if not val:
|
||
return ''
|
||
s = str(val)
|
||
s = s.replace('\ufeff', '').replace('"', '').replace('=', '').strip()
|
||
return s
|
||
|
||
|
||
def parse_int(val):
|
||
if not val or str(val).upper() == 'NA':
|
||
return 0
|
||
try:
|
||
return int(float(val))
|
||
except:
|
||
return 0
|
||
|
||
|
||
def parse_float(val):
|
||
if not val or str(val).upper() == 'NA':
|
||
return 0.0
|
||
try:
|
||
return float(val)
|
||
except:
|
||
return 0.0
|
||
|
||
|
||
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 read_csv_safe(filepath):
|
||
"""尝试多种编码读取 CSV，返回 (cleaned_fieldnames, rows_dict_list) 或 (None, None)"""
|
||
for enc in ['utf-8-sig', 'gbk', 'gb18030']:
|
||
try:
|
||
with open(filepath, 'r', encoding=enc) as f:
|
||
reader = csv.DictReader(f)
|
||
fnames = [clean_header(h) for h in reader.fieldnames]
|
||
rows = list(reader)
|
||
if len(rows) > 0:
|
||
return fnames, rows
|
||
except:
|
||
continue
|
||
return None, None
|
||
|
||
|
||
def import_file(filepath, data_date, warehouse_code, src='wdt'):
|
||
"""导入单个库存文件，返回 (inserted, fail_count, fail_samples)"""
|
||
fieldnames, rows = read_csv_safe(filepath)
|
||
if fieldnames is None or rows is None:
|
||
return 0, 0, ["无法读取CSV文件"]
|
||
|
||
print(f" 读取到 {len(rows)} 条记录, 表头: {fieldnames}")
|
||
|
||
conn = pymysql.connect(**DB)
|
||
inserted = 0
|
||
fail_count = 0
|
||
fail_samples = []
|
||
|
||
try:
|
||
with conn.cursor() as cur:
|
||
for row in rows:
|
||
try:
|
||
goods_code = (row.get('商家编码') or '').strip()
|
||
product_code = (row.get('货品编号') or '').strip()
|
||
if is_invalid_product_code(product_code):
|
||
fail_count += 1
|
||
continue
|
||
product_name = (row.get('货品名称') or '').strip()
|
||
if is_invalid_product_name(product_name):
|
||
fail_count += 1
|
||
continue
|
||
brand = (row.get('品牌') or '').strip()
|
||
cost_price = parse_float(row.get('成本价', 0))
|
||
# 特殊成本价规则：命中的商品不使用CSV/库存表成本价，使用单独指定的正确成本价
|
||
cost_price = SPECIAL_COST_PRICES.get(product_code, cost_price)
|
||
avail_qty = parse_int(row.get('库存', 0))
|
||
prod_transit = parse_int(row.get('生产在途量', 0))
|
||
purch_transit = parse_int(row.get('采购在途', 0))
|
||
cat1 = (row.get('一级类目') or '').strip()
|
||
cat2 = (row.get('二级类目') or '').strip()
|
||
cat3 = (row.get('三级类目') or '').strip()
|
||
|
||
created = None
|
||
ct = (row.get('单品创建时间') or '').strip()
|
||
if ct and ct.upper() != 'NA' and ct:
|
||
try:
|
||
created = datetime.strptime(ct, '%Y-%m-%d %H:%M:%S')
|
||
except:
|
||
pass
|
||
|
||
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, src, warehouse_code,
|
||
goods_code, product_code, product_name,
|
||
brand, cost_price, avail_qty, prod_transit, purch_transit,
|
||
cat1, cat2, cat3, created,
|
||
))
|
||
inserted += 1
|
||
|
||
except Exception as e:
|
||
fail_count += 1
|
||
if fail_count <= 3:
|
||
fail_samples.append(f"行异常: {e}")
|
||
|
||
# ── 导入后校验（在 cursor 生命周期内执行）──
|
||
if inserted > 0:
|
||
cur.execute("""
|
||
SELECT COUNT(*) AS total,
|
||
SUM(CASE WHEN product_code IS NULL OR product_code = '' THEN 1 ELSE 0 END) AS empty_code
|
||
FROM inventory_detail
|
||
WHERE data_date=%s AND warehouse_code=%s
|
||
""", (data_date, warehouse_code))
|
||
r = cur.fetchone()
|
||
total = r['total'] or 0
|
||
empty_code = r['empty_code'] or 0
|
||
if total > 0:
|
||
ratio = empty_code / total * 100
|
||
if ratio > EMPTY_CODE_THRESHOLD:
|
||
conn.rollback()
|
||
return 0, fail_count, [
|
||
f"货品编号空置率 {ratio:.1f}% ({empty_code}/{total}) 超过阈值 {EMPTY_CODE_THRESHOLD}%，已回滚"
|
||
]
|
||
|
||
conn.commit()
|
||
return inserted, fail_count, fail_samples
|
||
|
||
except Exception as e:
|
||
conn.rollback()
|
||
return 0, fail_count, [str(e)]
|
||
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
# ── 1. 查已有数据 ──
|
||
conn = pymysql.connect(**DB)
|
||
with conn.cursor() as cur:
|
||
cur.execute("SELECT DISTINCT data_date FROM inventory_detail")
|
||
existing_dates = set()
|
||
for r in cur.fetchall():
|
||
d = r['data_date']
|
||
d_str = d.strftime('%Y-%m-%d') if hasattr(d, 'strftime') else str(d)[:10]
|
||
existing_dates.add(d_str)
|
||
|
||
cur.execute("SELECT DISTINCT CONCAT(data_date, '|', LPAD(warehouse_code, 2, '0')) AS dw FROM inventory_detail")
|
||
existing_dw = set(r['dw'] for r in cur.fetchall())
|
||
conn.close()
|
||
|
||
print(f"已有库存日期: {sorted(existing_dates)}")
|
||
|
||
# ── 2. 获取钉盘库存文件列表（分页，limit 最大 50）──
|
||
drive_items = []
|
||
cursor = None
|
||
while True:
|
||
args = ["dws", "drive", "list", "--space-id", "28814565790",
|
||
"--folder", "NkDwLng8ZLGrXmrnFxYq3vRqVKMEvZBY",
|
||
"--limit", "50", "--format", "json"]
|
||
if cursor:
|
||
args.extend(["--cursor", cursor])
|
||
result = subprocess.run(args, capture_output=True, text=True, timeout=30)
|
||
try:
|
||
data = json.loads(result.stdout)
|
||
if not data.get('success', True):
|
||
raise ValueError(data.get('error', {}).get('message', 'unknown'))
|
||
drive_items.extend(data.get('result', {}).get('items', []))
|
||
except Exception as e:
|
||
print(f"❌ 获取钉盘文件列表失败: {e}")
|
||
sys.exit(1)
|
||
cursor = data.get('result', {}).get('nextToken')
|
||
if not cursor:
|
||
break
|
||
print(f"扫描到 {len(drive_items)} 个钉盘文件")
|
||
|
||
kcmx_files = []
|
||
for node in drive_items:
|
||
name = node['name']
|
||
if not name.startswith('2026') or 'kcmx' not in name:
|
||
continue
|
||
m = re.match(r'(\d{8})-wdt-kcmx-(\d+)', name)
|
||
if m:
|
||
date_str = f"{m.group(1)[:4]}-{m.group(1)[4:6]}-{m.group(1)[6:8]}"
|
||
wh = m.group(2) # 保留原始两位格式(01~29)，避免去前导零
|
||
kcmx_files.append((name, date_str, wh, node['fileId']))
|
||
|
||
kcmx_files.sort(key=lambda x: (x[1], int(x[2])))
|
||
print(f"共找到 {len(kcmx_files)} 个库存文件")
|
||
|
||
# ── 3. 过滤新文件 ──
|
||
new_files = []
|
||
for name, date_str, wh, node_id in kcmx_files:
|
||
dw_key = f"{date_str}|{wh.zfill(2)}"
|
||
if dw_key in existing_dw:
|
||
print(f" SKIP {name} (已存在)")
|
||
else:
|
||
print(f" NEW {name}")
|
||
new_files.append((name, date_str, wh, node_id))
|
||
|
||
if not new_files:
|
||
print("\n✅ 所有库存数据已是最新")
|
||
sys.exit(0)
|
||
|
||
print(f"\n需要导入: {len(new_files)} 个文件")
|
||
|
||
# ── 4. 下载并导入 ──
|
||
total = 0
|
||
has_errors = False
|
||
|
||
for i, (name, date_str, wh, node_id) in enumerate(new_files, 1):
|
||
print(f"\n[{i}/{len(new_files)}] {name} ({date_str} 仓库{wh})")
|
||
|
||
# 下载
|
||
local_path = os.path.join(TMP_DIR, f"{name}")
|
||
if os.path.exists(local_path) and os.path.getsize(local_path) > 100:
|
||
print(f" 使用缓存: {name}")
|
||
else:
|
||
r = subprocess.run(
|
||
["dws", "drive", "download", "--node", node_id,
|
||
"--output", local_path, "--format", "json"],
|
||
capture_output=True, text=True, timeout=60
|
||
)
|
||
if not (os.path.exists(local_path) and os.path.getsize(local_path) > 50):
|
||
print(f" FAIL 下载失败: {r.stderr[:200]}")
|
||
has_errors = True
|
||
continue
|
||
print(f" 已下载 ({os.path.getsize(local_path)} bytes)")
|
||
|
||
# 导入
|
||
inserted, fails, samples = import_file(local_path, date_str, wh)
|
||
if inserted > 0:
|
||
print(f" ✅ 导入 {inserted} 条", end='')
|
||
if fails > 0:
|
||
print(f", 跳过 {fails} 条异常")
|
||
for s in samples[:3]:
|
||
print(f" 示例: {s}")
|
||
else:
|
||
print()
|
||
total += inserted
|
||
else:
|
||
print(f" ❌ 导入失败")
|
||
for s in samples[:3]:
|
||
print(f" 原因: {s}")
|
||
has_errors = True
|
||
|
||
print(f"\n{'='*40}")
|
||
print(f"库存导入完成: 共 {total} 条记录")
|
||
if has_errors:
|
||
print(f"⚠️ 部分文件导入失败，请检查上方日志")
|
||
else:
|
||
# 导入脚本独立于 Gunicorn 进程运行，必须清除共享缓存文件，
|
||
# 让各 worker 放弃导入前的库存/计划预测结果。
|
||
try:
|
||
from services.cache import refresh_all_caches
|
||
refreshed = refresh_all_caches()
|
||
print(f"✅ 导入后计算缓存已清除（{len(refreshed)} 个模块）")
|
||
except Exception as e:
|
||
print(f"⚠️ 导入后计算缓存清除失败: {e}")
|
||
print(f"{'='*40}")
|