diff --git a/.env.example b/.env.example index b40580d..f39f8b0 100644 --- a/.env.example +++ b/.env.example @@ -25,6 +25,11 @@ REDIS_CONSUMER_NAME=purchase_worker_dev REDIS_DELAY_ZSET_KEY=jp_purchase_delay_task_dev REDIS_DELAY_POLL_INTERVAL=3600 REDIS_DELAY_BATCH_SIZE=10 +REDIS_NEWS_HASH_KEY=jp_surugaya_news_dev + +# 通知定时抓取 +NEWS_FETCH_ENABLED=true +NEWS_FETCH_INTERVAL_SECONDS=300 # 浏览器配置 BROWSER_HEADLESS=false diff --git a/src/config/settings.py b/src/config/settings.py index 93ea6cf..4e2fbd2 100644 --- a/src/config/settings.py +++ b/src/config/settings.py @@ -40,8 +40,15 @@ class Settings(BaseSettings): 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") + # 通知（お知らせ）抓取结果写入的 Redis Hash，field=data_id，中心 API 据此读取。 + # 默认值直接写字面量（不引用 redis_keys，兼容尚未发布该常量的旧版 surugaya-common）。 + redis_news_hash_key: str = Field(default="jp_surugaya_news", alias="REDIS_NEWS_HASH_KEY") + + # 通知定时抓取调度 + news_fetch_enabled: bool = Field(default=True, alias="NEWS_FETCH_ENABLED") + news_fetch_interval_seconds: int = Field(default=300, alias="NEWS_FETCH_INTERVAL_SECONDS") + - # 浏览器控制 browser_headless: bool = Field(default=True, alias="BROWSER_HEADLESS") browser_executable_path: str | None = Field(default=None, alias="BROWSER_EXECUTABLE_PATH") diff --git a/src/main.py b/src/main.py index 2e51fe6..5d55119 100644 --- a/src/main.py +++ b/src/main.py @@ -20,6 +20,7 @@ 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.news_scheduler import NewsFetchScheduler from src.task_queue.redis_manager import get_redis_manager from src.tasks.surugaya_auth import login_precheck from src.api.server import start_api_server @@ -60,6 +61,7 @@ async def main() -> None: stream_listener = RedisStreamListener(dispatcher) delay_listener = RedisZSetDelayListener(dispatcher) trade_fetch_listener = TradeFetchStreamListener(dispatcher) + news_scheduler = NewsFetchScheduler(dispatcher) # 4.1 启动截图查看 API 服务 api_runner = await start_api_server() @@ -73,6 +75,7 @@ async def main() -> None: asyncio.create_task(stream_listener.stop()) asyncio.create_task(delay_listener.stop()) asyncio.create_task(trade_fetch_listener.stop()) + asyncio.create_task(news_scheduler.stop()) # Windows 默认的 Proactor 事件循环不支持 add_signal_handler。 if sys.platform != "win32": @@ -90,7 +93,8 @@ async def main() -> None: stream_task = asyncio.create_task(stream_listener.listen()) delay_task = asyncio.create_task(delay_listener.listen()) trade_fetch_task = asyncio.create_task(trade_fetch_listener.listen()) - await asyncio.gather(stream_task, delay_task, trade_fetch_task) + news_task = asyncio.create_task(news_scheduler.listen()) + await asyncio.gather(stream_task, delay_task, trade_fetch_task, news_task) except asyncio.CancelledError: logger.warning("监听器轮询执行被事件循环取消。") except Exception as e: @@ -101,6 +105,7 @@ async def main() -> None: stream_listener.stop(), delay_listener.stop(), trade_fetch_listener.stop(), + news_scheduler.stop(), get_browser_manager().close_all(), get_redis_manager().close(), api_runner.cleanup(), diff --git a/src/parsers/__init__.py b/src/parsers/__init__.py new file mode 100644 index 0000000..8bdc500 --- /dev/null +++ b/src/parsers/__init__.py @@ -0,0 +1 @@ +"""worker 本地页面解析器。""" diff --git a/src/parsers/surugaya_news.py b/src/parsers/surugaya_news.py new file mode 100644 index 0000000..0b7832d --- /dev/null +++ b/src/parsers/surugaya_news.py @@ -0,0 +1,159 @@ +"""骏河屋「お知らせ一覧」（action_news）通知列表解析器。 + +页面地址：https://www.suruga-ya.jp/pcmypage/action_news +通知表格由 DataTables 异步渲染，每行结构为： + 2026/06/12 + 【販売】(取引番号：S2606115133)発送準備中です。 + +解析拆成几层： +- `extract_raw_rows(page)`：仅负责在浏览器侧抓取原始单元格（依赖 Playwright）。 +- `build_news_items(raw_rows)`：纯函数，把原始行规整为通知字典并从 data-url 解析 trade_code， + 无浏览器依赖，便于单元测试。 +- `parse_news_page(page)`：解析「当前已渲染页」的通知。 +- `collect_all_news(page)`：驱动 DataTables 逐页翻页，按 data_id 去重累积全部通知。 +""" +from __future__ import annotations + +from loguru import logger +from typing import Any +from urllib.parse import parse_qs, urlsplit + +from playwright.async_api import Page +from playwright.async_api import TimeoutError as PlaywrightTimeoutError + +# 通知列表所在的「お知らせ（取引）」标签页表格。 +NEWS_TABLE_ROWS_SELECTOR = "#table_news tbody tr" +# 行内承载通知数据的链接（带 data-id 才是有效通知行，可借此过滤空占位行）。 +NEWS_ROW_LINK_SELECTOR = "td a[data-id]" +# DataTables 翻页「次へ」按钮；最后一页时带 disabled 类。 +NEWS_NEXT_BUTTON_SELECTOR = "#table_news_next" + +# 等待 DataTables 异步渲染出首条通知链接的超时（毫秒）。 +NEWS_TABLE_RENDER_TIMEOUT_MS = 15000 +# 翻页后等待新一页渲染完成的超时（毫秒）。 +NEWS_PAGE_TURN_TIMEOUT_MS = 15000 +# 翻页安全上限，兜底防止异常情况下死循环（远高于真实页数）。 +NEWS_MAX_PAGES = 100 + + +def parse_trade_code(data_url: str | None) -> str | None: + """从 data-url 的查询串中解析 trade_code，非交易类通知返回 None。""" + if not data_url: + return None + query = urlsplit(data_url).query + values = parse_qs(query).get("trade_code") + if not values: + return None + trade_code = values[0].strip() + return trade_code or None + + +def build_news_items(raw_rows: list[dict[str, Any]]) -> list[dict[str, Any]]: + """把浏览器侧抓取的原始行规整为通知字典列表。 + + 缺少 data_id 的行视为无效（空占位/异常行）直接跳过；data_id 是站点侧 + 单条通知的唯一且单调递增标识，作为 Redis Hash 的 field 用于去重与排序。 + """ + items: list[dict[str, Any]] = [] + for row in raw_rows: + data_id = (row.get("data_id") or "").strip() + if not data_id: + continue + + data_url = (row.get("data_url") or "").strip() or None + items.append( + { + "data_id": data_id, + "date": (row.get("date") or "").strip(), + "message": (row.get("message") or "").strip(), + "data_type": (row.get("data_type") or "").strip(), + "data_url": data_url, + "trade_code": parse_trade_code(data_url), + } + ) + return items + + +async def extract_raw_rows(page: Page) -> list[dict[str, Any]]: + """在浏览器侧抓取通知表格的原始单元格数据（不做业务清洗）。""" + return await page.locator(NEWS_TABLE_ROWS_SELECTOR).evaluate_all( + """rows => rows.map((row) => { + const dateCell = row.querySelector(':scope > td'); + const link = row.querySelector('td a[data-id]'); + return { + date: dateCell ? (dateCell.innerText || '').trim() : '', + message: link ? (link.innerText || '').trim() : '', + data_type: link ? (link.getAttribute('data-type') || '') : '', + data_url: link ? (link.getAttribute('data-url') || '') : '', + data_id: link ? (link.getAttribute('data-id') || '') : '', + }; + })""" + ) + + +async def parse_news_page(page: Page) -> list[dict[str, Any]]: + """解析「当前已渲染页」的通知字典列表。""" + raw_rows = await extract_raw_rows(page) + return build_news_items(raw_rows) + + +async def _first_data_id(page: Page) -> str | None: + """当前页首行通知的 data-id；无有效行返回 None。""" + return await page.locator(NEWS_ROW_LINK_SELECTOR).first.get_attribute( + "data-id", timeout=1000 + ) + + +async def _next_button_disabled(page: Page) -> bool: + """「次へ」按钮是否处于禁用态（即已是最后一页）。""" + cls = await page.locator(NEWS_NEXT_BUTTON_SELECTOR).get_attribute("class") or "" + return "disabled" in cls.split() + + +async def collect_all_news(page: Page, max_pages: int = NEWS_MAX_PAGES) -> list[dict[str, Any]]: + """驱动 DataTables 逐页翻页，按 data_id 去重累积全部通知。 + + 通知表由 DataTables 异步渲染并分页（翻页走 ajax/JS，仅渲染当前页的 ）。 + 逐页解析后点击「次へ」，以「首行 data-id 变化」作为新页渲染完成信号； + 「次へ」带 disabled 类即最后一页，停止。始终无通知时返回空列表。 + """ + # 等待首页渲染出有效通知行；超时按「无通知」处理，不报错。 + try: + await page.locator(NEWS_ROW_LINK_SELECTOR).first.wait_for( + state="attached", timeout=NEWS_TABLE_RENDER_TIMEOUT_MS + ) + except PlaywrightTimeoutError: + logger.warning("等待通知表格渲染超时，按无通知处理。") + return [] + + collected: dict[str, dict[str, Any]] = {} + for page_index in range(max_pages): + for item in await parse_news_page(page): + collected.setdefault(item["data_id"], item) + + if await _next_button_disabled(page): + break + + before_id = await _first_data_id(page) + await page.locator(NEWS_NEXT_BUTTON_SELECTOR).click() + + # 等待新一页渲染：首行 data-id 与翻页前不同。 + try: + await page.wait_for_function( + """(beforeId) => { + const link = document.querySelector('#table_news tbody tr td a[data-id]'); + const cur = link ? link.getAttribute('data-id') : null; + return !!cur && cur !== beforeId; + }""", + arg=before_id, + timeout=NEWS_PAGE_TURN_TIMEOUT_MS, + ) + except PlaywrightTimeoutError: + logger.warning(f"翻页第 {page_index + 2} 页等待渲染超时，停止翻页。") + break + else: + logger.warning(f"通知翻页达到安全上限 {max_pages} 页，停止翻页。") + + return list(collected.values()) diff --git a/src/task_queue/dispatcher.py b/src/task_queue/dispatcher.py index 45b3461..26ff96f 100644 --- a/src/task_queue/dispatcher.py +++ b/src/task_queue/dispatcher.py @@ -5,11 +5,13 @@ from loguru import logger from src.tasks.surugaya import SurugayaPurchaseTask from src.tasks.surugaya_trade_monitor import SurugayaTradeMonitorTask +from src.tasks.surugaya_news_monitor import SurugayaNewsMonitorTask # topic 到采购任务类的映射注册表 TOPIC_TO_TASK_MAP: Dict[str, Type[Any]] = { "surugaya": SurugayaPurchaseTask, # 骏河屋采购任务 "surugaya_trade_monitor": SurugayaTradeMonitorTask, # 骏河屋交易订单获取任务 + "surugaya_news": SurugayaNewsMonitorTask, # 骏河屋通知（お知らせ）抓取任务 } diff --git a/src/task_queue/news_scheduler.py b/src/task_queue/news_scheduler.py new file mode 100644 index 0000000..b47af56 --- /dev/null +++ b/src/task_queue/news_scheduler.py @@ -0,0 +1,63 @@ +import asyncio + +from loguru import logger + +from src.config.accounts import load_accounts_config +from src.config.settings import get_settings +from src.task_queue.dispatcher import TaskDispatcher +from src.task_queue.task_messages import build_news_monitor_task + + +class NewsFetchScheduler: + """ + 通知（お知らせ）定时抓取调度器。 + + 与 Redis 监听器不同，这是纯时间驱动：每隔 news_fetch_interval_seconds， + 为每个已配置的骏河屋账号派发一条 surugaya_news 任务（启动后立即先跑一轮）。 + 任务经 TaskDispatcher 的同账号串行锁排队，不会与采购/订单任务争抢浏览器。 + """ + + def __init__(self, dispatcher: TaskDispatcher) -> None: + self.settings = get_settings() + self.dispatcher = dispatcher + self.running = False + self.interval_seconds = self.settings.news_fetch_interval_seconds + + async def listen(self) -> None: + if not self.settings.news_fetch_enabled: + logger.info("通知定时抓取已禁用（NEWS_FETCH_ENABLED=false），跳过。") + return + + self.running = True + logger.info( + f"通知定时抓取调度器已启动，间隔 {self.interval_seconds}s。" + ) + + while self.running: + try: + await self._dispatch_round() + except Exception as e: + logger.exception(f"通知定时抓取调度发生未预期错误: {e}") + + # 可中断的间隔等待：停机时尽快退出循环。 + for _ in range(self.interval_seconds): + if not self.running: + break + await asyncio.sleep(1) + + async def _dispatch_round(self) -> None: + accounts = load_accounts_config().sites.surugaya + if not accounts: + logger.debug("未配置骏河屋账号，本轮通知抓取跳过。") + return + + for account in accounts: + message = build_news_monitor_task(account.account_id) + logger.debug( + f"派发通知抓取任务: account='{account.account_id}', task_id='{message['task_id']}'。" + ) + await self.dispatcher.dispatch(message) + + async def stop(self) -> None: + logger.info("正在停止通知定时抓取调度器...") + self.running = False diff --git a/src/task_queue/task_messages.py b/src/task_queue/task_messages.py index fdf9249..8057b71 100644 --- a/src/task_queue/task_messages.py +++ b/src/task_queue/task_messages.py @@ -6,6 +6,23 @@ TRADE_MONITOR_ACCOUNT = "trdian023@tokugen.jp" TRADE_MONITOR_SITE_ID = "7" +def build_news_monitor_task(account_id: str) -> Dict[str, Any]: + """ + 构造骏河屋通知抓取任务消息（TaskDispatcher 兼容结构）。 + + 供通知定时调度器（NewsFetchScheduler）使用：每个配置账号一条任务， + 与采购/订单任务共用同账号串行锁，避免并发占用同一浏览器上下文。 + """ + return { + "task_id": f"news_{int(time.time())}", + "topic": "surugaya_news", + "purchase_account": account_id, + "site_id": TRADE_MONITOR_SITE_ID, + "record_video": "", + "payload": {}, + } + + def build_trade_monitor_task(trade_code: Any) -> Dict[str, Any]: """ 构造骏河屋交易订单获取任务消息（TaskDispatcher 兼容结构）。 diff --git a/src/tasks/surugaya_news_monitor.py b/src/tasks/surugaya_news_monitor.py new file mode 100644 index 0000000..9348466 --- /dev/null +++ b/src/tasks/surugaya_news_monitor.py @@ -0,0 +1,98 @@ +import json +import time +from typing import Any + +from loguru import logger + +from src.tasks import surugaya_auth +from src.tasks.base import BasePurchaseTask +from src.config.accounts import get_account_config +from src.parsers.surugaya_news import collect_all_news +from src.task_queue.redis_manager import get_redis_client + +# 「お知らせ一覧」页面地址。 +NEWS_URL = "https://www.suruga-ya.jp/pcmypage/action_news" + + +class SurugayaNewsMonitorTask(BasePurchaseTask): + """ + 骏河屋「お知らせ一覧」通知抓取任务。 + + 入参约定： + - topic: surugaya_news + - purchase_account: 抓取通知所属账号 + - payload: 无额外字段 + + 行为：打开 action_news 页面（必要时自动登录），解析首页最新通知， + 按 data_id 去重写入 Redis Hash（field=data_id），供中心 API 读取。 + """ + + async def execute(self) -> dict[str, Any]: + self.report_event = "surugaya_news" + self.is_report = False + + logger.debug(f"[任务 {self.task_id}] 骏河屋通知抓取工作流已启动。") + if not self.page: + raise ValueError("页面对象未初始化，无法执行通知抓取任务。") + + page = self.page + + await page.goto(NEWS_URL, wait_until="domcontentloaded", timeout=30000) + + # 登录兜底：长时间运行登录态可能过期，未登录则自动登录后重新进入。 + if not await self._ensure_login(page): + raise RuntimeError(f"账号 '{self.account_id}' 登录失败，无法抓取通知。") + + news_items = await collect_all_news(page) + logger.info(f"[任务 {self.task_id}] 账号 '{self.account_id}' 翻页解析到 {len(news_items)} 条通知。") + + written = await self._save_news_to_redis(news_items) + + return { + "account_id": self.account_id, + "parsed_count": len(news_items), + "written_count": written, + } + + async def _notify_task_failed(self) -> None: + """通知抓取是每 5 分钟运行的常规轮询，单次失败下轮即重试，不发失败告警邮件， + 避免登录抖动等偶发问题造成告警刷屏（失败现场截图仍由基类保留）。""" + logger.warning( + f"[任务 {self.task_id}] 通知抓取失败（不告警，等待下一轮重试）: {self.error_message}" + ) + + async def _ensure_login(self, page) -> bool: + log_prefix = f"[任务 {self.task_id}] " + if await surugaya_auth.check_login_status(page, log_prefix=log_prefix): + return True + + logger.warning(f"{log_prefix}账号 '{self.account_id}' 未登录，执行登录流程。") + account_config = get_account_config("surugaya", self.account_id) + if not account_config: + raise ValueError(f"未找到骏河屋账号配置: {self.account_id}") + + await surugaya_auth.surugaya_login(page, account_config, log_prefix=log_prefix) + await page.goto(NEWS_URL, wait_until="domcontentloaded", timeout=30000) + return await surugaya_auth.check_login_status(page, log_prefix=log_prefix) + + async def _save_news_to_redis(self, news_items: list[dict[str, Any]]) -> int: + """按 data_id 作为 field 写入 Redis Hash；返回写入条数。""" + if not news_items: + return 0 + + redis_client = get_redis_client() + hash_key = self.settings.redis_news_hash_key + fetched_at = int(time.time()) + + mapping = { + item["data_id"]: json.dumps( + {**item, "account_id": self.account_id, "fetched_at": fetched_at}, + ensure_ascii=False, + ) + for item in news_items + } + await redis_client.hset(hash_key, mapping=mapping) + logger.info( + f"[任务 {self.task_id}] 已将 {len(mapping)} 条通知写入 Redis Hash '{hash_key}'。" + ) + return len(mapping)