UT拦截统计

.env.example

@@ -63,3 +63,6 @@ AI_ANALYZE_RATE_LIMIT_WINDOW_SECONDS=60
 AI_ANALYZE_RATE_LIMIT_MAX_REQUESTS=10
 # dt-report -> AIFA 请求超时（秒）
 AI_ANALYZE_TIMEOUT_SECONDS=180
+
+# UT 门禁 Jenkins 上报：与 Jenkins Credentials 注入值一致；留空则 POST /api/v1/ut-gate-runs 返回 401
+UT_GATE_INTEGRATION_TOKEN=

backend/api/router.py

@@ -1,6 +1,6 @@
 from fastapi import APIRouter
 
-from backend.api.v1 import auth, dashboard, overview, history, analysis, cases, report, notification, admin
+from backend.api.v1 import auth, dashboard, overview, history, analysis, cases, report, notification, admin, ut_gate_run
 
 api_router = APIRouter(prefix="/api/v1")
 
@@ -13,3 +13,4 @@
 api_router.include_router(report.router)
 api_router.include_router(notification.router)
 api_router.include_router(admin.router)
+api_router.include_router(ut_gate_run.router)

backend/api/v1/ut_gate_run.py

@@ -0,0 +1,78 @@
+import logging
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from fastapi.responses import JSONResponse
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from backend.core.database import get_db
+from backend.core.dependencies import get_current_user, verify_ut_gate_integration_token
+from backend.schemas.common import PageResponse
+from backend.schemas.ut_gate_run import UtGateRunCreate, UtGateRunItem, UtGateRunQuery
+from backend.services.ut_gate_run_service import UtGateIdempotencyConflict, create_ut_gate_run, list_ut_gate_runs
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/ut-gate-runs", tags=["UT门禁上报"])
+
+
+@router.get(
+    "",
+    response_model=PageResponse[UtGateRunItem],
+    summary="分页查询 UT 门禁上报记录",
+    description="筛选 `reported_at` 使用 `start_time`/`end_time`（与 History 批次 `start_time` 无关）。规约见 `spec/17_ut_gate_runs_get_api_spec.md`。",
+)
+async def get_ut_gate_runs(
+    query: UtGateRunQuery = Depends(),
+    db: AsyncSession = Depends(get_db),
+    _payload: dict = Depends(get_current_user),
+):
+    rows, total = await list_ut_gate_runs(db, query)
+    return PageResponse(
+        items=[UtGateRunItem.model_validate(r) for r in rows],
+        total=total,
+        page=query.page,
+        page_size=query.page_size,
+    )
+
+
+@router.post(
+    "",
+    response_model=UtGateRunItem,
+    response_model_exclude_none=False,
+    responses={
+        200: {"description": "幂等键已存在且内容一致（spec/16 §5.2）"},
+        201: {"description": "新建记录"},
+        409: {"description": "幂等键已存在且请求内容不一致"},
+    },
+    summary="上报 UT 门禁单次构建结果",
+    description="实现规约见 `spec/16_ut_gate_report_post_api_spec.md`。",
+)
+async def post_ut_gate_run(
+    body: UtGateRunCreate,
+    db: AsyncSession = Depends(get_db),
+    _: None = Depends(verify_ut_gate_integration_token),
+):
+    try:
+        row, http_status = await create_ut_gate_run(db, body)
+    except UtGateIdempotencyConflict:
+        logger.warning(
+            "UT 门禁上报幂等冲突 idempotency_key=%s job_name=%s build_number=%s",
+            body.idempotency_key,
+            body.job_name,
+            body.build_number,
+        )
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail="幂等键已存在且请求内容不一致",
+        )
+
+    logger.info(
+        "UT 门禁上报成功 http_status=%s id=%s idempotency_key=%s job_name=%s build_number=%s",
+        http_status,
+        row.id,
+        body.idempotency_key,
+        body.job_name,
+        body.build_number,
+    )
+    payload = UtGateRunItem.model_validate(row).model_dump(mode="json")
+    return JSONResponse(status_code=http_status, content=payload)

backend/core/config.py

@@ -43,6 +43,9 @@ class Settings(BaseSettings):
     # dt-report -> AIFA 单次调用超时（秒）
     AI_ANALYZE_TIMEOUT_SECONDS: int = 180
 
+    # UT 门禁 Jenkins 上报：固定集成 Token（Authorization: Bearer），与 Jenkins Credentials 一致；空则拒绝上报（401）
+    UT_GATE_INTEGRATION_TOKEN: str = ""
+
     # LDAP 域登录（LDAP_HOST 留空则使用 MVP 密码模式）
     LDAP_HOST: str = ""
     LDAP_PORT: int = 389

backend/core/dependencies.py

@@ -1,12 +1,20 @@
+import hmac
+import logging
+from typing import Optional
+
 from fastapi import Depends, HTTPException, status
-from fastapi.security import HTTPBearer
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from backend.core.config import settings
 from backend.core.database import get_db
 from backend.core.security import verify_token
 from backend.services.auth_service import get_user_role
 
+logger = logging.getLogger(__name__)
+
 bearer_scheme = HTTPBearer(auto_error=False)
+_ut_gate_bearer = HTTPBearer(auto_error=False)
 
 
 async def get_db_session() -> AsyncSession:  # type: ignore[misc]
@@ -34,3 +42,36 @@ async def require_apply_failure_reason_permission(
     if role not in ("user", "admin"):
         raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="无权限执行该操作")
     return payload
+
+
+async def verify_ut_gate_integration_token(
+    credentials: Optional[HTTPAuthorizationCredentials] = Depends(_ut_gate_bearer),
+) -> None:
+    """
+    UT 门禁 Jenkins 上报专用：Bearer 与 UT_GATE_INTEGRATION_TOKEN 一致（spec/16 §3）。
+    Token 未配置或非 Bearer 时返回 401；不在日志中输出 Token。
+    """
+    expected = (settings.UT_GATE_INTEGRATION_TOKEN or "").strip()
+    if not expected:
+        logger.warning("UT 门禁上报被拒绝：UT_GATE_INTEGRATION_TOKEN 未配置")
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="UT 门禁上报未启用或密钥未配置",
+        )
+    if credentials is None or (credentials.scheme or "").lower() != "bearer":
+        logger.warning("UT 门禁上报鉴权失败：缺少或非法的 Authorization")
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="无效或不存在的认证信息",
+        )
+    received = credentials.credentials or ""
+    try:
+        ok = hmac.compare_digest(received.encode("utf-8"), expected.encode("utf-8"))
+    except ValueError:
+        ok = False
+    if not ok:
+        logger.warning("UT 门禁上报鉴权失败：Token 不匹配")
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="认证失败",
+        )

backend/models/__init__.py

@@ -10,6 +10,7 @@
 from backend.models.sys_audit_log import SysAuditLog
 from backend.models.report_snapshot import ReportSnapshot
 from backend.models.history_search_template import HistorySearchTemplate
+from backend.models.ut_gate_run import UtGateRun
 
 __all__ = [
     "Base",
@@ -24,4 +25,5 @@
     "SysAuditLog",
     "ReportSnapshot",
     "HistorySearchTemplate",
+    "UtGateRun",
 ]

backend/models/ut_gate_run.py

@@ -0,0 +1,45 @@
+from datetime import datetime
+from typing import Optional
+
+from sqlalchemy import Boolean, DateTime, Index, Integer, String, UniqueConstraint, text
+from sqlalchemy.dialects.mysql import BIGINT, INTEGER
+from sqlalchemy.orm import Mapped, mapped_column
+
+from backend.models.base import Base
+
+
+class UtGateRun(Base):
+    """Jenkins UT 门禁上报记录，与 database/V1.1.2__create_ut_gate_run.sql 一致。"""
+
+    __tablename__ = "ut_gate_run"
+    __table_args__ = (
+        UniqueConstraint("idempotency_key", name="uk_idempotency"),
+        Index("idx_created_at", "created_at"),
+        Index("idx_mr_url_created", "mr_url", "created_at", mysql_length={"mr_url": 191}),
+        Index("idx_is_intercepted_created", "is_intercepted", "created_at"),
+        Index("idx_job_build", "job_name", "build_number"),
+        {"extend_existing": True},
+    )
+
+    id: Mapped[int] = mapped_column(BIGINT(unsigned=True), primary_key=True, autoincrement=True, comment="主键")
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime, nullable=False, server_default=text("CURRENT_TIMESTAMP"), comment="记录创建时间"
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime,
+        nullable=False,
+        server_default=text("CURRENT_TIMESTAMP"),
+        server_onupdate=text("CURRENT_TIMESTAMP"),
+        comment="更新时间",
+    )
+    reported_at: Mapped[datetime] = mapped_column(
+        DateTime, nullable=False, server_default=text("CURRENT_TIMESTAMP"), comment="门禁结束上报时间"
+    )
+    jenkins_base_url: Mapped[Optional[str]] = mapped_column(String(512), nullable=True, comment="Jenkins 根 URL")
+    job_name: Mapped[str] = mapped_column(String(256), nullable=False, comment="Job 名称")
+    build_number: Mapped[int] = mapped_column(INTEGER(unsigned=True), nullable=False, comment="构建号")
+    build_url: Mapped[Optional[str]] = mapped_column(String(1024), nullable=True, comment="本次构建页 URL")
+    mr_url: Mapped[Optional[str]] = mapped_column(String(1024), nullable=True, comment="MR 页面完整 URL")
+    idempotency_key: Mapped[str] = mapped_column(String(128), nullable=False, comment="幂等键")
+    is_intercepted: Mapped[bool] = mapped_column(Boolean, nullable=False, comment="是否拦截到失败用例")
+    ut_exit_code: Mapped[Optional[int]] = mapped_column(Integer, nullable=True, comment="cargo make test 退出码")

backend/schemas/ut_gate_run.py

@@ -0,0 +1,158 @@
+from datetime import datetime, time, timezone
+import re
+from typing import Optional
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
+
+from backend.schemas.common import PageRequest
+
+
+class UtGateRunCreate(BaseModel):
+    """POST /api/v1/ut-gate-runs 请求体；未知字段忽略（spec/16 §4.1）。"""
+
+    model_config = ConfigDict(extra="ignore")
+
+    idempotency_key: str = Field(..., min_length=1, max_length=128)
+    job_name: str = Field(..., min_length=1, max_length=256)
+    build_number: int = Field(...)
+    is_intercepted: bool
+    ut_exit_code: Optional[int] = None
+    build_url: Optional[str] = Field(None, max_length=1024)
+    jenkins_base_url: Optional[str] = Field(None, max_length=512)
+    mr_url: Optional[str] = Field(None, max_length=1024)
+
+    @field_validator("idempotency_key", "job_name")
+    @classmethod
+    def strip_required_strings(cls, v: str) -> str:
+        s = v.strip()
+        if not s:
+            raise ValueError("不能为空或仅空白")
+        return s
+
+    @field_validator("build_url", "jenkins_base_url", "mr_url", mode="before")
+    @classmethod
+    def strip_optional_urls(cls, v: Optional[str]) -> Optional[str]:
+        if v is None:
+            return None
+        if not isinstance(v, str):
+            return v
+        s = v.strip()
+        return s if s else None
+
+    @field_validator("build_number")
+    @classmethod
+    def build_number_unsigned_int(cls, v: int) -> int:
+        if v < 0 or v > 4294967295:
+            raise ValueError("build_number 须在 0～4294967295（INT UNSIGNED）范围内")
+        return v
+
+
+class UtGateRunItem(BaseModel):
+    """单条 ut_gate_run 响应，与表字段一致（spec/16 §6）。"""
+
+    id: int
+    created_at: datetime
+    updated_at: datetime
+    reported_at: datetime
+    jenkins_base_url: Optional[str] = None
+    job_name: str
+    build_number: int
+    build_url: Optional[str] = None
+    mr_url: Optional[str] = None
+    idempotency_key: str
+    is_intercepted: bool
+    ut_exit_code: Optional[int] = None
+
+    model_config = {"from_attributes": True}
+
+
+_DATE_ONLY_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$")
+
+
+def _is_date_only_string(s: str) -> bool:
+    return bool(_DATE_ONLY_RE.match(s.strip()))
+
+
+class UtGateRunQuery(PageRequest):
+    """GET /api/v1/ut-gate-runs 查询参数（spec/17）。"""
+
+    model_config = ConfigDict(extra="ignore")
+
+    page_size: int = Field(20, ge=1, le=100, description="每页条数，最大 100")
+    start_time: Optional[str] = Field(
+        None,
+        description="reported_at 下限（闭区间）；YYYY-MM-DD 或 ISO8601，须与 end_time 同为日期或同为带时间格式",
+    )
+    end_time: Optional[str] = Field(
+        None,
+        description="reported_at 上限（闭区间）；YYYY-MM-DD 或 ISO8601",
+    )
+    is_intercepted: Optional[bool] = Field(None, description="是否拦截到（true/false）；省略则不过滤")
+    mr_url: Optional[str] = Field(None, max_length=1024, description="mr_url 精确匹配（与 mr_url_contains 互斥）")
+    mr_url_contains: Optional[str] = Field(None, max_length=200, description="mr_url 子串匹配（LIKE 转义）")
+    job_name_contains: Optional[str] = Field(None, max_length=200, description="job_name 子串匹配（LIKE 转义）")
+    sort_field: Optional[str] = Field(
+        None,
+        description="排序列：reported_at（默认）、created_at、id",
+    )
+    sort_order: Optional[str] = Field(None, description="asc / desc，默认 desc")
+
+    parsed_reported_at_start: Optional[datetime] = Field(default=None, exclude=True)
+    parsed_reported_at_end: Optional[datetime] = Field(default=None, exclude=True)
+
+    @field_validator("mr_url", "mr_url_contains", "job_name_contains", mode="before")
+    @classmethod
+    def strip_optional_query_strings(cls, v: Optional[str]) -> Optional[str]:
+        if v is None:
+            return None
+        if not isinstance(v, str):
+            return v
+        s = v.strip()
+        return s if s else None
+
+    @model_validator(mode="after")
+    def validate_mr_mutual_and_times_and_sort(self) -> "UtGateRunQuery":
+        mu = self.mr_url
+        mc = self.mr_url_contains
+        if (mu or "").strip() and (mc or "").strip():
+            raise ValueError("mr_url 与 mr_url_contains 互斥，不能同时指定")
+
+        st_raw = self.start_time
+        en_raw = self.end_time
+        if (st_raw or "").strip() and (en_raw or "").strip():
+            if _is_date_only_string(st_raw) != _is_date_only_string(en_raw):
+                raise ValueError("start_time 与 end_time 须同为 YYYY-MM-DD 日期或同为带时间的 ISO8601")
+
+        def _parse_one(raw: Optional[str], *, is_end: bool) -> Optional[datetime]:
+            if raw is None or not str(raw).strip():
+                return None
+            s = str(raw).strip()
+            if _is_date_only_string(s):
+                d = datetime.strptime(s, "%Y-%m-%d").date()
+                if is_end:
+                    return datetime.combine(d, time(23, 59, 59))
+                return datetime.combine(d, time.min)
+            iso = s.replace("Z", "+00:00") if s.endswith("Z") else s
+            try:
+                dt = datetime.fromisoformat(iso)
+            except ValueError as e:
+                raise ValueError("时间格式无效，请使用 YYYY-MM-DD 或 ISO8601") from e
+            if dt.tzinfo is not None:
+                dt = dt.astimezone(timezone.utc).replace(tzinfo=None)
+            return dt
+
+        ps = _parse_one(st_raw, is_end=False)
+        pe = _parse_one(en_raw, is_end=True)
+        if ps is not None and pe is not None and ps > pe:
+            raise ValueError("start_time 不能晚于 end_time")
+
+        sf = (self.sort_field or "").strip() or None
+        if sf is not None and sf not in ("reported_at", "created_at", "id"):
+            raise ValueError("sort_field 仅支持 reported_at、created_at、id")
+        so = (self.sort_order or "").strip().lower() or "desc"
+        if so not in ("asc", "desc"):
+            raise ValueError("sort_order 仅支持 asc 或 desc")
+
+        object.__setattr__(self, "parsed_reported_at_start", ps)
+        object.__setattr__(self, "parsed_reported_at_end", pe)
+        return self

backend/services/schema_check_service.py

@@ -27,6 +27,7 @@
     "case_offline_type": "V1.0.8__create_case_offline_type.sql",
     "sys_audit_log": "V1.0.9__create_sys_audit_log.sql",
     "report_snapshot": "V1.1.0__create_report_snapshot.sql",
+    "ut_gate_run": "V1.1.2__create_ut_gate_run.sql",
 }
 
 TABLE_RE = re.compile(r"CREATE\s+TABLE\s+`([^`]+)`\s*\(", re.IGNORECASE)