Compare commits
2 Commits
ada64a68ad
...
3b8bdf09f7
| Author | SHA1 | Date | |
|---|---|---|---|
| 3b8bdf09f7 | |||
| bc0f7161c0 |
@@ -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
|
||||
|
||||
|
||||
@@ -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()
|
||||
|
||||
|
||||
@@ -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):
|
||||
|
||||
+104
-36
@@ -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:
|
||||
|
||||
@@ -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 挑战在浏览器侧完成
|
||||
|
||||
+53
-31
@@ -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
|
||||
|
||||
@@ -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
|
||||
@@ -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 == {}
|
||||
Reference in New Issue
Block a user