204 lines
10 KiB
Python
204 lines
10 KiB
Python
#!/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()
|