123 lines
8.0 KiB
Python
123 lines
8.0 KiB
Python
#!/usr/bin/env python3
|
|
"""把 #280 规范 CSV 分批送往 Admin 预检或安全入库。
|
|
|
|
默认 --dry-run 调用 Admin 的只读数据库预检；只有显式 --apply 才会写入。
|
|
Token 只从环境变量读取，报告中不会保存 Token 或完整请求体。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse, csv, hashlib, json, os, sys, tempfile
|
|
from collections import Counter
|
|
from pathlib import Path
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.request import Request, urlopen
|
|
|
|
MAX_PRODUCTS, MAX_SKUS, MAX_BYTES = 500, 5000, 5 << 20
|
|
|
|
def error(message): raise RuntimeError(message)
|
|
|
|
def request(endpoint, token, payload):
|
|
body=json.dumps(payload,ensure_ascii=False,separators=(',',':')).encode()
|
|
req=Request(endpoint,data=body,method='POST',headers={'Authorization':f'Bearer {token}','Content-Type':'application/json','Accept':'application/json'})
|
|
try:
|
|
with urlopen(req,timeout=90) as response: return json.loads(response.read().decode())
|
|
except HTTPError as exc:
|
|
try: detail=json.loads(exc.read().decode()).get('error',{})
|
|
except Exception: detail={}
|
|
error(f"HTTP {exc.code} {detail.get('code','ERROR')}: {detail.get('message','预检/导入失败')}")
|
|
except URLError as exc: error(f"网络失败：{exc.reason}")
|
|
|
|
def save_report(path, value):
|
|
"""每批完成后原子保存进度；中断后不会留下半份 JSON。"""
|
|
path=Path(path); path.parent.mkdir(parents=True,exist_ok=True)
|
|
with tempfile.NamedTemporaryFile('w',encoding='utf-8',delete=False,dir=path.parent,suffix='.tmp') as f:
|
|
json.dump(value,f,ensure_ascii=False,indent=2); temp=f.name
|
|
Path(temp).replace(path)
|
|
|
|
def product_from(row):
|
|
return {'goods_id':row['shopee_goods_id'],'title':row['shopee_title'],'status':row['shopee_status'],'main_sku_code':row['shopee_main_sku_code'],'image_url':row['shopee_image_url'],'shop_name':row['shopee_shop_name']}
|
|
def sku_from(row):
|
|
return {'sku_id':row['shopee_sku_id'],'goods_id':row['shopee_goods_id'],'spec_raw':row['spec_raw'],'color':row['color'],'size':row['size'],'advice':row['advice'],'parse_ok':row['parse_ok']=='true','sku_code':row['shopee_sku_code']}
|
|
def pdd_from(row): return {'goods_id':row['pdd_goods_id'],'url':row['pdd_goods_url'],'title':'','shop_name':'','dimensions':[],'skus':[]}
|
|
def spec_key(raw): return ' '.join(raw.split())
|
|
|
|
def groups(root):
|
|
"""流式读 CSV；来源按商品连续导出，同商品不会一次占用整个文件。"""
|
|
for path in sorted(root.rglob('*.csv')):
|
|
current, rows = None, []
|
|
with path.open(encoding='utf-8-sig',newline='') as handle:
|
|
for row in csv.DictReader(handle):
|
|
key=row['shopee_goods_id']
|
|
if current is not None and key != current:
|
|
yield current, rows
|
|
rows=[]
|
|
current=key; rows.append(row)
|
|
if rows: yield current, rows
|
|
|
|
def make_group(rows):
|
|
first=rows[0]; skus=[]; specs=set(); pdds={}; associations=[]
|
|
for row in rows:
|
|
key=spec_key(row['spec_raw'])
|
|
if key and key not in specs:
|
|
specs.add(key); skus.append(sku_from(row))
|
|
if row['pdd_importable']=='true' and row['pdd_goods_id'] not in pdds:
|
|
pdds[row['pdd_goods_id']]=pdd_from(row)
|
|
associations.append({'shopee_goods_id':first['shopee_goods_id'],'pdd_goods_id':row['pdd_goods_id']})
|
|
return product_from(first),skus,pdds,associations
|
|
|
|
def batches(root, observed_at, dry_run, keep_existing):
|
|
selected=[]; seen_pdd={}; sku_count=0
|
|
def emit():
|
|
nonlocal selected,seen_pdd,sku_count
|
|
# 同一商品可能在不同源 CSV 再次出现；接口禁止批内重复身份，
|
|
# 所以在真正组装请求时按四类业务键再去重一次。
|
|
product_by_id={g[0]['goods_id']:g[0] for g in selected}
|
|
sku_by_key={}
|
|
association_by_key={}
|
|
for group in selected:
|
|
for sku in group[1]: sku_by_key.setdefault((sku['goods_id'],spec_key(sku['spec_raw'])),sku)
|
|
for association in group[3]:
|
|
if association['shopee_goods_id'] not in keep_existing:
|
|
association_by_key.setdefault((association['shopee_goods_id'],association['pdd_goods_id']),association)
|
|
products=list(product_by_id.values()); skus=list(sku_by_key.values())
|
|
pdds=list(seen_pdd.values()); assocs=list(association_by_key.values())
|
|
seed=json.dumps([products,skus,pdds,assocs],ensure_ascii=False,sort_keys=True,separators=(',',':')).encode()
|
|
payload={'schema_version':1,'batch_id':'thirdparty-csv-v1-'+hashlib.sha256(seed).hexdigest()[:24],'observed_at':observed_at,'update_policy':'fill_missing','dry_run':dry_run,'shopee_products':products,'shopee_skus':skus,'pdd_products':pdds,'associations':assocs}
|
|
if len(json.dumps(payload,ensure_ascii=False,separators=(',',':')).encode()) > MAX_BYTES:
|
|
error('单批请求超过 5 MB；请缩小每批商品数后重试')
|
|
selected=[];seen_pdd={};sku_count=0
|
|
return payload
|
|
for _, rows in groups(root):
|
|
group=make_group(rows); new_pdd=set(group[2])-set(seen_pdd)
|
|
if selected and (len(selected)+1+len(seen_pdd)+len(new_pdd)>MAX_PRODUCTS or sku_count+len(group[1])>MAX_SKUS): yield emit()
|
|
selected.append(group); seen_pdd.update(group[2]); sku_count+=len(group[1])
|
|
if selected: yield emit()
|
|
|
|
def main():
|
|
p=argparse.ArgumentParser(); p.add_argument('csv_dir'); p.add_argument('--base-url',required=True); p.add_argument('--observed-at',default='2026-08-20T00:00:00+08:00'); mode=p.add_mutually_exclusive_group(required=True); mode.add_argument('--dry-run',action='store_true'); mode.add_argument('--apply',action='store_true'); p.add_argument('--report',required=True); p.add_argument('--keep-existing-association',action='append',default=[],help='保留该蝦皮商品的数据库既有关联，不写来源新关联；可重复传入'); p.add_argument('--max-batches',type=int,default=0,help='本次最多处理几批；0 表示全部'); a=p.parse_args()
|
|
token=os.environ.get('CMAUTOBUY_CATALOG_TOKEN','').strip()
|
|
if not token: error('缺少环境变量 CMAUTOBUY_CATALOG_TOKEN')
|
|
root=Path(a.csv_dir)
|
|
endpoint=a.base_url.rstrip('/')+'/api/v1/integrations/catalog/batches'
|
|
total=Counter(); conflicts=[]; count=0; completed=[]
|
|
if Path(a.report).is_file():
|
|
previous=json.loads(Path(a.report).read_text(encoding='utf-8'))
|
|
if previous.get('mode') != ('preview' if a.dry_run else 'apply'): error('现有报告的模式不同，不能续跑')
|
|
completed=list(previous.get('completed_batch_ids',[])); total.update(previous.get('counts',{})); conflicts=list(previous.get('association_conflicts',[]))
|
|
completed_set=set(completed)
|
|
def checkpoint(finished):
|
|
save_report(a.report,{'mode':'preview' if a.dry_run else 'apply','completed':finished,'batches':len(completed),'counts':dict(total),'kept_existing_association_goods_ids':sorted(keep_existing),'association_conflicts':conflicts,'completed_batch_ids':completed})
|
|
keep_existing={value.strip() for value in a.keep_existing_association if value.strip()}
|
|
for payload in batches(root,a.observed_at,a.dry_run,keep_existing):
|
|
if payload['batch_id'] in completed_set: continue
|
|
if a.max_batches and count >= a.max_batches: break
|
|
count+=1; response=request(endpoint,token,payload)
|
|
if response.get('status') not in {'previewed','succeeded'}: error('接口未返回预期状态')
|
|
total.update(response.get('counts',{})); conflicts.extend(response.get('conflicts',[])); completed.append(payload['batch_id']); completed_set.add(payload['batch_id']); checkpoint(False)
|
|
print(f"[{count}] {response['status']} {payload['batch_id']}")
|
|
checkpoint(not any(payload['batch_id'] not in completed_set for payload in batches(root,a.observed_at,a.dry_run,keep_existing)))
|
|
print(f"本次完成：{count} 批；累计 {len(completed)} 批；报告：{a.report}")
|
|
if __name__=='__main__':
|
|
try: main()
|
|
except RuntimeError as exc: print(f'失败：{exc}',file=sys.stderr); raise SystemExit(1)
|