From 11efe91ce91cdd2b71ce2f3af80cb36535f98110 Mon Sep 17 00:00:00 2001 From: Profeto Agent Date: Sat, 19 Sep 2026 04:38:53 +0000 Subject: [PATCH] =?UTF-8?q?fix:=20=E5=85=A8=E6=96=B9=E4=BD=8D=E5=AE=A1?= =?UTF-8?q?=E6=9F=A5=E9=97=AE=E9=A2=98=E4=BF=AE=E5=A4=8D=EF=BC=88P0~P3?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P0: ORM-迁移同步 - 新增 RawEvent/IngestFailure/DataQualityCheck/DataLineage 4 个模型类 - MatchStats 补充 xg_source/xg_updated_at/xg_source_record_id 字段 - 修复 server_default=_utcnow → func.now()（4 处） - BigInteger 导入 P1: - log_buffer.py 移除 threading.Lock（asyncio 单线程下无需锁） - 确认 context_builder/understat/injuries 等已有修复 P2: - Dockerfile 新增非 root 用户 + .dockerignore - 前端 fetchDashboard 修复 total 字段（改用 items.length） - 前端 fetchSystemConfig 改用真实 /admin/settings 端点 - 修复 validation.py return self"" → return self P3: - predict.py 缓存加 _CACHE_MAX_SIZE=200 淘汰 - eval.py get_eval_summary 加 limit 参数（默认 1000）+ 返回 total_settled - orchestrator.py agent provider 配置缓存 60s --- .dockerignore | 37 +++++++++++++ Dockerfile | 9 ++++ frontend/src/admin/dal.ts | 40 +++++++++------ src/api/routes/eval.py | 8 +-- src/core/log_buffer.py | 8 +-- src/db/models.py | 94 ++++++++++++++++++++++++++++++++++ src/llm/agents/orchestrator.py | 19 +++++++ src/llm/eval.py | 18 +++++-- src/llm/predict.py | 6 ++- src/llm/validation.py | 2 +- 10 files changed, 209 insertions(+), 32 deletions(-) create mode 100644 .dockerignore diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..5b3f7e5 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,37 @@ +# Python +__pycache__ +*.py[cod] +*$py.class +*.egg-info +dist +build + +# 虚拟环境 +.venv +venv + +# 环境配置(含敏感信息,绝不能打入镜像) +.env +.env.* + +# 测试与工具 +tests +.pytest_cache +.mypy_cache +.ruff_cache + +# IDE +.idea +.vscode + +# Git +.git +.gitignore + +# 文档(运行时不需要) +docs +*.md +LICENSE* +.dockerignore +docker-compose.yml +nginx.conf diff --git a/Dockerfile b/Dockerfile index bf86fb8..2410ade 100644 --- a/Dockerfile +++ b/Dockerfile @@ -5,12 +5,21 @@ WORKDIR /app # 直连官方源不稳定,固定使用清华 PyPI 镜像 ENV PIP_INDEX_URL=https://pypi.tuna.tsinghua.edu.cn/simple +# P2-5: 创建非 root 用户(容器安全最佳实践) +RUN groupadd --system profeto && useradd --system --gid profeto profeto + RUN pip install --no-cache-dir hatchling COPY pyproject.toml README.md ./ COPY src ./src RUN pip install --no-cache-dir . +# 将工作目录所有权移交给非 root 用户 +RUN chown -R profeto:profeto /app + EXPOSE 8000 +# 以非 root 用户运行 +USER profeto + CMD ["uvicorn", "src.api.app:app", "--host", "0.0.0.0", "--port", "8000"] diff --git a/frontend/src/admin/dal.ts b/frontend/src/admin/dal.ts index 147bdcd..0d797ed 100644 --- a/frontend/src/admin/dal.ts +++ b/frontend/src/admin/dal.ts @@ -27,20 +27,26 @@ import type { /** * 从多个端点聚合仪表盘数据。 * 后端暂无专用仪表盘端点,这里组合 health + 各列表端点。 + * + * P2-8 修复: 使用 items.length 替代不存在的 total 字段, + * 并扩大 limit 以获得更有参考价值的数量。 */ export async function fetchDashboard(): Promise { // 并行获取各端点数据 + // matches 返回 {items, next_cursor, has_more}, predictions 返回数组 const [leagues, matches, predictions, health] = await Promise.allSettled([ api.get(`${API_BASE}/leagues`), - api.get(`${API_BASE}/matches?limit=1`), - api.get(`${API_BASE}/predictions?limit=1`), + api.get<{ items: Match[]; has_more: boolean }>(`${API_BASE}/matches?limit=100`), + api.get(`${API_BASE}/predictions?limit=100`), api.get<{ status: string }>('/health'), ]) return { leagues: leagues.status === 'fulfilled' ? leagues.value : [], - total_matches: matches.status === 'fulfilled' ? (matches.value as any)?.total ?? 0 : 0, - total_predictions: predictions.status === 'fulfilled' ? (predictions.value as any)?.total ?? 0 : 0, + // P2-8: matches 无 total 字段,用 items.length 近似(上限 100) + total_matches: matches.status === 'fulfilled' ? (matches.value as any)?.items?.length ?? 0 : 0, + // predictions 直接返回数组 + total_predictions: predictions.status === 'fulfilled' ? (predictions.value as any)?.length ?? 0 : 0, health: health.status === 'fulfilled' ? (health.value as any).status : 'unknown', db_tables: [], // 后端暂无表统计端点 last_collection: [], // 后端暂无采集历史端点 @@ -270,18 +276,20 @@ export async function fetchLLMUsageStats(): Promise { // ── 系统配置 ──────────────────────────────────────────────────── /** - * 获取系统配置列表 — 后端暂无配置端点,返回静态信息 + * P2-9 修复: 获取系统配置列表,从后端 /admin/settings 读取真实值(脱敏)。 + * 字段名对齐 Config.tsx 中使用的 { key, value_masked, description, is_sensitive } 格式。 */ export async function fetchSystemConfig(): Promise { - return [ - { key: 'LLM_PROVIDER', value_masked: 'openai', description: 'LLM 提供商', is_sensitive: false }, - { key: 'LLM_MODEL', value_masked: 'gpt-4o', description: 'LLM 模型', is_sensitive: false }, - { key: 'LLM_BASE_URL', value_masked: 'https://api.openai.com/v1', description: 'API 基础地址', is_sensitive: false }, - { key: 'LLM_API_KEY', value_masked: 'sk-****...****', description: 'LLM API 密钥', is_sensitive: true }, - { key: 'ADMIN_PASSWORD', value_masked: '••••••(已配置)', description: '管理后台登录密码', is_sensitive: true }, - { key: 'ADMIN_API_KEY', value_masked: '未配置时脚本调用不可用', description: '接口鉴权密钥 (X-API-Key)', is_sensitive: true }, - { key: 'BZZOIRO_KEY', value_masked: 'bz****...****', description: 'Bzzoiro 数据源密钥(可在「数据源」页在线配置)', is_sensitive: true }, - { key: 'DATABASE_URL', value_masked: 'postgresql://****@localhost/profeto', description: '数据库连接', is_sensitive: true }, - { key: 'LOG_LEVEL', value_masked: 'INFO', description: '日志级别', is_sensitive: false }, - ] + try { + const settings = await fetchSettings() + return settings.map(s => ({ + key: s.key, + value_masked: s.masked, + description: s.description, + is_sensitive: s.sensitive, + })) + } catch { + // 后端不可用时返回空列表,Config.tsx 会显示空状态 + return [] + } } diff --git a/src/api/routes/eval.py b/src/api/routes/eval.py index 06b4282..a7128e9 100644 --- a/src/api/routes/eval.py +++ b/src/api/routes/eval.py @@ -3,7 +3,7 @@ from __future__ import annotations import logging -from fastapi import APIRouter, Depends, HTTPException +from fastapi import APIRouter, Depends, HTTPException, Query from src.api.deps import require_admin from src.api.schemas import EvalSummaryOut, SettleRequest @@ -30,6 +30,6 @@ async def settle(req: SettleRequest, db: AsyncSession = Depends(get_db)): @router.get("/eval/summary", response_model=EvalSummaryOut, dependencies=[Depends(require_admin)]) -async def eval_summary(): - """提供商/模型准确率对比。""" - return await get_eval_summary() +async def eval_summary(limit: int = Query(1000, ge=1, le=10000, description="最大评估条数")): + """提供商/模型准确率对比。P3-4: 默认评估最近 1000 条,可通过 limit 调整。""" + return await get_eval_summary(limit=limit) diff --git a/src/core/log_buffer.py b/src/core/log_buffer.py index 51e50db..4cefe7b 100644 --- a/src/core/log_buffer.py +++ b/src/core/log_buffer.py @@ -6,11 +6,9 @@ from __future__ import annotations import logging -import threading from collections import deque _BUFFER: deque[dict] = deque(maxlen=2000) -_LOCK = threading.Lock() _LEVEL_ORDER = {"DEBUG": 10, "INFO": 20, "WARNING": 30, "ERROR": 40, "CRITICAL": 50} @@ -31,8 +29,7 @@ class MemoryLogHandler(logging.Handler): "logger": record.name, "message": self.format(record), } - with _LOCK: - _BUFFER.append(entry) + _BUFFER.append(entry) except Exception: # noqa: BLE001 日志采集绝不影响业务 self.handleError(record) @@ -54,8 +51,7 @@ def get_entries( """按条件查询缓冲日志,最新在前。""" min_no = _LEVEL_ORDER.get((min_level or "").upper(), 0) kw = (keyword or "").strip().lower() - with _LOCK: - items = list(_BUFFER) + items = list(_BUFFER) items.reverse() out: list[dict] = [] for e in items: diff --git a/src/db/models.py b/src/db/models.py index 8c04f5a..5a91184 100644 --- a/src/db/models.py +++ b/src/db/models.py @@ -4,6 +4,7 @@ from __future__ import annotations from datetime import date, datetime, timezone from sqlalchemy import ( + BigInteger, Boolean, CheckConstraint, Date, @@ -131,6 +132,10 @@ class MatchStats(Base): source_record_id: Mapped[str | None] = mapped_column(String(100)) retrieved_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) available_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + # xG 数据血缘:单独追踪 xG 字段的来源与更新时间(xG 可能独立于其他统计被更新) + xg_source: Mapped[str | None] = mapped_column(String(30)) + xg_updated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + xg_source_record_id: Mapped[str | None] = mapped_column(String(100)) match: Mapped[Match] = relationship(back_populates="stats") @@ -228,3 +233,92 @@ class AppSetting(Base): key: Mapped[str] = mapped_column(String(100), primary_key=True) value: Mapped[str] = mapped_column(Text, nullable=False) updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow, onupdate=_utcnow) + + +# ── 数据管线基础设施(对应迁移 0008) ────────────────────────────────── +# Bronze 层、死信、质量监控、血缘追踪 4 张表。 + + +class RawEvent(Base): + """Bronze 层:采集到的原始事件存档,便于重放与审计。""" + __tablename__ = "raw_events" + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True) + source_system: Mapped[str] = mapped_column(String(50), nullable=False) + source_record_id: Mapped[str] = mapped_column(String(100), nullable=False) + raw_payload: Mapped[dict] = mapped_column(JSONB, nullable=False) + ingested_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) + ingest_batch_id: Mapped[str | None] = mapped_column(String(36)) + + __table_args__ = ( + UniqueConstraint("source_system", "source_record_id", name="uq_raw_event"), + Index("ix_raw_event_batch", "ingest_batch_id"), + ) + + +class IngestFailure(Base): + """采集失败死信:记录失败原因、重试次数与下次重试时间。""" + __tablename__ = "ingest_failures" + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True) + source_system: Mapped[str] = mapped_column(String(50), nullable=False) + entity_type: Mapped[str] = mapped_column(String(50), nullable=False) + source_record_id: Mapped[str | None] = mapped_column(String(100)) + error_type: Mapped[str] = mapped_column(String(50), nullable=False) + error_detail: Mapped[str | None] = mapped_column(Text) + raw_payload: Mapped[dict | None] = mapped_column(JSONB) + retry_count: Mapped[int] = mapped_column(Integer, server_default="0") + next_retry_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + status: Mapped[str] = mapped_column(String(20), server_default="pending") + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) + resolved_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + + __table_args__ = ( + Index("ix_ingest_failure_status", "status", "next_retry_at"), + CheckConstraint( + "status IN ('pending', 'retrying', 'resolved', 'abandoned')", + name="ck_ingest_failure_status", + ), + ) + + +class DataQualityCheck(Base): + """数据质量监控:记录每次质量检查的结果。""" + __tablename__ = "data_quality_checks" + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True) + check_name: Mapped[str] = mapped_column(String(100), nullable=False) + entity_type: Mapped[str] = mapped_column(String(50), nullable=False) + entity_id: Mapped[int | None] = mapped_column(Integer) + expected_value: Mapped[float | None] = mapped_column(Float) + actual_value: Mapped[float] = mapped_column(Float, nullable=False) + passed: Mapped[bool] = mapped_column(Boolean, nullable=False) + severity: Mapped[str] = mapped_column(String(10), server_default="warning") + detail: Mapped[dict | None] = mapped_column(JSONB) + checked_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) + + __table_args__ = ( + Index("ix_dqc_checked_at", "checked_at"), + Index("ix_dqc_entity", "entity_type", "entity_id"), + ) + + +class DataLineage(Base): + """ETL 血缘追踪:记录从源到目标的转换过程。""" + __tablename__ = "data_lineage" + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True) + source_system: Mapped[str] = mapped_column(String(50), nullable=False) + source_record_id: Mapped[str] = mapped_column(String(100), nullable=False) + target_table: Mapped[str] = mapped_column(String(50), nullable=False) + target_id: Mapped[int | None] = mapped_column(Integer) + transform_name: Mapped[str] = mapped_column(String(50), nullable=False) + transform_detail: Mapped[dict | None] = mapped_column(JSONB) + batch_id: Mapped[str | None] = mapped_column(String(36)) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) + + __table_args__ = ( + Index("ix_lineage_source", "source_system", "source_record_id"), + Index("ix_lineage_target", "target_table", "target_id"), + Index("ix_lineage_batch", "batch_id"), + ) diff --git a/src/llm/agents/orchestrator.py b/src/llm/agents/orchestrator.py index 069c121..bcf6129 100644 --- a/src/llm/agents/orchestrator.py +++ b/src/llm/agents/orchestrator.py @@ -30,6 +30,11 @@ from src.llm.provider import LLMProvider, get_default_provider logger = logging.getLogger(__name__) +# P3-2: agent provider 配置缓存(TTL 60s),避免每次 _agent_provider 都多次查 DB +_AGENT_PROVIDER_CACHE: dict[str, tuple[float, LLMProvider]] = {} +_AGENT_PROVIDER_CACHE_TTL = 60.0 + + # ── 5 个专家 agent 定义 ── # A=近期状态 B=攻防数据 C=主客因素 D=阵容完整性 E=历史交锋 SPECIALIST_SPECS: list[AgentSpec] = [ @@ -103,7 +108,16 @@ async def _agent_provider(agent_id: str, *, tier: str) -> LLMProvider: 覆盖优先级: 模型: AGENT_MODEL_{ID}(运行时) → 层级默认(LLM_SPECIALIST/AGGREGATOR_MODEL) → 全局 LLM_MODEL 地址/密钥: AGENT_BASE_URL_{ID} / AGENT_API_KEY_{ID}(运行时) → 全局 LLM_BASE_URL / LLM_API_KEY + + P3-2: 结果缓存 60 秒,避免每次预测都多次查询运行时配置 DB。 """ + cache_key = f"{agent_id}:{tier}" + cached = _AGENT_PROVIDER_CACHE.get(cache_key) + if cached is not None: + ts, provider = cached + if time.time() - ts < _AGENT_PROVIDER_CACHE_TTL: + return provider + pfx = f"AGENT_{agent_id.upper()}_" p = await get_default_provider() tier_model = settings.LLM_SPECIALIST_MODEL if tier == "specialist" else settings.LLM_AGGREGATOR_MODEL @@ -118,6 +132,11 @@ async def _agent_provider(agent_id: str, *, tier: str) -> LLMProvider: key = await get_runtime_value(f"{pfx}API_KEY") if key: p.api_key = key + + _AGENT_PROVIDER_CACHE[cache_key] = (time.time(), p) + # 简单淘汰:超过 20 条时清空(60s TTL 下不会累积太多) + if len(_AGENT_PROVIDER_CACHE) > 20: + _AGENT_PROVIDER_CACHE.clear() return p diff --git a/src/llm/eval.py b/src/llm/eval.py index 8f8b63b..609f920 100644 --- a/src/llm/eval.py +++ b/src/llm/eval.py @@ -3,7 +3,7 @@ from __future__ import annotations import logging -from sqlalchemy import select +from sqlalchemy import func, select from src.db.models import Prediction from src.db.unit_of_work import get_uow @@ -32,12 +32,22 @@ def _actual_1x2(home: int, away: int) -> str: return "2" -async def get_eval_summary() -> dict: - """按 provider × 模型聚合评估。""" +async def get_eval_summary(limit: int = 1000) -> dict: + """按 provider × 模型聚合评估。 + + P3-4: 默认限制评估最近 1000 条已结算预测,避免全表加载导致内存压力。 + 可通过 eval 路由的 query 参数调整。 + """ async with get_uow() as session: + # 先统计全量已结算数,用于前端展示"共 X 条,评估 Y 条" + total_settled = (await session.execute( + select(func.count()).where(Prediction.settled == True) + )).scalar_one() stmt = ( select(Prediction) .where(Prediction.settled == True) + .order_by(Prediction.id.desc()) + .limit(limit) ) result = await session.execute(stmt) rows = list(result.scalars().all()) @@ -76,4 +86,4 @@ async def get_eval_summary() -> dict: "avg_score_rmse": round(avg_err, 2) if avg_err is not None else None, "avg_subjective_confidence": round(avg_conf, 2) if avg_conf is not None else None, }) - return {"summary": summary} + return {"summary": summary, "total_settled": total_settled, "evaluated": len(rows)} diff --git a/src/llm/predict.py b/src/llm/predict.py index 614cfe7..ddeb6ec 100644 --- a/src/llm/predict.py +++ b/src/llm/predict.py @@ -8,7 +8,6 @@ import time from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path -from threading import Lock from src.core.config import settings from sqlalchemy import select @@ -25,6 +24,7 @@ _PROMPT_DIR = Path(__file__).resolve().parent / "prompts" # ── LLM 响应缓存(match+provider+model+version → 结果) ── _CACHE_TTL_SEC = 300 # 5 分钟 +_CACHE_MAX_SIZE = 200 # P3-1: 有上限,避免长期运行内存无限增长 # P1-5: 缓存仅在 asyncio 协程内同步访问(dict 操作 GIL 原子),无需 threading.Lock。 # 删除 _cache_lock,避免同步锁阻塞事件循环;dict 的 get/set 在 CPython 下原子。 _cache: dict[str, tuple[float, PredictResult]] = {} @@ -56,6 +56,10 @@ def _set_cached(match_id: int, provider: str, model: str, version: str, tpl_hash # P1-5: 无锁写入。同上,dict set 原子。 key = _cache_key(match_id, provider, model, version, tpl_hash) _cache[key] = (time.time(), result) + # P3-1: 超过上限时淘汰最旧条目(按时间戳排序) + if len(_cache) > _CACHE_MAX_SIZE: + oldest_key = min(_cache, key=lambda k: _cache[k][0]) + _cache.pop(oldest_key, None) def clear_prompt_cache() -> None: diff --git a/src/llm/validation.py b/src/llm/validation.py index 571a4cc..6a81b27 100644 --- a/src/llm/validation.py +++ b/src/llm/validation.py @@ -75,7 +75,7 @@ class PredictionOutputSchema(BaseModel): ): self.alt_pred_home_goals = None self.alt_pred_away_goals = None - return self"" + return self @field_validator("pred_1x2") @classmethod