Compare commits

...

11 Commits

Author SHA1 Message Date
q792602257 ef7eb9dbb1 branch_number检测逻辑 2026-07-10 20:27:58 +08:00
q792602257 118667d01f 下单接口支持其他店铺
/api/order/get_shipping_fee 与购物车加入逻辑此前只认官方自营详情页,加盟店运费分桶
仅靠自由文本 supplier 判断,未与其他店铺列表的 tenpo_cd 打通。补充 OrderItem/
CargoItemRequest 的 tenpo_cd/branch_number 字段,运费计算优先按 tenpo_cd 分桶
(无则回退 supplier 保持兼容),cargo_service 加购 URL 同步带上店铺参数。

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-07-08 22:31:03 +08:00
q792602257 fe2a3f3768 支持其他店铺 2026-07-08 21:55:05 +08:00
q792602257 7ddb41cea1 支持其他店铺 2026-07-08 21:53:20 +08:00
q792602257 af8d3a859f 文档调整 2026-06-25 14:07:10 +08:00
q792602257 3c299899c3 部署文档 2026-06-25 13:43:39 +08:00
q792602257 d9a157c03d 浏览器自动恢复逻辑 2026-06-15 10:39:47 +08:00
q792602257 2bcaab1626 通知 2026-06-13 20:00:45 +08:00
q792602257 09f12c6c47 trade/refresh 2026-06-11 18:27:21 +08:00
q792602257 02a4484211 common扩展 2026-06-11 14:38:23 +08:00
q792602257 919840a893 uv workspace 改造 2026-06-10 09:44:59 +08:00
30 changed files with 2271 additions and 598 deletions
+2 -2
View File
@@ -38,8 +38,8 @@ SURUGAYA_PROXY_USERNAME=
# 代理密码(如代理需要认证则填写) # 代理密码(如代理需要认证则填写)
SURUGAYA_PROXY_PASSWORD= SURUGAYA_PROXY_PASSWORD=
# Redis(可选;用于跨进程/跨实例复用会话与限流状态) # Redis(可选;用于跨进程/跨实例复用会话与限流状态;并为 /api/news 读取 worker 抓取的通知
# 留空表示不使用 Redis(退回进程内存) # 留空表示不使用 Redis(退回进程内存,且 /api/news 不可用
SURUGAYA_REDIS_HOST= SURUGAYA_REDIS_HOST=
SURUGAYA_REDIS_PORT=6379 SURUGAYA_REDIS_PORT=6379
SURUGAYA_REDIS_PASSWORD= SURUGAYA_REDIS_PASSWORD=
+21 -8
View File
@@ -10,19 +10,32 @@ ENV PYTHONDONTWRITEBYTECODE=1 \
PYTHONUNBUFFERED=1 \ PYTHONUNBUFFERED=1 \
TZ=Asia/Tokyo TZ=Asia/Tokyo
# 4. 复制依赖文件并安装 Python 依赖 # 4. 引入 uv(项目使用 uv workspace + uv.lock 管理依赖
COPY requirements.txt /app/ COPY --from=ghcr.io/astral-sh/uv:0.10.12 /uv /usr/local/bin/uv
RUN pip install --no-cache-dir -r requirements.txt
# 5. 安装 Playwright 的 Chromium 浏览器内核 # 5. 安装依赖到 /app 之外的独立 venv
# 虽然基础镜像自带了一些依赖,但为了确保版本匹配,建议执行 install # - venv 放在 /opt/venv:docker-compose 用 `.:/app` 挂载宿主目录,
# 若 venv 落在 /app 下会被卷覆盖,故必须放到 /app 外。
# - 先只复制依赖清单与 workspace 包,最大化利用构建缓存。
# - uv sync 会按 uv.lock 安装第三方依赖,并把本地的 surugaya-common 一并装入,
# 从而修复运行时 `ModuleNotFoundError: No module named 'surugaya_common'`。
ENV UV_PROJECT_ENVIRONMENT=/opt/venv \
UV_LINK_MODE=copy
COPY pyproject.toml uv.lock ./
COPY packages ./packages
RUN uv sync --frozen --no-dev --no-install-project
# 让后续命令与运行时默认使用该 venv
ENV PATH="/opt/venv/bin:$PATH"
# 6. 安装 Playwright 的 Chromium 浏览器内核(确保与 venv 内 playwright 版本匹配)
RUN playwright install chromium RUN playwright install chromium
# 6. 复制项目代码 # 7. 复制项目代码
COPY . /app/ COPY . /app/
# 7. 暴露端口 # 8. 暴露端口
EXPOSE 31106 EXPOSE 31106
# 8. 启动命令 # 9. 启动命令
CMD ["python", "-m", "app.main"] CMD ["python", "-m", "app.main"]
+4
View File
@@ -16,6 +16,10 @@
- `POST /api/search` - `POST /api/search`
- `POST /api/item_detail` - `POST /api/item_detail`
## 部署
生产环境及更多部署方式请参考 [docs/DEPLOYMENT.md](docs/DEPLOYMENT.md)。
## 安装 ## 安装
```bash ```bash
+72
View File
@@ -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,
)
+80 -5
View File
@@ -18,6 +18,7 @@ from app.models.scrape import (
DetailRequest, DetailRequest,
OrderShippingFeeData, OrderShippingFeeData,
OrderShippingFeeRequest, OrderShippingFeeRequest,
OtherShopListData,
ProductDetailData, ProductDetailData,
PurchaseTaskRequest, PurchaseTaskRequest,
SearchRequest, SearchRequest,
@@ -34,6 +35,9 @@ from surugaya_common.models import SurugayaOrderDetail
router = APIRouter(prefix="/api", tags=["scrape"]) router = APIRouter(prefix="/api", tags=["scrape"])
# 同一 trade_code 在该窗口内不因 /trade 未命中重复下发抓取任务
TRADE_PENDING_TTL_SECONDS = 3600
def _parse_callback_response_body(content_type: str, body_bytes: bytes) -> Any: def _parse_callback_response_body(content_type: str, body_bytes: bytes) -> Any:
if not body_bytes: if not body_bytes:
@@ -119,6 +123,29 @@ async def item_detail(
) )
@router.post(
"/item_other_shops",
response_model=ApiResponse[OtherShopListData],
dependencies=[Depends(require_bearer_token)],
)
async def item_other_shops(
payload: DetailRequest,
container: ServiceContainer = Depends(get_container),
) -> ApiResponse[OtherShopListData]:
"""获取商品「他のショップ」全部店铺报价列表
根据商品 ID 抓取骏河屋 /product/other/{id} 页面,返回全部店铺(含官方自营+加盟店/市场店铺)报价列表;
列表中每项的 detail_url 已带上对应 tenpo_cd/branch_number,可直接用于 /api/item_detail 跳转查看该店铺报价。
"""
data = await container.surugaya_client.fetch_other_shops(payload)
return ApiResponse[OtherShopListData](
success=True,
msg="success",
data=data,
code=0,
)
@router.get( @router.get(
"/categories", "/categories",
response_model=ApiResponse[list[CategoryData]], response_model=ApiResponse[list[CategoryData]],
@@ -149,9 +176,10 @@ async def order_get_shipping_fee(payload: OrderShippingFeeRequest) -> ApiRespons
"""计算订单运费与手续费。 """计算订单运费与手续费。
规则: 规则:
- 官方店铺(supplier 为空)运费: 按官方部分商品总额套阶梯价(440/385/0) - 官方店铺(店铺标识为空)运费: 按官方部分商品总额套阶梯价(440/385/0)
* 订单中无任何官方商品时,官方运费 0 * 订单中无任何官方商品时,官方运费 0
- 加盟店运费(supplier 非空,同视为同一家): 每家统一 800 日元 - 加盟店运费(店铺标识非空,同标识视为同一家): 每家统一 800 日元
* 店铺标识优先取 tenpo_cd(与 other_shop_list 中的 tenpo_cd 一致),未提供时回退取 supplier
- 通贩手续费: 按全部商品(含官方+加盟店)总额套阶梯价(240/0) - 通贩手续费: 按全部商品(含官方+加盟店)总额套阶梯价(240/0)
""" """
official_amount = 0 official_amount = 0
@@ -160,10 +188,11 @@ async def order_get_shipping_fee(payload: OrderShippingFeeRequest) -> ApiRespons
for item in payload.item_list: for item in payload.item_list:
subtotal = item.goods_number * item.goods_price subtotal = item.goods_number * item.goods_price
goods_amount += subtotal goods_amount += subtotal
if item.supplier == "": shop_key = item.tenpo_cd or item.supplier
if shop_key == "":
official_amount += subtotal official_amount += subtotal
else: else:
franchise_stores.add(item.supplier) franchise_stores.add(shop_key)
official_shipping_fee = get_shipping_fee(official_amount) if official_amount > 0 else 0 official_shipping_fee = get_shipping_fee(official_amount) if official_amount > 0 else 0
franchise_shipping_fee = get_franchise_shipping_fee(len(franchise_stores)) franchise_shipping_fee = get_franchise_shipping_fee(len(franchise_stores))
@@ -317,7 +346,7 @@ async def trade(
dedup_key = redis_keys.trade_pending_key(trade_code) dedup_key = redis_keys.trade_pending_key(trade_code)
dedup_exists = await container.session_store._redis.get(dedup_key) dedup_exists = await container.session_store._redis.get(dedup_key)
if not dedup_exists: if not dedup_exists:
await container.session_store._redis.setex(dedup_key, 3600, "1") await container.session_store._redis.setex(dedup_key, TRADE_PENDING_TTL_SECONDS, "1")
await container.session_store._redis.xadd( await container.session_store._redis.xadd(
name=redis_keys.trade_fetch_stream(), name=redis_keys.trade_fetch_stream(),
fields={"payload": json.dumps({"trade_code": trade_code})}, fields={"payload": json.dumps({"trade_code": trade_code})},
@@ -352,6 +381,52 @@ async def trade(
) )
@router.get(
"/trade/refresh",
response_model=ApiResponse[None],
dependencies=[Depends(require_bearer_token)],
)
async def trade_refresh(
trade_code: str,
container: ServiceContainer = Depends(get_container),
) -> ApiResponse[None]:
"""
立即下发订单详情刷新任务(绕过 /trade 未命中时的去重窗口,强制重抓)
"""
if not trade_code or trade_code.strip() == "":
return ApiResponse[None](
success=False,
msg="trade_code 不能为空",
data=None,
code=400,
)
if container.session_store._redis is None:
return ApiResponse[None](
success=False,
msg="Redis 未配置",
data=None,
code=500,
)
# 强制下发并重置去重窗口,避免随后 /trade 未命中时再次重复下发
await container.session_store._redis.setex(
redis_keys.trade_pending_key(trade_code), TRADE_PENDING_TTL_SECONDS, "1"
)
await container.session_store._redis.xadd(
name=redis_keys.trade_fetch_stream(),
fields={"payload": json.dumps({"trade_code": trade_code})},
maxlen=100,
approximate=True,
)
return ApiResponse[None](
success=True,
msg="success",
data=None,
code=0,
)
@router.post( @router.post(
"/test/check_task_callback_url", "/test/check_task_callback_url",
+7 -8
View File
@@ -9,6 +9,8 @@ from typing import Literal
from pydantic_settings import BaseSettings, SettingsConfigDict from pydantic_settings import BaseSettings, SettingsConfigDict
from pathlib import Path from pathlib import Path
from surugaya_common import browser_profile, urls
BASE_DIR = Path(__file__).resolve().parent.parent.parent BASE_DIR = Path(__file__).resolve().parent.parent.parent
@@ -60,13 +62,10 @@ class Settings(BaseSettings):
cloudflare_wait_timeout_seconds: float = 60.0 # 等待 Cloudflare 挑战超时 cloudflare_wait_timeout_seconds: float = 60.0 # 等待 Cloudflare 挑战超时
cloudflare_reuse_wait_timeout_seconds: float = 8.0 # 复用会话时消化挑战的短超时,超时则放弃复用直接重建 cloudflare_reuse_wait_timeout_seconds: float = 8.0 # 复用会话时消化挑战的短超时,超时则放弃复用直接重建
session_ttl_seconds: int = 900 # 会话过期时间(秒) session_ttl_seconds: int = 900 # 会话过期时间(秒)
browser_locale: str = "ja-JP" # 指纹参数默认值取自共享包,与 worker 保持一致
browser_timezone_id: str = "Asia/Tokyo" browser_locale: str = browser_profile.LOCALE
browser_user_agent: str = ( browser_timezone_id: str = browser_profile.TIMEZONE_ID
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) " browser_user_agent: str = browser_profile.DEFAULT_USER_AGENT
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/124.0.0.0 Safari/537.36"
)
browser_stealth_enabled: bool = True # 是否启用浏览器 Stealth 模式 browser_stealth_enabled: bool = True # 是否启用浏览器 Stealth 模式
# ---- Redis 配置(可选,未配置时使用进程内存)---- # ---- Redis 配置(可选,未配置时使用进程内存)----
@@ -82,7 +81,7 @@ class Settings(BaseSettings):
proxy_password: str | None = None proxy_password: str | None = None
# ---- 目标站点 ---- # ---- 目标站点 ----
base_url: str = "https://www.suruga-ya.jp" base_url: str = urls.BASE_URL
@property @property
def browser_headless_effective(self) -> bool: def browser_headless_effective(self) -> bool:
+2
View File
@@ -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.health import router as health_router
from app.api.routes.scrape import router as scrape_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.config import get_settings
from app.core.container import ServiceContainer from app.core.container import ServiceContainer
from app.core.errors import AppError from app.core.errors import AppError
@@ -90,6 +91,7 @@ def create_app() -> FastAPI:
app = FastAPI(title="Surugaya Scraper Service", lifespan=lifespan) app = FastAPI(title="Surugaya Scraper Service", lifespan=lifespan)
app.include_router(health_router) app.include_router(health_router)
app.include_router(scrape_router) app.include_router(scrape_router)
app.include_router(news_router)
@app.exception_handler(AppError) @app.exception_handler(AppError)
async def app_error_handler(_: Request, exc: AppError) -> JSONResponse: async def app_error_handler(_: Request, exc: AppError) -> JSONResponse:
+48
View File
@@ -40,6 +40,9 @@ class DetailRequest(BaseModel):
"""商品详情请求参数,支持 id 或 product_url""" """商品详情请求参数,支持 id 或 product_url"""
id: str | None = Field(default=None, min_length=3, max_length=40) id: str | None = Field(default=None, min_length=3, max_length=40)
product_url: HttpUrl | None = None product_url: HttpUrl | None = None
tenpo_cd: str | None = None # 指定店铺代码;跳转查看某个加盟店/市场店铺的报价时传入
branch_number: str | None = None # 指定分支编号,与 tenpo_cd 搭配使用
include_other_shops: bool = False # 为 True 时,随详情额外抓取并返回 other_shop_list(多付出一次页面请求)
class ProductSummary(BaseModel): class ProductSummary(BaseModel):
@@ -67,6 +70,24 @@ class SearchResultData(BaseModel):
items: list[ProductSummary] items: list[ProductSummary]
class OtherShopItem(BaseModel):
"""商品「他のショップ」列表中的单个店铺报价
来自 /product/other/{goods_id} 页面,含官方自营与加盟店/市场店铺。
"""
tenpo_cd: str = "" # 店铺代码;空串 = 官方自营
shop_name: str = "" # 店铺名称;官方自营固定为"駿河屋"
shop_url: str = "" # 店铺主页链接
is_official: bool = False # 是否为官方自营(tenpo_cd 为空)
condition: str = "" # 成色/规格描述,如"中古"
price: int = 0 # 该店铺售价(日元)
branch_number: str = "" # 分支编号
rating_score: str = "" # 店铺评分,如"5.0"
rating_count: int = 0 # 评价数量
shipping_note: str = "" # 发货时效描述,如"1日〜3日以内に発送します"
detail_url: str = "" # 该报价对应的商品详情页链接(含 tenpo_cd/branch_number)
class ProductDetailData(BaseModel): class ProductDetailData(BaseModel):
"""商品详情数据""" """商品详情数据"""
goods_id: str goods_id: str
@@ -84,6 +105,13 @@ class ProductDetailData(BaseModel):
sku_list: list[dict[str, Any]] # 规格列表 sku_list: list[dict[str, Any]] # 规格列表
sku: str = "" # 规格 sku: str = "" # 规格
breadcrumb_list: list[dict[str, Any]] = [] # 面包屑 breadcrumb_list: list[dict[str, Any]] = [] # 面包屑
other_shop_list: list[OtherShopItem] = [] # 仅当请求 include_other_shops=True 时才会填充,否则为空列表
class OtherShopListData(BaseModel):
"""商品「他のショップ」全部店铺(含官方自营+加盟店/市场店铺)报价列表"""
goods_id: str
other_shop_list: list[OtherShopItem] = []
class HealthData(BaseModel): class HealthData(BaseModel):
@@ -100,6 +128,8 @@ class CargoItemRequest(BaseModel):
id: str id: str
number: int = 1 number: int = 1
price_limit: int | None = None price_limit: int | None = None
tenpo_cd: str | None = None # 指定店铺代码;跳转加入指定加盟店/市场店铺的商品时传入
branch_number: str | None = None # 指定分支编号,与 tenpo_cd 搭配使用
class CargoPayload(BaseModel): class CargoPayload(BaseModel):
@@ -134,6 +164,10 @@ class OrderItem(BaseModel):
goods_price: int = Field(ge=0) goods_price: int = Field(ge=0)
# 店铺标识: 空串 = 官方店铺;非空 = 加盟店标识(同名视为同一家加盟店) # 店铺标识: 空串 = 官方店铺;非空 = 加盟店标识(同名视为同一家加盟店)
supplier: str = "" supplier: str = ""
# 店铺代码,来自 other_shop_list 中的 tenpo_cd;非空 = 加盟店/市场店铺。
# 优先于 supplier 用于判定"同一家店铺"(同 tenpo_cd 视为同一家);未提供时回退使用 supplier。
tenpo_cd: str = ""
branch_number: str = "" # 分支编号,可选,与 tenpo_cd 搭配使用
class OrderShippingFeeRequest(BaseModel): class OrderShippingFeeRequest(BaseModel):
@@ -207,3 +241,17 @@ class CheckTaskCallbackUrlData(BaseModel):
"""测试回调地址响应数据""" """测试回调地址响应数据"""
response_status: int response_status: int
response_body: Any 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 秒)
+82 -45
View File
@@ -15,6 +15,7 @@ from typing import Any
from app.core.config import Settings from app.core.config import Settings
from app.core.errors import BrowserUnavailableError, ResourceBusyError from app.core.errors import BrowserUnavailableError, ResourceBusyError
from surugaya_common.browser_profile import DEFAULT_PAGE_HEADERS, stealth_launch_args
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -37,6 +38,8 @@ class BrowserPool:
self._contexts_since_start = 0 self._contexts_since_start = 0
self._browser_started_at: float | None = None self._browser_started_at: float | None = None
self._active_contexts = 0 self._active_contexts = 0
# 浏览器实例「代次」:每次成功 start 自增,用于并发重建时判断当前实例是否已被他人换新
self._browser_generation = 0
@property @property
def enabled(self) -> bool: def enabled(self) -> bool:
@@ -80,28 +83,17 @@ class BrowserPool:
# 用 playwright-stealth 钩子包装 playwright:后续 new_context/new_page # 用 playwright-stealth 钩子包装 playwright:后续 new_context/new_page
# 会自动应用 stealth(含 HTTP sec-ch-ua 客户端提示),并令 navigator.languages # 会自动应用 stealth(含 HTTP sec-ch-ua 客户端提示),并令 navigator.languages
# 与日语环境保持一致,避免指纹自相矛盾。 # 与日语环境保持一致,避免指纹自相矛盾。
from playwright_stealth import Stealth from surugaya_common.stealth import create_stealth
stealth = Stealth( self._playwright_cm = create_stealth().use_async(async_playwright())
navigator_languages_override=("ja-JP", "ja", "en-US", "en"),
)
self._playwright_cm = stealth.use_async(async_playwright())
self._playwright = await self._playwright_cm.__aenter__() self._playwright = await self._playwright_cm.__aenter__()
else: else:
self._playwright_cm = None self._playwright_cm = None
self._playwright = await async_playwright().start() self._playwright = await async_playwright().start()
chromium = self._playwright.chromium chromium = self._playwright.chromium
launch_args = [ launch_args = stealth_launch_args() + [
"--disable-blink-features=AutomationControlled",
"--disable-dev-shm-usage",
"--disable-web-security", "--disable-web-security",
"--no-first-run", "--no-first-run",
"--disable-infobars",
"--no-sandbox",
"--disable-setuid-sandbox",
"--disable-gpu",
"--use-mock-keychain",
"--password-store=basic",
] ]
# 本地可视化调试时强制窗口化,避免浏览器启动后不可见 # 本地可视化调试时强制窗口化,避免浏览器启动后不可见
if not headless: if not headless:
@@ -115,6 +107,7 @@ class BrowserPool:
self._started = True self._started = True
self._startup_error = None self._startup_error = None
self._contexts_since_start = 0 self._contexts_since_start = 0
self._browser_generation += 1
self._browser_started_at = asyncio.get_running_loop().time() self._browser_started_at = asyncio.get_running_loop().time()
logger.info("浏览器启动完成") logger.info("浏览器启动完成")
except Exception as exc: except Exception as exc:
@@ -169,10 +162,20 @@ class BrowserPool:
self._playwright_cm = None self._playwright_cm = None
self._started = False self._started = False
async def _restart_browser(self, reason: str) -> None: async def _restart_browser(self, reason: str, *, observed_generation: int | None = None) -> None:
"""在浏览器掉线后串行重建 Playwright 运行时。""" """在浏览器掉线后串行重建 Playwright 运行时。
observed_generation 为调用方观察到异常时的浏览器代次:
- 传入时:若持锁后代次已变化,说明其他协程已重建,直接返回;否则强制重建
(此场景已确认 new_context 因 driver 断链失败,不再依赖 is_connected 探测,
规避 connection_lost 事件滞后导致的误判)。
- 不传时:沿用 is_connected 探测,连通则跳过(用于获取页面前的预检)。
"""
async with self._restart_lock: async with self._restart_lock:
if self._is_browser_connected(): if observed_generation is not None:
if self._browser_generation != observed_generation:
return
elif self._is_browser_connected():
return return
logger.warning("检测到浏览器连接不可用,准备重建浏览器实例:reason=%s", reason) logger.warning("检测到浏览器连接不可用,准备重建浏览器实例:reason=%s", reason)
@@ -237,14 +240,22 @@ class BrowserPool:
raise BrowserUnavailableError(self._startup_error or "Browser recycle failed") raise BrowserUnavailableError(self._startup_error or "Browser recycle failed")
@staticmethod @staticmethod
def _should_retry_context_creation(exc: Exception) -> bool: def is_disconnect_error(exc: Exception) -> bool:
"""识别 Playwright driver 断链类错误,允许执行一次重建后重试。""" """识别 Playwright driver 断链类错误,作为「重建后重试」的统一判据。
浏览器池内部与上层(CloudflareSessionManager)共用此判据,集中维护标记,
避免各处散落的字符串匹配漂移。
"""
message = str(exc).lower() message = str(exc).lower()
retry_markers = ( retry_markers = (
"connection closed while reading from the driver", "connection closed while reading from the driver",
"browser has been closed", "browser has been closed",
"target page, context or browser has been closed", "target page, context or browser has been closed",
"connection closed", "connection closed",
# driver 子进程掉线导致 asyncio 传输管道关闭,写入命令时同步报错
"the handler is closed",
"transport closed",
"unable to perform operation",
) )
return any(marker in message for marker in retry_markers) return any(marker in message for marker in retry_markers)
@@ -272,16 +283,32 @@ class BrowserPool:
except Exception: except Exception:
logger.exception("关闭旧的保留调试上下文失败") logger.exception("关闭旧的保留调试上下文失败")
async def _build_context_obj(self, storage_state: dict[str, Any] | None) -> Any: async def _run_with_reconnect(self, factory: Any, what: str) -> Any:
"""创建浏览器上下文(含断链重试),仅负责创建本身 """执行 factory();若因 driver 断链失败,重建浏览器后重试一次
factory 必须是「可重复调用、每次从零创建上下文/页面」的协程工厂——因为
重建浏览器会令旧上下文一并失效,恢复的唯一正确粒度是重新创建。
借助 _browser_generation 守卫,并发场景下只重建一次,后到者复用新实例。
"""
for attempt in range(2):
observed_generation = self._browser_generation
try:
return await factory()
except Exception as exc:
if attempt == 0 and self.is_disconnect_error(exc):
logger.warning("%s 因浏览器断链失败,重建后重试一次:err=%s", what, exc)
await self._restart_browser(
f"{what} failed: {exc}", observed_generation=observed_generation
)
continue
raise
async def _new_context_raw(self, storage_state: dict[str, Any] | None) -> Any:
"""仅创建浏览器上下文本身并计入回收统计,不含断链重试与生命周期管理。
计入回收阈值统计;不负责并发信号量与关闭,由调用方管理生命周期。
启用 stealth 时,playwright 已被 use_async 钩子包装,new_context 会自动 启用 stealth 时,playwright 已被 use_async 钩子包装,new_context 会自动
应用 stealth(含 HTTP sec-ch-ua),此处无需再手动 apply。 应用 stealth(含 HTTP sec-ch-ua),此处无需再手动 apply。
""" """
context = None
for attempt in range(2):
try:
context = await self._browser.new_context( context = await self._browser.new_context(
locale=self._settings.browser_locale, locale=self._settings.browser_locale,
timezone_id=self._settings.browser_timezone_id, timezone_id=self._settings.browser_timezone_id,
@@ -289,29 +316,28 @@ class BrowserPool:
viewport=None if not self._settings.browser_headless_effective else {"width": 1440, "height": 900}, viewport=None if not self._settings.browser_headless_effective else {"width": 1440, "height": 900},
storage_state=storage_state, storage_state=storage_state,
) )
break
except Exception as exc:
if attempt == 0 and self._should_retry_context_creation(exc):
logger.warning("创建浏览器上下文失败,尝试重建浏览器后重试一次:err=%s", exc)
await self._restart_browser(f"new_context failed: {exc}")
continue
raise
# 上下文创建成功,计入回收阈值统计 # 上下文创建成功,计入回收阈值统计
self._contexts_since_start += 1 self._contexts_since_start += 1
return context return context
async def _open_context_and_page(self, storage_state: dict[str, Any] | None) -> tuple[Any, Any]:
"""创建上下文并在其内开页面,作为「断链重试」的最小恢复单元。
页面创建失败时关闭半开上下文,避免重建/重试时泄漏。
"""
context = await self._new_context_raw(storage_state)
try:
page = await self._configure_page(context)
except Exception:
await self.close_context(context)
raise
return context, page
@staticmethod @staticmethod
async def _configure_page(context: Any) -> Any: async def _configure_page(context: Any) -> Any:
"""在上下文中创建页面并设置反检测请求头。""" """在上下文中创建页面并设置反检测请求头。"""
page = await context.new_page() page = await context.new_page()
await page.set_extra_http_headers( await page.set_extra_http_headers(DEFAULT_PAGE_HEADERS)
{
"Accept-Language": "ja-JP,ja;q=0.9,en-US;q=0.8,en;q=0.7",
"Cache-Control": "no-cache",
"Pragma": "no-cache",
}
)
return page return page
async def create_persistent_context(self, storage_state: dict[str, Any] | None = None) -> Any: async def create_persistent_context(self, storage_state: dict[str, Any] | None = None) -> Any:
@@ -322,7 +348,16 @@ class BrowserPool:
调用方需在不再使用时调用 close_context 释放。 调用方需在不再使用时调用 close_context 释放。
""" """
await self._ensure_browser_ready() await self._ensure_browser_ready()
return await self._build_context_obj(storage_state) return await self._run_with_reconnect(
lambda: self._new_context_raw(storage_state), "创建长存上下文"
)
async def ensure_ready(self) -> None:
"""对外暴露的浏览器就绪保证:连接断开时自动重建。
供上层在捕获断链异常后、重试前主动调用,将重建判定收敛到浏览器池内。
"""
await self._ensure_browser_ready()
@asynccontextmanager @asynccontextmanager
async def page_in_context(self, context: Any) -> AsyncIterator[Any]: async def page_in_context(self, context: Any) -> AsyncIterator[Any]:
@@ -412,12 +447,14 @@ class BrowserPool:
try: try:
# 创建上下文前检查是否需要主动回收(仅在空闲时执行,不打断进行中的会话) # 创建上下文前检查是否需要主动回收(仅在空闲时执行,不打断进行中的会话)
await self._maybe_recycle() await self._maybe_recycle()
context = await self._build_context_obj(storage_state) # 上下文与页面创建作为一个整体进行断链重试:context 与 browser 实例绑定,
# 上下文创建成功,标记为在用(回收阈值计数已在 _build_context_obj 内完成) # 重建 browser 后必须连同 context 一起重新创建,二者不可拆开重试。
context, page = await self._run_with_reconnect(
lambda: self._open_context_and_page(storage_state), "打开页面会话"
)
# 创建成功后才标记为在用(回收阈值计数已在 _new_context_raw 内完成)
self._active_contexts += 1 self._active_contexts += 1
context_counted = True context_counted = True
page = await self._configure_page(context)
yield context, page yield context, page
except Exception: except Exception:
if self._settings.browser_keep_open_on_error_effective and context is not None: if self._settings.browser_keep_open_on_error_effective and context is not None:
+12 -76
View File
@@ -10,6 +10,9 @@ from selectolax.parser import HTMLParser
from app.core.config import Settings from app.core.config import Settings
from app.models.scrape import CargoRequest from app.models.scrape import CargoRequest
from app.utils.parse_util import extract_product_id, format_price from app.utils.parse_util import extract_product_id, format_price
from surugaya_common.browser_profile import DEFAULT_PAGE_HEADERS, stealth_launch_args
from surugaya_common.browser_utils import clear_cart
from surugaya_common.urls import CARGO_DETAIL_URL, product_detail_url
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -29,16 +32,7 @@ class CargoService:
""" """
async with async_playwright() as p: async with async_playwright() as p:
headless = self._settings.browser_headless_effective headless = self._settings.browser_headless_effective
launch_args = [ launch_args = stealth_launch_args()
"--disable-blink-features=AutomationControlled",
"--disable-infobars",
"--no-sandbox",
"--disable-setuid-sandbox",
"--disable-dev-shm-usage",
"--disable-gpu",
"--use-mock-keychain",
"--password-store=basic",
]
if not headless: if not headless:
launch_args.extend(["--start-maximized", "--window-position=0,0"]) launch_args.extend(["--start-maximized", "--window-position=0,0"])
@@ -61,21 +55,17 @@ class CargoService:
if self._settings.browser_stealth_enabled: if self._settings.browser_stealth_enabled:
try: try:
from playwright_stealth import Stealth # launch_persistent_context 之外的裸 context 用 apply_stealth_async;
await Stealth().apply_stealth_async(context) # 共享工厂带日语 languages override,与 locale/Accept-Language 对齐。
from surugaya_common.stealth import create_stealth
await create_stealth().apply_stealth_async(context)
except Exception as e: except Exception as e:
logger.warning("应用 playwright-stealth 失败: %s", e) logger.warning("应用 playwright-stealth 失败: %s", e)
page = await context.new_page() page = await context.new_page()
await page.set_extra_http_headers( await page.set_extra_http_headers(DEFAULT_PAGE_HEADERS)
{
"Accept-Language": "ja-JP,ja;q=0.9,en-US;q=0.8,en;q=0.7",
"Cache-Control": "no-cache",
"Pragma": "no-cache",
}
)
target_url = "https://www.suruga-ya.jp/cargo/detail" target_url = CARGO_DETAIL_URL
# 先进入购物车页面 # 先进入购物车页面
await page.goto(target_url, wait_until="domcontentloaded", timeout=15000) await page.goto(target_url, wait_until="domcontentloaded", timeout=15000)
@@ -85,7 +75,7 @@ class CargoService:
# 清空购物车 # 清空购物车
await asyncio.sleep(random.uniform(1, 3)) # 加入随机延迟,模拟人工操作间隔 await asyncio.sleep(random.uniform(1, 3)) # 加入随机延迟,模拟人工操作间隔
if cart_goods_num > 0: if cart_goods_num > 0:
await self.clear_cart(page) await clear_cart(page)
# 将待采购的商品加入购物车 # 将待采购的商品加入购物车
item_list = request.payload.item_list item_list = request.payload.item_list
@@ -101,7 +91,7 @@ class CargoService:
continue continue
logger.debug("进入商品 %s 详情页,准备购买数量: %s...", product_id, number) logger.debug("进入商品 %s 详情页,准备购买数量: %s...", product_id, number)
detail_url = f"https://www.suruga-ya.jp/product/detail/{product_id}" detail_url = product_detail_url(product_id, item.tenpo_cd, item.branch_number)
await page.goto(detail_url, wait_until="domcontentloaded", timeout=20000) await page.goto(detail_url, wait_until="domcontentloaded", timeout=20000)
# 判断是否缺货 # 判断是否缺货
@@ -327,57 +317,3 @@ class CargoService:
"total_commission_fee": total_commission_fee "total_commission_fee": total_commission_fee
} }
async def clear_cart(self, page):
"""
清空购物车
"""
cart_goods_num = await page.locator("table.item_box tbody tr.item").count()
if cart_goods_num < 1:
return
is_delete = 0
# 购物车商品数量超过3个后会出现"全削除"按钮
if cart_goods_num > 3:
try:
delete_button = page.locator("#total_box .delbtn-all").get_by_role("button", name="全削除")
# 3. 显式等待按钮在页面上可见(容错处理:比如给它最多 3000 毫秒的加载时间)
# 如果按钮是一开始就在静态 HTML 里的,这一步也可以省略,直接用下面的 is_visible()
await delete_button.wait_for(state="visible", timeout=3000)
if await delete_button.is_visible():
logger.debug("检测到‘全削除’按钮存在,正在执行点击...")
await delete_button.click()
is_delete = 1
except Exception:
logger.debug("未检测到‘全削除’按钮。")
if is_delete == 1:
return
# 遍历删除
item_selector = "table.item_box tbody tr.item"
remove_selector = "table.item_box a.remove_product"
while True:
cart_goods_num = await page.locator(item_selector).count()
if cart_goods_num == 0:
break
remove_link = page.locator(remove_selector).first
try:
await remove_link.click()
# 删除按钮不会触发弹窗,直接等待购物车条目数减少即可。
await page.wait_for_function(
"""({selector, expectedCount}) => {
return document.querySelectorAll(selector).length < expectedCount;
}""",
arg={"selector": item_selector, "expectedCount": cart_goods_num},
timeout=15_000,
)
await page.wait_for_timeout(random.randint(800, 1500))
except PlaywrightTimeoutError:
latest_count = await page.locator(item_selector).count()
if latest_count < cart_goods_num:
continue
await page.wait_for_timeout(1000)
+24 -2
View File
@@ -96,7 +96,10 @@ class CloudflareSessionManager:
return self._to_verified_session(existing) return self._to_verified_session(existing)
logger.debug("创建新会话:session_name=%s force_refresh=%s", session_name, force_refresh) logger.debug("创建新会话:session_name=%s force_refresh=%s", session_name, force_refresh)
session = await self._create_verified_session(session_name=session_name, target_url=target_url) session = await self._build_with_browser_retry(
f"创建验证会话({session_name})",
lambda: self._create_verified_session(session_name=session_name, target_url=target_url),
)
return session return session
async def fetch_html(self, target_url: str) -> tuple[str, VerifiedSession]: async def fetch_html(self, target_url: str) -> tuple[str, VerifiedSession]:
@@ -135,9 +138,28 @@ class CloudflareSessionManager:
return html, self._warm_to_session(warm) return html, self._warm_to_session(warm)
logger.info("创建保活上下文:session_name=%s", session_name) logger.info("创建保活上下文:session_name=%s", session_name)
warm, html = await self._build_warm_context(session_name, target_url) warm, html = await self._build_with_browser_retry(
f"创建保活上下文({session_name})",
lambda: self._build_warm_context(session_name, target_url),
)
return html, self._warm_to_session(warm) return html, self._warm_to_session(warm)
async def _build_with_browser_retry(self, what: str, factory: Any) -> Any:
"""执行 factory();若因浏览器断链失败,重建浏览器后整段重试一次。
factory 每次都从零创建上下文与页面(_build_warm_context / _create_verified_session
均如此),故断链重建后重试是安全的。此层覆盖 new_page、goto、content 等
「上下文创建之后」才发生的断链——这些阶段浏览器池内逐操作的重试无法触及。
"""
try:
return await factory()
except Exception as exc:
if not BrowserPool.is_disconnect_error(exc):
raise
logger.warning("%s 因浏览器断链失败,重建后重试一次:err=%s", what, exc)
await self._browser_pool.ensure_ready()
return await factory()
async def _reuse_warm(self, session_name: str, target_url: str) -> tuple[str, WarmContext] | None: async def _reuse_warm(self, session_name: str, target_url: str) -> tuple[str, WarmContext] | None:
"""尝试复用已有保活上下文。 """尝试复用已有保活上下文。
+150 -9
View File
@@ -19,13 +19,14 @@ from selectolax.parser import HTMLParser
from app.core.config import Settings from app.core.config import Settings
from app.core.errors import ScrapeParseError, UpstreamBlockedError from app.core.errors import ScrapeParseError, UpstreamBlockedError
from app.models.scrape import CategoryData, DetailRequest, ProductDetailData, ProductSummary, SearchRequest, \ from app.models.scrape import CategoryData, DetailRequest, OtherShopItem, OtherShopListData, ProductDetailData, \
SearchResultData ProductSummary, SearchRequest, SearchResultData
from app.services.browser_pool import BrowserPool from app.services.browser_pool import BrowserPool
from app.services.cloudflare_session import CloudflareSessionManager from app.services.cloudflare_session import CloudflareSessionManager
from app.services.session_store import SessionStore from app.services.session_store import SessionStore
from app.utils.parse_util import format_price, get_shipping_fee, surugaya_photo_url_to_cdn from app.utils.parse_util import format_price, get_shipping_fee, surugaya_photo_url_to_cdn
from surugaya_common.app_utils import is_empty_str from surugaya_common.app_utils import is_empty_str
from surugaya_common.urls import BASE_URL, product_other_url
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -146,7 +147,7 @@ class SurugayaClient:
if is_empty_str(goods_id): if is_empty_str(goods_id):
raise ScrapeParseError("id is empty") raise ScrapeParseError("id is empty")
url = self._build_product_url(goods_id) url = self._build_product_url(goods_id, tenpo_cd=payload.tenpo_cd, branch_number=payload.branch_number)
logger.info("开始抓取详情页:goods_id=%s", goods_id) logger.info("开始抓取详情页:goods_id=%s", goods_id)
logger.debug("详情页URL:url=%s", url) logger.debug("详情页URL:url=%s", url)
@@ -264,6 +265,11 @@ class SurugayaClient:
# 提取面包屑 # 提取面包屑
breadcrumb_list = self.get_breadcrumb_list(tree) breadcrumb_list = self.get_breadcrumb_list(tree)
# 仅当调用方显式要求时,才额外抓取「他のショップ」页面(多付出一次 Cloudflare 保护页面请求)
other_shop_list: list[OtherShopItem] = []
if payload.include_other_shops:
other_shop_list = await self._fetch_other_shop_list(goods_id)
detail = ProductDetailData( detail = ProductDetailData(
goods_id=goods_id, goods_id=goods_id,
goods_name=goods_name, goods_name=goods_name,
@@ -278,10 +284,132 @@ class SurugayaClient:
price_note=price_note, price_note=price_note,
shipping_comission_fee=shipping_comission_fee, shipping_comission_fee=shipping_comission_fee,
breadcrumb_list=breadcrumb_list, breadcrumb_list=breadcrumb_list,
other_shop_list=other_shop_list,
) )
logger.debug("详情页解析完成:goods_id=%s goods_name=%s ", detail.goods_id, detail.goods_name) logger.debug("详情页解析完成:goods_id=%s goods_name=%s ", detail.goods_id, detail.goods_name)
return detail return detail
async def fetch_other_shops(self, payload: DetailRequest) -> OtherShopListData:
"""获取商品「他のショップ」全部店铺报价列表
独立于 fetch_detail,仅在需要展示/跳转其他店铺时按需调用,
避免每次详情请求都多付出一次 Cloudflare 保护页面抓取的成本。
Args:
payload: 详情请求参数,仅使用其中的 id
Returns:
OtherShopListData: 全部店铺(含官方自营+加盟店/市场店铺)报价列表
Raises:
ScrapeParseError: 当 id 未提供时
"""
goods_id = payload.id or ""
if is_empty_str(goods_id):
raise ScrapeParseError("id is empty")
other_shop_list = await self._fetch_other_shop_list(goods_id)
return OtherShopListData(goods_id=goods_id, other_shop_list=other_shop_list)
async def _fetch_other_shop_list(self, goods_id: str) -> list[OtherShopItem]:
"""抓取并解析商品「他のショップ」页面,返回全部店铺(含官方自营)报价列表
该页面独立于详情页,需额外一次 Cloudflare 保护页面请求;
解析失败或抓取异常时降级为空列表,不影响详情页主流程。
"""
other_url = product_other_url(goods_id)
try:
other_html, _ = await self._session_manager.fetch_html(other_url)
except Exception as exc:
logger.warning("抓取他のショップ页面失败:goods_id=%s err=%s", goods_id, exc)
return []
try:
return self._parse_other_shop_list(HTMLParser(other_html))
except Exception:
logger.exception("解析他のショップ页面失败:goods_id=%s", goods_id)
return []
@staticmethod
def _parse_other_shop_list(tree: HTMLParser) -> list[OtherShopItem]:
"""解析「他のショップ」页面的全部店铺(tbl_all 标签页)报价列表
每行对应一个店铺报价;tenpo_cd 为空表示官方自营(駿河屋本店)。
"""
items: list[OtherShopItem] = []
for row in tree.css("#tbl_all tr.item"):
tds = row.css("td")
if len(tds) < 3:
continue
price_node = tds[0].css_first("strong.text-red")
price = format_price(price_node.text().strip()) or 0 if price_node else 0
condition_node = tds[1].css_first("h2.title_product")
condition = condition_node.text().strip() if condition_node else ""
detail_link_node = tds[1].css_first("a[href*='/product/detail/']")
detail_href = detail_link_node.attributes.get("href") if detail_link_node else ""
detail_url = f"{BASE_URL}{detail_href}" if detail_href else ""
shop_col = tds[2]
shop_link_node = shop_col.css_first("a[href^='/shop/']")
tenpo_cd = ""
shop_name = "駿河屋"
shop_url = f"{BASE_URL}/"
if shop_link_node:
shop_href = shop_link_node.attributes.get("href") or ""
tenpo_cd = shop_href.rsplit("/", 1)[-1]
shop_name = shop_link_node.text().strip()
shop_url = f"{BASE_URL}{shop_href}"
rating_score = ""
rating_count = 0
rating_node = shop_col.css_first("div:not([class])")
if rating_node:
rating_text = rating_node.text().strip()
rating_match = re.match(r"([\d.]+)\((\d+)", rating_text)
if rating_match:
rating_score = rating_match.group(1)
rating_count = int(rating_match.group(2))
shipping_note_node = row.css_first("ul.shipping_campaing_info li.padT5")
shipping_note = shipping_note_node.text().strip() if shipping_note_node else ""
# branch_number 优先从 detail_url 的查询参数中提取(与 detail_url 保持一致),
# data-zaiko_data 仅作为兜底(极少数情况下 href 未携带 branch_number 时)。
branch_number = ""
if detail_href:
branch_number = (parse_qs(urlparse(detail_href).query).get("branch_number") or [""])[0]
if not branch_number:
campaign_node = row.css_first("li.ajax-campaign-placeholder")
if campaign_node:
zaiko_raw = campaign_node.attributes.get("data-zaiko_data") or ""
if zaiko_raw:
try:
zaiko = json.loads(zaiko_raw)
branch_number = str(zaiko.get("branch_number", ""))
except json.JSONDecodeError:
pass
items.append(
OtherShopItem(
tenpo_cd=tenpo_cd,
shop_name=shop_name,
shop_url=shop_url,
is_official=tenpo_cd == "",
condition=condition,
price=price,
branch_number=branch_number,
rating_score=rating_score,
rating_count=rating_count,
shipping_note=shipping_note,
detail_url=detail_url,
)
)
return items
async def fetch_categories(self) -> list[CategoryData]: async def fetch_categories(self) -> list[CategoryData]:
"""获取商品分类(带缓存) """获取商品分类(带缓存)
@@ -303,7 +431,7 @@ class SurugayaClient:
logger.info("开始获取并解析商品分类...") logger.info("开始获取并解析商品分类...")
top_categories: list[CategoryData] = [] top_categories: list[CategoryData] = []
target_url = "https://www.suruga-ya.jp/" target_url = f"{BASE_URL}/"
session = await self._session_manager.get_verified_session(target_url, True) session = await self._session_manager.get_verified_session(target_url, True)
html_content = "" html_content = ""
@@ -366,7 +494,7 @@ class SurugayaClient:
async with self._browser_pool.page_session(storage_state=session.storage_state) as (_, page): async with self._browser_pool.page_session(storage_state=session.storage_state) as (_, page):
for top_category in top_categories: for top_category in top_categories:
try: try:
category_url = f"https://www.suruga-ya.jp{top_category.href}" category_url = f"{BASE_URL}{top_category.href}"
response = await page.goto( response = await page.goto(
category_url, category_url,
wait_until="domcontentloaded", wait_until="domcontentloaded",
@@ -469,12 +597,25 @@ class SurugayaClient:
return 1 return 1
return current_page if current_page >= 1 else 1 return current_page if current_page >= 1 else 1
def _build_product_url(self, product_id: str | None) -> str: def _build_product_url(
"""根据商品 ID 构建详情页 URL""" self,
product_id: str | None,
tenpo_cd: str | None = None,
branch_number: str | None = None,
) -> str:
"""根据商品 ID 构建详情页 URL;tenpo_cd/branch_number 用于跳转查看指定店铺的报价"""
if not product_id: if not product_id:
raise ScrapeParseError("Either product_id or product_url must be provided") raise ScrapeParseError("Either product_id or product_url must be provided")
base_url = self._settings.base_url.rstrip("/") base_url = self._settings.base_url.rstrip("/")
return f"{base_url}/product/detail/{product_id}" url = f"{base_url}/product/detail/{product_id}"
params = []
if tenpo_cd:
params.append(f"tenpo_cd={tenpo_cd}")
if branch_number:
params.append(f"branch_number={branch_number}")
if params:
url += "?" + "&".join(params)
return url
def _parse_search_items(self, tree: HTMLParser) -> list[ProductSummary]: def _parse_search_items(self, tree: HTMLParser) -> list[ProductSummary]:
"""解析搜索结果列表 """解析搜索结果列表
@@ -568,7 +709,7 @@ class SurugayaClient:
"""将相对 URL 补全为完整的 https URL""" """将相对 URL 补全为完整的 https URL"""
if url.startswith("http://") or url.startswith("https://"): if url.startswith("http://") or url.startswith("https://"):
return url return url
return f"https://www.suruga-ya.jp{url}" return f"{BASE_URL}{url}"
def get_breadcrumb_list(self, tree): def get_breadcrumb_list(self, tree):
dom_list = tree.css(".block-blocksurugayabreadcrum nav .breadcrumb .breadcrumb-item") dom_list = tree.css(".block-blocksurugayabreadcrum nav .breadcrumb .breadcrumb-item")
+364
View File
@@ -0,0 +1,364 @@
# 骏河屋抓取服务部署文档
本文档说明如何将 `jp-surugaya`(Surugaya Scraper Service)部署到生产或开发环境。
## 1. 概述
本项目是一个基于 Python FastAPI + Playwright 的 HTTP API 服务,用于抓取骏河屋(suruga-ya.jp)的搜索列表、商品详情及通知信息。生产部署的核心要点:
- 必须能运行 Chromium 浏览器(Playwright)。
- 目标站点受 Cloudflare 保护,建议使用日本 IP 代理以提高稳定性。
- **必须部署 Redis**:用于 Cloudflare 会话复用,以及与下游采购 worker(`jp_pruchase_zdh_5`)共享任务队列、通知与订单状态。
- 默认监听 `0.0.0.0:31106`
## 2. 部署要求
### 2.1 运行环境
- Python >= 3.12
- 支持 Docker 与 Docker Compose(推荐)
- 可访问日本 IP 的代理(推荐,非必须)
- **必须:Redis >= 5.0**(本服务强依赖 Redis,部署时必须同时部署)
### 2.2 端口
- 服务默认暴露 `31106`
- 若修改端口,请同步调整 `docker-compose.yml` 的端口映射与 `.env` 中的 `SURUGAYA_APP_PORT`
## 2.3 Redis 部署
本服务强依赖 Redis,用于:
- 跨请求/跨实例复用通过 Cloudflare 后的会话(`cf_clearance`)。
- 与下游采购 worker(`jp_pruchase_zdh_5`)共享采购任务队列、通知(news)与订单(trade)状态。
- 作为 `/api/news``/api/trade`、任务投递等接口的数据源。
因此部署 `jp_surugaya` 时必须**同时部署并启动 Redis**。
### Windows 裸机部署 Redis
1. 下载 Windows 原生 Redis 二进制,例如 [redis-windows/releases](https://github.com/redis-windows/redis-windows/releases)。
2. 解压到 `C:\jp-deploy\redis`
3. 编辑 `redis.windows.conf`(可选):
- 设置 `requirepass your_redis_password`
- 设置 `bind 0.0.0.0` 或保持 `127.0.0.1` 按需监听。
4. 启动:
```powershell
cd C:\jp-deploy\redis
.\redis-server.exe redis.windows.conf
```
5. 验证:
```powershell
.\redis-cli.exe ping
# 若设置了密码
.\redis-cli.exe -a your_redis_password ping
```
### Docker Compose 部署 Redis
在 `docker-compose.yml` 中增加 `redis` 服务,例如:
```yaml
services:
redis:
image: redis:7-alpine
container_name: surugaya-redis
restart: unless-stopped
command: redis-server --requirepass your_redis_password
networks:
- jp
ports:
- "6379:6379"
volumes:
- ./redis-data:/data
jp_surugaya:
# ... 原有配置
environment:
SURUGAYA_REDIS_HOST: redis
SURUGAYA_REDIS_PASSWORD: your_redis_password
```
> 生产环境建议将 Redis 数据目录持久化,并限制 `6379` 端口仅内网可访问。
## 3. 环境变量配置
所有配置项均通过 `SURUGAYA_` 前缀的环境变量注入,也可写入项目根目录的 `.env` 文件,由 `pydantic-settings` 自动加载。生产部署至少需要修改以下变量:
| 变量名 | 说明 | 生产建议 |
|---|---|---|
| `SURUGAYA_APP_HOST` | 监听地址 | `0.0.0.0` |
| `SURUGAYA_APP_PORT` | 监听端口 | `31106` |
| `SURUGAYA_APP_ENV` | 运行环境 | `prod` |
| `SURUGAYA_BEARER_TOKEN` | API 鉴权 Token | 必须替换为强随机字符串 |
| `SURUGAYA_BROWSER_HEADLESS` | 是否无头模式 | `true` |
| `SURUGAYA_BROWSER_CHANNEL` | 浏览器通道 | 容器内留空,使用 bundled Chromium |
| `SURUGAYA_PROXY_SERVER` | 代理地址 | 建议配置日本 IP 代理 |
| `SURUGAYA_PROXY_USERNAME` | 代理用户名 | 按需配置 |
| `SURUGAYA_PROXY_PASSWORD` | 代理密码 | 按需配置 |
| `SURUGAYA_REDIS_HOST` | Redis 主机 | **必须**,例如 `127.0.0.1` 或 `redis` |
完整示例见项目根目录 `.env.example`。
### 3.1 生产环境 `.env` 示例
```ini
SURUGAYA_APP_HOST=0.0.0.0
SURUGAYA_APP_PORT=31106
SURUGAYA_APP_ENV=prod
SURUGAYA_LOG_LEVEL=INFO
SURUGAYA_BEARER_TOKEN=your-strong-random-token-here
SURUGAYA_BROWSER_ENABLED=true
SURUGAYA_BROWSER_HEADLESS=true
SURUGAYA_BROWSER_CHANNEL=
SURUGAYA_BROWSER_KEEP_OPEN_ON_ERROR=false
SURUGAYA_PROXY_SERVER=http://your-jp-proxy:7890
SURUGAYA_PROXY_USERNAME=
SURUGAYA_PROXY_PASSWORD=
SURUGAYA_REDIS_HOST=redis
SURUGAYA_REDIS_PORT=6379
SURUGAYA_REDIS_PASSWORD=
SURUGAYA_REDIS_DB=0
```
## 4. 部署方式一:Docker Compose(推荐)
项目已提供 `Dockerfile` 与 `docker-compose.yml`,基于官方 Playwright Python 镜像构建,使用 `uv` 管理依赖。
### 4.1 步骤
1. 克隆仓库并进入项目根目录。
2. 确保 Redis 已部署:
- 若使用 Docker Compose,请在 `docker-compose.yml` 中加入 Redis 服务(参考 [2.3 Redis 部署](#23-redis-部署))。
- 若使用外部 Redis,在 `.env` 中正确填写 `SURUGAYA_REDIS_HOST` 与密码。
3. 复制环境变量模板并修改:
```bash
cp .env.example .env
# 编辑 .env,重点修改 SURUGAYA_BEARER_TOKEN、Redis 连接与代理配置
```
4. 构建并启动容器:
```bash
docker compose up -d --build
```
5. 查看日志:
```bash
docker compose logs -f
```
### 4.2 说明
- `docker-compose.yml` 使用 `.:/app` 挂载宿主目录,因此 `venv` 被放置在 `/opt/venv`,避免被卷覆盖。
- 容器内 `SURUGAYA_BROWSER_CHANNEL` 已强制留空,使用 Playwright 自带的 Chromium。
- 时区设置为 `Asia/Tokyo`。
### 4.3 停止与重启
```bash
docker compose down
docker compose up -d --build
```
## 5. 部署方式二:裸机/虚拟机
### 5.1 前置条件:部署 Redis
裸机/虚拟机部署前,请先部署 Redis:
- **Windows**:参考 [2.3 Redis 部署](#23-redis-部署) 中的 Windows 裸机方式。
- **Linux**:安装并启动 Redis:
```bash
sudo apt-get install redis-server
sudo systemctl enable redis-server --now
```
- 确认 `.env` 中的 `SURUGAYA_REDIS_HOST`、`SURUGAYA_REDIS_PORT`、`SURUGAYA_REDIS_PASSWORD` 配置正确。
### 5.2 安装依赖
推荐使用 `uv` 作为包管理器:
```bash
# 安装 uv(若未安装)
curl -LsSf https://astral.sh/uv/install.sh | sh
# 同步依赖(包含 dev 依赖)
uv sync --extra dev
# 安装 Chromium 浏览器
uv run playwright install chromium
```
> 说明:`surugaya-common` 位于本仓库 `packages/surugaya_common` 目录下,通过 `tool.uv.workspace` 作为本地 workspace 包被引用。`uv sync` 会自动安装该本地包,无需额外配置私有索引。
若使用传统 pip:
```bash
python -m pip install -e ".[dev]"
python -m playwright install chromium
```
### 5.3 启动服务
```bash
# 方式 1:直接运行
python -m app.main
# 方式 2:使用 uvicorn
uvicorn app.main:app --host 0.0.0.0 --port 31106
```
### 5.4 后台运行(systemd 示例)
创建 `/etc/systemd/system/surugaya.service`:
```ini
[Unit]
Description=Surugaya Scraper Service
After=network.target
[Service]
Type=simple
User=app
WorkingDirectory=/opt/jp_surugaya
Environment=PYTHONUNBUFFERED=1
EnvironmentFile=/opt/jp_surugaya/.env
ExecStart=/opt/jp_surugaya/.venv/bin/python -m app.main
Restart=on-failure
RestartSec=5
[Install]
WantedBy=multi-user.target
```
启用并启动:
```bash
sudo systemctl daemon-reload
sudo systemctl enable surugaya
sudo systemctl start surugaya
```
## 6. 部署方式三:Windows 本地开发
Windows 本地开发前,请先按 [2.3 Redis 部署](#23-redis-部署) 启动 Redis。
启动脚本:
```powershell
.\start-server.ps1
```
可选参数:
- `-Port 31106`:指定端口。
- `-HostName 0.0.0.0`:指定监听地址。
- `-NoKillExisting`:端口占用时不自动结束旧进程。
- `-DryRun`:仅做检查,不实际启动。
## 7. 启动验证
### 7.1 健康检查
```bash
curl http://127.0.0.1:31106/health
```
### 7.2 API 鉴权测试
```bash
curl -H "Authorization: Bearer <SURUGAYA_BEARER_TOKEN>" \
-H "Content-Type: application/json" \
-X POST "http://127.0.0.1:31106/api/search" \
-d '{"search_word":"cat","category":"","page":1}'
```
## 8. 生产环境建议
### 8.1 代理与 IP
- 骏河屋对非日本 IP 访问可能触发更严格的 Cloudflare 挑战,建议稳定使用日本 IP 代理。
- 代理配置变更后需重启服务生效。
### 8.2 浏览器回收
长时间运行会导致 Chromium 内存累积,建议保留默认回收策略:
- `SURUGAYA_BROWSER_RECYCLE_AFTER_CONTEXTS=200`
- `SURUGAYA_BROWSER_RECYCLE_AFTER_SECONDS=3600`
如需关闭回收,可设为 `0`。
### 8.3 会话存储
- **Redis 是必选项**:本服务必须与 Redis 一起部署,用于会话复用、与下游采购 worker 共享任务与状态。
- 配置 Redis 后,`cf_clearance` 等会话可在实例间共享;未配置 Redis 时,部分接口(如 `/api/news`、`/api/trade`、任务投递)将不可用。
### 8.4 日志
生产环境默认写文件到 `logs/` 目录,支持按大小/时间切割与保留清理。`.env.example` 示例按 `100 MB` 切割、保留 `14 days`。可通过以下变量调整:
- `SURUGAYA_LOG_TO_FILE`
- `SURUGAYA_LOG_DIR`
- `SURUGAYA_LOG_ROTATION`
- `SURUGAYA_LOG_RETENTION`
- `SURUGAYA_LOG_COMPRESSION`
## 9. 故障排查
### 9.1 浏览器启动失败
- 确认已执行 `playwright install chromium`。
- 容器内确认 `SURUGAYA_BROWSER_CHANNEL=` 留空。
- Linux 服务器需安装系统依赖,Playwright 镜像已包含。
### 9.2 Cloudflare 验证反复失败
- 检查代理是否为日本 IP。
- 检查浏览器指纹配置与 stealth 是否启用(`SURUGAYA_BROWSER_STEALTH_ENABLED` 默认 `true`)。
- 查看日志中是否有 `cf_clearance` 获取成功/失败的记录。
### 9.3 端口占用
- 裸机部署时使用 `start-server.ps1` 会自动结束占用进程。
- Linux 可使用 `lsof -i :31106` 或 `ss -tlnp | grep 31106` 定位占用进程。
### 9.4 Redis 连接失败
- 确认 Redis 服务已启动:`redis-cli ping`(Linux)或 `.\redis-cli.exe ping`(Windows)。
- 检查 `.env` 中的 `SURUGAYA_REDIS_HOST`、`SURUGAYA_REDIS_PORT`、`SURUGAYA_REDIS_PASSWORD` 是否正确。
- 确认防火墙未拦截 Redis 端口(默认 `6379`)。
### 9.5 内存持续增长
- 检查 `SURUGAYA_BROWSER_RECYCLE_AFTER_CONTEXTS` 与 `SURUGAYA_BROWSER_RECYCLE_AFTER_SECONDS` 是否被误关闭。
- 长时间运行后观察 `logs/` 中浏览器回收日志。
## 10. 升级维护
1. 拉取最新代码。
2. 若依赖有变更,重新同步:
```bash
uv sync --extra dev
```
3. Docker Compose 部署:
```bash
docker compose down
docker compose up -d --build
```
4. 裸机部署:
```bash
sudo systemctl restart surugaya
```
+104
View File
@@ -0,0 +1,104 @@
# 通知(お知らせ)查询接口
读取 worker 定时抓取的骏河屋「お知らせ一覧」(action_news) 通知。worker 翻页抓取后按
`data_id` 去重写入 Redis,本接口只读返回,不删除、不消费,可重复拉取。
## 请求
| 项 | 值 |
| --- | --- |
| 方法 | `GET` |
| 路径 | `/api/news` |
| 鉴权 | 请求头 `Authorization: Bearer <token>`(token 由服务端 `SURUGAYA_BEARER_TOKEN` 配置) |
### Query 参数
| 参数 | 类型 | 必填 | 说明 |
| --- | --- | --- | --- |
| `limit` | int | 否 | 返回条数上限,范围 `1`~`500`;省略则返回全部。结果按 `data_id` 倒序(新→旧),`limit` 取最新的若干条 |
### 示例
```
GET /api/news?limit=20
Authorization: Bearer NilsO8Il8yoT1wazJFs2eeOmlPRfWuSx
```
## 成功响应 `200`
```json
{
"success": true,
"msg": "success",
"code": 0,
"data": [
{
"data_id": "869431007",
"date": "2026/06/12",
"message": "【販売】(取引番号:S2606115133)発送準備中です。",
"data_type": "2",
"data_url": "/pcmypage/action_sell_search/detail?trade_code=S2606115133",
"trade_code": "S2606115133",
"account_id": "trdian023@tokugen.jp",
"fetched_at": 1781348570
},
{
"data_id": "868997716",
"date": "2026/06/05",
"message": "【販売】(取引番号:M2606056999)商品の注文を受け付けました。",
"data_type": "2",
"data_url": "/pcmypage/action_sell_search/detail?trade_code=M2606056999",
"trade_code": "M2606056999",
"account_id": "trdian023@tokugen.jp",
"fetched_at": 1781348570
},
{
"data_id": "860001234",
"date": "2026/01/01",
"message": "【お知らせ】メンテナンスのお知らせ",
"data_type": "1",
"data_url": "/pcmypage/action_news_detail?id=860001234",
"trade_code": null,
"account_id": "trdian023@tokugen.jp",
"fetched_at": 1781348570
}
]
}
```
### data 字段说明
| 字段 | 类型 | 说明 |
| --- | --- | --- |
| `data_id` | string | 骏河屋侧通知唯一标识(单调递增),去重与排序依据 |
| `date` | string | 通知日期,格式 `YYYY/MM/DD` |
| `message` | string | 通知文案原文(日文) |
| `data_type` | string | 骏河屋侧 `data-type` 原值(如 `2`=交易类通知) |
| `data_url` | string \| null | 骏河屋侧 `data-url`(详情相对路径) |
| `trade_code` | string \| null | 从 `data_url` 解析的交易单号;非交易类通知为 `null` |
| `account_id` | string | 该通知所属账号(骏河屋登录邮箱) |
| `fetched_at` | int | worker 抓取写入时间(Unix epoch 秒) |
> 说明
> - 列表按 `data_id` 数值倒序返回(最新在前)。
> - 同一 `data_id` 只保留一条,重复抓取会以最新内容覆盖,调用方可安全重复拉取。
> - 需要区分账号时按 `account_id` 过滤即可(当前为登录邮箱)。
## 错误响应
统一响应体结构:`success=false``code` 为业务错误码,`data=null`
| HTTP | code | 场景 | 示例 msg |
| --- | --- | --- | --- |
| 401 | 1001 | 缺少/无效 Bearer Token | `Invalid token` |
| 422 | 1002 | `limit` 越界(不在 1~500) | `limit: Input should be less than or equal to 500` |
| 500 | 500 | 服务端未配置 Redis,通知不可读 | `Redis 未配置` |
```json
{
"success": false,
"msg": "Redis 未配置",
"code": 500,
"data": null
}
```
+6 -3
View File
@@ -4,8 +4,8 @@ build-backend = "hatchling.build"
[project] [project]
name = "surugaya-common" name = "surugaya-common"
version = "0.2.0" version = "0.4.0"
description = "Shared contracts (Redis keys, future models) for jp_surugaya services" description = "Shared contracts and browser utilities (Redis keys, models, parsers, fingerprint profile) for jp_surugaya services"
requires-python = ">=3.12" requires-python = ">=3.12"
dependencies = [ dependencies = [
"pydantic>=2.0.0,<3.0.0", "pydantic>=2.0.0,<3.0.0",
@@ -15,7 +15,10 @@ dependencies = [
# playwright 仅作为可选 extra:只有解析器/浏览器工具需要它,核心安装保持 light。 # playwright 仅作为可选 extra:只有解析器/浏览器工具需要它,核心安装保持 light。
# 范围取 jp_surugaya(>=1.54,<2) 与 worker(>=1.43) 的交集,确保 workspace 解析到单一 playwright。 # 范围取 jp_surugaya(>=1.54,<2) 与 worker(>=1.43) 的交集,确保 workspace 解析到单一 playwright。
[project.optional-dependencies] [project.optional-dependencies]
browser = ["playwright>=1.54.0,<2.0.0"] browser = [
"playwright>=1.54.0,<2.0.0",
"playwright-stealth>=1.0.0,<3.0.0",
]
[tool.hatch.build.targets.wheel] [tool.hatch.build.targets.wheel]
packages = ["src/surugaya_common"] packages = ["src/surugaya_common"]
@@ -1,5 +1,10 @@
"""Shared contracts for jp_surugaya services (Redis keys, future models).""" """Shared contracts and utilities for jp_surugaya services.
from surugaya_common import redis_keys 核心安装(无 extra)可用:redis_keys、urls、browser_profile、parse_utils、
app_utils、http_utils、models。
需安装 surugaya_common[browser]:browser_utils、stealth、parsers。
"""
__all__ = ["redis_keys"] from surugaya_common import browser_profile, parse_utils, redis_keys, urls
__all__ = ["browser_profile", "parse_utils", "redis_keys", "urls"]
@@ -0,0 +1,43 @@
"""浏览器指纹与启动参数共享配置(纯常量,core 安装即可用,不依赖 playwright)。
API(jp_surugaya)与 worker(jp_pruchase_zdh_5)访问同一目标站点,
指纹参数必须保持一致:UA、语言、时区任何一处自相矛盾都可能触发风控
(参见 cf_clearance 复用因指纹不一致被判 403 的教训)。
"""
# 与日语环境一致的指纹参数
LOCALE = "ja-JP"
TIMEZONE_ID = "Asia/Tokyo"
NAVIGATOR_LANGUAGES = ("ja-JP", "ja", "en-US", "en")
ACCEPT_LANGUAGE = "ja-JP,ja;q=0.9,en-US;q=0.8,en;q=0.7"
# 两项目统一使用的 Windows Chrome User-Agent
DEFAULT_USER_AGENT = (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/124.0.0.0 Safari/537.36"
)
# 页面级反检测请求头(page.set_extra_http_headers 用)
DEFAULT_PAGE_HEADERS = {
"Accept-Language": ACCEPT_LANGUAGE,
"Cache-Control": "no-cache",
"Pragma": "no-cache",
}
# 两项目共用的 Chromium 反检测启动参数
STEALTH_LAUNCH_ARGS = (
"--disable-blink-features=AutomationControlled", # 浏览器层面移除 navigator.webdriver
"--disable-infobars", # 隐藏「Chrome 正受到自动测试软件控制」提示条
"--no-sandbox", # Docker/服务器环境兼容
"--disable-setuid-sandbox",
"--disable-dev-shm-usage", # 内存受限环境避免资源崩溃
"--disable-gpu", # 服务器无头模式稳定渲染
"--use-mock-keychain", # 避免本地密钥环弹窗(macOS/Linux)
"--password-store=basic",
)
def stealth_launch_args() -> list[str]:
"""返回共享启动参数的可变副本,调用方可在其上追加进程自有参数。"""
return list(STEALTH_LAUNCH_ARGS)
@@ -1,14 +1,81 @@
"""浏览器拟真交互工具(需 playwright,安装 surugaya_common[browser])。 """浏览器拟真交互工具(需 playwright,安装 surugaya_common[browser])。
从 jp_pruchase_zdh_5/src/utils/page_utils.py 忠实迁移 human_scroll,逻辑保持一致。 - human_scroll:从 jp_pruchase_zdh_5/src/utils/page_utils.py 忠实迁移,逻辑保持一致。
注:page_utils.py 中的纯函数 format_japanese_price 未迁移——它与 API 端 - clear_cart:worker 与 API cargo_service 各自的 clear_cart 近乎逐字重复,
app/utils/parse_util.py 的 format_price 语义不同,合并属行为变更,单独处理 统一为本实现(以 worker 版为准,保留拟真随机等待)
- format_japanese_price 已迁至 surugaya_common.parse_utils(纯 stdlib,不依赖 playwright)。
""" """
from playwright.async_api import Page from playwright.async_api import Page
from playwright.async_api import TimeoutError as PlaywrightTimeoutError
import asyncio import asyncio
import logging
import random import random
logger = logging.getLogger(__name__)
# 购物车页(/cargo/detail)的关键选择器
CART_ITEM_SELECTOR = "table.item_box tbody tr.item"
CART_REMOVE_LINK_SELECTOR = "table.item_box a.remove_product"
async def count_cart_items(page: Page) -> int:
"""统计购物车页当前的商品条目数(需已位于 /cargo/detail 页面)。"""
return await page.locator(CART_ITEM_SELECTOR).count()
async def clear_cart(page: Page) -> None:
"""清空购物车(需已位于 /cargo/detail 页面)。
- 商品数量超过 3 个时优先点击「全削除」按钮一次性清空
- 否则逐条点击删除链接,等待条目数减少后继续,含拟真随机等待
"""
logger.debug("正在清空购物车...")
cart_goods_num = await count_cart_items(page)
if cart_goods_num < 1:
return
# 购物车商品数量超过3个后会出现"全削除"按钮
if cart_goods_num > 3:
try:
delete_button = page.locator("#total_box .delbtn-all").get_by_role("button", name="全削除")
# 显式等待按钮在页面上可见(容错处理:给它最多 3000 毫秒的加载时间)
await delete_button.wait_for(state="visible", timeout=3000)
if await delete_button.is_visible():
logger.debug("检测到'全削除'按钮存在,正在执行点击...")
await delete_button.click()
return
except Exception:
logger.debug("未检测到'全削除'按钮。")
# 遍历删除
await page.wait_for_timeout(random.randint(1000, 3000))
while True:
cart_goods_num = await page.locator(CART_ITEM_SELECTOR).count()
if cart_goods_num == 0:
break
remove_link = page.locator(CART_REMOVE_LINK_SELECTOR).first
try:
await remove_link.click()
# 删除按钮不会触发弹窗,直接等待购物车条目数减少即可。
await page.wait_for_function(
"""({selector, expectedCount}) => {
return document.querySelectorAll(selector).length < expectedCount;
}""",
arg={"selector": CART_ITEM_SELECTOR, "expectedCount": cart_goods_num},
timeout=15_000,
)
await page.wait_for_timeout(random.randint(800, 1500))
except PlaywrightTimeoutError:
latest_count = await page.locator(CART_ITEM_SELECTOR).count()
if latest_count < cart_goods_num:
logger.debug("购物车商品已删除,当前购物车数量:%s", latest_count)
continue
logger.warning("删除商品后数量未变化,等待后重试")
await page.wait_for_timeout(1000)
async def human_scroll(page: Page, mode: str, value: int = 0): async def human_scroll(page: Page, mode: str, value: int = 0):
""" """
@@ -0,0 +1,24 @@
"""纯文本解析工具(仅 stdlib)。
从 jp_pruchase_zdh_5/src/utils/page_utils.py 忠实迁移 format_japanese_price。
"""
import re
def format_japanese_price(price_str: str | None) -> int:
"""
将日本电商网站的价格字符串格式化为标准整数
支持的格式示例: "600円 (税込)", "1,250円", "¥ 3,000", "300"
注意:实现为提取字符串中所有数字字符后拼接,与 jp_surugaya API 端
parse_util.format_price(货币符号定位 + 价格范围取低价)语义不同,勿混用。
"""
if not price_str:
return 0
digits = re.findall(r'\d', price_str)
if not digits:
return 0
return int("".join(digits))
@@ -10,6 +10,7 @@ Current literals (as of this extraction):
trade hash : jp_surugaya_trade trade hash : jp_surugaya_trade
trade pending dedup : jp_surugaya_trade_pending:{trade_code} trade pending dedup : jp_surugaya_trade_pending:{trade_code}
trade fetch stream : jp_surugaya_trade_fetch trade fetch stream : jp_surugaya_trade_fetch
news hash : jp_surugaya_news[_dev]
""" """
from __future__ import annotations from __future__ import annotations
@@ -21,6 +22,7 @@ TRADE_DELAY_ZSET: str = "jp_surugaya_trade_delay"
TRADE_HASH: str = "jp_surugaya_trade" TRADE_HASH: str = "jp_surugaya_trade"
TRADE_PENDING_PREFIX: str = "jp_surugaya_trade_pending" TRADE_PENDING_PREFIX: str = "jp_surugaya_trade_pending"
TRADE_FETCH_STREAM: str = "jp_surugaya_trade_fetch" TRADE_FETCH_STREAM: str = "jp_surugaya_trade_fetch"
NEWS_HASH: str = "jp_surugaya_news"
# Suffix appended to dev-mode keys. # Suffix appended to dev-mode keys.
_DEV_SUFFIX: str = "_dev" _DEV_SUFFIX: str = "_dev"
@@ -78,3 +80,14 @@ def trade_pending_key(trade_code: object) -> str:
def trade_fetch_stream() -> str: def trade_fetch_stream() -> str:
"""Redis Stream key for trade fetch tasks.""" """Redis Stream key for trade fetch tasks."""
return TRADE_FETCH_STREAM 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
@@ -0,0 +1,24 @@
"""playwright-stealth 共享工厂(需安装 surugaya_common[browser])。
统一两项目的 stealth 指纹配置:navigator.languages 覆盖必须与日语
locale / Accept-Language 对齐,否则带 cf_clearance 的复用请求会因
指纹自相矛盾被 Cloudflare 判 403。
"""
from playwright_stealth import Stealth
from surugaya_common.browser_profile import NAVIGATOR_LANGUAGES
def create_stealth() -> Stealth:
"""创建与日语环境对齐的 Stealth 实例。
用法约定:
- browser.new_context 流程:用 stealth.use_async(async_playwright()) 钩子
包装 playwright,new_context/new_page 自动应用 stealth(含 HTTP
sec-ch-ua 客户端提示)。jp_surugaya BrowserPool 即此用法。
- launch_persistent_context 流程:use_async 钩子不覆盖它(playwright-stealth
已知缺口),必须显式 await stealth.apply_stealth_async(context),
此路径只注入 init script,无 sec-ch-ua 等 HTTP 头伪装。
"""
return Stealth(navigator_languages_override=NAVIGATOR_LANGUAGES)
@@ -0,0 +1,32 @@
"""suruga-ya.jp 站点 URL 契约:两项目共享的页面地址与构造器。"""
BASE_URL = "https://www.suruga-ya.jp"
# 购物车详情页
CARGO_DETAIL_URL = f"{BASE_URL}/cargo/detail"
# 会员中心(登录入口)
MYPAGE_URL = f"{BASE_URL}/pcmypage"
def product_detail_url(product_id: str, tenpo_cd: str | None = None, branch_number: str | None = None) -> str:
"""商品详情页地址;tenpo_cd/branch_number 用于指定加盟店/市场店铺的具体报价。"""
url = f"{BASE_URL}/product/detail/{product_id}"
params = []
if tenpo_cd:
params.append(f"tenpo_cd={tenpo_cd}")
if branch_number:
params.append(f"branch_number={branch_number}")
if params:
url += "?" + "&".join(params)
return url
def product_other_url(product_id: str) -> str:
"""商品「他のショップ」页地址:列出该商品全部店铺(官方自营+加盟店/市场店铺)报价。"""
return f"{BASE_URL}/product/other/{product_id}"
def trade_detail_url(trade_code: str) -> str:
"""交易(订单)详情页地址。"""
return f"{BASE_URL}/pcmypage/action_sell_search/detail?trade_code={trade_code}"
+2 -3
View File
@@ -6,15 +6,14 @@ requires-python = ">=3.12"
dependencies = [ dependencies = [
"fastapi>=0.115.0,<1.0.0", "fastapi>=0.115.0,<1.0.0",
"loguru>=0.7.0,<1.0.0", "loguru>=0.7.0,<1.0.0",
"playwright>=1.54.0,<2.0.0",
"playwright-stealth>=1.0.0,<3.0.0",
"pydantic>=2.0.0,<3.0.0", "pydantic>=2.0.0,<3.0.0",
"pydantic-settings>=2.4.0,<3.0.0", "pydantic-settings>=2.4.0,<3.0.0",
"redis==8.0.0", "redis==8.0.0",
"selectolax>=0.3.21,<1.0.0", "selectolax>=0.3.21,<1.0.0",
"starlette>=0.37.0,<1.0.0", "starlette>=0.37.0,<1.0.0",
"uvicorn[standard]>=0.30.0,<1.0.0", "uvicorn[standard]>=0.30.0,<1.0.0",
"surugaya-common", # playwright 与 playwright-stealth 由 [browser] extra 提供,与 worker 保持同一来源
"surugaya-common[browser]",
] ]
[project.optional-dependencies] [project.optional-dependencies]
+162
View File
@@ -0,0 +1,162 @@
"""浏览器池 driver 断链自动恢复逻辑的单元测试(不启动真实 Chromium)。
覆盖本次修复的核心:
- is_disconnect_error 对 WriteUnixTransport "the handler is closed" 等断链错误的识别
- _run_with_reconnect 在断链后重建一次并重试成功
- _browser_generation 守卫:并发下他人已重建时不重复重建
"""
from __future__ import annotations
import pytest
from app.core.config import Settings
from app.services.browser_pool import BrowserPool
class _FakeBrowser:
def __init__(self, connected: bool = True):
self._connected = connected
def is_connected(self) -> bool:
return self._connected
def _make_pool(**overrides) -> BrowserPool:
settings = Settings(**overrides)
pool = BrowserPool(settings)
pool._browser = _FakeBrowser()
pool._started = True
pool._browser_generation = 1
return pool
# 复现线上崩溃日志里的真实报错文本
_HANDLER_CLOSED = (
"Stealth.hooked_new_context_async: unable to perform operation on "
"<WriteUnixTransport closed=True reading=False 0x7f9a27185a40>; the handler is closed"
)
def test_is_disconnect_error_recognizes_closed_transport():
assert BrowserPool.is_disconnect_error(RuntimeError(_HANDLER_CLOSED)) is True
assert BrowserPool.is_disconnect_error(Exception("connection closed")) is True
# 业务异常不应被误判为断链
assert BrowserPool.is_disconnect_error(ValueError("invalid selector")) is False
@pytest.mark.asyncio
async def test_run_with_reconnect_rebuilds_then_retries(monkeypatch):
"""首次因断链失败 → 重建一次 → 第二次成功。"""
pool = _make_pool()
rebuilt: list[str] = []
async def fake_restart(reason, *, observed_generation=None):
rebuilt.append(reason)
pool._browser_generation += 1 # 模拟 start() 自增代次
monkeypatch.setattr(pool, "_restart_browser", fake_restart)
attempts = {"n": 0}
async def factory():
attempts["n"] += 1
if attempts["n"] == 1:
raise RuntimeError(_HANDLER_CLOSED)
return "ok"
result = await pool._run_with_reconnect(factory, "测试操作")
assert result == "ok"
assert attempts["n"] == 2
assert len(rebuilt) == 1
@pytest.mark.asyncio
async def test_run_with_reconnect_gives_up_after_one_retry(monkeypatch):
"""连续两次断链:只重试一次,第二次仍失败则抛出。"""
pool = _make_pool()
async def fake_restart(reason, *, observed_generation=None):
pool._browser_generation += 1
monkeypatch.setattr(pool, "_restart_browser", fake_restart)
attempts = {"n": 0}
async def factory():
attempts["n"] += 1
raise RuntimeError(_HANDLER_CLOSED)
with pytest.raises(RuntimeError):
await pool._run_with_reconnect(factory, "测试操作")
assert attempts["n"] == 2 # 初次 + 一次重试
@pytest.mark.asyncio
async def test_run_with_reconnect_passes_through_non_disconnect(monkeypatch):
"""非断链异常不触发重建,直接抛出。"""
pool = _make_pool()
rebuilt: list[str] = []
async def fake_restart(reason, *, observed_generation=None):
rebuilt.append(reason)
monkeypatch.setattr(pool, "_restart_browser", fake_restart)
async def factory():
raise ValueError("业务错误")
with pytest.raises(ValueError):
await pool._run_with_reconnect(factory, "测试操作")
assert rebuilt == []
@pytest.mark.asyncio
async def test_restart_browser_skips_when_generation_changed():
"""持锁后代次已变化(他人已重建)→ 直接返回,不重复重建。"""
pool = _make_pool()
closed: list[str] = []
async def fake_close():
closed.append("close")
pool.close = fake_close # type: ignore[method-assign]
# 调用方观察到的是旧代次,而当前代次已被他人推进
await pool._restart_browser("stale", observed_generation=pool._browser_generation - 1)
assert closed == [] # 未触发关闭/重建
@pytest.mark.asyncio
async def test_open_context_and_page_closes_context_on_page_failure(monkeypatch):
"""页面创建失败时关闭半开上下文,避免泄漏。"""
pool = _make_pool()
class _Ctx:
def __init__(self):
self.closed = False
async def new_page(self):
raise RuntimeError("new_page boom")
ctx = _Ctx()
async def fake_new_context_raw(storage_state):
return ctx
closed: list[object] = []
async def fake_close_context(c):
closed.append(c)
monkeypatch.setattr(pool, "_new_context_raw", fake_new_context_raw)
monkeypatch.setattr(pool, "close_context", fake_close_context)
with pytest.raises(RuntimeError, match="new_page boom"):
await pool._open_context_and_page(None)
assert closed == [ctx]
+75
View File
@@ -0,0 +1,75 @@
"""surugaya_common 共享契约回归测试:URL 构造与价格解析必须与迁移前字面量/行为一致。"""
from surugaya_common import browser_profile
from surugaya_common.parse_utils import format_japanese_price
from surugaya_common.urls import (
BASE_URL,
CARGO_DETAIL_URL,
MYPAGE_URL,
product_detail_url,
trade_detail_url,
)
class TestUrls:
def test_literals_match_pre_migration(self):
assert BASE_URL == "https://www.suruga-ya.jp"
assert CARGO_DETAIL_URL == "https://www.suruga-ya.jp/cargo/detail"
assert MYPAGE_URL == "https://www.suruga-ya.jp/pcmypage"
def test_product_detail_url(self):
assert product_detail_url("128001234") == "https://www.suruga-ya.jp/product/detail/128001234"
assert (
product_detail_url("128001234", tenpo_cd="400496")
== "https://www.suruga-ya.jp/product/detail/128001234?tenpo_cd=400496"
)
# 空字符串 tenpo_cd 与未提供等价(与迁移前 if tenpo_cd 判断一致)
assert product_detail_url("128001234", tenpo_cd="") == "https://www.suruga-ya.jp/product/detail/128001234"
assert (
product_detail_url("128001234", tenpo_cd="400496", branch_number="0003")
== "https://www.suruga-ya.jp/product/detail/128001234?tenpo_cd=400496&branch_number=0003"
)
assert (
product_detail_url("128001234", branch_number="0003")
== "https://www.suruga-ya.jp/product/detail/128001234?branch_number=0003"
)
def test_trade_detail_url(self):
assert (
trade_detail_url("T123456")
== "https://www.suruga-ya.jp/pcmypage/action_sell_search/detail?trade_code=T123456"
)
class TestFormatJapanesePrice:
def test_supported_formats(self):
assert format_japanese_price("600円 (税込)") == 600
assert format_japanese_price("1,250円") == 1250
assert format_japanese_price("¥ 3,000") == 3000
assert format_japanese_price("300") == 300
def test_empty_and_no_digit(self):
assert format_japanese_price(None) == 0
assert format_japanese_price("") == 0
assert format_japanese_price("税込") == 0
class TestBrowserProfile:
def test_fingerprint_consistency(self):
# UA / 语言 / 时区必须保持日语环境自洽,且与迁移前两仓字面量一致
assert browser_profile.DEFAULT_USER_AGENT == (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/124.0.0.0 Safari/537.36"
)
assert browser_profile.LOCALE == "ja-JP"
assert browser_profile.TIMEZONE_ID == "Asia/Tokyo"
assert browser_profile.NAVIGATOR_LANGUAGES == ("ja-JP", "ja", "en-US", "en")
assert browser_profile.ACCEPT_LANGUAGE.startswith("ja-JP,ja;")
assert browser_profile.DEFAULT_PAGE_HEADERS["Accept-Language"] == browser_profile.ACCEPT_LANGUAGE
def test_stealth_launch_args_returns_copy(self):
args = browser_profile.stealth_launch_args()
assert "--disable-blink-features=AutomationControlled" in args
args.append("--extra")
assert "--extra" not in browser_profile.stealth_launch_args()
+64
View File
@@ -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
+68
View File
@@ -0,0 +1,68 @@
"""/api/order/get_shipping_fee 中"店铺标识"分桶逻辑的单元测试:
- 店铺标识优先取 tenpo_cd(与 other_shop_list 一致),未提供时回退取 supplier
- 同 tenpo_cd(或同 supplier)视为同一家加盟店,只收一次 800 日元
"""
from __future__ import annotations
import pytest
from app.api.routes.scrape import order_get_shipping_fee
from app.models.scrape import OrderItem, OrderShippingFeeRequest
@pytest.mark.asyncio
async def test_official_only_uses_official_ladder() -> None:
request = OrderShippingFeeRequest(item_list=[OrderItem(goods_id="1", goods_number=1, goods_price=800)])
resp = await order_get_shipping_fee(request)
assert resp.data.shipping_fee == 440
assert resp.data.goods_amount == 800
@pytest.mark.asyncio
async def test_tenpo_cd_groups_same_franchise_store_once() -> None:
request = OrderShippingFeeRequest(
item_list=[
OrderItem(goods_id="1", goods_number=1, goods_price=1000, tenpo_cd="400451"),
OrderItem(goods_id="2", goods_number=1, goods_price=2000, tenpo_cd="400451", branch_number="0003"),
]
)
resp = await order_get_shipping_fee(request)
assert resp.data.shipping_fee == 800 # 同一 tenpo_cd 只收一次加盟店运费
assert resp.data.goods_amount == 3000
@pytest.mark.asyncio
async def test_different_tenpo_cd_charged_separately() -> None:
request = OrderShippingFeeRequest(
item_list=[
OrderItem(goods_id="1", goods_number=1, goods_price=1000, tenpo_cd="400451"),
OrderItem(goods_id="2", goods_number=1, goods_price=1000, tenpo_cd="400999"),
]
)
resp = await order_get_shipping_fee(request)
assert resp.data.shipping_fee == 1600 # 两家不同加盟店各收 800
@pytest.mark.asyncio
async def test_tenpo_cd_takes_precedence_over_supplier() -> None:
request = OrderShippingFeeRequest(
item_list=[
OrderItem(goods_id="1", goods_number=1, goods_price=1000, supplier="山口大学前店", tenpo_cd="400451"),
OrderItem(goods_id="2", goods_number=1, goods_price=1000, supplier="其他名称", tenpo_cd="400451"),
]
)
resp = await order_get_shipping_fee(request)
assert resp.data.shipping_fee == 800 # 尽管 supplier 文本不同,tenpo_cd 相同仍视为同一家
@pytest.mark.asyncio
async def test_falls_back_to_supplier_when_tenpo_cd_missing() -> None:
request = OrderShippingFeeRequest(
item_list=[
OrderItem(goods_id="1", goods_number=1, goods_price=1000, supplier="老加盟店"),
OrderItem(goods_id="2", goods_number=1, goods_price=1000, supplier="老加盟店"),
]
)
resp = await order_get_shipping_fee(request)
assert resp.data.shipping_fee == 800 # 无 tenpo_cd 时按 supplier 分桶,保持向后兼容
+168
View File
@@ -0,0 +1,168 @@
"""SurugayaClient._parse_other_shop_list 的解析单元测试。
固定 HTML 节选自真实的 /product/other/{goods_id} 页面结构(tbl_all 标签页),
覆盖官方自营行(无 tenpo_cd/无评分)与加盟店/市场店铺行(含 tenpo_cd/评分)。
"""
from __future__ import annotations
from selectolax.parser import HTMLParser
from app.services.surugaya_client import SurugayaClient
OTHER_SHOP_HTML = """
<div class="tab-item" id="tabs-all">
<table id="tbl_all">
<tr class="thead"><th>头</th></tr>
<tr class="item">
<td><p><strong class="text-red text-bold mgnL10">1,370円</strong></p></td>
<td>
<a href="/product/detail/220035970?tenpo_cd=400451&branch_number=0003">
<h2 class="title_product">中古箱・ジャケット・ケース不備(中)</h2>
</a>
</td>
<td>
<div class="space_text_1 mgnB5"><strong><a href="/shop/400451">駿河屋 山口大学前店</a></strong></div>
<div class="star_group mgnB5"></div>
<div>5.0(1334件)</div>
</td>
<td>
<ul class="shipping_campaing_info">
<li class="padT5">1日〜3日以内に発送します</li>
<li class="all ajax-campaign-placeholder"
data-zaiko_data="{&quot;shinaban&quot;:&quot;220035970&quot;,&quot;branch_number&quot;:&quot;0003&quot;,&quot;tenpo_cd&quot;:&quot;400451&quot;,&quot;baika&quot;:&quot;1370&quot;}">
</li>
</ul>
</td>
</tr>
<tr class="item">
<td><p><strong class="text-red text-bold mgnL10">2,000円</strong></p></td>
<td>
<a href="/product/detail/220035970?branch_number=0001">
<h2 class="title_product">中古</h2>
</a>
</td>
<td>
<div><a href="/"><img src="/database/images/top-logo.png" class="w100"></a></div>
</td>
<td>
<ul class="shipping_campaing_info">
<li class="padT5">1日〜7日以内に発送します</li>
<li class="all ajax-campaign-placeholder"
data-zaiko_data="{&quot;shinaban&quot;:&quot;220035970&quot;,&quot;branch_number&quot;:&quot;0001&quot;,&quot;tenpo_cd&quot;:&quot;&quot;,&quot;baika&quot;:&quot;2000&quot;}">
</li>
</ul>
</td>
</tr>
<tr class="item">
<td><p><strong class="text-red text-bold mgnL10">2,744円</strong></p></td>
<td>
<a href="/product/detail/220035970?tenpo_cd=410118&branch_number=9001">
<h2 class="title_product">新品新古品(店頭在庫品) ※ショップ情報をご確認下さい</h2>
</a>
</td>
<td>
<div class="space_text_1 mgnB5"><strong><a href="/shop/410118">玉光堂 駿河屋マーケットプレイス店</a></strong></div>
<div class="star_group mgnB5"></div>
<div>5.0(46件)</div>
</td>
<td>
<ul class="shipping_campaing_info">
<li class="padT5">2日〜7日以内に発送します</li>
<li class="all ajax-campaign-placeholder"
data-zaiko_data="{&quot;shinaban&quot;:&quot;220035970&quot;,&quot;branch_number&quot;:&quot;9001&quot;,&quot;tenpo_cd&quot;:&quot;410118&quot;,&quot;baika&quot;:&quot;2744&quot;}">
</li>
</ul>
</td>
</tr>
<tr class="item">
<td><p><strong class="text-red text-bold mgnL10">980円</strong></p></td>
<td>
<a href="/product/detail/602274031?tenpo_cd=400567&branch_number=0003">
<h2 class="title_product">中古</h2>
</a>
</td>
<td>
<div class="space_text_1 mgnB5"><strong><a href="/shop/400567">駿河屋 テスト店</a></strong></div>
<div class="star_group mgnB5"></div>
<div>4.5(10件)</div>
</td>
<td>
<ul class="shipping_campaing_info">
<li class="padT5">1日〜3日以内に発送します</li>
</ul>
</td>
</tr>
</table>
</div>
"""
def test_parse_other_shop_list_extracts_all_rows() -> None:
"""四行分别为:加盟店、官方自营、市场店铺、无 data-zaiko_data 的加盟店,均应被正确解析。"""
tree = HTMLParser(OTHER_SHOP_HTML)
items = SurugayaClient._parse_other_shop_list(tree)
assert len(items) == 4
def test_parse_other_shop_list_franchise_row() -> None:
"""加盟店行:tenpo_cd/店名/评分/价格/发货时效均应正确填充。"""
tree = HTMLParser(OTHER_SHOP_HTML)
item = SurugayaClient._parse_other_shop_list(tree)[0]
assert item.tenpo_cd == "400451"
assert item.is_official is False
assert item.shop_name == "駿河屋 山口大学前店"
assert item.shop_url == "https://www.suruga-ya.jp/shop/400451"
assert item.price == 1370
assert item.condition == "中古箱・ジャケット・ケース不備(中)"
assert item.branch_number == "0003"
assert item.rating_score == "5.0"
assert item.rating_count == 1334
assert item.shipping_note == "1日〜3日以内に発送します"
assert item.detail_url == (
"https://www.suruga-ya.jp/product/detail/220035970?tenpo_cd=400451&branch_number=0003"
)
def test_parse_other_shop_list_official_row_has_no_tenpo_cd() -> None:
"""官方自营行:无 /shop/ 链接、tenpo_cd 为空、is_official 应为 True,且无评分。"""
tree = HTMLParser(OTHER_SHOP_HTML)
item = SurugayaClient._parse_other_shop_list(tree)[1]
assert item.tenpo_cd == ""
assert item.is_official is True
assert item.shop_name == "駿河屋"
assert item.price == 2000
assert item.rating_score == ""
assert item.rating_count == 0
def test_parse_other_shop_list_marketplace_row() -> None:
"""市场店铺行:与加盟店行结构一致,同样应正确解析。"""
tree = HTMLParser(OTHER_SHOP_HTML)
item = SurugayaClient._parse_other_shop_list(tree)[2]
assert item.tenpo_cd == "410118"
assert item.is_official is False
assert item.shop_name == "玉光堂 駿河屋マーケットプレイス店"
assert item.price == 2744
assert item.rating_score == "5.0"
assert item.rating_count == 46
def test_parse_other_shop_list_branch_number_falls_back_to_detail_url() -> None:
"""无 ajax-campaign-placeholder(无 data-zaiko_data)时,branch_number 应从 detail_url 查询参数回退提取。"""
tree = HTMLParser(OTHER_SHOP_HTML)
item = SurugayaClient._parse_other_shop_list(tree)[3]
assert item.branch_number == "0003"
assert item.detail_url == (
"https://www.suruga-ya.jp/product/detail/602274031?tenpo_cd=400567&branch_number=0003"
)
def test_parse_other_shop_list_empty_html_returns_empty_list() -> None:
"""页面结构不包含 tbl_all 时(如解析失败/改版),应返回空列表而非抛异常。"""
tree = HTMLParser("<html><body>no table here</body></html>")
assert SurugayaClient._parse_other_shop_list(tree) == []
+111
View File
@@ -0,0 +1,111 @@
"""SurugayaClient 中「指定店铺」相关能力的单元测试:
- _build_product_url 携带 tenpo_cd/branch_number 构造跳转到指定店铺报价的详情页 URL
- fetch_other_shops 独立于 fetch_detail,仅返回 other_shop_list(解耦双倍抓取成本)
"""
from __future__ import annotations
import pytest
from app.core.config import Settings
from app.models.scrape import DetailRequest, OtherShopItem
from app.services.surugaya_client import SurugayaClient
def _make_client() -> SurugayaClient:
return SurugayaClient(
settings=Settings(_env_file=None),
browser_pool=None,
session_manager=None,
session_store=None,
)
class TestBuildProductUrl:
def test_without_shop_params(self) -> None:
client = _make_client()
assert client._build_product_url("220035970") == "https://www.suruga-ya.jp/product/detail/220035970"
def test_with_tenpo_cd_and_branch_number(self) -> None:
client = _make_client()
url = client._build_product_url("220035970", tenpo_cd="400451", branch_number="0003")
assert url == "https://www.suruga-ya.jp/product/detail/220035970?tenpo_cd=400451&branch_number=0003"
def test_missing_product_id_raises(self) -> None:
client = _make_client()
with pytest.raises(Exception):
client._build_product_url(None)
class TestFetchOtherShops:
@pytest.mark.asyncio
async def test_returns_goods_id_and_shop_list(self, monkeypatch: pytest.MonkeyPatch) -> None:
client = _make_client()
fake_items = [OtherShopItem(tenpo_cd="400451", shop_name="駿河屋 山口大学前店", price=1370)]
async def fake_fetch_other_shop_list(goods_id: str) -> list[OtherShopItem]:
assert goods_id == "220035970"
return fake_items
monkeypatch.setattr(client, "_fetch_other_shop_list", fake_fetch_other_shop_list)
result = await client.fetch_other_shops(DetailRequest(id="220035970"))
assert result.goods_id == "220035970"
assert result.other_shop_list == fake_items
@pytest.mark.asyncio
async def test_empty_id_raises(self) -> None:
client = _make_client()
with pytest.raises(Exception):
await client.fetch_other_shops(DetailRequest(id=None))
class TestFetchDetailIncludeOtherShops:
"""DetailRequest.include_other_shops 控制 fetch_detail 是否顺带抓取 other_shop_list。"""
MINIMAL_DETAIL_HTML = '<html><body><h1 id="item_title">测试商品</h1></body></html>'
@pytest.mark.asyncio
async def test_flag_false_skips_other_shop_fetch(self, monkeypatch: pytest.MonkeyPatch) -> None:
client = _make_client()
monkeypatch.setattr(client, "_session_manager", _FakeSessionManager(self.MINIMAL_DETAIL_HTML))
called = False
async def fake_fetch_other_shop_list(goods_id: str) -> list[OtherShopItem]:
nonlocal called
called = True
return []
monkeypatch.setattr(client, "_fetch_other_shop_list", fake_fetch_other_shop_list)
detail = await client.fetch_detail(DetailRequest(id="220035970"))
assert called is False
assert detail.other_shop_list == []
@pytest.mark.asyncio
async def test_flag_true_includes_other_shop_list(self, monkeypatch: pytest.MonkeyPatch) -> None:
client = _make_client()
monkeypatch.setattr(client, "_session_manager", _FakeSessionManager(self.MINIMAL_DETAIL_HTML))
fake_items = [OtherShopItem(tenpo_cd="400451", shop_name="駿河屋 山口大学前店", price=1370)]
async def fake_fetch_other_shop_list(goods_id: str) -> list[OtherShopItem]:
assert goods_id == "220035970"
return fake_items
monkeypatch.setattr(client, "_fetch_other_shop_list", fake_fetch_other_shop_list)
detail = await client.fetch_detail(DetailRequest(id="220035970", include_other_shops=True))
assert detail.other_shop_list == fake_items
class _FakeSessionManager:
def __init__(self, html: str) -> None:
self._html = html
async def fetch_html(self, url: str):
return self._html, None
Generated
+426 -428
View File
File diff suppressed because it is too large Load Diff