add order item constraint to order parse

2025-12-15 15:22:22 +08:00
parent 651de05bb5
commit 47b5fe2f8a
5 changed files with 358 additions and 42 deletions
@@ -0,0 +1,53 @@
+<?php
+
+declare(strict_types=1);
+
+namespace App\Platform;
+
+use App\Entity\Parse\EntityParse;
+use InvalidArgumentException;
+
+/**
+ * 订单解析器抽象基类
+ *
+ * 继承 EntityParse 的通用能力，并实现 OrderContract 契约
+ * 提供订单解析的通用方法和辅助函数
+ */
+abstract class AbstractOrderParse extends EntityParse implements OrderContract
+{
+    /**
+     * 解析订单子项数据（抽象方法，由子类实现）
+     *
+     * 每个平台有自己的订单子项数据结构，需要各自实现具体的解析逻辑
+     *
+     * @param array $orderItems 原始订单子项数组
+     * @param string $platformOrderId 平台订单 ID
+     * @return array 解析后的订单子项数据数组
+     */
+    abstract public function parseOrderItems(array $orderItems, string $platformOrderId): array;
+
+    /**
+     * 通用辅助方法：验证订单必需字段
+     *
+     * @param array $orderData 订单原始数据
+     * @param array $requiredFields 必需字段列表
+     * @throws InvalidArgumentException
+     */
+    protected function validateOrderFields(array $orderData, array $requiredFields = []): void
+    {
+    }
+
+    /**
+     * 通用辅助方法：格式化货币金额
+     *
+     * @param float|int|string|null $amount 金额
+     * @param int $decimals 小数位数，默认 2 位
+     * @return float 格式化后的金额
+     */
+    protected function formatAmount(float|int|string|null $amount, int $decimals = 2): float
+    {
+        return round((float)($amount ?? 0), $decimals);
+    }
+
+
+}
@@ -14,6 +14,7 @@ use Hyperf\Amqp\Producer;
 use PhpAmqpLib\Message\AMQPMessage;
 use Hyperf\Di\Annotation\Inject;
 use Hyperf\DbConnection\Db;
+use App\Model\OrderItem;
 use Hyperf\Context\ApplicationContext;
 use Throwable;

@@ -112,18 +113,34 @@ class OrderConsumer extends ConsumerMessage

            dump("Processing " . count($dataToUpsert) . " order(s) with batch upsert");

-            Db::beginTransaction();
+            // 分离订单数据和子项数据
+            $ordersData = [];
+            $itemsByPlatformOrderId = [];

-            // 使用 upsert 批量处理插入和更新
+            foreach ($dataToUpsert as $data) {
+                $ordersData[] = $data['order'];
+                $itemsByPlatformOrderId[$data['order']['platform_order_id']] = $data['items'] ?? [];
+            }
+
+            Db::beginTransaction();
+            // @attention 为考虑数据写入的时效性和执行效率，采用批量写入 + 事务方式处理
+            // @attention 如果多条记录中有个别记录请求失败，可能会导致该批次写入失败，此时则需要判断和修复
+
+            // 1. 使用 upsert 批量处理订单插入和更新
            // 利用数据库唯一索引自动判断是插入还是更新
            // 解决了重复订单推送的问题：存在则更新，不存在则插入
            $entity->newQuery()->upsert(
-                $dataToUpsert,
+                $ordersData,
                $parse->getUniqueBy(),        // 从解析器获取唯一键字段
                $parse->getUpdateFields()     // 从解析器获取可更新字段
            );

+            // 2. 处理订单子项
+            // 鉴于定义子项为了保留足够的灵活性，因此每次订单更新，我们都需要完整更新 OrderItem
+            $this->processOrderItems($itemsByPlatformOrderId);
+
            Db::commit();
+            // @TODO 触发事件通知，更新自动聚合任务

            // 在数据库事务中尝试对  $entityMapResult 中的元素进行持久化，如果没有问题， 则返回 ACK，否则这是 NACK 且 回滚事务。
            return Result::ACK;
@@ -244,4 +261,133 @@ class OrderConsumer extends ConsumerMessage
            ]);
        }
    }
+
+    /**
+     * 处理订单子项的批量同步（优化版本）
+     *
+     * 策略优化：使用业务键 (store_id, platform_order_id, sub_order_id) 作为唯一性约束
+     * 1. 直接批量 upsert OrderItems（无需先查询 order_id）
+     * 2. 批量更新 order_id（通过 JOIN orders 表）
+     * 3. 删除不在新数据中的旧 OrderItem（完全同步）
+     *
+     * 性能优势：减少一次批量查询，直接使用业务键进行 upsert
+     *
+     * @param array $itemsByPlatformOrderId 以 platform_order_id 为键的子项数据数组
+     * @return void
+     */
+    protected function processOrderItems(array $itemsByPlatformOrderId): void
+    {
+        if (empty($itemsByPlatformOrderId)) {
+            dump('No order items to process');
+            return;
+        }
+
+        // 1. 构建所有子项数据（无需查询 order_id）
+        $allItemsToUpsert = [];
+        $itemSubOrderIdsByPlatformOrderId = []; // 记录每个平台订单的新子项 ID 列表
+
+        foreach ($itemsByPlatformOrderId as $platformOrderId => $items) {
+            $subOrderIds = [];
+
+            foreach ($items as $item) {
+                // order_id 暂时设为 0，后续批量更新
+                $item['order_id'] = 0;
+
+                $allItemsToUpsert[] = $item;
+                $subOrderIds[] = $item['sub_order_id'];
+            }
+
+            $itemSubOrderIdsByPlatformOrderId[$platformOrderId] = $subOrderIds;
+        }
+
+        if (empty($allItemsToUpsert)) {
+            dump('No valid order items to upsert');
+            return;
+        }
+
+        dump("Upserting " . count($allItemsToUpsert) . " order items");
+
+        // 2. 批量 upsert OrderItems（使用业务键作为唯一性约束）
+        OrderItem::query()->upsert(
+            $allItemsToUpsert,
+            ['store_id', 'platform_order_id', 'sub_order_id'],  // 唯一键（业务键）
+            [
+                'company_id',
+                'platform_id',
+                'sub_order_type_id',
+                'product_id',
+                'platform_product_id',
+                'product_sku',
+                'product_barcode',
+                'unit_price',
+                'quantity',
+                'discount',
+                'total',
+                'ext',
+                'updated_at',
+            ]  // 可更新字段（不包括 order_id，需要单独更新）
+        );
+
+        // 3. 批量更新 order_id（通过 JOIN orders 表）
+        // UPDATE order_items SET order_id = orders.id
+        // FROM orders
+        // WHERE order_items.store_id = orders.store_id
+        //   AND order_items.platform_order_id = orders.platform_order_id
+        Db::update('
+            UPDATE order_items
+            SET order_id = orders.id
+            FROM orders
+            WHERE order_items.store_id = orders.store_id
+              AND order_items.platform_order_id = orders.platform_order_id
+              AND order_items.order_id = 0
+        ');
+
+        dump("Updated order_id for all items");
+
+        // 4. 批量删除不在新数据中的旧 OrderItem（完全同步策略）
+        // 优化：一次性删除所有不匹配的旧子项，而不是逐个订单处理
+        // 此部分业务应该很少被调用，订单子项新增或删减的情况很少见
+
+        if (!empty($allItemsToUpsert)) {
+            // 构建本次更新的所有 (store_id, platform_order_id) 组合（用于限定删除范围）
+            $updatedOrders = [];
+            foreach ($itemsByPlatformOrderId as $platformOrderId => $items) {
+                if (!empty($items)) {
+                    $storeId = (int)$items[0]['store_id'];
+                    $platformOrderId = addslashes($platformOrderId);
+                    $updatedOrders[] = "({$storeId}, '{$platformOrderId}')";
+                }
+            }
+
+            // 构建本次更新的所有 (store_id, platform_order_id, sub_order_id) 组合（保留的记录）
+            $validItems = [];
+            foreach ($allItemsToUpsert as $item) {
+                $storeId = (int)$item['store_id'];
+                $platformOrderId = addslashes($item['platform_order_id']);
+                $subOrderId = addslashes($item['sub_order_id']);
+                $validItems[] = "({$storeId}, '{$platformOrderId}', '{$subOrderId}')";
+            }
+
+            if (!empty($updatedOrders) && !empty($validItems)) {
+                $ordersIn = implode(', ', $updatedOrders);
+                $itemsNotIn = implode(', ', $validItems);
+
+                // 批量删除：删除在本次更新订单范围内，但不在新数据中的旧子项
+                // DELETE FROM order_items
+                // WHERE (store_id, platform_order_id) IN ((1, 'order1'), (1, 'order2'))
+                //   AND (store_id, platform_order_id, sub_order_id) NOT IN ((1, 'order1', 'item1'), ...)
+                $deleted = Db::delete("
+                    DELETE FROM order_items
+                    WHERE (store_id, platform_order_id) IN ({$ordersIn})
+                      AND (store_id, platform_order_id, sub_order_id) NOT IN ({$itemsNotIn})
+                ");
+
+                if ($deleted > 0) {
+                    dump("Batch deleted {$deleted} obsolete order items");
+                }
+            }
+        }
+
+        dump("Order items processing completed");
+    }
 }
@@ -0,0 +1,28 @@
+<?php
+
+declare(strict_types=1);
+
+namespace App\Platform;
+
+/**
+ * 订单解析器契约接口
+ *
+ * 定义所有电商平台订单解析器必须实现的方法
+ * 适用于：Shopee、Tmall、JD 等所有电商平台
+ */
+interface OrderContract
+{
+    /**
+     * 解析订单子项数据
+     *
+     * @param array $orderItems 原始订单子项数组
+     * @param string $platformOrderId 平台订单 ID
+     * @return array 解析后的订单子项数据数组，每个元素包含：
+     *               - company_id, platform_id, store_id
+     *               - platform_order_id, sub_order_id
+     *               - product_id, platform_product_id
+     *               - unit_price, quantity, discount, total
+     *               等字段
+     */
+    public function parseOrderItems(array $orderItems, string $platformOrderId): array;
+}
@@ -7,7 +7,7 @@ namespace App\Platform\Shopee\EntityParse;
 use App\Model\Company;
 use App\Model\Model as Entity;
 use App\Model\Store;
-use App\Entity\Parse\EntityParse;
+use App\Platform\AbstractOrderParse;
 use App\Platform\Shopee\Constants\OrderStatus;
 use App\Platform\Shopee\Constants\PaymentMethod;
 use Carbon\Carbon;
@@ -19,9 +19,10 @@ use InvalidArgumentException;
 /**
 * Shopee 订单解析器
 *
- * 展示如何继承 EntityParse 实现自定义解析逻辑
+ * 继承 AbstractOrderParse，实现 Shopee 平台特定的订单解析逻辑
+ * 实现 OrderContract 契约中定义的 parseOrderItems 方法
 */
-class Order extends EntityParse
+class Order extends AbstractOrderParse
 {
    /**
     * 公司作用域匹配
@@ -97,11 +98,18 @@ class Order extends EntityParse
            $records = isset($rawData[0]) ? $rawData : [$rawData];

            foreach ($records as $record) {
+                $_origin_order_item = $record['order_list']; // 'order_list' 为来自 shopee 原始数据的 订单子项信息
+
+                // 解析订单子项数据，传入 platform_order_id（实现 OrderContract 契约方法）
+                $order_items = $this->parseOrderItems($_origin_order_item, $record['order_sn']);

                // 根据实际业务需求映射其他字段
                // 映射每条原始数据到 Order Model 可用的数组格式
                // @attention 此处业务决定了 电商平台数据 和 数据仓库 之间的映射关系
+
+                // 返回包含订单数据和子项数据的结构
                yield [
+                    'order' => [
                        'company_id' => $this->getCompany()->id,
                        'platform_id' => 25,
                        'store_id' => $this->getStore()->id,
@@ -134,13 +142,56 @@ class Order extends EntityParse
                        'shipping_date' => $record['pickup_done_time'] > 0 ? Carbon::createFromTimestamp($record['pickup_done_time'], $this->getStore()->getTimezoneString()) : null,
                        'country' => $record['region'], // address 等地理位置信息因 shopee 数据保护协议已被隐藏, 目前仅提供 ISO 格式国家代码
                        'raw' => json_encode($record['raw']),
+                    ],
+                    'items' => $order_items,
                ];
            }
        });
    }

-    private function orderItemParse( array $order_items) : array {
-        return [];
+    /**
+     * 解析订单子项数据（实现 OrderContract 契约方法）
+     *
+     * Shopee API 文档：https://open.shopee.com/documents/v2/v2.order.get_order_detail?module=94&type=1
+     *
+     * @param array $orderItems Shopee 原始订单子项数组
+     * @param string $platformOrderId 平台订单 ID (order_sn)
+     * @return array 解析后的订单子项数据数组
+     */
+    public function parseOrderItems(array $orderItems, string $platformOrderId): array
+    {
+        $parsedItems = [];
+
+        foreach ($orderItems as $item) {
+            // 价格和数量
+            $originalPrice = (float)($item['model_original_price'] ?? 0);
+            $discountedPrice = (float)($item['model_discounted_price'] ?? $originalPrice);
+            $quantity = (int)($item['model_quantity_purchased'] ?? 1);
+
+            // SKU: 优先使用 model_sku（变体SKU），如果为空则使用 item_sku（商品SKU）
+            $sku = !empty($item['model_sku']) ? $item['model_sku'] : ($item['item_sku'] ?? '');
+
+            $parsedItems[] = [
+                'company_id' => $this->getCompany()->id,
+                'platform_id' => 25,
+                'store_id' => $this->getStore()->id,
+                // order_id 将在 Consumer 中补充
+                'platform_order_id' => $platformOrderId, // 平台订单 ID（从 order_sn 传入）
+                'sub_order_id' => (string)$item['order_item_id'], // 使用 order_item_id 作为订单子项的唯一标识
+                'sub_order_type_id' => 0,
+                'product_id' => 0, // 需要根据 platform_product_id 匹配，暂设为 0
+                'platform_product_id' => (string)$item['item_id'], // Shopee 商品 ID
+                'product_sku' => $sku,
+                'product_barcode' => '',
+                'unit_price' => $discountedPrice, // 实际成交单价（折扣后）
+                'quantity' => $quantity,
+                'discount' => ($originalPrice - $discountedPrice) * $quantity, // 总折扣金额
+                'total' => $discountedPrice * $quantity, // 总金额 = 折扣价 × 数量
+                'ext' => '', // 原始信息已在 Order.raw 中记录，此处留空
+            ];
+        }
+
+        return $parsedItems;
    }

    /**
@@ -0,0 +1,38 @@
+<?php
+
+use Hyperf\Database\Schema\Schema;
+use Hyperf\Database\Schema\Blueprint;
+use Hyperf\Database\Migrations\Migration;
+
+return new class extends Migration
+{
+    /**
+     * Run the migrations.
+     */
+    public function up(): void
+    {
+        Schema::table('order_items', function (Blueprint $table) {
+            // 索引1：基于数据库 ID 的唯一性（用于外键关联和基于 order_id 的查询）
+            $table->unique(['order_id', 'sub_order_id'], 'order_items_order_sub_order_unique');
+
+            // 索引2：基于业务键的唯一性（用于 upsert 操作，无需先查询 order_id）
+            // 与 orders 表的 (store_id, platform_order_id) 唯一索引配合使用
+            $table->unique(
+                ['store_id', 'platform_order_id', 'sub_order_id'],
+                'order_items_store_platform_sub_unique'
+            );
+        });
+    }
+
+    /**
+     * Reverse the migrations.
+     */
+    public function down(): void
+    {
+        Schema::table('order_items', function (Blueprint $table) {
+            // 删除两个唯一索引
+            $table->dropUnique('order_items_order_sub_order_unique');
+            $table->dropUnique('order_items_store_platform_sub_unique');
+        });
+    }
+};