refactor(scheduler): 整理代码格式与导入，新增调度器配置与handler注册框架
本次提交包含多类改进： 1. 清理冗余空行、修复导入格式与类型检查占位 2. 新增external_systems与scheduler上下文的handler注册框架 3. 补全调度器相关配置项与环境变量覆盖逻辑 4. 新增配置校验与敏感字段脱敏逻辑 5. 简化协议方法声明格式
This commit is contained in:
parent
3ad986e033
commit
2a401450cb
@ -8,7 +8,7 @@ from typing import Any
|
||||
|
||||
import tomli
|
||||
import tomli_w
|
||||
from pydantic import BaseModel, Field, PrivateAttr
|
||||
from pydantic import BaseModel, Field, PrivateAttr, model_validator
|
||||
|
||||
from yuxi.utils.logging_config import logger
|
||||
|
||||
@ -55,6 +55,48 @@ class Config(BaseModel):
|
||||
ssrf_blocked_domains: list[str] = Field(default_factory=list, description="SSRF 域名黑名单")
|
||||
ssrf_dns_pinned_ttl: int = Field(default=300, description="SSRF DNS 缓存时间（秒）")
|
||||
|
||||
# === 定时任务调度配置（扁平字段，对齐既有模式，见设计方案 §13） ===
|
||||
scheduler_enabled: bool = Field(default=True, description="定时任务总开关")
|
||||
scheduler_tick_interval_seconds: int = Field(default=60, description="tick 频率（秒）")
|
||||
scheduler_max_consecutive_errors: int = Field(default=5, description="死信阈值")
|
||||
scheduler_task_timeout_seconds: int = Field(default=600, description="单任务超时（秒）")
|
||||
scheduler_max_payload_size_kb: int = Field(default=64, description="payload 大小上限（KB）")
|
||||
scheduler_max_active_tasks_per_owner: int = Field(default=100, description="业务方任务数上限")
|
||||
scheduler_stagger_max_seconds: int = Field(default=300, description="整点抖动上限（秒）")
|
||||
scheduler_run_log_retention_days: int = Field(default=90, description="run_logs 保留天数")
|
||||
scheduler_idempotency_retention_hours: int = Field(default=24, description="幂等记录保留小时数")
|
||||
scheduler_min_cron_interval_minutes: int = Field(default=10, description="cron 最小间隔（分钟）")
|
||||
scheduler_backoff_schedule: list[int] = Field(
|
||||
default_factory=lambda: [10, 30, 120, 600, 3600],
|
||||
description="指数退避序列（秒）",
|
||||
)
|
||||
|
||||
# === Web 搜索配置（扁平字段，见设计方案 §10.1.1） ===
|
||||
web_search_provider_chain: list[str] = Field(
|
||||
default_factory=lambda: ["searxng", "tavily"],
|
||||
description="Web 搜索 Provider 优先级链",
|
||||
)
|
||||
web_search_searxng_base_url: str = Field(default="http://searxng:8080", description="SearXNG 服务地址")
|
||||
web_search_searxng_timeout: int = Field(default=10, description="SearXNG 调用超时（秒）")
|
||||
web_search_searxng_engines: list[str] = Field(default_factory=list, description="SearXNG 指定引擎列表（空=使用默认）")
|
||||
web_search_searxng_language: str = Field(default="auto", description="SearXNG 搜索语言")
|
||||
web_search_searxng_safesearch: int = Field(default=0, description="SearXNG 安全搜索级别（0/1/2）")
|
||||
web_search_tavily_api_key: str = Field(default="", description="Tavily API Key（空=读 TAVILY_API_KEY 环境变量）")
|
||||
web_search_tavily_timeout: int = Field(default=10, description="Tavily 调用超时（秒）")
|
||||
web_search_default_max_results: int = Field(default=5, description="Web 搜索默认结果数")
|
||||
web_search_rate_limit_per_minute: int = Field(default=20, description="单 agent Web 搜索每分钟调用上限")
|
||||
|
||||
# === Web 抓取配置（扁平字段，见设计方案 §10.1.2） ===
|
||||
crawl4ai_base_url: str = Field(default="http://crawl4ai:11235", description="Crawl4AI 服务地址")
|
||||
crawl4ai_timeout: int = Field(default=30, description="Crawl4AI 单页抓取超时（秒）")
|
||||
crawl4ai_api_token: str = Field(default="", description="Crawl4AI API Token（预留 JWT 认证）")
|
||||
web_crawl_default_max_depth: int = Field(default=2, description="Web 抓取默认递归深度")
|
||||
web_crawl_default_max_pages: int = Field(default=20, description="Web 抓取默认最大页数")
|
||||
web_crawl_default_concurrency: int = Field(default=5, description="Web 抓取默认并发数")
|
||||
web_crawl_domain_whitelist: list[str] = Field(default_factory=list, description="Web 抓取域名白名单（空=允许所有公网域名）")
|
||||
web_crawl_respect_robots: bool = Field(default=True, description="Web 抓取是否尊重 robots.txt")
|
||||
web_crawl_rate_limit_per_minute: int = Field(default=10, description="单 agent Web 抓取每分钟调用上限")
|
||||
|
||||
_config_file: Path | None = PrivateAttr(default=None)
|
||||
_user_modified_fields: set[str] = PrivateAttr(default_factory=set)
|
||||
|
||||
@ -121,6 +163,93 @@ class Config(BaseModel):
|
||||
if ssrf_env:
|
||||
self.ssrf_allowed_domains = [d.strip() for d in ssrf_env.split(",") if d.strip()]
|
||||
|
||||
# === scheduler 环境变量覆盖（对齐既有三段式写法，见设计方案 §13） ===
|
||||
self.scheduler_enabled = (
|
||||
os.getenv("SCHEDULER_ENABLED", "true").lower() in ("true", "1", "yes")
|
||||
)
|
||||
self.scheduler_tick_interval_seconds = int(
|
||||
os.getenv("SCHEDULER_TICK_INTERVAL_SECONDS") or self.scheduler_tick_interval_seconds or 60
|
||||
)
|
||||
self.scheduler_max_consecutive_errors = int(
|
||||
os.getenv("SCHEDULER_MAX_CONSECUTIVE_ERRORS") or self.scheduler_max_consecutive_errors or 5
|
||||
)
|
||||
self.scheduler_task_timeout_seconds = int(
|
||||
os.getenv("SCHEDULER_TASK_TIMEOUT_SECONDS") or self.scheduler_task_timeout_seconds or 600
|
||||
)
|
||||
self.scheduler_max_payload_size_kb = int(
|
||||
os.getenv("SCHEDULER_MAX_PAYLOAD_SIZE_KB") or self.scheduler_max_payload_size_kb or 64
|
||||
)
|
||||
self.scheduler_max_active_tasks_per_owner = int(
|
||||
os.getenv("SCHEDULER_MAX_ACTIVE_TASKS_PER_OWNER")
|
||||
or self.scheduler_max_active_tasks_per_owner
|
||||
or 100
|
||||
)
|
||||
self.scheduler_stagger_max_seconds = int(
|
||||
os.getenv("SCHEDULER_STAGGER_MAX_SECONDS") or self.scheduler_stagger_max_seconds or 300
|
||||
)
|
||||
self.scheduler_run_log_retention_days = int(
|
||||
os.getenv("SCHEDULER_RUN_LOG_RETENTION_DAYS") or self.scheduler_run_log_retention_days or 90
|
||||
)
|
||||
self.scheduler_idempotency_retention_hours = int(
|
||||
os.getenv("SCHEDULER_IDEMPOTENCY_RETENTION_HOURS")
|
||||
or self.scheduler_idempotency_retention_hours
|
||||
or 24
|
||||
)
|
||||
self.scheduler_min_cron_interval_minutes = int(
|
||||
os.getenv("SCHEDULER_MIN_CRON_INTERVAL_MINUTES")
|
||||
or self.scheduler_min_cron_interval_minutes
|
||||
or 10
|
||||
)
|
||||
|
||||
# === Web 搜索环境变量覆盖 ===
|
||||
self.web_search_searxng_base_url = (
|
||||
os.getenv("WEB_SEARCH_SEARXNG_BASE_URL") or self.web_search_searxng_base_url
|
||||
)
|
||||
self.web_search_searxng_timeout = int(
|
||||
os.getenv("WEB_SEARCH_SEARXNG_TIMEOUT") or self.web_search_searxng_timeout
|
||||
)
|
||||
self.web_search_searxng_language = (
|
||||
os.getenv("WEB_SEARCH_SEARXNG_LANGUAGE") or self.web_search_searxng_language
|
||||
)
|
||||
self.web_search_searxng_safesearch = int(
|
||||
os.getenv("WEB_SEARCH_SEARXNG_SAFESEARCH") or self.web_search_searxng_safesearch
|
||||
)
|
||||
self.web_search_tavily_api_key = (
|
||||
os.getenv("TAVILY_API_KEY") or self.web_search_tavily_api_key
|
||||
)
|
||||
self.web_search_tavily_timeout = int(
|
||||
os.getenv("WEB_SEARCH_TAVILY_TIMEOUT") or self.web_search_tavily_timeout
|
||||
)
|
||||
|
||||
# === Web 抓取环境变量覆盖 ===
|
||||
self.crawl4ai_base_url = (
|
||||
os.getenv("CRAWL4AI_BASE_URL") or self.crawl4ai_base_url
|
||||
)
|
||||
self.crawl4ai_timeout = int(
|
||||
os.getenv("CRAWL4AI_TIMEOUT") or self.crawl4ai_timeout
|
||||
)
|
||||
self.crawl4ai_api_token = (
|
||||
os.getenv("CRAWL4AI_API_TOKEN") or self.crawl4ai_api_token
|
||||
)
|
||||
|
||||
@model_validator(mode="after")
|
||||
def _validate_web_search_crawl_config(self):
|
||||
"""校验 web_search 与 web_crawl 配置合法性。"""
|
||||
valid_providers = {"searxng", "tavily"}
|
||||
for provider in self.web_search_provider_chain:
|
||||
if provider not in valid_providers:
|
||||
raise ValueError(
|
||||
f"web_search_provider_chain 含未知 Provider 名: {provider}，"
|
||||
f"合法值: {valid_providers}"
|
||||
)
|
||||
if not 1 <= self.web_search_default_max_results <= 10:
|
||||
raise ValueError("web_search_default_max_results 必须在 [1, 10] 范围内")
|
||||
if not 1 <= self.web_crawl_default_max_depth <= 3:
|
||||
raise ValueError("web_crawl_default_max_depth 必须在 [1, 3] 范围内")
|
||||
if not 1 <= self.web_crawl_default_max_pages <= 100:
|
||||
raise ValueError("web_crawl_default_max_pages 必须在 [1, 100] 范围内")
|
||||
return self
|
||||
|
||||
def save(self) -> None:
|
||||
if not self._config_file:
|
||||
logger.warning("Config file path not set")
|
||||
@ -146,6 +275,11 @@ class Config(BaseModel):
|
||||
|
||||
def dump_config(self) -> dict[str, Any]:
|
||||
config_dict = self.model_dump()
|
||||
# 敏感字段脱敏
|
||||
sensitive_keys = {"web_search_tavily_api_key", "crawl4ai_api_token"}
|
||||
for key in sensitive_keys:
|
||||
if key in config_dict and config_dict[key]:
|
||||
config_dict[key] = "***"
|
||||
fields_info = {}
|
||||
for field_name, field_info in Config.model_fields.items():
|
||||
if field_info.exclude:
|
||||
|
||||
@ -0,0 +1,14 @@
|
||||
"""external_systems 的定时任务 handler 集合。
|
||||
|
||||
本子包存放 external_systems 模块接入 scheduler 限界上下文的 handler 实现。
|
||||
每个 handler 实现 ``yuxi.scheduler.core.contracts.TaskHandler`` 协议，
|
||||
由 ``external_systems/scheduler.py`` 的 ``register_scheduler_handlers``
|
||||
统一注册到 worker 进程级 ``HandlerRegistry``。
|
||||
|
||||
依赖方向：仅依赖 ``scheduler/core/contracts``（``TaskHandler`` 协议）+
|
||||
``storage/postgres/manager``（db 会话）+ 本模块的 use_cases / framework/runtime。
|
||||
不依赖 ``scheduler`` 的 use_cases / adapters / infrastructure。
|
||||
|
||||
会话隔离：handler 通过构造函数注入 ``session_factory``，每次 ``execute``
|
||||
创建独立 db 会话，避免 worker 进程长生命周期会话导致的连接失效问题。
|
||||
"""
|
||||
@ -0,0 +1,45 @@
|
||||
"""external_systems 模块的定时任务 handler 注册入口。
|
||||
|
||||
本模块是 external_systems 限界上下文接入 scheduler 限界上下文的唯一登记点，
|
||||
对齐 scheduler 上下文设计方案 §14.3 的"业务域 container 主动注册"模式。
|
||||
|
||||
与 ``infrastructure/container.py`` 同级，遵循 external_systems 的分层约定：
|
||||
装配逻辑集中在 ``infrastructure/``，包根不出现装配代码。
|
||||
|
||||
对外暴露：
|
||||
|
||||
- ``register_scheduler_handlers(registry)``：业务模块注册函数，由 worker
|
||||
启动钩子（``services/run_worker.py`` 的 ``_worker_startup``）统一调用。
|
||||
内部实例化本模块的 handler 并调用 ``registry.register(handler)``。
|
||||
|
||||
边界规范：
|
||||
- 本模块是 external_systems 接入 scheduler 的**唯一装配点**
|
||||
- handler 的依赖（如 ``session_factory``）在函数内部构造，不暴露给装配层
|
||||
- 新增 handler 时，在本函数内追加 ``registry.register(...)`` 即可，无需改
|
||||
``run_worker.py``（前提：本模块已登记在 ``_BUSINESS_HANDLER_REGISTRARS``）
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from yuxi.scheduler.framework.runtime import HandlerRegistry
|
||||
from yuxi.storage.postgres.manager import pg_manager
|
||||
|
||||
__all__ = ["register_scheduler_handlers"]
|
||||
|
||||
|
||||
def register_scheduler_handlers(registry: HandlerRegistry) -> None:
|
||||
"""向 scheduler 注册 external_systems 模块的定时任务 handler。
|
||||
|
||||
由 worker 启动钩子在装配阶段统一调用。新增 handler 时在此函数内追加
|
||||
``registry.register(...)`` 即可。
|
||||
|
||||
Args:
|
||||
registry: worker 进程级 handler 注册表。
|
||||
"""
|
||||
session_factory = pg_manager.get_async_session_context
|
||||
|
||||
# 在此追加 external_systems 的定时任务 handler 注册
|
||||
# registry.register(
|
||||
# XxxHandler(session_factory=session_factory),
|
||||
# )
|
||||
_ = session_factory # 暂无 handler，占位避免未使用告警
|
||||
@ -21,14 +21,22 @@ from typing import TYPE_CHECKING, Any
|
||||
|
||||
from yuxi.scheduler.core.models import (
|
||||
DailyStat as DailyStatDC,
|
||||
)
|
||||
from yuxi.scheduler.core.models import (
|
||||
HandlerSummary as HandlerSummaryDC,
|
||||
)
|
||||
from yuxi.scheduler.core.models import (
|
||||
ScheduledTask as ScheduledTaskDC,
|
||||
)
|
||||
from yuxi.scheduler.core.models import (
|
||||
ScheduledTaskRunLog as ScheduledTaskRunLogDC,
|
||||
)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from yuxi.storage.postgres.models_scheduler import (
|
||||
ScheduledTask as ScheduledTaskORM,
|
||||
)
|
||||
from yuxi.storage.postgres.models_scheduler import (
|
||||
ScheduledTaskRunLog as ScheduledTaskRunLogORM,
|
||||
)
|
||||
|
||||
|
||||
@ -23,6 +23,8 @@ from yuxi.repositories.scheduled.scheduled_task_repository import (
|
||||
)
|
||||
from yuxi.scheduler.core.models import (
|
||||
HandlerSummary as HandlerSummaryDC,
|
||||
)
|
||||
from yuxi.scheduler.core.models import (
|
||||
ScheduledTask as ScheduledTaskDC,
|
||||
)
|
||||
|
||||
|
||||
@ -11,7 +11,7 @@ from typing import TYPE_CHECKING
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from yuxi.scheduler.core.contracts import UnitOfWork
|
||||
pass
|
||||
|
||||
|
||||
class SqlAlchemyUnitOfWork:
|
||||
|
||||
@ -37,7 +37,6 @@ from yuxi.scheduler.core.models import (
|
||||
ScheduledTaskRunLog,
|
||||
)
|
||||
|
||||
|
||||
# ─── 仓储端口 ────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
|
||||
@ -184,9 +184,7 @@ class SchedulerServicePort(Protocol):
|
||||
"""
|
||||
...
|
||||
|
||||
async def list_all_run_logs(
|
||||
self, input_dto: ListAllRunLogsInput
|
||||
) -> ListRunLogsOutput:
|
||||
async def list_all_run_logs(self, input_dto: ListAllRunLogsInput) -> ListRunLogsOutput:
|
||||
"""跨任务列出执行日志（FR-ST-09）。
|
||||
|
||||
``task_id`` 非空时按任务维度查询（等价于 ``list_run_logs``）；``task_id``
|
||||
@ -195,9 +193,7 @@ class SchedulerServicePort(Protocol):
|
||||
"""
|
||||
...
|
||||
|
||||
async def list_daily_stats(
|
||||
self, input_dto: ListDailyStatsInput
|
||||
) -> ListDailyStatsOutput:
|
||||
async def list_daily_stats(self, input_dto: ListDailyStatsInput) -> ListDailyStatsOutput:
|
||||
"""列出日聚合统计（FR-ST-09）。
|
||||
|
||||
按日期范围 + handler 维度查询日聚合统计，返回含 ``total_count`` /
|
||||
@ -264,9 +260,7 @@ class SchedulerServicePort(Protocol):
|
||||
# 任务回收站（async，DTO 入参）
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
async def list_deleted_tasks(
|
||||
self, input_dto: ListDeletedTasksInput
|
||||
) -> ListTasksOutput:
|
||||
async def list_deleted_tasks(self, input_dto: ListDeletedTasksInput) -> ListTasksOutput:
|
||||
"""列出已删除任务（回收站，FR-ST-05）。
|
||||
|
||||
返回软删除的任务列表，按 ``deleted_at`` 降序排序，支持时间范围过滤与分页。
|
||||
|
||||
@ -20,7 +20,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
from datetime import UTC, date, datetime, timedelta
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from typing import TYPE_CHECKING, Any
|
||||
from zoneinfo import ZoneInfo
|
||||
|
||||
@ -556,9 +556,7 @@ class SchedulerService(SchedulerServicePort):
|
||||
)
|
||||
return to_run_log_output(log)
|
||||
|
||||
async def list_all_run_logs(
|
||||
self, input_dto: ListAllRunLogsInput
|
||||
) -> ListRunLogsOutput:
|
||||
async def list_all_run_logs(self, input_dto: ListAllRunLogsInput) -> ListRunLogsOutput:
|
||||
"""跨任务列出执行日志（FR-ST-09）。
|
||||
|
||||
``task_id`` 非空时按任务维度查询；``task_id`` 为空时按 ``status`` 跨任务
|
||||
@ -627,9 +625,7 @@ class SchedulerService(SchedulerServicePort):
|
||||
)
|
||||
return to_list_upcoming_output(tasks)
|
||||
|
||||
async def list_daily_stats(
|
||||
self, input_dto: ListDailyStatsInput
|
||||
) -> ListDailyStatsOutput:
|
||||
async def list_daily_stats(self, input_dto: ListDailyStatsInput) -> ListDailyStatsOutput:
|
||||
"""列出日聚合统计（FR-ST-09）。
|
||||
|
||||
解析 ``start_date`` / ``end_date`` 为 ``date``，校验范围 ≤ 90 天，
|
||||
@ -1029,9 +1025,7 @@ class SchedulerService(SchedulerServicePort):
|
||||
# 任务回收站
|
||||
# ==================================================================
|
||||
|
||||
async def list_deleted_tasks(
|
||||
self, input_dto: ListDeletedTasksInput
|
||||
) -> ListTasksOutput:
|
||||
async def list_deleted_tasks(self, input_dto: ListDeletedTasksInput) -> ListTasksOutput:
|
||||
"""列出已删除任务（回收站，FR-ST-05）。
|
||||
|
||||
``list_deleted`` 不返回总数，``total`` 取当前页条数（客户端可按
|
||||
|
||||
Loading…
Reference in New Issue
Block a user