From 2bcaab1626bbda3c1659907a51230d0f7e9b5747 Mon Sep 17 00:00:00 2001 From: Jerry Yan <792602257@qq.com> Date: Sat, 13 Jun 2026 20:00:45 +0800 Subject: [PATCH] =?UTF-8?q?=E9=80=9A=E7=9F=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .env.example | 4 +- app/api/routes/news.py | 72 +++++++++++++++++++ app/main.py | 2 + app/models/scrape.py | 14 ++++ .../src/surugaya_common/redis_keys.py | 13 ++++ tests/test_news.py | 64 +++++++++++++++++ 6 files changed, 167 insertions(+), 2 deletions(-) create mode 100644 app/api/routes/news.py create mode 100644 tests/test_news.py diff --git a/.env.example b/.env.example index f26159f..ca5a44a 100644 --- a/.env.example +++ b/.env.example @@ -38,8 +38,8 @@ SURUGAYA_PROXY_USERNAME= # 代理密码（如代理需要认证则填写） SURUGAYA_PROXY_PASSWORD= -# Redis（可选；用于跨进程/跨实例复用会话与限流状态） -# 留空表示不使用 Redis（退回进程内存） +# Redis（可选；用于跨进程/跨实例复用会话与限流状态；并为 /api/news 读取 worker 抓取的通知） +# 留空表示不使用 Redis（退回进程内存，且 /api/news 不可用） SURUGAYA_REDIS_HOST= SURUGAYA_REDIS_PORT=6379 SURUGAYA_REDIS_PASSWORD= diff --git a/app/api/routes/news.py b/app/api/routes/news.py new file mode 100644 index 0000000..08a5bcc --- /dev/null +++ b/app/api/routes/news.py @@ -0,0 +1,72 @@ +"""通知（お知らせ）路由：从 Redis 读取 worker 抓取的账号通知。 + +worker（pruchase_jhw）定时抓取骏河屋 action_news 页面，把每条通知按 data_id +作为 field 写入 Redis Hash（key 见 redis_keys.news_hash）。本路由读取该 Hash， +按 data_id 倒序（新→旧）返回，只读不删除。Redis 读取沿用 session_store 复用的连接， +与 scrape 路由保持一致。 +""" +from __future__ import annotations + +import json +import logging +from typing import Any + +from fastapi import APIRouter, Depends, Query + +from app.api.dependencies import get_container, require_bearer_token +from app.core.container import ServiceContainer +from app.models.scrape import ApiResponse, NewsItemData +from surugaya_common import redis_keys + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/api", tags=["news"]) + + +def _news_sort_key(item: dict[str, Any]) -> tuple[int, int]: + """排序键：优先按 data_id 数值倒序，非数值退化到最小，保证排序稳定。""" + raw = str(item.get("data_id", "")).strip() + if raw.isdigit(): + return (1, int(raw)) + return (0, 0) + + +@router.get( + "/news", + response_model=ApiResponse[list[NewsItemData]], + dependencies=[Depends(require_bearer_token)], +) +async def list_news( + limit: int | None = Query(default=None, ge=1, le=500, description="返回条数上限，省略则返回全部"), + container: ServiceContainer = Depends(get_container), +) -> ApiResponse[list[NewsItemData]]: + """读取通知列表（按 data_id 倒序，新→旧），只读不删除。""" + if container.session_store._redis is None: + return ApiResponse[list[NewsItemData]]( + success=False, + msg="Redis 未配置", + data=None, + code=500, + ) + + # dev/prod 由本服务 app_env 推导，需与 worker 写入端（APP_ENV）保持一致。 + is_dev = "1" if container.settings.app_env == "dev" else "0" + raw_map = await container.session_store._redis.hgetall(redis_keys.news_hash(is_dev)) + + items: list[dict[str, Any]] = [] + for data_id, raw_value in raw_map.items(): + try: + items.append(json.loads(raw_value)) + except (json.JSONDecodeError, TypeError): + logger.warning("通知 data_id=%s 的 JSON 解析失败，已跳过。", data_id) + + items.sort(key=_news_sort_key, reverse=True) + if limit is not None and limit > 0: + items = items[:limit] + + return ApiResponse[list[NewsItemData]]( + success=True, + msg="success", + data=[NewsItemData(**item) for item in items], + code=0, + ) diff --git a/app/main.py b/app/main.py index 4674439..6436ecf 100644 --- a/app/main.py +++ b/app/main.py @@ -18,6 +18,7 @@ from starlette.exceptions import HTTPException as StarletteHTTPException from app.api.routes.health import router as health_router from app.api.routes.scrape import router as scrape_router +from app.api.routes.news import router as news_router from app.core.config import get_settings from app.core.container import ServiceContainer from app.core.errors import AppError @@ -90,6 +91,7 @@ def create_app() -> FastAPI: app = FastAPI(title="Surugaya Scraper Service", lifespan=lifespan) app.include_router(health_router) app.include_router(scrape_router) + app.include_router(news_router) @app.exception_handler(AppError) async def app_error_handler(_: Request, exc: AppError) -> JSONResponse: diff --git a/app/models/scrape.py b/app/models/scrape.py index 7d69152..f77ddfa 100644 --- a/app/models/scrape.py +++ b/app/models/scrape.py @@ -207,3 +207,17 @@ class CheckTaskCallbackUrlData(BaseModel): """测试回调地址响应数据""" response_status: int response_body: Any + + +class NewsItemData(BaseModel): + """单条「お知らせ」通知，由 worker 解析 action_news 页面后写入 Redis。""" + model_config = ConfigDict(extra="allow") + + data_id: str # 站点侧通知唯一标识（单调递增） + date: str = "" # 通知日期，如 2026/06/12 + message: str = "" # 通知文案，如 【販売】(取引番号：S2606115133)発送準備中です。 + data_type: str = "" # 站点 data-type + data_url: str | None = None # 站点 data-url + trade_code: str | None = None # 从 data-url 解析的交易单号（非交易通知为 null） + account_id: str = "" # 该通知所属抓取账号 + fetched_at: int = 0 # worker 抓取时间（epoch 秒） diff --git a/packages/surugaya_common/src/surugaya_common/redis_keys.py b/packages/surugaya_common/src/surugaya_common/redis_keys.py index 999797c..b427755 100644 --- a/packages/surugaya_common/src/surugaya_common/redis_keys.py +++ b/packages/surugaya_common/src/surugaya_common/redis_keys.py @@ -10,6 +10,7 @@ Current literals (as of this extraction): trade hash : jp_surugaya_trade trade pending dedup : jp_surugaya_trade_pending:{trade_code} trade fetch stream : jp_surugaya_trade_fetch + news hash : jp_surugaya_news[_dev] """ from __future__ import annotations @@ -21,6 +22,7 @@ TRADE_DELAY_ZSET: str = "jp_surugaya_trade_delay" TRADE_HASH: str = "jp_surugaya_trade" TRADE_PENDING_PREFIX: str = "jp_surugaya_trade_pending" TRADE_FETCH_STREAM: str = "jp_surugaya_trade_fetch" +NEWS_HASH: str = "jp_surugaya_news" # Suffix appended to dev-mode keys. _DEV_SUFFIX: str = "_dev" @@ -78,3 +80,14 @@ def trade_pending_key(trade_code: object) -> str: def trade_fetch_stream() -> str: """Redis Stream key for trade fetch tasks.""" return TRADE_FETCH_STREAM + + +def news_hash(is_dev: object = None) -> str: + """Redis Hash key storing account notifications keyed by data_id. + + Reproduces: + "jp_surugaya_news_dev" if is_dev == "1" else "jp_surugaya_news" + """ + if _is_dev(is_dev): + return f"{NEWS_HASH}{_DEV_SUFFIX}" + return NEWS_HASH diff --git a/tests/test_news.py b/tests/test_news.py new file mode 100644 index 0000000..c6c4fdf --- /dev/null +++ b/tests/test_news.py @@ -0,0 +1,64 @@ +"""通知（お知らせ）读取链路测试：redis_keys 契约 + /api/news 路由排序/截断/降级。""" +from __future__ import annotations + +import json +from types import SimpleNamespace + +from app.api.routes.news import list_news +from app.core.config import Settings +from surugaya_common import redis_keys + + +class FakeRedis: + """最小化异步 Redis 替身，仅实现 hgetall（值已 decode 为 str）。""" + + def __init__(self, mapping: dict[str, str]): + self._mapping = mapping + + async def hgetall(self, key: str) -> dict[str, str]: + return dict(self._mapping) + + +def _container(redis_obj, app_env: str = "prod") -> SimpleNamespace: + settings = Settings(_env_file=None, app_env=app_env) + session_store = SimpleNamespace(_redis=redis_obj) + return SimpleNamespace(settings=settings, session_store=session_store) + + +def _news_value(data_id: str, **extra) -> str: + return json.dumps({"data_id": data_id, "message": f"msg-{data_id}", **extra}, ensure_ascii=False) + + +class TestNewsHashKey: + def test_literals(self): + assert redis_keys.NEWS_HASH == "jp_surugaya_news" + assert redis_keys.news_hash() == "jp_surugaya_news" + assert redis_keys.news_hash("0") == "jp_surugaya_news" + assert redis_keys.news_hash("1") == "jp_surugaya_news_dev" + + +async def test_list_news_sorted_desc_by_data_id(): + redis_obj = FakeRedis( + {"100": _news_value("100"), "300": _news_value("300", trade_code="S300"), "200": _news_value("200")} + ) + resp = await list_news(limit=None, container=_container(redis_obj)) + assert resp.success and resp.code == 0 + assert [i.data_id for i in resp.data] == ["300", "200", "100"] + assert resp.data[0].trade_code == "S300" + + +async def test_list_news_limit(): + redis_obj = FakeRedis({str(i): _news_value(str(i)) for i in range(1, 11)}) + resp = await list_news(limit=3, container=_container(redis_obj)) + assert [i.data_id for i in resp.data] == ["10", "9", "8"] + + +async def test_list_news_skips_bad_json(): + redis_obj = FakeRedis({"1": _news_value("1"), "2": "not-json"}) + resp = await list_news(limit=None, container=_container(redis_obj)) + assert [i.data_id for i in resp.data] == ["1"] + + +async def test_list_news_redis_not_configured(): + resp = await list_news(limit=None, container=_container(None)) + assert not resp.success and resp.code == 500 and resp.data is None