You've already forked pruchase_jhw
99 lines
3.8 KiB
Python
99 lines
3.8 KiB
Python
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)
|