| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677 |
- """搜索服务封装"""
- 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}"
|
