fix: deduplicate knowledge content search results

ai_knowledge_base_v2/apps/backend/app/api/admin_knowledge_lifecycle.py

@@ -3,7 +3,7 @@ from __future__ import annotations
 import json
 
 from fastapi import APIRouter, Depends, HTTPException, Query, status
-from sqlalchemy import func, literal, or_, select, union_all
+from sqlalchemy import and_, func, literal, or_, select, union_all
 from sqlalchemy.orm import Session
 
 from app.core.database import get_db
@@ -290,9 +290,43 @@ def content_search(
         )
     )
     search = union_all(section_query, card_query, chunk_query).subquery()
-    total = db.scalar(select(func.count()).select_from(search)) or 0
+    grouped = (
+        select(
+            search.c.knowledge_id,
+            search.c.version_id,
+            search.c.section_id,
+            func.max(search.c.knowledge_status).label("knowledge_status"),
+            func.min(search.c.sort_order).label("sort_order"),
+            func.min(search.c.item_rank).label("item_rank"),
+            func.min(search.c.item_id).label("item_id"),
+        )
+        .group_by(search.c.knowledge_id, search.c.version_id, search.c.section_id)
+        .subquery()
+    )
+    total = db.scalar(select(func.count()).select_from(grouped)) or 0
+    page_groups = (
+        select(grouped.c.knowledge_id, grouped.c.version_id, grouped.c.section_id)
+        .order_by(
+            grouped.c.knowledge_status.desc(),
+            grouped.c.knowledge_id.desc(),
+            grouped.c.sort_order.asc(),
+            grouped.c.item_rank.asc(),
+            grouped.c.item_id.asc(),
+        )
+        .offset((page - 1) * pageSize)
+        .limit(pageSize)
+        .subquery()
+    )
     rows = db.execute(
         select(search)
+        .join(
+            page_groups,
+            and_(
+                page_groups.c.knowledge_id == search.c.knowledge_id,
+                page_groups.c.version_id == search.c.version_id,
+                page_groups.c.section_id == search.c.section_id,
+            ),
+        )
         .order_by(
             search.c.knowledge_status.desc(),
             search.c.knowledge_id.desc(),
@@ -300,8 +334,6 @@ def content_search(
             search.c.item_rank.asc(),
             search.c.item_id.asc(),
         )
-        .offset((page - 1) * pageSize)
-        .limit(pageSize)
     ).mappings().all()
     section_ids = [int(row["section_id"]) for row in rows if row["section_id"]]
     chunk_section_ids = set()
@@ -309,8 +341,8 @@ def content_search(
         chunk_section_ids = set(
             db.scalars(select(KnowledgeChunk.section_id).where(KnowledgeChunk.section_id.in_(section_ids))).all()
         )
-    items = [_content_search_item(row, keyword, int(row["section_id"]) in chunk_section_ids) for row in rows]
-    return api_success(page_result(items, total=total, page=page, page_size=pageSize))
+    merged = _merge_content_search_rows(rows, keyword, chunk_section_ids)
+    return api_success(page_result(merged, total=total, page=page, page_size=pageSize))
 
 
 @router.get("/knowledge/{knowledge_id}/sync-jobs")
@@ -610,6 +642,25 @@ def _escape_like(value: str) -> str:
     return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
 
 
+def _merge_content_search_rows(rows, keyword: str, chunk_section_ids: set[int]) -> list[dict]:
+    grouped: dict[tuple[int, int, int], dict] = {}
+    for row in rows:
+        key = (int(row["knowledge_id"]), int(row["version_id"]), int(row["section_id"]))
+        item_type = str(row["item_type"])
+        current = grouped.get(key)
+        if current is None:
+            current = _content_search_item(row, keyword, int(row["section_id"]) in chunk_section_ids)
+            current["itemTypes"] = []
+            grouped[key] = current
+        if item_type not in current["itemTypes"]:
+            current["itemTypes"].append(item_type)
+        for matched in _matched_in(str(row["title"] or ""), str(row["content"] or ""), keyword):
+            if matched not in current["matchedIn"]:
+                current["matchedIn"].append(matched)
+        current["itemType"] = _primary_item_type(current["itemTypes"])
+    return list(grouped.values())
+
+
 def _content_search_item(row, keyword: str, has_chunks: bool) -> dict:
     title = str(row["title"] or "")
     content = str(row["content"] or "")
@@ -637,12 +688,20 @@ def _content_search_item(row, keyword: str, has_chunks: bool) -> dict:
         "title": title,
         "snippet": _snippet(f"{title}\n{content}", keyword),
         "matchedIn": _matched_in(title, content, keyword),
+        "itemTypes": [row["item_type"]],
         "hasChunks": has_chunks,
         "isCurrentVersion": True,
         "canAgentUse": is_active_current and has_chunks,
     }
 
 
+def _primary_item_type(item_types: list[str]) -> str:
+    for item_type in ("section", "card", "chunk"):
+        if item_type in item_types:
+            return item_type
+    return item_types[0] if item_types else "section"
+
+
 def _matched_in(title: str, content: str, keyword: str) -> list[str]:
     result: list[str] = []
     lowered = keyword.lower()

ai_knowledge_base_v2/apps/backend/tests/test_knowledge_admin.py

@@ -325,8 +325,10 @@ def test_content_search_finds_current_sections_cards_and_chunks():
         response = content_search("静水流深", includeClosed=True, page=1, pageSize=10, db=db, current_admin=_admin())
 
         data = response["data"]
-        assert data["total"] == 3
-        assert {item["itemType"] for item in data["items"]} == {"section", "card", "chunk"}
+        assert data["total"] == 1
+        assert len(data["items"]) == 1
+        assert data["items"][0]["itemType"] == "section"
+        assert data["items"][0]["itemTypes"] == ["section", "card", "chunk"]
         assert data["items"][0]["sectionKey"] == "S0098"
         assert data["items"][0]["hasChunks"] is True
         assert data["items"][0]["canAgentUse"] is True