fux: mdy_plate_order sync pk conflict

2026-01-21 18:16:43 +08:00
parent 43399385a8
commit 49ac17cf37
6 changed files with 414 additions and 9 deletions
--- a/api_v1/mdy_plate_order_sync.py
+++ b/api_v1/mdy_plate_order_sync.py
@@ -4,6 +4,8 @@ from datetime import datetime
 import time
 from typing import Literal

+from django.db import connection
+from django.db.utils import IntegrityError
 from django.utils import timezone

 from api_v1 import models as api_models
@@ -19,6 +21,42 @@ from flower.utils.mingdaoyun.relations import (

 logger = logging.getLogger(__name__)

+_MDY_PLATE_ORDER_STAGING_PK_CONSTRAINT = "api_mdy_plate_order_staging_pkey"
+
+
+def _reset_mdy_plate_order_staging_id_sequence() -> None:
+    """修复 PostgreSQL 自增序列落后导致的主键重复（duplicate key violates ..._pkey）。
+
+    常见场景：数据库 restore/copy 后，sequence 未随数据同步到最新值。
+    """
+    table_name = api_models.MDYPlateOrderStaging._meta.db_table  # 'api_mdy_plate_order_staging'
+    pk_column = api_models.MDYPlateOrderStaging._meta.pk.column  # 'id'
+
+    with connection.cursor() as cursor:
+        cursor.execute(f"SELECT COALESCE(MAX({pk_column}), 1) FROM {table_name}")
+        max_id = cursor.fetchone()[0] or 1
+
+        cursor.execute("SELECT pg_get_serial_sequence(%s, %s)", [table_name, pk_column])
+        seq_name = cursor.fetchone()[0]
+        if not seq_name:
+            logger.warning("未找到 %s.%s 的 serial sequence，跳过 setval", table_name, pk_column)
+            return
+
+        # is_called=true => nextval 会返回 max_id+1
+        cursor.execute("SELECT setval(%s, %s, true)", [seq_name, max_id])
+        logger.warning(
+            "已重置 sequence=%s 到 max(id)=%s（下次插入将从 %s 开始）",
+            seq_name,
+            max_id,
+            max_id + 1,
+        )
+
+
+def _is_pk_duplicate_error(exc: Exception) -> bool:
+    """判断是否为主键约束冲突（多见于 sequence 未对齐）。"""
+    msg = str(exc) or ""
+    return _MDY_PLATE_ORDER_STAGING_PK_CONSTRAINT in msg
+

 def _parse_mdy_datetime(value: str | None):
    if not value:
@@ -196,15 +234,33 @@ def sync_mdy_plate_orders_to_staging(
            if not rowid:
                continue

-            api_models.MDYPlateOrderStaging.objects.update_or_create(
-                mdy_rowid=rowid,
-                defaults={
-                    "ctime": _parse_mdy_datetime(row.get("ctime")),
-                    "utime": _parse_mdy_datetime(row.get("utime")),
-                    "raw": row,
-                    "related": related_map.get(rowid, []),
-                },
-            )
+            defaults = {
+                "ctime": _parse_mdy_datetime(row.get("ctime")),
+                "utime": _parse_mdy_datetime(row.get("utime")),
+                "raw": row,
+                "related": related_map.get(rowid, []),
+            }
+
+            try:
+                api_models.MDYPlateOrderStaging.objects.update_or_create(
+                    mdy_rowid=rowid,
+                    defaults=defaults,
+                )
+            except IntegrityError as exc:
+                # 仅对“主键重复（通常为 sequence 未对齐）”做一次自愈重试。
+                if _is_pk_duplicate_error(exc):
+                    logger.warning(
+                        "检测到暂存表主键冲突，尝试重置序列后重试一次：rowid=%s err=%s",
+                        rowid,
+                        exc,
+                    )
+                    _reset_mdy_plate_order_staging_id_sequence()
+                    api_models.MDYPlateOrderStaging.objects.update_or_create(
+                        mdy_rowid=rowid,
+                        defaults=defaults,
+                    )
+                else:
+                    raise
            synced_rows += 1

            record_ctime = _parse_mdy_datetime(row.get("ctime"))
--- a/docs/2026-01-20_summary.md
+++ b/docs/2026-01-20_summary.md
@@ -39,3 +39,7 @@
 前端取值建议：

 - 若腾讯云 `SearchImage` 返回的 `ImageInfos[i].CustomContent` 可用，则直接作为可访问 URL 使用
+
+### 3) 补充独立文档（便于长期维护/不再翻工作日志）
+
+- 新增：`docs/tiia_image_gallery.md`（包含配置、上传函数、定时任务、失败落库、手动跑批命令、搜图 API、常见排障）
--- a/docs/2026-01-21_summary.md
+++ b/docs/2026-01-21_summary.md
@@ -0,0 +1,12 @@
+### 2026-01-21 工作记录
+
+#### 1) MDY（明道云）PlateOrder 同步任务/命令梳理
+
+- 确认 `PlateOrder`（开版数据表）同步能力已存在：既有 Celery task（`api_v1.tasks.sync_mdy_plate_orders`）也有 management command（`sync_mdy_plate_orders`）
+- 确认当前 `CELERY_BEAT_SCHEDULE` 未配置 PlateOrder 同步定时（已配置的仅有 mdy 产品/客户同步）
+- 补充独立说明文档：`docs/mdy_plate_order_sync.md`
+  - 包含：同步目标表/增量 checkpoint、是否定时、手动 command 用法、Celery task 触发方式、以及暂存查询 API 入口
+- 修复同步时偶发 `django.db.utils.IntegrityError: duplicate key value violates unique constraint "api_mdy_plate_order_staging_pkey"`：
+  - 根因：PostgreSQL 暂存表 `api_mdy_plate_order_staging` 的自增序列与现有最大 `id` 不一致
+  - 处理：在同步 service 中检测该类主键冲突后自动重置 sequence 并重试一次（避免任务/命令直接失败）
+
--- a/docs/api_v1_mdy_plate_order_staging.md
+++ b/docs/api_v1_mdy_plate_order_staging.md
@@ -113,3 +113,6 @@ GET /api/v1/mdy-plate-order-staging/?limit=20&offset=0&design_no=82724
 - 路由注册：`api_v1/urls.py`（`mdy-plate-order-staging`）
 - 字段映射：`flower/utils/mingdaoyun/mappings.py`（`plate_order_field_definitions`）
 - 碾平工具：`flower/utils/mingdaoyun/relations.py`（`flatten_row_by_field_definitions` / `normalize_mdy_value`）
+
+### 相关文档（同步如何跑）
+- 同步任务/command/是否定时：`docs/mdy_plate_order_sync.md`
--- a/docs/mdy_plate_order_sync.md
+++ b/docs/mdy_plate_order_sync.md
@@ -0,0 +1,169 @@
+# MDY（明道云）PlateOrder 同步：任务 / Command / 数据查询说明
+
+## 这份文档解决什么问题
+
+当你忘记：
+
+- PlateOrder（开版数据表）是否有定时同步？
+- 手动怎么跑同步？
+- 同步的数据落到哪里、怎么查？
+
+直接看本文档即可。
+
+---
+
+## 一、同步“是什么”：从明道云到哪里？
+
+### 1) 同步目标表（本地暂存）
+
+同步会把明道云“开版数据表”的行写入本地暂存表：
+
+- 模型：`api_v1.models.MDYPlateOrderStaging`
+- 字段：
+  - `mdy_rowid`：明道云 rowid（唯一）
+  - `raw`：整行原始数据（JSON）
+  - `related`：可选，跨表关联数据（list[dict]，已做碾平）
+  - `ctime/utime`：明道云系统时间字段（用于排序/增量）
+
+### 2) 增量游标（checkpoint）
+
+同步进度记录在：
+
+- 模型：`api_v1.models.DataSync`
+- 表名枚举：`DataSync.TableName.PLATE_ORDER`
+
+Command 默认会读取/写入 checkpoint；可以通过参数关闭。
+
+---
+
+## 二、有没有“定时任务”（Celery Beat）？
+
+结论：**当前没有把 PlateOrder 同步加入 `CELERY_BEAT_SCHEDULE`**。
+
+目前 `flower/settings.py` 里看到的 mdy 定时任务只有：
+
+- `mdy_product_sync`：每 10 分钟（`api_v1.tasks.sync_mdy_products`）
+- `mdy_customer_sync`：每 5 分钟（`api_v1.tasks.sync_mdy_customers`）
+
+但代码里确实存在 PlateOrder 的 Celery task（见下文），只是 **未配置到 beat schedule**。
+
+---
+
+## 三、如何手动运行（推荐：management command）
+
+### Command：`sync_mdy_plate_orders`
+
+入口文件：
+
+- `api_v1/management/commands/sync_mdy_plate_orders.py`
+
+最常用用法：
+
+- 默认增量同步（使用 DataSync checkpoint；默认抓关联表数据）：
+  - `uv run python manage.py sync_mdy_plate_orders`
+
+常用参数：
+
+- `--page-size 100`：每页拉取条数（默认 100）
+- `--max-pages 2`：本次最多处理多少页（不是最大页码）
+- `--max-records 200`：本次最多处理多少条
+- `--with-related / --without-related`：是否抓取跨表关联数据（默认 with-related）
+- `--max-related-per-type 5`：每种关联表最多拉多少条 rowid
+- `--request-interval-seconds 0.02`：每次明道云请求之间最小间隔（限流；0.02≈50qps）
+- `--use-checkpoint / --skip-checkpoint`：是否读取 DataSync 游标（默认 use）
+- `--update-checkpoint / --skip-checkpoint-write`：是否写入 DataSync 记录（默认写）
+- `--sort-direction asc|desc`：按 ctime 升/降序抓取（默认 asc）
+
+示例：
+
+- 只拉主表、不拉关联（更快）：
+  - `uv run python manage.py sync_mdy_plate_orders --without-related`
+- 从头全量跑一段（不读 checkpoint，且跑完不写 checkpoint）：
+  - `uv run python manage.py sync_mdy_plate_orders --skip-checkpoint --skip-checkpoint-write`
+- 控制单次任务规模（避免跑太久）：
+  - `uv run python manage.py sync_mdy_plate_orders --page-size 100 --max-pages 3 --max-records 250`
+
+---
+
+## 四、如何通过 Celery 运行（可选）
+
+### Celery Task：`api_v1.tasks.sync_mdy_plate_orders`
+
+位置：
+
+- `api_v1/tasks.py`（task 名：`sync_mdy_plate_orders`）
+
+说明：
+
+- 该 task 内部调用 `api_v1.mdy_plate_order_sync.sync_mdy_plate_orders_to_staging`
+- 该 task **目前不在 beat schedule**，但可以手动触发
+
+两种触发方式（任选其一）：
+
+1) 通过 celery call（适合已有 worker 环境）：
+   - `uv run celery -A flower call api_v1.tasks.sync_mdy_plate_orders --kwargs='{"page_size":100,"max_pages":2,"with_related":true,"request_interval_seconds":0.02}'`
+
+2) 通过 Django shell（适合本地临时验证）：
+   - `uv run python manage.py shell -c "from api_v1.tasks import sync_mdy_plate_orders; r=sync_mdy_plate_orders.delay(page_size=100,max_pages=2); print(r.id)"`
+
+> 如果你确实需要“定时同步 PlateOrder”，可以把该 task 加进 `CELERY_BEAT_SCHEDULE`；但这属于策略选择（数据量/频率/接口限流），当前项目默认没开。
+
+---
+
+## 五、同步后如何查询（API）
+
+### 暂存查询接口（只读）
+
+文档：
+
+- `docs/api_v1_mdy_plate_order_staging.md`
+
+接口：
+
+- List：`GET /api/v1/mdy-plate-order-staging/?limit=20&offset=0`
+- Detail：`GET /api/v1/mdy-plate-order-staging/{id}/`
+
+说明：
+
+- 支持 LimitOffset 分页
+- 支持按“内部字段名”的 query param 过滤（对前端隐藏 JSONField 细节）
+
+---
+
+## 六、相关实现入口（定位用）
+
+- 同步逻辑（写入暂存+checkpoint）：`api_v1/mdy_plate_order_sync.py`
+- 拉取明道云 rows：`flower/utils/mingdaoyun/fetch.py`（`fetch_plate_orders_from_mingdaoyun`）
+- 明道云客户端：`flower/utils/mingdaoyun/client.py`（当前 appKey/sign 仍是硬编码）
+- 暂存查询 API：`api_v1/views/mingdaoyun/plate_order_staging.py`
+
+---
+
+## 七、常见报错：`duplicate key value violates ... api_mdy_plate_order_staging_pkey`
+
+### 这到底是哪种“唯一性冲突”？
+
+`api_mdy_plate_order_staging_pkey` 是 **主键（id）** 的约束名，不是 `mdy_rowid` 的 unique。
+
+这类报错通常意味着：**PostgreSQL 自增序列（sequence）落后于当前表内的最大 id**（常见于 restore/copy/import 后）。
+
+### 当前代码如何处理
+
+`sync_mdy_plate_orders_to_staging` 在写入暂存表时，如果遇到该主键冲突，会：
+
+- 自动重置该表的 id sequence 到 `MAX(id)`（使下一次插入从 `MAX(id)+1` 开始）
+- 然后对当前 row 再重试一次写入
+
+对应实现：`api_v1/mdy_plate_order_sync.py` 的 `_reset_mdy_plate_order_staging_id_sequence()`。
+
+### 手动修复（可选）
+
+如果你想一次性手动修复，可在 PostgreSQL 执行（把 sequence 名替换为实际值）：
+
+```sql
+SELECT setval(
+  pg_get_serial_sequence('api_mdy_plate_order_staging', 'id'),
+  COALESCE((SELECT MAX(id) FROM api_mdy_plate_order_staging), 1),
+  true
+);
+```
--- a/docs/tiia_image_gallery.md
+++ b/docs/tiia_image_gallery.md
@@ -0,0 +1,161 @@
+# 腾讯云 TIIA 图库（PlateOrder 图片上传 & 搜图）说明
+
+## 背景与目标
+
+本项目接入腾讯云 TIIA（Image and Video AI）的“图库”能力，用于：
+
+- 将 `printing.PlateOrder` 的图片（通过 **ImageUrl**）上传到 TIIA 图库（CreateImage）
+- 通过输入图片（URL 或 Base64）在图库中做相似图检索（SearchImage）
+
+---
+
+## 一、配置（settings + .env）
+
+配置集中在 `flower/settings.py`，建议通过 `.env` 注入：
+
+- `TENCENTCLOUD_SECRET_ID`
+- `TENCENTCLOUD_SECRET_KEY`
+- `TENCENTCLOUD_TOKEN`（可选，STS）
+- `TENCENTCLOUD_TIIA_GROUP_ID`（当前默认 `168`）
+- `TENCENTCLOUD_TIIA_REGION`（默认 `ap-guangzhou`）
+- `TENCENTCLOUD_TIIA_ENDPOINT`（默认 `tiia.tencentcloudapi.com`）
+- `TENCENTCLOUD_TIIA_PIC_NAME_PREFIX`（默认 `plate_order`）
+- `TENCENTCLOUD_TIIA_QPS`（默认 `10`，用于限速）
+
+说明：
+
+- 本项目通过 `django-environ` 读取 `.env`（`Env.read_env(BASE_DIR / '.env')`）
+- 本项目依赖 `tencentcloud-sdk-python`；在容器/服务启动方式不同的情况下，请确保运行使用的是同一套依赖环境（推荐 `uv run ...`）。
+
+---
+
+## 二、上传到图库（CreateImage）
+
+### 2.1 基础上传函数（上传单张 ImageUrl）
+
+位置：`api_v1/utils/tencentcloud_tiia.py`
+
+- `upload_image_url_to_tencent_tiia(image_url, entity_id, tags=None, custom_content=None)`
+
+行为（当前约定）：
+
+- `GroupId` 固定从 settings 读取
+- `EntityId` 由调用方传入（在 PlateOrder 场景中= `str(plate_order_id)`）
+- `PicName` 自动生成（带 `TENCENTCLOUD_TIIA_PIC_NAME_PREFIX`）
+- `ImageUrl` 使用传入的 `image_url`
+- **`CustomContent` 默认写入完整 `image_url`**
+- **`Tags` 默认不写入**（避免触发“标签值长度过长”）
+
+> 注意：腾讯云 `SearchImage` 的返回结构默认不包含 `ImageUrl`；如果你希望在搜图结果里拿到可访问 URL，建议依赖 `CustomContent`（前提是腾讯云搜图返回里会回传该字段）。
+
+### 2.2 PlateOrder 批量上传（读取 plate_image）
+
+位置：`api_v1/utils/tencentcloud_tiia.py`
+
+- `upload_plate_order_images_to_tencent_tiia(plate_order_id, rate_limiter=None)`
+
+行为：
+
+- 查询 `printing.PlateOrder(id=plate_order_id)`
+- 从 `plate_image(JSONField)` 提取所有图片 URL/Path（兼容多种字段名与格式），并标准化为公网可访问 URL
+- 对每张图片调用 `upload_image_url_to_tencent_tiia(...)`
+- 单张失败不抛出，返回每张图片的结果列表（便于落库/排查）
+
+---
+
+## 三、定时任务（Celery Beat：每天 03:00 上传“昨天创建的订单”）
+
+### 3.1 Task 本体
+
+位置：`printing/tasks.py`
+
+- `upload_yesterday_plate_order_images_to_tencent_tiia`
+
+行为：
+
+- 计算“昨天”的日期范围 `[start, end)`
+- 遍历 `created_at` 落在该范围内的所有 `PlateOrder`
+- 逐单上传 `plate_image` 中所有图片
+- 遇错：记录失败并继续处理下一个订单
+- 使用 `SimpleRateLimiter(qps=TENCENTCLOUD_TIIA_QPS)` 控制 QPS（默认 10）
+
+### 3.2 Beat 调度配置
+
+位置：`flower/settings.py`
+
+- `CELERY_BEAT_SCHEDULE['daily_plate_order_tiia_image_upload']`
+  - `task = 'printing.tasks.upload_yesterday_plate_order_images_to_tencent_tiia'`
+  - `schedule = crontab(hour=3, minute=0)`
+
+说明：
+
+- 仅配置 schedule 不代表一定在跑；需要线上同时运行 **celery worker** + **celery beat**。
+
+---
+
+## 四、失败记录（可追踪/可重试）
+
+位置：`printing/models.py`
+
+- `PlateOrderTiiaUploadFailure`
+  - `run_date / plate_order_id / error / details / attempts / last_attempt_at`
+  - 唯一约束：`(run_date, plate_order_id)`，同日重复失败会累加 `attempts`
+
+可在 Django Admin 中查看（`printing/admin.py` 已注册）。
+
+---
+
+## 五、手动跑批（推荐：management command）
+
+位置：`printing/management/commands/tiia_upload_plate_order_images.py`
+
+常用：
+
+- 全量跑批：`uv run python manage.py tiia_upload_plate_order_images --all`
+- 跑昨天：`uv run python manage.py tiia_upload_plate_order_images --yesterday`
+- 跑某天：`uv run python manage.py tiia_upload_plate_order_images --date 2026-01-15`
+- 跑单个：`uv run python manage.py tiia_upload_plate_order_images --plate-order-id 123`
+- 小批量验证：`uv run python manage.py tiia_upload_plate_order_images --all --limit 20`
+- Dry-run：`uv run python manage.py tiia_upload_plate_order_images --all --dry-run`
+
+说明：
+
+- 同样遵守 `TENCENTCLOUD_TIIA_QPS` 的限速，并会落库失败记录。
+
+---
+
+## 六、搜图 API（简化版：SearchImage）
+
+### 6.1 接口
+
+- `POST /api/v1/tiia/search-image/`
+
+参数（JSON body）：
+
+- `imageUrl`：图片 URL（优先）
+- `imageBase64`：Base64（可选，若无 imageUrl）
+- `limit`：可选，1~100（默认腾讯云 10；本项目已透传）
+- `offset`：可选，>=0
+- `matchThreshold`：可选，0~100
+
+### 6.2 返回
+
+- 直接返回腾讯云 `SearchImage` 的原始响应 JSON
+
+### 6.3 已知限制/注意点
+
+- 腾讯云 `SearchImage` 返回结果通常以 `EntityId/PicName/Score/Tags/CustomContent` 为主，不一定直接提供可访问 `ImageUrl`
+- 如果期望在搜图结果里拿到可访问 URL，请优先依赖上传时写入的 `CustomContent`
+
+---
+
+## 七、排障清单（常见问题）
+
+- SDK 导入失败（`No module named 'tencentcloud'`）：
+  - 确认使用 `uv run python ...`（确保解释器与依赖一致）
+- 腾讯云返回“标签值长度过长”：
+  - 不要默认写入 `Tags`；本项目已改为默认不写 `Tags`
+- 任务未触发：
+  - 检查是否同时运行 celery worker + celery beat
+  - 检查 `CELERY_BROKER_URL / CELERY_RESULT_BACKEND` 是否可用
+