diff --git a/tools/import_thirdparty_catalog_csv.py b/tools/import_thirdparty_catalog_csv.py index 5385e1d..c80896b 100644 --- a/tools/import_thirdparty_catalog_csv.py +++ b/tools/import_thirdparty_catalog_csv.py @@ -6,7 +6,7 @@ Token 只从环境变量读取,报告中不会保存 Token 或完整请求体 """ from __future__ import annotations -import argparse, csv, hashlib, json, os, sys +import argparse, csv, hashlib, json, os, sys, tempfile from collections import Counter from pathlib import Path from urllib.error import HTTPError, URLError @@ -27,6 +27,13 @@ def request(endpoint, token, payload): 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): @@ -87,20 +94,29 @@ def batches(root, observed_at, dry_run, keep_existing): 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='保留该蝦皮商品的数据库既有关联,不写来源新关联;可重复传入'); a=p.parse_args() + 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 + 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',[])) + 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']}") - Path(a.report).write_text(json.dumps({'mode':'preview' if a.dry_run else 'apply','batches':count,'counts':total,'kept_existing_association_goods_ids':sorted(keep_existing),'association_conflicts':conflicts},ensure_ascii=False,indent=2),encoding='utf-8') - print(f"完成:{count} 批;报告:{a.report}") + 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)