支持补查订单

src/task_queue/delay_zset_listener.py

@@ -8,6 +8,7 @@ from loguru import logger
 
 from src.config.settings import get_settings
 from src.task_queue.dispatcher import TaskDispatcher
+from src.task_queue.task_messages import build_trade_monitor_task
 
 _CLAIM_DUE_MEMBERS_LUA = """
 local key = KEYS[1]
@@ -99,19 +100,7 @@ class RedisZSetDelayListener:
             data = json.loads(member_str)
             if not isinstance(data, dict):
                 raise ValueError("member JSON 不是对象结构")
-            now = int(time.time())
-            task_data = {
-                "task_id": f"id_{now}",
-                "topic": "surugaya_trade_monitor",
-                "purchase_account": "trdian023@tokugen.jp",
-                "site_id": "7",
-                "record_video": "",
-                "payload": {
-                    "trade_code": data.get("trade_code"),
-                },
-            }
-
-            return task_data
+            return build_trade_monitor_task(data.get("trade_code"))
         except Exception as e:
             raise ValueError(f"延时队列 member 解析失败: {e} | member={member_str}")
 

src/task_queue/listener.py

@@ -16,16 +16,23 @@ class RedisStreamListener:
     - 连接异常时安全重连。
     """
 
-    def __init__(self, dispatcher: TaskDispatcher) -> None:
+    def __init__(
+            self,
+            dispatcher: TaskDispatcher,
+            *,
+            stream_key: str | None = None,
+            group_name: str | None = None,
+            consumer_name: str | None = None,
+    ) -> None:
         self.settings = get_settings()
         self.dispatcher = dispatcher
         self.redis: aioredis.Redis | None = None
         self.running = False
 
-        # 从全局配置读取 Stream 相关参数
-        self.stream_key = self.settings.redis_stream_key
-        self.group_name = self.settings.redis_consumer_group
-        self.consumer_name = self.settings.redis_consumer_name
+        # 默认从全局配置读取 Stream 相关参数，子类/调用方可覆盖以监听其他 Stream
+        self.stream_key = stream_key or self.settings.redis_stream_key
+        self.group_name = group_name or self.settings.redis_consumer_group
+        self.consumer_name = consumer_name or self.settings.redis_consumer_name
 
     async def initialize(self) -> None:
         """
@@ -69,7 +76,7 @@ class RedisStreamListener:
             await self.initialize()
 
         self.running = True
-        logger.info("Redis Stream 监听循环已启动。")
+        logger.info(f"Redis Stream 监听循环已启动，stream='{self.stream_key}'。")
 
         # 1. 先恢复分配给本消费者但未 ACK 的 Pending 消息（崩溃重启清理）
         await self._consume_pending_messages()

src/task_queue/task_messages.py

@@ -0,0 +1,25 @@
+import time
+from typing import Any, Dict
+
+# surugaya_trade_monitor 任务的固定执行账号与站点
+TRADE_MONITOR_ACCOUNT = "trdian023@tokugen.jp"
+TRADE_MONITOR_SITE_ID = "7"
+
+
+def build_trade_monitor_task(trade_code: Any) -> Dict[str, Any]:
+    """
+    构造骏河屋交易订单获取任务消息（TaskDispatcher 兼容结构）。
+
+    供延时队列（trade_delay zset）与补抓队列（trade_fetch stream）
+    两条入口复用，保证 topic / 账号等路由字段只维护一份。
+    """
+    return {
+        "task_id": f"id_{int(time.time())}",
+        "topic": "surugaya_trade_monitor",
+        "purchase_account": TRADE_MONITOR_ACCOUNT,
+        "site_id": TRADE_MONITOR_SITE_ID,
+        "record_video": "",
+        "payload": {
+            "trade_code": trade_code,
+        },
+    }

src/task_queue/trade_fetch_listener.py

@@ -0,0 +1,32 @@
+from typing import Any, Dict
+
+from src.config.settings import get_settings
+from src.task_queue.dispatcher import TaskDispatcher
+from src.task_queue.listener import RedisStreamListener
+from src.task_queue.task_messages import build_trade_monitor_task
+
+
+class TradeFetchStreamListener(RedisStreamListener):
+    """
+    订单详情补抓 Stream 监听器。
+
+    消费 jp_surugaya API 侧 /api/trade 未命中订单详情时下发的
+    jp_surugaya_trade_fetch 消息（payload 仅含 trade_code），
+    包装为 surugaya_trade_monitor 任务后交给 TaskDispatcher。
+    """
+
+    def __init__(self, dispatcher: TaskDispatcher) -> None:
+        settings = get_settings()
+        super().__init__(
+            dispatcher,
+            stream_key=settings.redis_trade_fetch_stream_key,
+            group_name=settings.redis_trade_fetch_consumer_group,
+            consumer_name=settings.redis_trade_fetch_consumer_name,
+        )
+
+    def _parse_message(self, msg_id: bytes, raw_data: Dict[bytes, bytes]) -> Dict[str, Any]:
+        decoded_data = super()._parse_message(msg_id, raw_data)
+        task_data = build_trade_monitor_task(decoded_data.get("trade_code"))
+        # 保留 Stream 消息 ID，供任务完成后 ACK 使用
+        task_data["_stream_msg_id"] = decoded_data.get("_stream_msg_id")
+        return task_data