Compare commits

..

2 Commits

Author SHA1 Message Date
q792602257 3b8bdf09f7 优化浏览器context复用 2026-06-09 16:04:44 +08:00
q792602257 bc0f7161c0 运费逻辑 2026-06-09 14:55:00 +08:00
8 changed files with 635 additions and 127 deletions
+27 -3
View File
@@ -23,7 +23,11 @@ from app.models.scrape import (
SearchResultData, TradeMonitorRequest, SearchResultData, TradeMonitorRequest,
) )
from app.utils.http_utils import generate_signed_headers 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"]) router = APIRouter(prefix="/api", tags=["scrape"])
@@ -139,8 +143,28 @@ async def categories(
dependencies=[Depends(require_bearer_token)], dependencies=[Depends(require_bearer_token)],
) )
async def order_get_shipping_fee(payload: OrderShippingFeeRequest) -> ApiResponse[OrderShippingFeeData]: 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) handle_amount = get_handle_fee(goods_amount)
order_amount = goods_amount + shipping_fee + handle_amount order_amount = goods_amount + shipping_fee + handle_amount
+1
View File
@@ -80,6 +80,7 @@ async def lifespan(app: FastAPI):
try: try:
yield yield
finally: finally:
await container.session_manager.close()
await container.browser_pool.close() await container.browser_pool.close()
await container.session_store.close() await container.session_store.close()
+2
View File
@@ -132,6 +132,8 @@ class OrderItem(BaseModel):
goods_id: str = Field(min_length=1, max_length=40) goods_id: str = Field(min_length=1, max_length=40)
goods_number: int = Field(ge=1, le=9999) goods_number: int = Field(ge=1, le=9999)
goods_price: int = Field(ge=0) goods_price: int = Field(ge=0)
# 店铺标识: 空串 = 官方店铺;非空 = 加盟店标识(同名视为同一家加盟店)
supplier: str = ""
class OrderShippingFeeRequest(BaseModel): class OrderShippingFeeRequest(BaseModel):
+104 -36
View File
@@ -250,6 +250,107 @@ class BrowserPool:
except Exception: except Exception:
logger.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 @asynccontextmanager
async def page_session( async def page_session(
self, self,
@@ -295,45 +396,12 @@ class BrowserPool:
try: try:
# 创建上下文前检查是否需要主动回收(仅在空闲时执行,不打断进行中的会话) # 创建上下文前检查是否需要主动回收(仅在空闲时执行,不打断进行中的会话)
await self._maybe_recycle() await self._maybe_recycle()
for attempt in range(2): context = await self._build_context_obj(storage_state)
try: # 上下文创建成功,标记为在用(回收阈值计数已在 _build_context_obj 内完成)
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
self._active_contexts += 1 self._active_contexts += 1
context_counted = True context_counted = True
if self._settings.browser_stealth_enabled: page = await self._configure_page(context)
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",
}
)
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:
+158 -45
View File
@@ -14,6 +14,7 @@ import hashlib
import logging import logging
from collections.abc import Iterable from collections.abc import Iterable
from dataclasses import dataclass from dataclasses import dataclass
from typing import Any
from urllib.parse import urlparse from urllib.parse import urlparse
from uuid import uuid4 from uuid import uuid4
from pathlib import Path from pathlib import Path
@@ -35,6 +36,24 @@ class VerifiedSession:
html: str | None = None # 新创建会话时携带的页面 HTML,复用场景为 None 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: class CloudflareSessionManager:
"""Cloudflare 会话管理器,负责会话的获取、验证和失效""" """Cloudflare 会话管理器,负责会话的获取、验证和失效"""
@@ -48,6 +67,10 @@ class CloudflareSessionManager:
self._browser_pool = browser_pool self._browser_pool = browser_pool
self._session_store = session_store 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: async def get_verified_session(self, target_url: str, force_refresh: bool = False) -> VerifiedSession:
"""获取一个已通过 Cloudflare 验证的会话 """获取一个已通过 Cloudflare 验证的会话
@@ -77,84 +100,171 @@ class CloudflareSessionManager:
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]:
"""获取目标页面 HTML,并屏蔽会话复用失败的细节 """获取目标页面 HTML,基于「保活上下文」复用通过验证的浏览器会话
统一处理三种情形,确保调用方拿到的要么是有效页面,要么是明确的上游异常 cf_clearance 无法跨上下文重放,因此保留通过挑战的上下文本身
1. 新建会话:直接复用创建时携带的页面 HTML,无需二次导航 1. 已有保活上下文且未过期:在其内开新页面访问目标页(同上下文 clearance 有效)
2. 复用会话:带 storage_state 重新导航,并消化可能出现的 Cloudflare 挑战 2. 复用失败或无保活上下文:在构建锁内创建新的保活上下文(顺带完成本次访问)
3. 复用失败:主动失效后强制重建一次,对调用方透明(避免「一次成功一次失败」交替)
Args: Args:
target_url: 目标页面 URL target_url: 目标页面 URL
Returns: Returns:
(html, session) 元组,session 为最终生效会话 (html, session) 元组,session 携带当前生效会话的 session_id 等信息
Raises: Raises:
UpstreamBlockedError: 会话重建后仍无法获取有效页面 UpstreamBlockedError: 重建后仍无法获取有效页面
CloudflareChallengeError: 等待 Cloudflare 挑战超时 CloudflareChallengeError: 等待 Cloudflare 挑战超时
ResourceBusyError: 等待页面槽位超时
""" """
session = await self.get_verified_session(target_url) session_name = self._session_name_for_url(target_url)
if session.html is not None:
return session.html, session
html = await self._try_fetch_with_session(target_url, 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: if html is not None:
return html, session return html, self._warm_to_session(warm)
await self._retire_warm(session_name, warm)
# 复用会话已失效:失效并强制重建,新会话自带页面 HTML # 2. 在构建锁内创建/复用保活上下文,避免并发重建风暴
logger.info("复用会话失败,强制重建会话:url=%s", target_url) lock = self._build_locks.setdefault(session_name, asyncio.Lock())
await self.invalidate(target_url) async with lock:
session = await self.get_verified_session(target_url, force_refresh=True) # 双重检查:可能已被其他协程在等待锁期间建好
if session.html is not None: warm = self._warm_contexts.get(session_name)
return session.html, session 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,缺失时再导航一次 logger.info("创建保活上下文:session_name=%s", session_name)
html = await self._try_fetch_with_session(target_url, session) warm, html = await self._build_warm_context(session_name, target_url)
if html is None: return html, self._warm_to_session(warm)
raise UpstreamBlockedError(f"会话重建后仍无法获取页面:url={target_url}")
return html, session
async def _try_fetch_with_session(self, target_url: str, session: VerifiedSession) -> str | None: def _warm_expired(self, warm: WarmContext) -> bool:
"""带已有会话导航并消化 Cloudflare 挑战 """保活上下文是否已退役或超过最大存活时长(沿用 session_ttl)。"""
if warm.retired:
return True
age = asyncio.get_running_loop().time() - warm.created_at
return age >= self._settings.session_ttl_seconds
与新建路径保持一致:导航后等待挑战自动通过,而非一遇非 200 就判死。 async def _fetch_via_warm(self, warm: WarmContext, target_url: str) -> str | None:
CF 对携带过期 cf_clearance 的请求常直接返回挑战页(首响应可能非 200), """在保活上下文内开新页面访问目标页。
因此即便首响应非 200,也先给页面一次自动通过 Turnstile 的机会,
仅当挑战超时/被硬阻断、或通过后仍无有效内容时才判定会话失效。
Returns: Returns:
通过验证的页面 HTML;若判定会话已失效,返回 None 通过验证的页面 HTML;若判定上下文已失效(非 200 / 挑战未过 / 上下文异常),返回 None
""" """
async with self._browser_pool.page_session(storage_state=session.storage_state) as (_, page): warm.in_use += 1
try:
async with self._browser_pool.page_in_context(warm.context) as page:
response = await page.goto( response = await page.goto(
target_url, target_url,
wait_until="domcontentloaded", wait_until="domcontentloaded",
timeout=int(self._settings.request_timeout_seconds * 1000), timeout=int(self._settings.request_timeout_seconds * 1000),
) )
status = response.status if response is not None else 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
# 首响应 200:极少数情况下仍是软挑战,用短超时给一次自动通过的机会
try: try:
# 复用使用短超时:快速验证能否通过,不行就放弃复用交由上层重建,
# 避免在无法自动通过的挑战上空等整个 cloudflare_wait_timeout_seconds。
await self._wait_for_clearance( await self._wait_for_clearance(
page, timeout_seconds=self._settings.cloudflare_reuse_wait_timeout_seconds page, timeout_seconds=self._settings.cloudflare_reuse_wait_timeout_seconds
) )
except (CloudflareChallengeError, UpstreamBlockedError): except (CloudflareChallengeError, UpstreamBlockedError):
logger.info( logger.info("保活上下文复用软挑战未通过,判定失效:url=%s", target_url)
"复用会话未通过挑战,判定失效:url=%s status=%s had_cf_clearance=%s",
target_url, status, bool(session.cf_clearance),
)
return None return None
html = await page.content() warm.last_used_at = asyncio.get_running_loop().time()
# 挑战已通过但页面仍是挑战壳(极少数 wait 提前返回的情况):判失效,触发重建 return await page.content()
if self._looks_like_challenge((await page.title()).lower(), html.lower()): except Exception as exc:
logger.info( # 上下文/浏览器已被关闭(如浏览器回收)或导航异常:视为失效,触发重建
"复用会话通过等待后仍无有效内容,判定失效:url=%s status=%s had_cf_clearance=%s", logger.info("保活上下文复用异常,判定失效:url=%s err=%s", target_url, exc)
target_url, status, bool(session.cf_clearance), 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),
)
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 return None
return html
async def invalidate(self, target_url: str) -> None: async def invalidate(self, target_url: str) -> None:
"""主动让会话失效 """主动让会话失效
@@ -167,6 +277,9 @@ class CloudflareSessionManager:
session_name = self._session_name_for_url(target_url) session_name = self._session_name_for_url(target_url)
logger.warning("会话失效:session_name=%s", session_name) logger.warning("会话失效:session_name=%s", session_name)
await self._session_store.invalidate_session(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: async def _create_verified_session(self, session_name: str, target_url: str) -> VerifiedSession:
"""通过真实浏览器访问目标页面,让 Cloudflare 挑战在浏览器侧完成 """通过真实浏览器访问目标页面,让 Cloudflare 挑战在浏览器侧完成
+53 -31
View File
@@ -122,42 +122,64 @@ def extract_product_id(url: str | None = None) -> str:
return "" 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日元 # 每 1 家加盟店,无论订单数量与金额,统一收 FRANCHISE_SHIPPING_FEE_PER_STORE 日元
• 订单金额低于1500日元:运费385日元 FRANCHISE_SHIPPING_FEE_PER_STORE = 800
• 订单金额1500日元及以上:免运费
*北海道、冲绳、离岛及其他部分地区可能需额外支付运费。 # --- 通贩手续费规则 ---
货到付款需另付550日元货到付款手续费。银行转账付款需由顾客承担转账手续费。信用卡付款无需支付手续费。 # 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 return 0
elif amount > 1000: if amount >= OFFICIAL_SHIPPING_FEE_LOW_THRESHOLD:
return 440 return OFFICIAL_SHIPPING_FEE_MID
else: return OFFICIAL_SHIPPING_FEE_LOW
return 385
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: 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 return 0
else: return HANDLE_FEE_BELOW
return 240
+154
View File
@@ -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
+124
View File
@@ -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 == {}