q792602257andClaude Opus 5
104d7fef6b
把需要账号登录态的链路从抓取服务里拆出成独立进程。分界线不是「要不要登录」， 而是抓取无状态、幂等、可多开实例，而交易的写操作不可逆、登录态全局唯一、 订单监控是常驻轮询——同进程时抓取一扩容就会复制出 N 份登录态与 N 个轮询， 同一账号会被并发操作。 - app/shared：配置、错误码、日志、ApiResponse 信封 + Bearer 鉴权 + 异常处理器、 导航请求头构造器 - app/scraping：站点常量、会话、解析器与 10 个抓取接口，:31107，可多开 - app/trading：登录态查询/重载与健康检查，:31108，只能单实例 - 依赖方向锁为 scraping→shared、trading→shared，两侧互不 import； tests/test_architecture.py 用 AST 检查 import 并校验两个 app 的路径不串 - 登录态 UA 在 trading 独立持有：与抓取 UA 值相同但变更理由不同，抓取 UA 为绕 反爬可随时调整，登录 UA 一改可能触发设备校验使已落盘 cookie 失效 - scripts/login.py 与 AuthSession 共用 auth_site.PROFILES 与 is_logged_in，判据只写一遍 - 同一镜像两个启动命令，交易容器覆盖 command 并设 RAKUTEN_HEALTH_PORT 同时带上此前未提交的 ラクマ 分类接口与登录态基础设施。 验证：239 个离线用例全绿；两个入口真实启动，/health 与鉴权正常。 未验证：真实探测登录态（当前开发机无外网，对站点的连接全部超时）。 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
110 lines
4.4 KiB
Python
110 lines
4.4 KiB
Python
"""分类树解析测试：基于真实页面状态样本"""
|
|
import pytest
|
|
|
|
from app.scraping.core import site
|
|
from app.shared.errors import ScrapeParseError
|
|
from app.scraping.parsers.genre import parse_genres
|
|
from app.scraping.utils.urls import build_genre_url
|
|
|
|
|
|
# ---- URL 构建 ----
|
|
|
|
def test_genre_url_uses_category_page_when_id_given():
|
|
assert build_genre_url("101205") == "https://www.rakuten.co.jp/category/101205/"
|
|
|
|
|
|
def test_genre_url_falls_back_to_sentinel_search_for_top_level():
|
|
"""顶层列表没有对应的分类页，改用查不到东西的哨兵关键词取干净的分类分面"""
|
|
url = build_genre_url(None)
|
|
assert url == f"https://search.rakuten.co.jp/search/mall/{site.GENRE_FACET_PROBE_KEYWORD}/"
|
|
|
|
|
|
# ---- 顶层分类 ----
|
|
|
|
def test_root_returns_all_top_level_genres(genre_root_state):
|
|
result = parse_genres(genre_root_state, genre_id=None)
|
|
assert result.genre_id == ""
|
|
assert result.ancestors == []
|
|
assert len(result.children) == 39
|
|
names = {child.name for child in result.children}
|
|
assert "テレビゲーム" in names
|
|
assert "レディースファッション" in names
|
|
|
|
|
|
def test_root_children_expose_id_shortcut_and_url(genre_root_state):
|
|
result = parse_genres(genre_root_state, genre_id=None)
|
|
game = next(child for child in result.children if child.name == "テレビゲーム")
|
|
assert game.genre_id == "101205"
|
|
assert game.shortcut == "game"
|
|
assert game.url == "https://www.rakuten.co.jp/category/101205/"
|
|
|
|
|
|
def test_root_omits_item_count_because_it_is_query_scoped(genre_root_state):
|
|
"""站点给的是「当前查询在该分类下的命中数」，不是分类商品总量，不能透出"""
|
|
result = parse_genres(genre_root_state, genre_id=None)
|
|
assert all(child.item_count is None for child in result.children)
|
|
|
|
|
|
# ---- 指定分类 ----
|
|
|
|
def test_genre_node_returns_itself_and_direct_children(genre_node_state):
|
|
result = parse_genres(genre_node_state, genre_id="101205")
|
|
assert result.genre_id == "101205"
|
|
assert result.name == "テレビゲーム"
|
|
assert result.is_leaf is False
|
|
assert result.url == "https://www.rakuten.co.jp/category/101205/"
|
|
assert len(result.children) == 27
|
|
child_names = {child.name for child in result.children}
|
|
assert "Nintendo Switch" in child_names
|
|
|
|
|
|
def test_genre_children_carry_item_count(genre_node_state):
|
|
result = parse_genres(genre_node_state, genre_id="101205")
|
|
switch = next(child for child in result.children if child.name == "Nintendo Switch")
|
|
assert switch.genre_id == "565950"
|
|
assert switch.item_count and switch.item_count > 0
|
|
|
|
|
|
def test_top_level_genre_has_no_ancestors_and_carries_genre_info(genre_node_state):
|
|
result = parse_genres(genre_node_state, genre_id="101205")
|
|
# 虚拟根 id=0 不是真实分类，不应出现在祖先里
|
|
assert result.ancestors == []
|
|
assert result.full_name == "テレビゲーム"
|
|
assert result.description
|
|
|
|
|
|
def test_nested_genre_reports_its_ancestor_chain(genre_node_state):
|
|
"""深层分类要能回报从顶层到父级的路径，方便上游拼面包屑"""
|
|
root = genre_node_state["state"]["data"]["genreTree"]["parent_category"]
|
|
game = root["children"][0]
|
|
switch = next(child for child in game["children"] if child["id"] == 565950)
|
|
# 模拟站点返回深层分类页时的链式结构：只保留通往目标的那条分支
|
|
game["children"] = [switch]
|
|
switch["children"] = [{"id": 566404, "name": "ソフト", "count": 100, "leaf": True, "children": []}]
|
|
|
|
result = parse_genres(genre_node_state, genre_id="565950")
|
|
assert result.genre_id == "565950"
|
|
assert [(a.genre_id, a.name) for a in result.ancestors] == [("101205", "テレビゲーム")]
|
|
assert [c.genre_id for c in result.children] == ["566404"]
|
|
|
|
|
|
def test_leaf_genre_reports_no_children(genre_node_state):
|
|
root = genre_node_state["state"]["data"]["genreTree"]["parent_category"]
|
|
root["children"] = [{"id": 999999, "name": "葉", "count": 5, "leaf": True, "children": []}]
|
|
|
|
result = parse_genres(genre_node_state, genre_id="999999")
|
|
assert result.is_leaf is True
|
|
assert result.children == []
|
|
|
|
|
|
# ---- 错误处理 ----
|
|
|
|
def test_missing_genre_tree_is_a_parse_error():
|
|
with pytest.raises(ScrapeParseError):
|
|
parse_genres({"state": {"data": {}}}, genre_id=None)
|
|
|
|
|
|
def test_unknown_genre_id_is_a_parse_error(genre_node_state):
|
|
with pytest.raises(ScrapeParseError):
|
|
parse_genres(genre_node_state, genre_id="1")
|