feat: 规范 CSV 预检与安全入库 (#281)

admin/repository/catalog_import.go

@@ -18,6 +18,48 @@ var (
 	ErrCatalogImportRunNotFound = errors.New("商品目录批次不存在")
 )
 
+// CatalogEntityExists 仅用于目录导入预检，不会修改任何记录。
+func CatalogEntityExists(q Execer, table, goodsID string) (bool, error) {
+	var found int
+	var query string
+	switch table {
+	case "shopee_products":
+		query = `SELECT 1 FROM shopee_products WHERE goods_id=? AND deleted_at IS NULL`
+	case "pdd_products":
+		query = `SELECT 1 FROM pdd_products WHERE goods_id=? AND deleted_at IS NULL`
+	default:
+		return false, fmt.Errorf("不支持的目录表: %s", table)
+	}
+	err := q.QueryRow(query, goodsID).Scan(&found)
+	if errors.Is(err, sql.ErrNoRows) {
+		return false, nil
+	}
+	return err == nil, err
+}
+
+// CatalogSKUExists 按与实际写入一致的商品+规格键判断 SKU 是否已存在。
+func CatalogSKUExists(q Execer, goodsID, specKey string) (bool, error) {
+	var found int
+	err := q.QueryRow(`SELECT 1 FROM shopee_skus WHERE goods_id=? AND spec_key=?`, goodsID, specKey).Scan(&found)
+	if errors.Is(err, sql.ErrNoRows) {
+		return false, nil
+	}
+	return err == nil, err
+}
+
+// CatalogAssociationCurrentPDD 返回当前关联；空字符串表示该蝦皮商品没有 PDD 关联。
+func CatalogAssociationCurrentPDD(q Execer, goodsID string) (string, bool, error) {
+	var current sql.NullString
+	err := q.QueryRow(`SELECT pdd_goods_id FROM shopee_products WHERE goods_id=? AND deleted_at IS NULL`, goodsID).Scan(&current)
+	if errors.Is(err, sql.ErrNoRows) {
+		return "", false, nil
+	}
+	if err != nil {
+		return "", false, err
+	}
+	return current.String, true, nil
+}
+
 // InsertCatalogImportRun 先登记 processing 批次；表的复合主键是并发幂等的最终防线。
 func InsertCatalogImportRun(q Execer, run model.CatalogImportRun) error {
 	_, err := q.Exec(`INSERT INTO catalog_import_runs (
@@ -169,13 +211,13 @@ type CatalogProductOutcome struct {
 
 // UpsertCatalogShopeeProduct 独立保护主图和店铺字段，绝不覆盖人工 PDD 关联。
 func UpsertCatalogShopeeProduct(q Execer, in CatalogShopeeProductInput) (out CatalogProductOutcome, err error) {
-	var oldObserved, oldShopID, imageURL, shopName, imageSource, imageObserved, shopSource, shopObserved sql.NullString
+	var oldObserved, oldShopID, title, status, mainSKU, imageURL, shopName, imageSource, imageObserved, shopSource, shopObserved sql.NullString
 	var imageManual, shopManual int
 	shopID, err := FindShopIDByName(q, in.ShopName)
 	if err != nil {
 		return out, err
 	}
-	err = q.QueryRow(`SELECT source_observed_at,shop_id,image_url,shopee_shop_name,image_source,image_observed_at,image_is_manual,shop_name_source,shop_name_observed_at,shop_name_is_manual FROM shopee_products WHERE goods_id=?`, in.GoodsID).Scan(&oldObserved, &oldShopID, &imageURL, &shopName, &imageSource, &imageObserved, &imageManual, &shopSource, &shopObserved, &shopManual)
+	err = q.QueryRow(`SELECT source_observed_at,shop_id,title,shopee_status,main_sku_code,image_url,shopee_shop_name,image_source,image_observed_at,image_is_manual,shop_name_source,shop_name_observed_at,shop_name_is_manual FROM shopee_products WHERE goods_id=?`, in.GoodsID).Scan(&oldObserved, &oldShopID, &title, &status, &mainSKU, &imageURL, &shopName, &imageSource, &imageObserved, &imageManual, &shopSource, &shopObserved, &shopManual)
 	if errors.Is(err, sql.ErrNoRows) {
 		_, err = q.Exec(`INSERT INTO shopee_products(goods_id,shop_id,title,shopee_status,main_sku_code,image_url,shopee_shop_name,image_source,image_observed_at,image_is_manual,shop_name_source,shop_name_observed_at,shop_name_is_manual,source,source_observed_at,created_at,updated_at) VALUES(?,NULLIF(?,''),?,NULLIF(?,''),NULLIF(?,''),NULLIF(?,''),NULLIF(?,''),CASE WHEN TRIM(?)='' THEN NULL ELSE ? END,CASE WHEN TRIM(?)='' THEN NULL ELSE ? END,0,CASE WHEN TRIM(?)='' THEN NULL ELSE ? END,CASE WHEN TRIM(?)='' THEN NULL ELSE ? END,0,'api',?,?,?)`, in.GoodsID, shopID, in.Title, in.Status, in.MainSKU, in.ImageURL, in.ShopName, in.ImageURL, in.Source, in.ImageURL, in.ObservedAt, in.ShopName, in.Source, in.ShopName, in.ObservedAt, in.ObservedAt, in.Now, in.Now)
 		out.Created = err == nil
@@ -184,7 +226,10 @@ func UpsertCatalogShopeeProduct(q Execer, in CatalogShopeeProductInput) (out Cat
 	if err != nil {
 		return out, err
 	}
-	baseUpdate := !oldObserved.Valid || oldObserved.String <= in.ObservedAt
+	baseUpdate := in.UpdatePolicy == "overwrite_same_source" && (!oldObserved.Valid || oldObserved.String <= in.ObservedAt)
+	fillTitle := in.UpdatePolicy == "fill_missing" && strings.TrimSpace(in.Title) != "" && strings.TrimSpace(title.String) == ""
+	fillStatus := in.UpdatePolicy == "fill_missing" && strings.TrimSpace(in.Status) != "" && strings.TrimSpace(status.String) == ""
+	fillMainSKU := in.UpdatePolicy == "fill_missing" && strings.TrimSpace(in.MainSKU) != "" && strings.TrimSpace(mainSKU.String) == ""
 	newImage, newShop := imageURL.String, shopName.String
 	imageChanged, shopChanged := false, false
 	apply := func(incoming string, current *string, source, observed sql.NullString, manual int) (changed bool) {
@@ -241,10 +286,10 @@ func UpsertCatalogShopeeProduct(q Execer, in CatalogShopeeProductInput) (out Cat
 		resolvedShopID = shopID
 		shopAssociationChanged = true
 	}
-	if !baseUpdate && !imageChanged && !shopChanged && !shopAssociationChanged {
+	if !baseUpdate && !fillTitle && !fillStatus && !fillMainSKU && !imageChanged && !shopChanged && !shopAssociationChanged {
 		return out, nil
 	}
-	_, err = q.Exec(`UPDATE shopee_products SET title=CASE WHEN ? THEN ? ELSE title END,shopee_status=CASE WHEN ? THEN NULLIF(?,'') ELSE shopee_status END,main_sku_code=CASE WHEN ? THEN NULLIF(?,'') ELSE main_sku_code END,source=CASE WHEN ? THEN 'api' ELSE source END,source_observed_at=CASE WHEN ? THEN ? ELSE source_observed_at END,shop_id=CASE WHEN ? THEN NULLIF(?,'') ELSE shop_id END,image_url=NULLIF(?,''),shopee_shop_name=NULLIF(?,''),image_source=CASE WHEN ? THEN ? ELSE image_source END,image_observed_at=CASE WHEN ? THEN ? ELSE image_observed_at END,shop_name_source=CASE WHEN ? THEN ? ELSE shop_name_source END,shop_name_observed_at=CASE WHEN ? THEN ? ELSE shop_name_observed_at END,updated_at=? WHERE goods_id=?`, baseUpdate, in.Title, baseUpdate, in.Status, baseUpdate, in.MainSKU, baseUpdate, baseUpdate, in.ObservedAt, shopAssociationChanged, resolvedShopID, newImage, newShop, imageChanged, in.Source, imageChanged, in.ObservedAt, shopChanged, in.Source, shopChanged, in.ObservedAt, in.Now, in.GoodsID)
+	_, err = q.Exec(`UPDATE shopee_products SET title=CASE WHEN ? OR ? THEN ? ELSE title END,shopee_status=CASE WHEN ? OR ? THEN NULLIF(?,'') ELSE shopee_status END,main_sku_code=CASE WHEN ? OR ? THEN NULLIF(?,'') ELSE main_sku_code END,source=CASE WHEN ? THEN 'api' ELSE source END,source_observed_at=CASE WHEN ? THEN ? ELSE source_observed_at END,shop_id=CASE WHEN ? THEN NULLIF(?,'') ELSE shop_id END,image_url=NULLIF(?,''),shopee_shop_name=NULLIF(?,''),image_source=CASE WHEN ? THEN ? ELSE image_source END,image_observed_at=CASE WHEN ? THEN ? ELSE image_observed_at END,shop_name_source=CASE WHEN ? THEN ? ELSE shop_name_source END,shop_name_observed_at=CASE WHEN ? THEN ? ELSE shop_name_observed_at END,updated_at=? WHERE goods_id=?`, baseUpdate, fillTitle, in.Title, baseUpdate, fillStatus, in.Status, baseUpdate, fillMainSKU, in.MainSKU, baseUpdate, baseUpdate, in.ObservedAt, shopAssociationChanged, resolvedShopID, newImage, newShop, imageChanged, in.Source, imageChanged, in.ObservedAt, shopChanged, in.Source, shopChanged, in.ObservedAt, in.Now, in.GoodsID)
 	out.Updated = err == nil
 	return out, err
 }
@@ -351,8 +396,7 @@ func UpsertCatalogShopeeSKU(q Execer, in CatalogShopeeSKUInput) (CatalogSKUOutco
 		sources, times := catalogFieldProvenance(current)
 		changed := external != current.ShopeeSKUID.String
 		canFill := func(field, current, incoming string) bool {
-			return sources[field] != "manual" && strings.TrimSpace(incoming) != "" &&
-				(current == "" || (sources[field] == "syb" && in.Source != "syb"))
+			return sources[field] != "manual" && strings.TrimSpace(incoming) != "" && current == ""
 		}
 		if canFill("color", color, in.Color) {
 			color = in.Color
@@ -444,7 +488,7 @@ func catalogFieldProvenance(row *catalogSKUStored) (map[string]string, map[strin
 }
 
 // UpsertCatalogPddProduct 写入结构化 PDD 数据转换后的规范 JSON；空规格不清除既有采集结果。
-func UpsertCatalogPddProduct(q Execer, goodsID, url, title, shopName, skusJSON, observedAt, now string) (created, updated bool, err error) {
+func UpsertCatalogPddProduct(q Execer, goodsID, url, title, shopName, skusJSON, observedAt, now, updatePolicy string) (created, updated bool, err error) {
 	var oldObserved, deletedAt sql.NullString
 	err = q.QueryRow(`SELECT source_observed_at,deleted_at FROM pdd_products WHERE goods_id=?`, goodsID).Scan(&oldObserved, &deletedAt)
 	status := "pending"
@@ -460,13 +504,17 @@ func UpsertCatalogPddProduct(q Execer, goodsID, url, title, shopName, skusJSON,
 	if err != nil {
 		return false, false, err
 	}
-	if oldObserved.Valid && oldObserved.String > observedAt {
+	if updatePolicy == "insert_only" || (oldObserved.Valid && oldObserved.String > observedAt && updatePolicy != "fill_missing") {
 		return false, false, nil
 	}
-	if skusJSON == "" {
-		_, err = q.Exec(`UPDATE pdd_products SET url=?,title=CASE WHEN TRIM(?)='' THEN title ELSE ? END,
-			shop_name=CASE WHEN TRIM(?)='' THEN shop_name ELSE ? END,deleted_at=NULL,source='api',source_observed_at=?,updated_at=? WHERE goods_id=?`,
-			url, title, title, shopName, shopName, observedAt, now, goodsID)
+	if skusJSON == "" || updatePolicy == "fill_missing" {
+		if updatePolicy == "fill_missing" {
+			_, err = q.Exec(`UPDATE pdd_products SET url=CASE WHEN TRIM(COALESCE(url,''))='' THEN ? ELSE url END,title=CASE WHEN TRIM(COALESCE(title,''))='' THEN NULLIF(?,'') ELSE title END,shop_name=CASE WHEN TRIM(COALESCE(shop_name,''))='' THEN NULLIF(?,'') ELSE shop_name END,updated_at=? WHERE goods_id=?`, url, title, shopName, now, goodsID)
+		} else {
+			_, err = q.Exec(`UPDATE pdd_products SET url=?,title=CASE WHEN TRIM(?)='' THEN title ELSE ? END,
+				shop_name=CASE WHEN TRIM(?)='' THEN shop_name ELSE ? END,deleted_at=NULL,source='api',source_observed_at=?,updated_at=? WHERE goods_id=?`,
+				url, title, title, shopName, shopName, observedAt, now, goodsID)
+		}
 	} else {
 		_, err = q.Exec(`UPDATE pdd_products SET url=?,title=NULLIF(?,''),shop_name=NULLIF(?,''),skus_json=?,
 			collect_status='collected',collect_msg=NULL,artifact_ref=NULL,collected_at=?,deleted_at=NULL,

admin/service/catalog_import.go

@@ -73,6 +73,8 @@ type CatalogBatchRequest struct {
 	ShopeeSKUs     []CatalogShopeeSKU     `json:"shopee_skus"`
 	PddProducts    []CatalogPddProduct    `json:"pdd_products"`
 	Associations   []CatalogAssociation   `json:"associations"`
+	// DryRun 只执行数据库预检；不得登记导入批次或写入任何业务数据。
+	DryRun         bool                   `json:"dry_run"`
 }
 type CatalogCounts struct {
 	ShopeeCreated                 int `json:"shopee_created"`
@@ -99,6 +101,14 @@ type CatalogBatchResponse struct {
 	Replayed   bool          `json:"replayed"`
 	Counts     CatalogCounts `json:"counts"`
 	FinishedAt string        `json:"finished_at"`
+	Conflicts  []CatalogAssociationConflict `json:"conflicts,omitempty"`
+}
+
+// CatalogAssociationConflict 是预检发现的既有人工关联冲突，不包含请求体或凭据。
+type CatalogAssociationConflict struct {
+	ShopeeGoodsID string `json:"shopee_goods_id"`
+	ExistingPddID string `json:"existing_pdd_goods_id"`
+	IncomingPddID string `json:"incoming_pdd_goods_id"`
 }
 
 type CatalogError struct {
@@ -177,10 +187,52 @@ func ValidateCatalogBatch(req CatalogBatchRequest) error {
 	return nil
 }
 
+// PreviewCatalogBatch 按当前数据库做只读预检。它不登记 catalog_import_runs，
+// 也不写业务表；正式执行仍会重新检查，避免预检后人工修改造成覆盖。
+func PreviewCatalogBatch(db *sql.DB, req CatalogBatchRequest) (CatalogBatchResponse, error) {
+	if err := ValidateCatalogBatch(req); err != nil {
+		return CatalogBatchResponse{}, err
+	}
+	tx, err := db.BeginTx(context.Background(), &sql.TxOptions{ReadOnly: true, Isolation: sql.LevelReadCommitted})
+	if err != nil {
+		return CatalogBatchResponse{}, err
+	}
+	defer tx.Rollback()
+	counts := CatalogCounts{}
+	for _, p := range req.ShopeeProducts {
+		exists, err := repository.CatalogEntityExists(tx, "shopee_products", p.GoodsID)
+		if err != nil { return CatalogBatchResponse{}, err }
+		if exists { counts.ShopeeUpdated++ } else { counts.ShopeeCreated++ }
+	}
+	for _, p := range req.PddProducts {
+		exists, err := repository.CatalogEntityExists(tx, "pdd_products", p.GoodsID)
+		if err != nil { return CatalogBatchResponse{}, err }
+		if exists { counts.PddUpdated++ } else { counts.PddCreated++ }
+	}
+	for _, s := range req.ShopeeSKUs {
+		key, _ := spec.SpecKey(s.SpecRaw)
+		exists, err := repository.CatalogSKUExists(tx, s.GoodsID, key)
+		if err != nil { return CatalogBatchResponse{}, err }
+		if exists { counts.SKUSkipped++ } else { counts.SKUCreated++ }
+	}
+	conflicts := make([]CatalogAssociationConflict, 0)
+	for _, a := range req.Associations {
+		current, exists, err := repository.CatalogAssociationCurrentPDD(tx, a.ShopeeGoodsID)
+		if err != nil { return CatalogBatchResponse{}, err }
+		if !exists || current == "" { counts.AssociationCreated++; continue }
+		if current == a.PddGoodsID { counts.AssociationUnchanged++; continue }
+		conflicts = append(conflicts, CatalogAssociationConflict{ShopeeGoodsID: a.ShopeeGoodsID, ExistingPddID: current, IncomingPddID: a.PddGoodsID})
+	}
+	return CatalogBatchResponse{BatchID: req.BatchID, Status: "previewed", Counts: counts, FinishedAt: model.NowISO(), Conflicts: conflicts}, nil
+}
+
 func ImportCatalogBatch(db *sql.DB, source string, req CatalogBatchRequest, raw []byte) (CatalogBatchResponse, error) {
 	if err := ValidateCatalogBatch(req); err != nil {
 		return CatalogBatchResponse{}, err
 	}
+	if req.DryRun {
+		return PreviewCatalogBatch(db, req)
+	}
 	observed, _ := time.Parse(time.RFC3339Nano, req.ObservedAt)
 	req.ObservedAt = observed.UTC().Format(model.TimeLayout)
 	policy, _ := normalizeCatalogUpdatePolicy(req.UpdatePolicy)
@@ -239,7 +291,7 @@ func ImportCatalogBatch(db *sql.DB, source string, req CatalogBatchRequest, raw
 			b, _ := json.Marshal(map[string]any{"schema_version": 1, "goods_id": p.GoodsID, "title": p.Title, "shop_name": p.ShopName, "dimensions": p.Dimensions, "skus": p.SKUs})
 			skus = string(b)
 		}
-		c, u, e := repository.UpsertCatalogPddProduct(tx, p.GoodsID, p.URL, p.Title, p.ShopName, skus, req.ObservedAt, now)
+		c, u, e := repository.UpsertCatalogPddProduct(tx, p.GoodsID, p.URL, p.Title, p.ShopName, skus, req.ObservedAt, now, policy)
 		if e != nil {
 			return fail(e)
 		}

admin/service/catalog_import_test.go

@@ -356,3 +356,44 @@ func TestValidateCatalogBatch_拒绝危险图片URL(t *testing.T) {
 		t.Fatalf("合法图片被拒绝：%v", err)
 	}
 }
+
+func TestPreviewCatalogBatch_只读且报告既有关联冲突(t *testing.T) {
+	db := newCatalogTestDB(t)
+	base := validCatalogBatch()
+	raw, _ := json.Marshal(base)
+	if _, err := ImportCatalogBatch(db, "script-a", base, raw); err != nil {
+		t.Fatal(err)
+	}
+	preview := validCatalogBatch()
+	preview.BatchID = "preview-only"
+	preview.DryRun = true
+	preview.Associations[0].PddGoodsID = "P-2"
+	preview.PddProducts[0].GoodsID = "P-2"
+	preview.PddProducts[0].URL = "https://mobile.yangkeduo.com/goods.html?goods_id=P-2"
+	got, err := ImportCatalogBatch(db, "script-a", preview, []byte("unused"))
+	if err != nil || got.Status != "previewed" || len(got.Conflicts) != 1 {
+		t.Fatalf("预检结果错误：%+v %v", got, err)
+	}
+	var runs, pdd int
+	_ = db.QueryRow(`SELECT COUNT(*) FROM catalog_import_runs WHERE batch_id='preview-only'`).Scan(&runs)
+	_ = db.QueryRow(`SELECT COUNT(*) FROM pdd_products WHERE goods_id='P-2'`).Scan(&pdd)
+	if runs != 0 || pdd != 0 {
+		t.Fatalf("预检写入了数据：runs=%d pdd=%d", runs, pdd)
+	}
+}
+
+func TestImportCatalogBatch_FillMissing不覆盖已有PDD和商品字段(t *testing.T) {
+	db := newCatalogTestDB(t)
+	base := validCatalogBatch()
+	raw, _ := json.Marshal(base)
+	if _, err := ImportCatalogBatch(db, "script-a", base, raw); err != nil { t.Fatal(err) }
+	_, _ = db.Exec(`UPDATE shopee_products SET title='人工标题',main_sku_code='人工货号' WHERE goods_id='S-1'`)
+	_, _ = db.Exec(`UPDATE pdd_products SET url='https://manual.example/P-1',title='已采集标题',shop_name='已采集店铺',collect_status='collected' WHERE goods_id='P-1'`)
+	fill := validCatalogBatch(); fill.BatchID="no-overwrite"; fill.ShopeeProducts[0].Title="来源标题"; fill.ShopeeProducts[0].MainSKUCode="来源货号"; fill.PddProducts[0].URL="https://mobile.yangkeduo.com/goods.html?goods_id=P-1"; fill.PddProducts[0].Title="来源PDD标题"; fill.PddProducts[0].ShopName="来源PDD店铺"
+	raw, _ = json.Marshal(fill)
+	if _, err := ImportCatalogBatch(db, "thirdparty-csv", fill, raw); err != nil { t.Fatal(err) }
+	var title, mainSKU, url, pddTitle, shop, status string
+	_ = db.QueryRow(`SELECT title,main_sku_code FROM shopee_products WHERE goods_id='S-1'`).Scan(&title,&mainSKU)
+	_ = db.QueryRow(`SELECT url,title,shop_name,collect_status FROM pdd_products WHERE goods_id='P-1'`).Scan(&url,&pddTitle,&shop,&status)
+	if title!="人工标题" || mainSKU!="人工货号" || url!="https://manual.example/P-1" || pddTitle!="已采集标题" || shop!="已采集店铺" || status!="collected" { t.Fatalf("fill_missing 覆盖了已有字段：%q/%q/%q/%q/%q/%q",title,mainSKU,url,pddTitle,shop,status) }
+}

docs/admin/10-商品目录接入接口.md

@@ -20,6 +20,7 @@ Token 通过 `CMAUTOBUY_CATALOG_TOKEN` 或未提交的 `config.yaml` 配置。
   "batch_id": "source-file-20260811-001",
   "observed_at": "2026-08-11T10:00:00+08:00",
   "update_policy": "fill_missing",
+  "dry_run": false,
   "shopee_products": [
     {"goods_id": "S-1", "title": "蝦皮上衣", "status": "NORMAL", "main_sku_code": "A01", "image_url": "https://img.example.com/S-1.jpg", "shop_name": "蝦皮示例店铺"}
   ],
@@ -67,6 +68,17 @@ HTTP/HTTPS URL，最长 2048 字节；店铺名最长 500 个字符。Admin 只
 - 空关联可以建立、相同关联不重复写；已有不同 PDD 关联返回 409，绝不静默替换。
 - 批次缺少某条记录不表示删除，接口没有“全量覆盖”语义。
 
+### 3.1 数据库预检
+
+请求中的 `dry_run` 设为 `true` 时，接口只在只读事务中检查当前数据库，不登记
+`catalog_import_runs`，也不写入商品、SKU、PDD 或关联。响应 `status` 为
+`previewed`，`counts` 用于显示预计新增或现有记录数量；`conflicts` 列出同一蝦皮商品
+已有不同 PDD 关联的业务 ID。正式导入仍会重新执行相同的非覆盖和关联冲突判断。
+
+第三方规范 CSV 导入固定使用 `fill_missing`：已有蝦皮商品、SKU 和 PDD 商品的非空字段
+不被覆盖；PDD 已采集的标题、店铺、规格、价格和状态不被第三方 CSV 修改。相同关联
+幂等跳过，不同关联返回冲突，绝不替换。
+
 成功响应包含批次状态以及 SKU 新增、补空、同来源更新、人工跳过、旧数据跳过等计数。相同请求重放时
 `replayed=true`。
 

tools/import_thirdparty_catalog_csv.py

@@ -0,0 +1,93 @@
+#!/usr/bin/env python3
+"""把 #280 规范 CSV 分批送往 Admin 预检或安全入库。
+
+默认 --dry-run 调用 Admin 的只读数据库预检；只有显式 --apply 才会写入。
+Token 只从环境变量读取，报告中不会保存 Token 或完整请求体。
+"""
+from __future__ import annotations
+
+import argparse, csv, hashlib, json, os, sys
+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 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 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:
+        if row['spec_raw'] and row['spec_raw'] not in specs:
+            specs.add(row['spec_raw']); 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):
+    selected=[]; seen_pdd={}; sku_count=0
+    def emit():
+        nonlocal selected,seen_pdd,sku_count
+        products=[g[0] for g in selected]; skus=[s for g in selected for s in g[1]]
+        pdds=list(seen_pdd.values()); assocs=[a for g in selected for a in g[3]]
+        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); 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
+    for payload in batches(root,a.observed_at,a.dry_run):
+        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',[]))
+        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,'association_conflicts':conflicts},ensure_ascii=False,indent=2),encoding='utf-8')
+    print(f"完成：{count} 批；报告：{a.report}")
+if __name__=='__main__':
+    try: main()
+    except RuntimeError as exc: print(f'失败：{exc}',file=sys.stderr); raise SystemExit(1)