fix: 支持目录导入断点续跑 (#281)

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)