"""搜索服务封装""" from app.database import get_connection, search_fts def search(config, query: str, page: int = 1, page_size: int = 20) -> dict: """全文搜索,返回带高亮摘要的分页结果""" conn = get_connection(config.db_path) try: limit = page * page_size + page_size results = search_fts(conn, query, limit=limit) # 提取摘要(包含关键词的上下文) snippet_length = 100 items = [] for r in results: body = _get_body(conn, r["id"]) snippet = _extract_snippet(body, query, snippet_length) items.append({ "id": r["id"], "body_snippet": snippet, "difficulty": r.get("difficulty", 0.5), "question_type": r.get("question_type", ""), "import_type": r.get("import_type", ""), "chapters": r.get("chapters", []), }) total = len(items) start = (page - 1) * page_size end = start + page_size return { "items": items[start:end], "total": total, "page": page, "page_size": page_size, } finally: conn.close() def _get_body(conn, problem_id: str) -> str: """获取题目的 FTS body""" row = conn.execute( "SELECT body FROM problems_fts WHERE problem_id = ?", (problem_id,) ).fetchone() return row["body"] if row else "" def _extract_snippet(text: str, query: str, length: int = 100) -> str: """从文本中提取包含查询关键词的摘要""" if not text: return "" # 查找关键词位置 idx = text.lower().find(query.lower()) if idx == -1: # 尝试部分匹配 for word in query.split(): idx = text.lower().find(word.lower()) if idx != -1: break if idx == -1: # 无匹配,返回开头 return text[:length] + ("..." if len(text) > length else "") # 提取上下文 start = max(0, idx - length // 2) end = min(len(text), idx + length // 2) snippet = text[start:end] prefix = "..." if start > 0 else "" suffix = "..." if end < len(text) else "" return f"{prefix}{snippet}{suffix}"