新闻

.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

src/config/settings.py

@@ -40,6 +40,13 @@ 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")
 
 
     # 浏览器控制

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(),

src/parsers/__init__.py

@@ -0,0 +1 @@
+"""worker 本地页面解析器。"""

src/parsers/surugaya_news.py

@@ -0,0 +1,159 @@
+"""骏河屋「お知らせ一覧」（action_news）通知列表解析器。
+
+页面地址：https://www.suruga-ya.jp/pcmypage/action_news
+通知表格由 DataTables 异步渲染，每行结构为：
+    <tr><td>2026/06/12</td>
+        <td><a data-type="2"
+               data-url="/pcmypage/action_sell_search/detail?trade_code=S2606115133"
+               data-id="869431007">【販売】(取引番号：S2606115133)発送準備中です。</a></td></tr>
+
+解析拆成几层：
+- `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，仅渲染当前页的 <tr>）。
+    逐页解析后点击「次へ」，以「首行 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())

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,  # 骏河屋通知（お知らせ）抓取任务
 }
 
 

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

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 兼容结构）。

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)