优化浏览器context复用

app/api/routes/scrape.py

@@ -23,7 +23,11 @@ from app.models.scrape import (
     SearchResultData, TradeMonitorRequest,
 )
 from app.utils.http_utils import generate_signed_headers
-from app.utils.parse_util import get_handle_fee, get_shipping_fee
+from app.utils.parse_util import (
+    get_franchise_shipping_fee,
+    get_handle_fee,
+    get_shipping_fee,
+)
 
 router = APIRouter(prefix="/api", tags=["scrape"])
 
@@ -139,8 +143,28 @@ async def categories(
     dependencies=[Depends(require_bearer_token)],
 )
 async def order_get_shipping_fee(payload: OrderShippingFeeRequest) -> ApiResponse[OrderShippingFeeData]:
-    goods_amount = sum(item.goods_number * item.goods_price for item in payload.item_list)
-    shipping_fee = get_shipping_fee(goods_amount)
+    """计算订单运费与手续费。
+
+    规则:
+      - 官方店铺(supplier 为空)运费: 按官方部分商品总额套阶梯价(440/385/0)
+        * 订单中无任何官方商品时,官方运费 0
+      - 加盟店运费(supplier 非空,同名视为同一家): 每家统一 800 日元
+      - 通贩手续费: 按全部商品(含官方+加盟店)总额套阶梯价(240/0)
+    """
+    official_amount = 0
+    franchise_stores: set[str] = set()
+    goods_amount = 0
+    for item in payload.item_list:
+        subtotal = item.goods_number * item.goods_price
+        goods_amount += subtotal
+        if item.supplier == "":
+            official_amount += subtotal
+        else:
+            franchise_stores.add(item.supplier)
+
+    official_shipping_fee = get_shipping_fee(official_amount) if official_amount > 0 else 0
+    franchise_shipping_fee = get_franchise_shipping_fee(len(franchise_stores))
+    shipping_fee = official_shipping_fee + franchise_shipping_fee
     handle_amount = get_handle_fee(goods_amount)
     order_amount = goods_amount + shipping_fee + handle_amount
 

app/main.py

@@ -80,6 +80,7 @@ async def lifespan(app: FastAPI):
     try:
         yield
     finally:
+        await container.session_manager.close()
         await container.browser_pool.close()
         await container.session_store.close()
 

app/models/scrape.py

@@ -132,6 +132,8 @@ class OrderItem(BaseModel):
     goods_id: str = Field(min_length=1, max_length=40)
     goods_number: int = Field(ge=1, le=9999)
     goods_price: int = Field(ge=0)
+    # 店铺标识: 空串 = 官方店铺;非空 = 加盟店标识(同名视为同一家加盟店)
+    supplier: str = ""
 
 
 class OrderShippingFeeRequest(BaseModel):

app/services/browser_pool.py

@@ -250,6 +250,107 @@ class BrowserPool:
             except Exception:
                 logger.exception("关闭旧的保留调试上下文失败")
 
+    async def _build_context_obj(self, storage_state: dict[str, Any] | None) -> Any:
+        """创建浏览器上下文（含断链重试与 stealth），仅负责创建本身。
+
+        计入回收阈值统计；不负责并发信号量与关闭，由调用方管理生命周期。
+        """
+        context = None
+        for attempt in range(2):
+            try:
+                context = await self._browser.new_context(
+                    locale=self._settings.browser_locale,
+                    timezone_id=self._settings.browser_timezone_id,
+                    user_agent=self._settings.browser_user_agent,
+                    viewport=None if not self._settings.browser_headless_effective else {"width": 1440, "height": 900},
+                    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
+
+        if self._settings.browser_stealth_enabled:
+            try:
+                from playwright_stealth import Stealth
+
+                await Stealth().apply_stealth_async(context)
+            except Exception as e:
+                logger.warning("应用 playwright-stealth 失败，隐身模式未生效, err=%s", e)
+
+        return context
+
+    @staticmethod
+    async def _configure_page(context: Any) -> Any:
+        """在上下文中创建页面并设置反检测请求头。"""
+        page = await context.new_page()
+        await page.set_extra_http_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
+
+    async def create_persistent_context(self, storage_state: dict[str, Any] | None = None) -> Any:
+        """创建一个由调用方负责生命周期的长存上下文。
+
+        用于 cf_clearance 保活复用：通过挑战后保留该上下文，后续请求在其内
+        开新页面访问目标页（同一上下文内 cf_clearance 有效）。调用方需在不再
+        使用时调用 close_context 释放。
+        """
+        await self._ensure_browser_ready()
+        return await self._build_context_obj(storage_state)
+
+    @asynccontextmanager
+    async def page_in_context(self, context: Any) -> AsyncIterator[Any]:
+        """在既有（长存）上下文中开一个页面，复用并发信号量。
+
+        结束时仅关闭页面，不关闭上下文（上下文生命周期由调用方维护）。
+
+        Raises:
+            ResourceBusyError: 等待可用槽位超时
+        """
+        await self._ensure_browser_ready()
+        try:
+            await asyncio.wait_for(
+                self._semaphore.acquire(),
+                timeout=self._settings.page_acquire_timeout_seconds,
+            )
+        except TimeoutError as exc:
+            raise ResourceBusyError() from exc
+
+        page = None
+        counted = False
+        try:
+            self._active_contexts += 1
+            counted = True
+            page = await self._configure_page(context)
+            yield page
+        finally:
+            if page is not None:
+                try:
+                    await page.close()
+                except Exception:
+                    logger.debug("关闭复用上下文页面失败", exc_info=True)
+            if counted:
+                self._active_contexts -= 1
+            self._semaphore.release()
+
+    async def close_context(self, context: Any) -> None:
+        """关闭一个长存上下文（忽略已关闭等异常）。"""
+        try:
+            await context.close()
+        except Exception:
+            logger.debug("关闭长存上下文失败", exc_info=True)
+
     @asynccontextmanager
     async def page_session(
         self,
@@ -295,45 +396,12 @@ class BrowserPool:
         try:
             # 创建上下文前检查是否需要主动回收（仅在空闲时执行，不打断进行中的会话）
             await self._maybe_recycle()
-            for attempt in range(2):
-                try:
-                    context = await self._browser.new_context(
-                        locale=self._settings.browser_locale,
-                        timezone_id=self._settings.browser_timezone_id,
-                        user_agent=self._settings.browser_user_agent,
-                        viewport=None if not self._settings.browser_headless_effective else {"width": 1440, "height": 900},
-                        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
+            context = await self._build_context_obj(storage_state)
+            # 上下文创建成功，标记为在用（回收阈值计数已在 _build_context_obj 内完成）
             self._active_contexts += 1
             context_counted = True
 
-            if self._settings.browser_stealth_enabled:
-                try:
-                    from playwright_stealth import Stealth
-
-                    await Stealth().apply_stealth_async(context)
-                    # logger.debug("已为浏览器上下文应用 Stealth 模式")
-                except Exception as e:
-                    logger.warning("应用 playwright-stealth 失败，隐身模式未生效, err=%s", e)
-
-            page = await context.new_page()
-            await page.set_extra_http_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",
-                }
-            )
+            page = await self._configure_page(context)
             yield context, page
         except Exception:
             if self._settings.browser_keep_open_on_error_effective and context is not None:

app/services/cloudflare_session.py

@@ -14,6 +14,7 @@ import hashlib
 import logging
 from collections.abc import Iterable
 from dataclasses import dataclass
+from typing import Any
 from urllib.parse import urlparse
 from uuid import uuid4
 from pathlib import Path
@@ -35,6 +36,24 @@ class VerifiedSession:
     html: str | None = None  # 新创建会话时携带的页面 HTML，复用场景为 None
 
 
+@dataclass(slots=True)
+class WarmContext:
+    """保活的浏览器上下文：通过 Cloudflare 挑战后长期驻留，供后续请求复用
+
+    cf_clearance 无法跨上下文重放，因此保留通过挑战的上下文本身，
+    后续请求在同一上下文内开新页面访问目标页（上下文内 clearance 有效）。
+    带引用计数以保证并发使用时安全退役。
+    """
+    session_id: str
+    context: Any  # Playwright BrowserContext
+    cf_clearance: str | None
+    created_at: float
+    last_used_at: float
+    in_use: int = 0  # 当前正在该上下文内使用页面的请求数
+    retired: bool = False  # 已从注册表摘除、等待空闲后关闭
+    close_done: bool = False  # 关闭只执行一次的标记
+
+
 class CloudflareSessionManager:
     """Cloudflare 会话管理器，负责会话的获取、验证和失效"""
 
@@ -48,6 +67,10 @@ class CloudflareSessionManager:
         self._browser_pool = browser_pool
         self._session_store = session_store
 
+        # 保活上下文注册表：session_name -> WarmContext；构建锁防止并发重建风暴
+        self._warm_contexts: dict[str, WarmContext] = {}
+        self._build_locks: dict[str, asyncio.Lock] = {}
+
     async def get_verified_session(self, target_url: str, force_refresh: bool = False) -> VerifiedSession:
         """获取一个已通过 Cloudflare 验证的会话
 
@@ -77,84 +100,171 @@ class CloudflareSessionManager:
         return session
 
     async def fetch_html(self, target_url: str) -> tuple[str, VerifiedSession]:
-        """获取目标页面 HTML，并屏蔽会话复用失败的细节
+        """获取目标页面 HTML，基于「保活上下文」复用通过验证的浏览器会话
 
-        统一处理三种情形，确保调用方拿到的要么是有效页面，要么是明确的上游异常：
-        1. 新建会话：直接复用创建时携带的页面 HTML，无需二次导航
-        2. 复用会话：带 storage_state 重新导航，并消化可能出现的 Cloudflare 挑战
-        3. 复用失败：主动失效后强制重建一次，对调用方透明（避免「一次成功一次失败」交替）
+        cf_clearance 无法跨上下文重放，因此保留通过挑战的上下文本身：
+        1. 已有保活上下文且未过期：在其内开新页面访问目标页（同上下文 clearance 有效）
+        2. 复用失败或无保活上下文：在构建锁内创建新的保活上下文（顺带完成本次访问）
 
         Args:
             target_url: 目标页面 URL
 
         Returns:
-            (html, session) 元组，session 为最终生效的会话
+            (html, session) 元组，session 携带当前生效会话的 session_id 等信息
 
         Raises:
-            UpstreamBlockedError: 会话重建后仍无法获取有效页面
+            UpstreamBlockedError: 重建后仍无法获取有效页面
             CloudflareChallengeError: 等待 Cloudflare 挑战超时
+            ResourceBusyError: 等待页面槽位超时
         """
-        session = await self.get_verified_session(target_url)
-        if session.html is not None:
-            return session.html, session
+        session_name = self._session_name_for_url(target_url)
 
-        html = await self._try_fetch_with_session(target_url, session)
-        if html is not None:
-            return html, session
+        # 1. 优先复用已有保活上下文
+        warm = self._warm_contexts.get(session_name)
+        if warm is not None and not self._warm_expired(warm):
+            html = await self._fetch_via_warm(warm, target_url)
+            if html is not None:
+                return html, self._warm_to_session(warm)
+            await self._retire_warm(session_name, warm)
 
-        # 复用会话已失效：失效并强制重建，新会话自带页面 HTML
-        logger.info("复用会话失败，强制重建会话：url=%s", target_url)
-        await self.invalidate(target_url)
-        session = await self.get_verified_session(target_url, force_refresh=True)
-        if session.html is not None:
-            return session.html, session
+        # 2. 在构建锁内创建/复用保活上下文，避免并发重建风暴
+        lock = self._build_locks.setdefault(session_name, asyncio.Lock())
+        async with lock:
+            # 双重检查：可能已被其他协程在等待锁期间建好
+            warm = self._warm_contexts.get(session_name)
+            if warm is not None and not self._warm_expired(warm):
+                html = await self._fetch_via_warm(warm, target_url)
+                if html is not None:
+                    return html, self._warm_to_session(warm)
+                await self._retire_warm(session_name, warm)
 
-        # 兜底：force_refresh 理论上必带 HTML，缺失时再导航一次
-        html = await self._try_fetch_with_session(target_url, session)
-        if html is None:
-            raise UpstreamBlockedError(f"会话重建后仍无法获取页面：url={target_url}")
-        return html, session
+            logger.info("创建保活上下文：session_name=%s", session_name)
+            warm, html = await self._build_warm_context(session_name, target_url)
+            return html, self._warm_to_session(warm)
 
-    async def _try_fetch_with_session(self, target_url: str, session: VerifiedSession) -> str | None:
-        """带已有会话导航并消化 Cloudflare 挑战
+    def _warm_expired(self, warm: WarmContext) -> bool:
+        """保活上下文是否已退役或超过最大存活时长（沿用 session_ttl）。"""
+        if warm.retired:
+            return True
+        age = asyncio.get_running_loop().time() - warm.created_at
+        return age >= self._settings.session_ttl_seconds
 
-        与新建路径保持一致：导航后等待挑战自动通过，而非一遇非 200 就判死。
-        CF 对携带过期 cf_clearance 的请求常直接返回挑战页（首响应可能非 200），
-        因此即便首响应非 200，也先给页面一次自动通过 Turnstile 的机会，
-        仅当挑战超时/被硬阻断、或通过后仍无有效内容时才判定会话失效。
+    async def _fetch_via_warm(self, warm: WarmContext, target_url: str) -> str | None:
+        """在保活上下文内开新页面访问目标页。
 
         Returns:
-            通过验证的页面 HTML；若判定会话已失效，返回 None
+            通过验证的页面 HTML；若判定上下文已失效（非 200 / 挑战未过 / 上下文异常），返回 None
         """
-        async with self._browser_pool.page_session(storage_state=session.storage_state) as (_, page):
-            response = await page.goto(
-                target_url,
-                wait_until="domcontentloaded",
-                timeout=int(self._settings.request_timeout_seconds * 1000),
-            )
-            status = response.status if response is not None else None
-            try:
-                # 复用使用短超时：快速验证能否通过，不行就放弃复用交由上层重建，
-                # 避免在无法自动通过的挑战上空等整个 cloudflare_wait_timeout_seconds。
-                await self._wait_for_clearance(
-                    page, timeout_seconds=self._settings.cloudflare_reuse_wait_timeout_seconds
+        warm.in_use += 1
+        try:
+            async with self._browser_pool.page_in_context(warm.context) as page:
+                response = await page.goto(
+                    target_url,
+                    wait_until="domcontentloaded",
+                    timeout=int(self._settings.request_timeout_seconds * 1000),
                 )
-            except (CloudflareChallengeError, UpstreamBlockedError):
-                logger.info(
-                    "复用会话未通过挑战，判定失效：url=%s status=%s had_cf_clearance=%s",
-                    target_url, status, bool(session.cf_clearance),
-                )
-                return None
+                status = response.status if response is not None else None
+                if status != 200:
+                    logger.info("保活上下文复用导航非 200，判定失效：url=%s status=%s", target_url, status)
+                    return None
 
-            html = await page.content()
-            # 挑战已通过但页面仍是挑战壳（极少数 wait 提前返回的情况）：判失效，触发重建
-            if self._looks_like_challenge((await page.title()).lower(), html.lower()):
-                logger.info(
-                    "复用会话通过等待后仍无有效内容，判定失效：url=%s status=%s had_cf_clearance=%s",
-                    target_url, status, bool(session.cf_clearance),
+                # 首响应 200：极少数情况下仍是软挑战，用短超时给一次自动通过的机会
+                try:
+                    await self._wait_for_clearance(
+                        page, timeout_seconds=self._settings.cloudflare_reuse_wait_timeout_seconds
+                    )
+                except (CloudflareChallengeError, UpstreamBlockedError):
+                    logger.info("保活上下文复用软挑战未通过，判定失效：url=%s", target_url)
+                    return None
+
+                warm.last_used_at = asyncio.get_running_loop().time()
+                return await page.content()
+        except Exception as exc:
+            # 上下文/浏览器已被关闭（如浏览器回收）或导航异常：视为失效，触发重建
+            logger.info("保活上下文复用异常，判定失效：url=%s err=%s", target_url, exc)
+            return None
+        finally:
+            warm.in_use -= 1
+            if warm.retired and warm.in_use <= 0:
+                await self._close_warm_context(warm)
+
+    async def _build_warm_context(self, session_name: str, target_url: str) -> tuple[WarmContext, str]:
+        """创建一个通过 Cloudflare 挑战的保活上下文，并返回本次访问的页面 HTML。"""
+        context = await self._browser_pool.create_persistent_context()
+        try:
+            async with self._browser_pool.page_in_context(context) as page:
+                response = await page.goto(
+                    target_url,
+                    wait_until="domcontentloaded",
+                    timeout=int(self._settings.request_timeout_seconds * 1000),
                 )
-                return None
-            return html
+                status = response.status if response is not None else None
+                if status != 200:
+                    screenshot_path = await self._capture_screenshot(page)
+                    raise UpstreamBlockedError(
+                        f"Upstream status={status}, 页面返回非200状态码, 截图地址：{screenshot_path}"
+                    )
+                await self._wait_for_clearance(page)
+                html_content = await page.content()
+                cookies = await context.cookies()
+                cf_clearance = self._get_cookie_value(cookies, "cf_clearance")
+        except Exception:
+            await self._browser_pool.close_context(context)
+            raise
+
+        now = asyncio.get_running_loop().time()
+        warm = WarmContext(
+            session_id=str(uuid4()),
+            context=context,
+            cf_clearance=cf_clearance,
+            created_at=now,
+            last_used_at=now,
+        )
+        self._warm_contexts[session_name] = warm
+        logger.info("保活上下文创建完成：session_name=%s cf_clearance=%s", session_name, bool(cf_clearance))
+        return warm, html_content
+
+    async def _retire_warm(self, session_name: str, warm: WarmContext) -> None:
+        """将保活上下文摘除注册表并退役（无在用页面时立即关闭，否则等空闲后关闭）。"""
+        if self._warm_contexts.get(session_name) is warm:
+            del self._warm_contexts[session_name]
+        warm.retired = True
+        if warm.in_use <= 0:
+            await self._close_warm_context(warm)
+
+    async def _close_warm_context(self, warm: WarmContext) -> None:
+        """关闭保活上下文（保证只关闭一次）。"""
+        if warm.close_done:
+            return
+        warm.close_done = True
+        await self._browser_pool.close_context(warm.context)
+
+    @staticmethod
+    def _warm_to_session(warm: WarmContext) -> VerifiedSession:
+        """将保活上下文映射为对外的 VerifiedSession（storage_state 不再对外暴露）。"""
+        return VerifiedSession(
+            session_id=warm.session_id,
+            storage_state={},
+            cf_clearance=warm.cf_clearance,
+            html=None,
+        )
+
+    async def close(self) -> None:
+        """关闭全部保活上下文，应在浏览器池关闭前调用。"""
+        for warm in list(self._warm_contexts.values()):
+            await self._close_warm_context(warm)
+        self._warm_contexts.clear()
+
+    async def _capture_screenshot(self, page: object) -> str | None:
+        """对当前页面截图用于排查，失败时返回 None。"""
+        try:
+            screenshot_dir = Path(__file__).resolve().parents[2] / self._settings.log_dir / "screenshots"
+            screenshot_dir.mkdir(parents=True, exist_ok=True)
+            screenshot_file = screenshot_dir / f"{uuid4().hex}.png"
+            await page.screenshot(path=str(screenshot_file), full_page=True)
+            return str(screenshot_file)
+        except Exception:
+            return None
 
     async def invalidate(self, target_url: str) -> None:
         """主动让会话失效
@@ -167,6 +277,9 @@ class CloudflareSessionManager:
         session_name = self._session_name_for_url(target_url)
         logger.warning("会话失效：session_name=%s", session_name)
         await self._session_store.invalidate_session(session_name)
+        warm = self._warm_contexts.get(session_name)
+        if warm is not None:
+            await self._retire_warm(session_name, warm)
 
     async def _create_verified_session(self, session_name: str, target_url: str) -> VerifiedSession:
         """通过真实浏览器访问目标页面，让 Cloudflare 挑战在浏览器侧完成

app/utils/parse_util.py

@@ -122,42 +122,64 @@ def extract_product_id(url: str | None = None) -> str:
     return ""
    
 
-def get_shipping_fee(amount: int) -> int:
-    """
-    根据定义的运费规则计算运费
+# --- 官方店铺日本境内运费规则 ---
+# amount < OFFICIAL_SHIPPING_FEE_LOW_THRESHOLD: OFFICIAL_SHIPPING_FEE_LOW
+# OFFICIAL_SHIPPING_FEE_LOW_THRESHOLD <= amount < OFFICIAL_SHIPPING_FEE_HIGH_THRESHOLD: OFFICIAL_SHIPPING_FEE_MID
+# amount >= OFFICIAL_SHIPPING_FEE_HIGH_THRESHOLD: 免运费(0)
+OFFICIAL_SHIPPING_FEE_LOW_THRESHOLD = 1000
+OFFICIAL_SHIPPING_FEE_HIGH_THRESHOLD = 1500
+OFFICIAL_SHIPPING_FEE_LOW = 440
+OFFICIAL_SHIPPING_FEE_MID = 385
 
-    日本境内运费统一为固定价格（部分地区除外）。
-    • 订单金额低于1000日元：运费440日元
-    • 订单金额低于1500日元：运费385日元
-    • 订单金额1500日元及以上：免运费
-    *北海道、冲绳、离岛及其他部分地区可能需额外支付运费。
-    货到付款需另付550日元货到付款手续费。银行转账付款需由顾客承担转账手续费。信用卡付款无需支付手续费。
-    *通过电商平台购买的商品，运费因店铺而异。
+# --- 加盟店日本境内运费规则 ---
+# 每 1 家加盟店,无论订单数量与金额,统一收 FRANCHISE_SHIPPING_FEE_PER_STORE 日元
+FRANCHISE_SHIPPING_FEE_PER_STORE = 800
+
+# --- 通贩手续费规则 ---
+# amount < HANDLE_FEE_FREE_THRESHOLD: HANDLE_FEE_BELOW
+# amount >= HANDLE_FEE_FREE_THRESHOLD: 免手续费(0)
+HANDLE_FEE_FREE_THRESHOLD = 5000
+HANDLE_FEE_BELOW = 240
+
+
+def get_shipping_fee(amount: int) -> int:
+    """计算官方店铺日本境内运费(本州/四国/九州统一价)。
+
+    规则:
+      - amount < 1000 日元: 440 日元
+      - 1000 <= amount < 1500 日元: 385 日元
+      - amount >= 1500 日元: 0(免运费)
+
+    备注: 北海道/冲绳/离岛可能需额外支付;此处按本州口径计算。
+          加盟店运费另由 get_franchise_shipping_fee 计算。
     """
-    if amount > 1500:
+    if amount >= OFFICIAL_SHIPPING_FEE_HIGH_THRESHOLD:
         return 0
-    elif amount > 1000:
-        return 440
-    else:
-        return 385
+    if amount >= OFFICIAL_SHIPPING_FEE_LOW_THRESHOLD:
+        return OFFICIAL_SHIPPING_FEE_MID
+    return OFFICIAL_SHIPPING_FEE_LOW
+
+
+def get_franchise_shipping_fee(franchise_store_count: int) -> int:
+    """计算加盟店日本境内运费。
+
+    每 1 家加盟店统一收 800 日元,与订单数量/金额无关。
+    franchise_store_count <= 0 时返回 0。
+    """
+    if franchise_store_count <= 0:
+        return 0
+    return franchise_store_count * FRANCHISE_SHIPPING_FEE_PER_STORE
+
 
 def get_handle_fee(amount: int) -> int:
-    """
-    根据定义的规则计算手续费
-    https://www.suruga-ya.jp/man/qa/hanbai_qa/shiharai.html#shiharai_1_2
-    A. 运费根据购买金额而定。
-    ■本州、四国、九州：
-    购买金额低于5,000日元…240日元；
-    购买金额5,000日元及以上…免运费。
-    ■北海道、冲绳：
-    购买金额低于5,000日元…570日元；
-    购买金额5,000日元至低于10,000日元…285日元；
-    购买金额10,000日元及以上…免
+    """计算通贩手续费(本州/四国/九州口径)。
 
-    运费。偏远岛屿及其他部分地区可能需额外收费。
-    这里按照本州计算手续费。
+    规则:
+      - amount < 5000 日元: 240 日元
+      - amount >= 5000 日元: 0(免手续费)
+
+    参考: https://www.suruga-ya.jp/man/qa/hanbai_qa/shiharai.html#shiharai_1_2
     """
-    if amount > 5000:
+    if amount >= HANDLE_FEE_FREE_THRESHOLD:
         return 0
-    else:
-        return 240
+    return HANDLE_FEE_BELOW

tests/test_parse_util.py

@@ -0,0 +1,154 @@
+"""parse_util 运费/手续费计算函数以及订单运费路由的单元测试。"""
+from __future__ import annotations
+
+import pytest
+
+from app.api.routes.scrape import order_get_shipping_fee
+from app.models.scrape import OrderItem, OrderShippingFeeRequest
+from app.utils.parse_util import (
+    FRANCHISE_SHIPPING_FEE_PER_STORE,
+    HANDLE_FEE_BELOW,
+    OFFICIAL_SHIPPING_FEE_LOW,
+    OFFICIAL_SHIPPING_FEE_MID,
+    get_franchise_shipping_fee,
+    get_handle_fee,
+    get_shipping_fee,
+)
+
+
+# --- 官方店铺运费规则: 边界与典型值 ---
+@pytest.mark.parametrize(
+    "amount, expected",
+    [
+        (0, OFFICIAL_SHIPPING_FEE_LOW),       # < 1000
+        (999, OFFICIAL_SHIPPING_FEE_LOW),     # < 1000
+        (1000, OFFICIAL_SHIPPING_FEE_MID),    # = 1000 -> 385
+        (1200, OFFICIAL_SHIPPING_FEE_MID),    # 1000 <= < 1500
+        (1499, OFFICIAL_SHIPPING_FEE_MID),    # < 1500
+        (1500, 0),                            # = 1500 -> 免运费
+        (2000, 0),                            # > 1500
+    ],
+)
+def test_get_shipping_fee_boundaries(amount: int, expected: int) -> None:
+    assert get_shipping_fee(amount) == expected
+
+
+# --- 加盟店运费规则: 按家数 ---
+@pytest.mark.parametrize(
+    "count, expected",
+    [
+        (-1, 0),
+        (0, 0),
+        (1, FRANCHISE_SHIPPING_FEE_PER_STORE),
+        (3, 3 * FRANCHISE_SHIPPING_FEE_PER_STORE),
+    ],
+)
+def test_get_franchise_shipping_fee(count: int, expected: int) -> None:
+    assert get_franchise_shipping_fee(count) == expected
+
+
+# --- 通贩手续费规则: 边界与典型值 ---
+@pytest.mark.parametrize(
+    "amount, expected",
+    [
+        (0, HANDLE_FEE_BELOW),
+        (4999, HANDLE_FEE_BELOW),
+        (5000, 0),       # = 5000 -> 免手续费
+        (10000, 0),
+    ],
+)
+def test_get_handle_fee_boundaries(amount: int, expected: int) -> None:
+    assert get_handle_fee(amount) == expected
+
+
+# ---- 路由 order_get_shipping_fee 端到端 ----
+# 直接调用 async 路由函数,绕开 FastAPI 应用初始化(浏览器/redis 等)。
+def _payload(*items: tuple[int, int, str]) -> OrderShippingFeeRequest:
+    """items: (goods_number, goods_price, supplier) 三元组列表"""
+    return OrderShippingFeeRequest(
+        item_list=[
+            OrderItem(
+                goods_id=f"g{idx}",
+                goods_number=n,
+                goods_price=p,
+                supplier=s,
+            )
+            for idx, (n, p, s) in enumerate(items)
+        ]
+    )
+
+
+async def test_route_all_official_below_threshold() -> None:
+    # 1 件 999 元官方商品 -> 官方运费 440, 手续费 240
+    resp = await order_get_shipping_fee(_payload((1, 999, "")))
+    data = resp.data
+    assert data.goods_amount == 999
+    assert data.shipping_fee == 440
+    assert data.handle_amount == 240
+    assert data.order_amount == 999 + 440 + 240
+
+
+async def test_route_all_official_at_free_shipping_boundary() -> None:
+    # 1 件 1500 元官方商品 -> 官方运费 0(>=1500), 手续费 240(<5000)
+    resp = await order_get_shipping_fee(_payload((1, 1500, "")))
+    data = resp.data
+    assert data.goods_amount == 1500
+    assert data.shipping_fee == 0
+    assert data.handle_amount == 240
+    assert data.order_amount == 1500 + 240
+
+
+async def test_route_only_franchise_no_official_shipping() -> None:
+    # 全部来自 1 家加盟店 -> 官方运费 0(无官方商品), 加盟运费 800
+    resp = await order_get_shipping_fee(_payload((2, 400, "ShopA")))
+    data = resp.data
+    assert data.goods_amount == 800
+    assert data.shipping_fee == 800
+    assert data.handle_amount == 240
+    assert data.order_amount == 800 + 800 + 240
+
+
+async def test_route_two_franchise_stores_dedup_by_supplier() -> None:
+    # 2 家加盟店各 1 件 + 官方满 1500 -> 官方 0 + 加盟 1600
+    resp = await order_get_shipping_fee(
+        _payload(
+            (1, 1500, ""),
+            (1, 500, "ShopA"),
+            (1, 500, "ShopA"),  # 同店,不重复计费
+            (1, 500, "ShopB"),
+        )
+    )
+    data = resp.data
+    assert data.goods_amount == 3000
+    assert data.shipping_fee == 0 + 2 * 800  # 官方 0 + 2 家加盟
+    assert data.handle_amount == 240  # 3000 < 5000
+    assert data.order_amount == 3000 + 1600 + 240
+
+
+async def test_route_total_amount_over_5000_handle_fee_free() -> None:
+    # 全部商品总额(官方+加盟)>= 5000 时手续费免
+    resp = await order_get_shipping_fee(
+        _payload(
+            (1, 3000, ""),       # 官方 3000
+            (1, 2000, "ShopA"),  # 加盟 2000
+        )
+    )
+    data = resp.data
+    assert data.goods_amount == 5000
+    # 官方 3000 -> 1500<=3000 -> 0 运费; 加盟 1 家 -> 800
+    assert data.shipping_fee == 0 + 800
+    assert data.handle_amount == 0  # 总额 5000 -> 免
+    assert data.order_amount == 5000 + 800
+
+
+async def test_route_backward_compatible_default_supplier() -> None:
+    # 旧调用方不传 supplier 字段 -> 默认空串 -> 全部按官方
+    payload = OrderShippingFeeRequest.model_validate(
+        {"item_list": [{"goods_id": "x", "goods_number": 1, "goods_price": 800}]}
+    )
+    resp = await order_get_shipping_fee(payload)
+    data = resp.data
+    # 800 < 1000 -> 官方运费 440, 手续费 240
+    assert data.shipping_fee == 440
+    assert data.handle_amount == 240
+    assert data.order_amount == 800 + 440 + 240

tests/test_warm_context.py

@@ -0,0 +1,124 @@
+"""保活上下文（cf_clearance 复用）逻辑的单元测试（不启动真实 Chromium）。"""
+from __future__ import annotations
+
+from contextlib import asynccontextmanager
+from types import SimpleNamespace
+
+import pytest
+
+from app.core.config import Settings
+from app.services.cloudflare_session import CloudflareSessionManager
+from app.services.session_store import SessionStore
+
+_DETAIL_URL = "https://www.suruga-ya.jp/product/detail/{}"
+
+
+class _FakePage:
+    """最小页面替身：goto 状态由所属上下文是否被标记失败决定。"""
+
+    def __init__(self, pool: "_FakeBrowserPool", context: "_FakeContext"):
+        self._pool = pool
+        self._context = context
+
+    async def goto(self, url: str, **_kw):
+        status = 403 if self._context.idx in self._pool.fail_idxs else 200
+        return SimpleNamespace(status=status)
+
+    async def title(self) -> str:
+        return "ok"
+
+    async def content(self) -> str:
+        return self._pool.content_html
+
+    async def screenshot(self, **_kw):  # pragma: no cover - 非 200 分支才用到
+        return None
+
+    async def wait_for_timeout(self, _ms):  # pragma: no cover
+        return None
+
+
+class _FakeContext:
+    def __init__(self, idx: int):
+        self.idx = idx
+        self.closed = False
+
+    async def cookies(self):
+        return [{"name": "cf_clearance", "value": f"cv-{self.idx}"}]
+
+    async def close(self):
+        self.closed = True
+
+
+class _FakeBrowserPool:
+    """记录上下文创建次数，并允许把指定上下文标记为「复用必失败」。"""
+
+    def __init__(self):
+        self.content_html = "<html>ok</html>"
+        self.create_calls = 0
+        self.contexts: list[_FakeContext] = []
+        self.fail_idxs: set[int] = set()
+
+    async def create_persistent_context(self, storage_state=None) -> _FakeContext:
+        self.create_calls += 1
+        ctx = _FakeContext(self.create_calls)
+        self.contexts.append(ctx)
+        return ctx
+
+    @asynccontextmanager
+    async def page_in_context(self, context: _FakeContext):
+        yield _FakePage(self, context)
+
+    async def close_context(self, context: _FakeContext):
+        await context.close()
+
+
+def _make_manager(pool: _FakeBrowserPool) -> CloudflareSessionManager:
+    settings = Settings(session_ttl_seconds=900, cloudflare_reuse_wait_timeout_seconds=8.0)
+    return CloudflareSessionManager(settings, pool, SessionStore(settings))
+
+
+@pytest.mark.asyncio
+async def test_warm_context_is_reused_without_rebuild():
+    pool = _FakeBrowserPool()
+    mgr = _make_manager(pool)
+
+    html1, sess1 = await mgr.fetch_html(_DETAIL_URL.format(1))
+    assert html1 == pool.content_html
+    assert pool.create_calls == 1  # 首次构建
+
+    # 第二次请求（不同商品）应复用同一保活上下文，不再新建
+    html2, sess2 = await mgr.fetch_html(_DETAIL_URL.format(2))
+    assert html2 == pool.content_html
+    assert pool.create_calls == 1
+    assert sess2.session_id == sess1.session_id
+
+
+@pytest.mark.asyncio
+async def test_warm_context_rebuilds_and_closes_on_failure():
+    pool = _FakeBrowserPool()
+    mgr = _make_manager(pool)
+
+    _, sess1 = await mgr.fetch_html(_DETAIL_URL.format(1))
+    assert pool.create_calls == 1
+
+    # 让首个上下文在后续复用导航时返回 403（模拟 clearance 失效）
+    pool.fail_idxs.add(1)
+
+    html2, sess2 = await mgr.fetch_html(_DETAIL_URL.format(2))
+    assert html2 == pool.content_html
+    assert pool.create_calls == 2                 # 复用失败后重建
+    assert pool.contexts[0].closed is True        # 旧上下文已退役关闭
+    assert sess2.session_id != sess1.session_id   # 换了新会话
+
+
+@pytest.mark.asyncio
+async def test_close_releases_all_warm_contexts():
+    pool = _FakeBrowserPool()
+    mgr = _make_manager(pool)
+
+    await mgr.fetch_html(_DETAIL_URL.format(1))
+    assert pool.contexts[0].closed is False
+
+    await mgr.close()
+    assert pool.contexts[0].closed is True
+    assert mgr._warm_contexts == {}