From fe64d3b3c713163b93b9f95dd8e0ff7c541c2117 Mon Sep 17 00:00:00 2001 From: chengma Date: Thu, 20 Aug 2026 14:41:01 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E8=A7=84=E8=8C=83=20CSV=20=E9=A2=84?= =?UTF-8?q?=E6=A3=80=E4=B8=8E=E5=AE=89=E5=85=A8=E5=85=A5=E5=BA=93=20(#281)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- admin/repository/catalog_import.go | 74 ++++++++++++++++---- admin/service/catalog_import.go | 54 ++++++++++++++- admin/service/catalog_import_test.go | 41 ++++++++++++ docs/admin/10-商品目录接入接口.md | 12 ++++ tools/import_thirdparty_catalog_csv.py | 93 ++++++++++++++++++++++++++ 5 files changed, 260 insertions(+), 14 deletions(-) create mode 100644 tools/import_thirdparty_catalog_csv.py diff --git a/admin/repository/catalog_import.go b/admin/repository/catalog_import.go index 38c2382..7ceea1b 100644 --- a/admin/repository/catalog_import.go +++ b/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(¤t) + 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, diff --git a/admin/service/catalog_import.go b/admin/service/catalog_import.go index 7263e85..3faed7f 100644 --- a/admin/service/catalog_import.go +++ b/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) } diff --git a/admin/service/catalog_import_test.go b/admin/service/catalog_import_test.go index 366d050..a2248fd 100644 --- a/admin/service/catalog_import_test.go +++ b/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) } +} diff --git a/docs/admin/10-商品目录接入接口.md b/docs/admin/10-商品目录接入接口.md index c307051..6801f92 100644 --- a/docs/admin/10-商品目录接入接口.md +++ b/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`。 diff --git a/tools/import_thirdparty_catalog_csv.py b/tools/import_thirdparty_catalog_csv.py new file mode 100644 index 0000000..4ac1212 --- /dev/null +++ b/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)