#!/usr/bin/env python3 """钉盘增量导入（通用 diff 版）：按钉盘 vs DB 差异补录出库与库存。""" import csv, os, json, subprocess, pymysql, importlib.util, sys from datetime import datetime sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) s = importlib.util.spec_from_file_location('config', 'config.py') m = importlib.util.module_from_spec(s); s.loader.exec_module(m) DB = m.DB_CONFIG SPACE_ID = '28814565790' OB_FOLDER = '3NwLYZXWynyBjRB4tZjwNxDvVkyEqBQm' IV_FOLDER = 'NkDwLng8ZLGrXmrnFxYq3vRqVKMEvZBY' TMP = 'tmp_csv/update_latest' os.makedirs(TMP, exist_ok=True) MIN_DATE = '2026-09-05' def si(v): try: return int(float(str(v or 0).replace('"','').replace('=','').strip())) except: return 0 def sf(v): try: return float(str(v or 0).replace('"','').replace('=','').strip()) except: return 0.0 def clean(v): return str(v or '').replace('"','').replace('=','').strip() def bad(c): n = clean(c).upper().rstrip(':') return not n or n in ('NA','N/A','合计') SPECIALS = {'POZ-FM-0184-TM-SM':.055,'POZ-XFS-9014-TM-MD':.062,'POZ-FM-9742-TM-MD':.181, 'POZ-MM-0580-TM-MD':.22,'POZ-SKS-3374-LD':.1,'POZ-SKS-3367-LD':.1, 'POZ-SKS-3350-LD':.1,'POZ-SKS-3343-LD':.1} def dws_list(folder): out=[]; cursor=None while True: args=['dws','drive','list','--space-id',SPACE_ID,'--folder',folder,'--limit','50','--format','json'] if cursor: args+=['--cursor',cursor] r=subprocess.run(args,capture_output=True,text=True) if r.returncode!=0: print(' [list err]',r.stderr[:150]); break d=json.loads(r.stdout).get('result',{}) out+=d.get('items',[]) cursor=d.get('nextToken') if not cursor: break return out def dl(name, fileId): dest=os.path.join(TMP,name) if os.path.exists(dest) and os.path.getsize(dest)>10: return dest r=subprocess.run(['dws','drive','download','--node',fileId,'--space-id',SPACE_ID,'--output',dest,'--format','json'],capture_output=True,text=True) if r.returncode!=0 or not os.path.exists(dest) or os.path.getsize(dest)<10: print(' [dl err]',name,r.stderr[:120]); return None return dest def import_outbound(conn, date, src, path): with conn.cursor() as x: x.execute('DELETE FROM daily_outbound WHERE 日期=%s AND 数据来源=%s',(date,src)) ins=0 with conn.cursor() as x: if src=='wdt': # wdt 文件编码可能为 UTF-8(BOM) 或 GBK，自动检测 raw = open(path, 'rb').read() text = None for enc in ('utf-8-sig', 'gb18030', 'gbk'): try: text = raw.decode(enc); break except UnicodeDecodeError: continue if text is None: print(' [编码识别失败]', path); return 0 import io as _io for row in csv.DictReader(_io.StringIO(text)): code=clean(row.get('货品编号')) if bad(code): continue x.execute('INSERT INTO daily_outbound (货品编号,货品名称,品牌,仓库,店铺,一级类目,二级类目,三级类目,货品数量,货品成本,日期,数据来源,数据类型) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)', (code,clean(row.get('货品名称')),clean(row.get('品牌')),clean(row.get('仓库')),clean(row.get('店铺')), clean(row.get('一级类目')),clean(row.get('二级类目')),clean(row.get('三级类目')), si(row.get('货品数量')),sf(row.get('货品成本')),date,src,'出库数据')); ins+=1 else: with open(path, encoding='utf-8-sig') as f: rr=list(csv.reader(f)) for raw in rr[1:]: if not raw or len(raw)<2: continue code=clean(raw[0]); qty=si(raw[1]) if bad(code): continue x.execute('SELECT product_name, brand, category1, category2, category3, cost_price FROM products WHERE product_code=%s LIMIT 1',(code,)) p=x.fetchone() if p: pname=p['product_name'] or '' if clean(pname) in ('-','--'): continue brand=p['brand'] or ''; c1=p['category1'] or ''; c2=p['category2'] or ''; c3=p['category3'] or '' cost=float(p['cost_price']) if p['cost_price'] else 0 else: pname=brand=c1=c2=c3=''; cost=0 x.execute('INSERT INTO daily_outbound (货品编号,货品名称,品牌,仓库,店铺,一级类目,二级类目,三级类目,货品数量,货品成本,日期,数据来源,数据类型) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)', (code,pname,brand,'','',c1,c2,c3,qty,cost,date,src,'出库数据')); ins+=1 return ins def import_inventory(conn, date, path, wh): # 必须在删除旧快照前验证列完整性。缺失采购在途≠采购在途为0； # 这种精简导出会把在途成本和依赖它的总成本错误地归零。 import io as _io raw = open(path, 'rb').read() text = None for enc in ('utf-8-sig', 'gb18030', 'gbk'): try: text = raw.decode(enc) break except UnicodeDecodeError: continue if text is None: raise ValueError(f'{path}: 无法识别文件编码') reader = csv.DictReader(_io.StringIO(text)) required = {'货品编号', '货品名称', '库存', '成本价', '采购在途', '生产在途量'} missing = required.difference(reader.fieldnames or ()) if missing: raise ValueError(f'{path}: 库存文件缺少必要列 {sorted(missing)}；本次不导入，以免将缺失数据写成0') with conn.cursor() as x: x.execute('DELETE FROM inventory_detail WHERE data_date=%s AND warehouse_code=%s',(date,wh)) ins=0 with conn.cursor() as x: for row in reader: code=clean(row.get('货品编号')) if bad(code) or clean(row.get('货品名称')) in ('-','--'): continue cost=SPECIALS.get(code, sf(row.get('成本价'))) ct=clean(row.get('单品创建时间')); dt=None for fmt in ('%Y-%m-%d %H:%M:%S','%Y-%m-%d'): try: dt=datetime.strptime(ct,fmt); break except: pass x.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)', (date,'wdt',wh,clean(row.get('商家编码')),code,clean(row.get('货品名称')),clean(row.get('品牌')),cost, si(row.get('库存')),si(row.get('生产在途量')),si(row.get('采购在途')), clean(row.get('一级类目')),clean(row.get('二级类目')),clean(row.get('三级类目')),dt)); ins+=1 return ins def main(): ob = dws_list(OB_FOLDER); iv = dws_list(IV_FOLDER) ob_files={x['name']:x['fileId'] for x in ob if x['type']=='FILE' and x['name'].endswith('.csv')} iv_files={x['name']:x['fileId'] for x in iv if x['type']=='FILE' and x['name'].endswith('.csv')} conn = pymysql.connect(**DB, cursorclass=pymysql.cursors.DictCursor) existing_ob=set() with conn.cursor() as x: x.execute("SELECT 日期,数据来源 FROM daily_outbound") for r in x.fetchall(): d=r['日期'].strftime('%Y-%m-%d') if hasattr(r['日期'],'strftime') else str(r['日期'])[:10] existing_ob.add((d,r['数据来源'])) x.execute("SELECT DISTINCT data_date FROM inventory_detail") existing_iv=set(str(r['data_date'])[:10] for r in x.fetchall()) need_ob=[]; need_iv=[] for name,fid in sorted(ob_files.items()): p=name.split('-') if len(p)<3 or not p[0].isdigit() or len(p[0])!=8: continue date=f'{p[0][:4]}-{p[0][4:6]}-{p[0][6:8]}'; src=p[1] if date < MIN_DATE: continue if (date,src) in existing_ob: continue need_ob.append((name,fid,date,src)) for name,fid in sorted(iv_files.items()): p=name.split('-') if not (len(p)>=3 and p[0].isdigit() and len(p[0])==8 and p[1]=='wdt' and p[2]=='kcmx'): continue date=f'{p[0][:4]}-{p[0][4:6]}-{p[0][6:8]}'; wh=name.rsplit('-',1)[1][:-4] if date < MIN_DATE: continue if date in existing_iv: continue need_iv.append((name,fid,date,wh)) print(f'需导入 出库 {len(need_ob)} 文件, 库存 {len(need_iv)} 文件') # 先检查本批所有库存文件的列，再进行任何数据库写入；禁止部分成功后提交。 for name,fid,date,wh in need_iv: p=dl(name,fid) if not p: raise RuntimeError(f'库存文件下载失败，取消本批导入: {name}') with open(p, 'rb') as f: raw=f.read() text=None for enc in ('utf-8-sig', 'gb18030', 'gbk'): try: text=raw.decode(enc) break except UnicodeDecodeError: continue if text is None: raise ValueError(f'{name}: 无法识别文件编码') columns=csv.DictReader(__import__('io').StringIO(text)).fieldnames or [] missing={'货品编号','货品名称','库存','成本价','采购在途','生产在途量'} - set(columns) if missing: raise ValueError(f'{name}: 缺少必要列 {sorted(missing)}；已中止导入，不会将缺失列记为0') ob_ins=0 for name,fid,date,src in need_ob: p=dl(name,fid) if not p: continue n=import_outbound(conn,date,src,p); ob_ins+=n print(f' 出库 {name} ({date} {src}) 导入 {n} 条') iv_ins=0 for name,fid,date,wh in need_iv: p=dl(name,fid) if not p: continue n=import_inventory(conn,date,p,wh); iv_ins+=n print(f' 库存 {name} (仓库{wh}) 导入 {n} 条') conn.commit(); conn.close() print(f'完成: 出库新增 {ob_ins} 条, 库存新增 {iv_ins} 条') if __name__=='__main__': main()