diff --git a/pyproject.toml b/pyproject.toml index d35638e..9e11cdf 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -9,10 +9,8 @@ authors = [ { name = "Antigravity Development Team" } ] dependencies = [ - # worker 自身已直接依赖 playwright，故 surugaya-common 无需 [browser] extra（extra 仅用于 - # 给没有 playwright 的消费者拉取 playwright），保持不带 extra 更简洁。 - "surugaya-common>=0.2.0", - "playwright>=1.54.0,<2.0.0", + # playwright 由 surugaya-common 的 [browser] extra 提供（>=1.54.0,<2.0.0），不再直接声明。 + "surugaya-common[browser]>=0.2.0", "pydantic>=2.7.0", "pydantic-settings>=2.2.0", "loguru>=0.7.2", diff --git a/src/config/settings.py b/src/config/settings.py index 05d1d24..01580a5 100644 --- a/src/config/settings.py +++ b/src/config/settings.py @@ -37,6 +37,9 @@ class Settings(BaseSettings): redis_delay_zset_key: str = Field(default=redis_keys.TRADE_DELAY_ZSET, alias="REDIS_DELAY_ZSET_KEY") redis_delay_poll_interval: int = Field(default=3600, alias="REDIS_DELAY_POLL_INTERVAL") redis_delay_batch_size: int = Field(default=10, alias="REDIS_DELAY_BATCH_SIZE") + redis_trade_fetch_stream_key: str = Field(default=redis_keys.TRADE_FETCH_STREAM, alias="REDIS_TRADE_FETCH_STREAM_KEY") + redis_trade_fetch_consumer_group: str = Field(default="surugaya_trade_fetch_group", alias="REDIS_TRADE_FETCH_CONSUMER_GROUP") + redis_trade_fetch_consumer_name: str = Field(default="surugaya_trade_fetch_worker", alias="REDIS_TRADE_FETCH_CONSUMER_NAME") # 浏览器控制 diff --git a/src/main.py b/src/main.py index 8f9e422..b826447 100644 --- a/src/main.py +++ b/src/main.py @@ -19,6 +19,7 @@ from src.utils.logger import setup_logger from src.task_queue.dispatcher import TaskDispatcher from src.task_queue.delay_zset_listener import RedisZSetDelayListener from src.task_queue.listener import RedisStreamListener +from src.task_queue.trade_fetch_listener import TradeFetchStreamListener from src.task_queue.redis_manager import get_redis_manager async def main() -> None: @@ -53,6 +54,7 @@ async def main() -> None: dispatcher = TaskDispatcher() stream_listener = RedisStreamListener(dispatcher) delay_listener = RedisZSetDelayListener(dispatcher) + trade_fetch_listener = TradeFetchStreamListener(dispatcher) # 5. 优雅停机信号处理 (跨平台支持) loop = asyncio.get_running_loop() @@ -62,6 +64,7 @@ async def main() -> None: # 触发监听器停止的协程任务 asyncio.create_task(stream_listener.stop()) asyncio.create_task(delay_listener.stop()) + asyncio.create_task(trade_fetch_listener.stop()) # Windows 默认的 Proactor 事件循环不支持 add_signal_handler。 if sys.platform != "win32": @@ -78,7 +81,8 @@ async def main() -> None: try: stream_task = asyncio.create_task(stream_listener.listen()) delay_task = asyncio.create_task(delay_listener.listen()) - await asyncio.gather(stream_task, delay_task) + trade_fetch_task = asyncio.create_task(trade_fetch_listener.listen()) + await asyncio.gather(stream_task, delay_task, trade_fetch_task) except asyncio.CancelledError: logger.warning("监听器轮询执行被事件循环取消。") except Exception as e: @@ -88,6 +92,7 @@ async def main() -> None: await asyncio.gather( stream_listener.stop(), delay_listener.stop(), + trade_fetch_listener.stop(), get_browser_manager().close_all(), get_redis_manager().close(), return_exceptions=True, diff --git a/src/task_queue/delay_zset_listener.py b/src/task_queue/delay_zset_listener.py index c5c1635..5f3bf48 100644 --- a/src/task_queue/delay_zset_listener.py +++ b/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}") diff --git a/src/task_queue/listener.py b/src/task_queue/listener.py index 56204db..7b1f5af 100644 --- a/src/task_queue/listener.py +++ b/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() diff --git a/src/task_queue/task_messages.py b/src/task_queue/task_messages.py new file mode 100644 index 0000000..fdf9249 --- /dev/null +++ b/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, + }, + } diff --git a/src/task_queue/trade_fetch_listener.py b/src/task_queue/trade_fetch_listener.py new file mode 100644 index 0000000..9e930a8 --- /dev/null +++ b/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 diff --git a/uv.lock b/uv.lock index 6d4c6d8..1555c27 100644 --- a/uv.lock +++ b/uv.lock @@ -165,14 +165,13 @@ dependencies = [ { name = "numpy" }, { name = "opencv-python" }, { name = "pillow" }, - { name = "playwright" }, { name = "pyautogui" }, { name = "pydantic" }, { name = "pydantic-settings" }, { name = "pyscreeze" }, { name = "pyyaml" }, { name = "redis" }, - { name = "surugaya-common" }, + { name = "surugaya-common", extra = ["browser"] }, ] [package.metadata] @@ -182,14 +181,13 @@ requires-dist = [ { name = "numpy", specifier = ">=2.4.6" }, { name = "opencv-python", specifier = ">=4.13.0.92" }, { name = "pillow", specifier = ">=12.2.0" }, - { name = "playwright", specifier = ">=1.54.0,<2.0.0" }, { name = "pyautogui", specifier = ">=0.9.54" }, { name = "pydantic", specifier = ">=2.7.0" }, { name = "pydantic-settings", specifier = ">=2.2.0" }, { name = "pyscreeze", specifier = ">=1.0.1" }, { name = "pyyaml", specifier = ">=6.0.1" }, { name = "redis", specifier = "==8.0.0" }, - { name = "surugaya-common", specifier = ">=0.2.0", index = "https://git.jerryyan.net/api/packages/jp/pypi/simple" }, + { name = "surugaya-common", extras = ["browser"], specifier = ">=0.2.0", index = "https://git.jerryyan.net/api/packages/jp/pypi/simple" }, ] [[package]] @@ -698,6 +696,11 @@ wheels = [ { url = "https://git.jerryyan.net/api/packages/jp/pypi/files/surugaya-common/0.2.0/surugaya_common-0.2.0-py3-none-any.whl", hash = "sha256:140587df6b60abf1f1fc3b592d0668a096ddd42f5216ee16d4b83a95783b58cd" }, ] +[package.optional-dependencies] +browser = [ + { name = "playwright" }, +] + [[package]] name = "typing-extensions" version = "4.15.0"