diff --git a/backend/package/yuxi/config/app.py b/backend/package/yuxi/config/app.py index 8f372ed5..c15ae935 100644 --- a/backend/package/yuxi/config/app.py +++ b/backend/package/yuxi/config/app.py @@ -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: diff --git a/backend/package/yuxi/external_systems/framework/handlers/__init__.py b/backend/package/yuxi/external_systems/framework/handlers/__init__.py new file mode 100644 index 00000000..c3dd66d9 --- /dev/null +++ b/backend/package/yuxi/external_systems/framework/handlers/__init__.py @@ -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 进程长生命周期会话导致的连接失效问题。 +""" diff --git a/backend/package/yuxi/external_systems/infrastructure/scheduler.py b/backend/package/yuxi/external_systems/infrastructure/scheduler.py new file mode 100644 index 00000000..ed77c6c3 --- /dev/null +++ b/backend/package/yuxi/external_systems/infrastructure/scheduler.py @@ -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，占位避免未使用告警 diff --git a/backend/package/yuxi/scheduler/adapters/persistence/mappers.py b/backend/package/yuxi/scheduler/adapters/persistence/mappers.py index c37babda..37c147fb 100644 --- a/backend/package/yuxi/scheduler/adapters/persistence/mappers.py +++ b/backend/package/yuxi/scheduler/adapters/persistence/mappers.py @@ -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, ) diff --git a/backend/package/yuxi/scheduler/adapters/persistence/sqlalchemy_task_repo.py b/backend/package/yuxi/scheduler/adapters/persistence/sqlalchemy_task_repo.py index 17cc9b5f..27ecaa0b 100644 --- a/backend/package/yuxi/scheduler/adapters/persistence/sqlalchemy_task_repo.py +++ b/backend/package/yuxi/scheduler/adapters/persistence/sqlalchemy_task_repo.py @@ -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, ) diff --git a/backend/package/yuxi/scheduler/adapters/persistence/unit_of_work.py b/backend/package/yuxi/scheduler/adapters/persistence/unit_of_work.py index e6db8e4f..cd23e159 100644 --- a/backend/package/yuxi/scheduler/adapters/persistence/unit_of_work.py +++ b/backend/package/yuxi/scheduler/adapters/persistence/unit_of_work.py @@ -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: diff --git a/backend/package/yuxi/scheduler/core/ports.py b/backend/package/yuxi/scheduler/core/ports.py index 5c0a8aba..778cb614 100644 --- a/backend/package/yuxi/scheduler/core/ports.py +++ b/backend/package/yuxi/scheduler/core/ports.py @@ -37,7 +37,6 @@ from yuxi.scheduler.core.models import ( ScheduledTaskRunLog, ) - # ─── 仓储端口 ──────────────────────────────────────────────────────── diff --git a/backend/package/yuxi/scheduler/use_cases/ports/scheduler_service_port.py b/backend/package/yuxi/scheduler/use_cases/ports/scheduler_service_port.py index bf0b3247..f2efed46 100644 --- a/backend/package/yuxi/scheduler/use_cases/ports/scheduler_service_port.py +++ b/backend/package/yuxi/scheduler/use_cases/ports/scheduler_service_port.py @@ -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`` 降序排序，支持时间范围过滤与分页。 diff --git a/backend/package/yuxi/scheduler/use_cases/services/scheduler_service.py b/backend/package/yuxi/scheduler/use_cases/services/scheduler_service.py index 1e372f9a..d3923f91 100644 --- a/backend/package/yuxi/scheduler/use_cases/services/scheduler_service.py +++ b/backend/package/yuxi/scheduler/use_cases/services/scheduler_service.py @@ -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`` 取当前页条数（客户端可按