feat(history): 优化工具栏搜索能力

backend/api/v1/history.py

@@ -193,6 +193,16 @@ async def get_history_list(
     code_branch: Optional[List[str]] = Query(None),    # 筛选代码分支（多选）
     failure_owner: Optional[List[str]] = Query(None),  # 筛选失败跟踪人（多选）
     failed_type: Optional[List[str]] = Query(None),    # 筛选失败原因（多选）
+    start_time_contains: Optional[str] = Query(None, max_length=200),
+    subtask_contains: Optional[str] = Query(None, max_length=200),
+    case_name_contains: Optional[str] = Query(None, max_length=200),
+    main_module_contains: Optional[str] = Query(None, max_length=200),
+    case_result_contains: Optional[str] = Query(None, max_length=200),
+    case_level_contains: Optional[str] = Query(None, max_length=200),
+    platform_contains: Optional[str] = Query(None, max_length=200),
+    code_branch_contains: Optional[str] = Query(None, max_length=200),
+    failure_owner_contains: Optional[str] = Query(None, max_length=200),
+    failed_type_contains: Optional[str] = Query(None, max_length=200),
     sort_field: Optional[str] = Query(None),           # 排序列
     sort_order: Optional[str] = Query(None),           # 排序方向：asc / desc
     # Depends(get_db) 是 FastAPI 的核心特性"依赖注入":
@@ -219,6 +229,16 @@ async def get_history_list(
         code_branch=code_branch,
         failure_owner=failure_owner,
         failed_type=failed_type,
+        start_time_contains=start_time_contains,
+        subtask_contains=subtask_contains,
+        case_name_contains=case_name_contains,
+        main_module_contains=main_module_contains,
+        case_result_contains=case_result_contains,
+        case_level_contains=case_level_contains,
+        platform_contains=platform_contains,
+        code_branch_contains=code_branch_contains,
+        failure_owner_contains=failure_owner_contains,
+        failed_type_contains=failed_type_contains,
         sort_field=sort_field,
         sort_order=sort_order,
     )

backend/schemas/history.py

@@ -12,7 +12,7 @@
 from typing import List, Optional
 
 # BaseModel: Pydantic 的基类，所有 Schema 都继承它以获得自动校验能力
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 
 # PageRequest: 我们自定义的分页请求基类（包含 page 和 page_size 字段）
 from backend.schemas.common import PageRequest
@@ -72,6 +72,17 @@ class HistoryQuery(PageRequest):
     code_branch: Optional[List[str]] = None      # 按代码分支筛选（多选）
     failure_owner: Optional[List[str]] = None    # 按失败跟踪人筛选（多选）
     failed_type: Optional[List[str]] = None      # 按失败原因筛选（多选）
+    # 子串筛选（与对应 IN 列表互斥；Service 层有非空 *_contains 时忽略同维度的 IN）
+    start_time_contains: Optional[str] = Field(None, max_length=200)
+    subtask_contains: Optional[str] = Field(None, max_length=200)
+    case_name_contains: Optional[str] = Field(None, max_length=200)
+    main_module_contains: Optional[str] = Field(None, max_length=200)
+    case_result_contains: Optional[str] = Field(None, max_length=200)
+    case_level_contains: Optional[str] = Field(None, max_length=200)
+    platform_contains: Optional[str] = Field(None, max_length=200)
+    code_branch_contains: Optional[str] = Field(None, max_length=200)
+    failure_owner_contains: Optional[str] = Field(None, max_length=200)
+    failed_type_contains: Optional[str] = Field(None, max_length=200)
     sort_field: Optional[str] = None             # 排序列（如 start_time, case_name）
     sort_order: Optional[str] = None             # 排序方向：asc / desc
 

backend/services/history_service.py

@@ -29,9 +29,34 @@
     "case_level", "analyzed", "platform", "code_branch", "created_at",
 }
 
+# LIKE … ESCAPE 使用单字符 `!`，避免反斜杠在 Python / 方言编译层被多重解释；与 _like_escape_literal 一致。
+_LIKE_ESCAPE_CHAR = "!"
+
+
+def _non_empty_str(value: Optional[str]) -> bool:
+    return value is not None and str(value).strip() != ""
+
+
+def _like_escape_literal(value: str) -> str:
+    return (
+        value.replace(_LIKE_ESCAPE_CHAR, _LIKE_ESCAPE_CHAR + _LIKE_ESCAPE_CHAR)
+        .replace("%", _LIKE_ESCAPE_CHAR + "%")
+        .replace("_", _LIKE_ESCAPE_CHAR + "_")
+    )
+
+
+def _like_substring(column, raw: Optional[str]) -> Optional[object]:
+    if not _non_empty_str(raw):
+        return None
+    inner = _like_escape_literal(str(raw).strip())
+    pattern = f"%{inner}%"
+    return column.like(pattern, escape=_LIKE_ESCAPE_CHAR)
+
 
 def _has_non_empty_case_name_filter(query: HistoryQuery) -> bool:
-    """Spec 08 §3.1.1：已选用例名且至少一项非空时，不注入默认 N 批。"""
+    """Spec 08 §3.1.1：已选用例名（IN 或子串）且有效时，不注入默认 N 批。"""
+    if _non_empty_str(query.case_name_contains):
+        return True
     if not query.case_name:
         return False
     for s in query.case_name:
@@ -40,6 +65,17 @@ def _has_non_empty_case_name_filter(query: HistoryQuery) -> bool:
     return False
 
 
+def _skip_default_start_time_injection(query: HistoryQuery) -> bool:
+    """已显式约束轮次（IN 或子串）或已选用例名（IN/子串）时，不注入默认最近 N 批。"""
+    if query.start_time:
+        return True
+    if _non_empty_str(query.start_time_contains):
+        return True
+    if _has_non_empty_case_name_filter(query):
+        return True
+    return False
+
+
 async def list_history(
     db: AsyncSession, query: HistoryQuery
 ) -> Tuple[
@@ -58,13 +94,13 @@ async def list_history(
 ]:
     """
     按 Spec 07：条件合并 + EXISTS 跨表筛选，禁止 JOIN、禁止结果集驱动。
-    按 Spec 08：未选 start_time 时注入默认最近 30 批；若已选非空 case_name 则不注入（§3.1.1，全时间范围）。
+    按 Spec 08：未选 start_time 时注入默认最近 30 批；若已传非空 start_time_contains 或已选非空 case_name（IN/子串）则不注入（显式筛选不叠默认批次）。
     1. 仅查 pipeline_history 主表，跨表条件通过 EXISTS 子查询
     2. 主查询完成后，根据当前页 (case_name, start_time, platform) 批量查 pfr 拼装 failure_owner、failed_type
     3. 根据当前页 main_module 批量查 ums_module_owner（及必要时 ums_email），拼装用例开发责任人展示串
     """
-    # ===== 第零步：未选 start_time 时注入默认最近 30 批（Spec 08）；§3.1.1 例外见 _has_non_empty_case_name_filter
-    if not query.start_time and not _has_non_empty_case_name_filter(query):
+    # ===== 第零步：未选 start_time 时注入默认最近 30 批（Spec 08）；显式轮次子串/用例名筛选时不注入
+    if not _skip_default_start_time_injection(query):
         default_batches_stmt = (
             select(ph.start_time)
             .where(ph.start_time.is_not(None))
@@ -82,34 +118,70 @@ async def list_history(
 
     # ===== 第一步：构建主表查询（无 JOIN）=====
     stmt = select(ph)
-    if query.start_time:
+    st_like = _like_substring(ph.start_time, query.start_time_contains)
+    if st_like is not None:
+        stmt = stmt.where(st_like)
+    elif query.start_time:
         stmt = stmt.where(ph.start_time.in_(query.start_time))
-    if query.subtask:
+    sub_like = _like_substring(ph.subtask, query.subtask_contains)
+    if sub_like is not None:
+        stmt = stmt.where(sub_like)
+    elif query.subtask:
         stmt = stmt.where(ph.subtask.in_(query.subtask))
-    if query.case_name:
+    cn_like = _like_substring(ph.case_name, query.case_name_contains)
+    if cn_like is not None:
+        stmt = stmt.where(cn_like)
+    elif query.case_name:
         stmt = stmt.where(ph.case_name.in_(query.case_name))
-    if query.main_module:
+    mm_like = _like_substring(ph.main_module, query.main_module_contains)
+    if mm_like is not None:
+        stmt = stmt.where(mm_like)
+    elif query.main_module:
         stmt = stmt.where(ph.main_module.in_(query.main_module))
-    if query.case_result:
+    cr_like = _like_substring(ph.case_result, query.case_result_contains)
+    if cr_like is not None:
+        stmt = stmt.where(cr_like)
+    elif query.case_result:
         stmt = stmt.where(ph.case_result.in_(query.case_result))
-    if query.case_level:
+    cl_like = _like_substring(ph.case_level, query.case_level_contains)
+    if cl_like is not None:
+        stmt = stmt.where(cl_like)
+    elif query.case_level:
         stmt = stmt.where(ph.case_level.in_(query.case_level))
     if query.analyzed:
         stmt = stmt.where(ph.analyzed.in_(query.analyzed))
-    if query.platform:
+    pl_like = _like_substring(ph.platform, query.platform_contains)
+    if pl_like is not None:
+        stmt = stmt.where(pl_like)
+    elif query.platform:
         stmt = stmt.where(ph.platform.in_(query.platform))
-    if query.code_branch:
+    cb_like = _like_substring(ph.code_branch, query.code_branch_contains)
+    if cb_like is not None:
+        stmt = stmt.where(cb_like)
+    elif query.code_branch:
         stmt = stmt.where(ph.code_branch.in_(query.code_branch))
     # 跨表筛选：EXISTS 子查询（Spec 4.2），执行键三字段精确匹配
-    if query.failure_owner or query.failed_type:
+    has_pfr_filters = (
+        query.failure_owner
+        or query.failed_type
+        or _non_empty_str(query.failure_owner_contains)
+        or _non_empty_str(query.failed_type_contains)
+    )
+    if has_pfr_filters:
         exists_conds = [
             pfr.case_name == ph.case_name,
             pfr.failed_batch == ph.start_time,
             pfr.platform == ph.platform,
         ]
-        if query.failed_type:
+        ft_like = _like_substring(pfr.failed_type, query.failed_type_contains)
+        if ft_like is not None:
+            exists_conds.append(ft_like)
+        elif query.failed_type:
             exists_conds.append(pfr.failed_type.in_(query.failed_type))
-        if query.failure_owner:
+        fo_like = _like_substring(pfr.owner, query.failure_owner_contains)
+        if fo_like is not None:
+            exists_conds.append(fo_like)
+        elif query.failure_owner:
             exists_conds.append(pfr.owner.in_(query.failure_owner))
         stmt = stmt.where(exists(select(1).select_from(pfr).where(and_(*exists_conds))))
 

docs/01_user_story_map.md

@@ -25,6 +25,7 @@
 
 * **Story 1.3: 多条件筛选与搜索**
 * **描述：** 作为开发人员，希望能够通过用例ID、模块名称、执行状态对结果进行组合筛选，以便从海量数据中过滤出想要观测分析的数据。
+* **补充（已实现）：** 详细执行历史支持各字符串维度 **IN 多选** 与 URL **`*_contains` 子串**（互斥）、下拉搜索后「全部」写入子串、子串以灰色 Tag 展示；规约见 `spec/07`、`spec/08`，与 `docs/02_prd.md` Story 1.3 一致。
 
 
 

docs/02_prd.md

@@ -93,8 +93,8 @@
 - **数据来源：** 批次级 → `pipeline_overview`（按 `batch` 过滤）；分组级 → `pipeline_overview`（按 `batch` + `subtask` 过滤）；用例级 → `pipeline_history`（按 `start_time` + `subtask` 过滤）。
 
 #### Story 1.3: 多条件筛选与搜索
-- **描述：** 作为开发人员，希望能够通过用例名称（`case_name`）、模块名称（`main_module` / `module`）、执行状态（`case_result`）、平台（`platform`）、批次（`start_time`）、是否已分析（`analyzed`）等字段对结果进行组合筛选，以便从海量数据中过滤出想要观测分析的数据。
-- **数据来源：** 主要查询 `pipeline_history` 表，可 JOIN `pipeline_failure_reason` 获取归因信息。
+- **描述：** 作为开发人员，希望能够通过用例名称（`case_name`）、模块名称（`main_module` / `module`）、执行状态（`case_result`）、平台（`platform`）、批次（`start_time`）、是否已分析（`analyzed`）等字段对结果进行组合筛选，以便从海量数据中过滤出想要观测分析的数据。详细执行历史页在 **IN 多选** 之外，对各字符串维度支持 **URL 参数 `*_contains` 子串筛选**（与对应维度 IN 互斥），下拉内可对候选项搜索并一键「全部」写入子串条件；子串条件在界面中以灰色标签展示（见 `spec/07`、`spec/08` 与 `docs/04_project_structure.md`）。
+- **数据来源：** 主要查询 `pipeline_history` 表；失败跟踪人、失败原因等跨表筛选由后端以 **EXISTS** 实现（禁止大表 JOIN），详见 `spec/07_history_filter_query_spec.md`。
 
 ---
 
@@ -283,10 +283,10 @@
 
 ### 5.4 详细执行历史列表
 
-- **数据源：** `pipeline_history`，可 LEFT JOIN `pipeline_failure_reason`
-- **列字段：** 批次、分组、用例名、主模块、执行结果、用例级别、负责人、是否已分析、平台、代码分支、创建时间
-- **筛选条件：** 批次、分组、用例名（模糊搜索）、主模块、执行结果、是否已分析、平台
-- **操作：** 点击行展开详情
+- **数据源：** `pipeline_history`；失败归因展示字段通过服务层批量查询 `pipeline_failure_reason` 拼装（**禁止**列表主查询与 `pipeline_failure_reason` 大结果集 JOIN，见 `spec/07`）。
+- **列字段：** 批次、分组、用例名、主模块、执行结果、用例级别、负责人（用例开发责任人展示）、是否已分析、平台、代码分支、创建时间
+- **筛选条件：** 批次、分组、用例名、主模块、执行结果、用例级别、是否已分析、平台、代码分支、失败跟踪人、失败原因等；各字符串维度支持 **多选 IN**（URL 多值）或 **子串 `*_contains`**（URL 单值，与同维 IN 互斥）；未选批次时列表默认最近 30 批，**已选用例名（IN 或 `case_name_contains`）且无批次时不注入**（全时间范围，见 `spec/08` §3.1.1、`spec/12` 钻取）。
+- **操作：** 点击行打开 Drawer 详情；用例名可链至钻取页
 
 ### 5.5 用例详情交互 (Detail Interaction)