134 lines
5.1 KiB
Python
134 lines
5.1 KiB
Python
"""应用配置：通过环境变量和 .env 文件加载所有配置项
|
|
|
|
配置项统一使用 SURUGAYA_ 前缀，例如 SURUGAYA_APP_PORT=31106。
|
|
支持 .env 文件自动加载。
|
|
"""
|
|
from functools import lru_cache
|
|
from typing import Literal
|
|
|
|
from pydantic_settings import BaseSettings, SettingsConfigDict
|
|
from pathlib import Path
|
|
|
|
from surugaya_common import browser_profile, urls
|
|
|
|
BASE_DIR = Path(__file__).resolve().parent.parent.parent
|
|
|
|
|
|
class Settings(BaseSettings):
|
|
"""应用全局配置
|
|
|
|
所有配置项均可通过环境变量覆盖，前缀为 SURUGAYA_。
|
|
例如：SURUGAYA_APP_PORT=31106 对应 app_port 配置项。
|
|
"""
|
|
|
|
model_config = SettingsConfigDict(
|
|
env_file=str(BASE_DIR / ".env"), # 使用绝对路径
|
|
env_file_encoding="utf-8",
|
|
env_prefix="SURUGAYA_",
|
|
extra="ignore",
|
|
)
|
|
|
|
# ---- 服务基本配置 ----
|
|
app_name: str = "Surugaya Scraper Service"
|
|
app_env: Literal["dev", "prod", "test"] = "dev"
|
|
app_host: str = "0.0.0.0"
|
|
app_port: int = 31106
|
|
|
|
# ---- 日志配置 ----
|
|
log_level: str = "DEBUG"
|
|
log_to_file: bool | None = None # None 表示根据环境自动决定
|
|
log_dir: str = "logs"
|
|
log_rotation: str = "10 MB"
|
|
log_retention: str = "14 days"
|
|
log_compression: str = "zip"
|
|
log_format: str = "{time:YYYY-MM-DD HH:mm:ss} {level} {message}"
|
|
log_enqueue: bool = True
|
|
|
|
# ---- 鉴权配置 ----
|
|
bearer_token: str = "NilsO8Il8yoT1wazJFs2eeOmlPRfWuSx"
|
|
|
|
# ---- 浏览器配置 ----
|
|
browser_enabled: bool = True
|
|
browser_headless: bool | None = None # None 表示根据环境自动决定
|
|
browser_keep_open: bool | None = None # None 表示根据环境自动决定
|
|
browser_keep_open_on_error: bool | None = None # None 表示根据环境自动决定
|
|
browser_channel: str | None = "chrome"
|
|
browser_pool_size: int = 3
|
|
max_site_concurrency: int = 5 # 最大并发页面数
|
|
browser_recycle_after_contexts: int = 200 # 浏览器累计创建上下文达到该值后主动回收，0 关闭
|
|
browser_recycle_after_seconds: float = 3600.0 # 浏览器运行超过该秒数后主动回收，0 关闭
|
|
page_acquire_timeout_seconds: float = 60.0 # 获取页面槽位超时
|
|
request_timeout_seconds: float = 120.0 # 页面请求超时
|
|
cloudflare_wait_timeout_seconds: float = 60.0 # 等待 Cloudflare 挑战超时
|
|
cloudflare_reuse_wait_timeout_seconds: float = 8.0 # 复用会话时消化挑战的短超时，超时则放弃复用直接重建
|
|
session_ttl_seconds: int = 900 # 会话过期时间（秒）
|
|
# 指纹参数默认值取自共享包，与 worker 保持一致
|
|
browser_locale: str = browser_profile.LOCALE
|
|
browser_timezone_id: str = browser_profile.TIMEZONE_ID
|
|
browser_user_agent: str = browser_profile.DEFAULT_USER_AGENT
|
|
browser_stealth_enabled: bool = True # 是否启用浏览器 Stealth 模式
|
|
|
|
# ---- Redis 配置（可选，未配置时使用进程内存）----
|
|
redis_host: str = ""
|
|
redis_port: int = 6379
|
|
redis_password: str | None = None
|
|
redis_db: int = 0
|
|
session_namespace: str = "jp_surugaya"
|
|
|
|
# ---- 代理配置（可选，用于日本 IP）----
|
|
proxy_server: str | None = None
|
|
proxy_username: str | None = None
|
|
proxy_password: str | None = None
|
|
|
|
# ---- 目标站点 ----
|
|
base_url: str = urls.BASE_URL
|
|
|
|
@property
|
|
def browser_headless_effective(self) -> bool:
|
|
"""浏览器无头模式：显式配置优先，否则开发环境使用有头模式方便调试"""
|
|
if self.browser_headless is not None:
|
|
return self.browser_headless
|
|
if self.app_env == "dev":
|
|
return False
|
|
return True
|
|
|
|
@property
|
|
def browser_keep_open_on_error_effective(self) -> bool:
|
|
"""出错时是否保留浏览器窗口：仅在开发环境有头模式下保留"""
|
|
if self.browser_keep_open_on_error is not None:
|
|
return self.browser_keep_open_on_error
|
|
return self.app_env == "dev" and not self.browser_headless_effective
|
|
|
|
@property
|
|
def browser_keep_open_effective(self) -> bool:
|
|
"""成功后是否保留浏览器窗口：仅在开发环境有头模式下保留"""
|
|
if self.browser_keep_open is not None:
|
|
return self.browser_keep_open
|
|
return self.app_env == "dev" and not self.browser_headless_effective
|
|
|
|
@property
|
|
def log_to_file_effective(self) -> bool:
|
|
"""是否写日志文件：显式配置优先，否则生产环境默认写文件"""
|
|
if self.log_to_file is not None:
|
|
return self.log_to_file
|
|
return self.app_env == "prod"
|
|
|
|
@property
|
|
def proxy(self) -> dict[str, str] | None:
|
|
"""构建 Playwright 代理配置字典"""
|
|
if not self.proxy_server:
|
|
return None
|
|
|
|
proxy: dict[str, str] = {"server": self.proxy_server}
|
|
if self.proxy_username:
|
|
proxy["username"] = self.proxy_username
|
|
if self.proxy_password:
|
|
proxy["password"] = self.proxy_password
|
|
return proxy
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def get_settings() -> Settings:
|
|
"""获取全局配置单例"""
|
|
return Settings()
|