feat:增加博客搜索功能

service/blog_elastic_service.py

@@ -1,7 +1,56 @@
 from datetime import datetime
+from typing import List
+
+from config.logging import logger
+
+from elasticsearch.helpers import bulk
+from sqlalchemy import select
+from sqlalchemy.orm import Session
 
 from config.elastic import es, BLOG_INDEX
-from schemas.blog_elastic import BlogElastic
+from models.blog import Blog, BlogCategory, BlogContent
+from schemas.blog_elastic import BlogElastic, BlogSearch
+
+
+def sync_all_blog(db: Session) -> bool:
+    query = select(
+        Blog.id,
+        Blog.title,
+        BlogCategory.name.label("category"),
+        BlogContent.content.label("content"),
+        Blog.create_time.label("createTime"),
+        Blog.update_time.label("updateTime"),
+    ).select_from(Blog).outerjoin(
+        BlogCategory, Blog.category_id == BlogCategory.id
+    ).outerjoin(
+        BlogContent, Blog.content_id == BlogContent.id
+    )
+    results = db.execute(query).fetchall()
+
+    actions = [
+        {
+            "_op_type": "index",  # 如果存在则更新，不存在则创建
+            "_index": BLOG_INDEX,
+            "_id": blog.id,
+            "_source": {
+                "title": blog.title,
+                "category": blog.category,
+                "content": blog.content.decode("utf-8"),
+                "createTime": blog.createTime.strftime("%Y-%m-%d %H:%M:%S"),
+                "updateTime": blog.updateTime.strftime("%Y-%m-%d %H:%M:%S")
+            }
+        }
+        for blog in results
+    ]
+
+    success_count, errors = bulk(es, actions=actions)
+
+    if errors:
+        logger.error(f"部分文档同步失败，成功同步 {success_count} 条")
+        return False
+
+    logger.info(f"成功同步 {success_count} 条博客到 Elasticsearch")
+    return True
 
 
 def add_blog_elastic(blog: BlogElastic) -> bool:
@@ -35,3 +84,55 @@ def delete_blog(blog_id: int) -> bool:
     es.delete(index=BLOG_INDEX, id=blog_id)
 
     return True
+
+
+def search_blog(keyword: str) -> List[BlogSearch]:
+    # 构造查询请求
+    body = {
+        "query": {
+            "multi_match": {
+                "query": keyword,
+                "fields": ["title^3", "content"],
+                "type": "phrase"
+            }
+        },
+        "highlight": {
+            "pre_tags": ["<mark>"],
+            "post_tags": ["</mark>"],
+            "fields": {
+                "title": {},
+                "content": {
+                    "fragment_size": 150,
+                    "number_of_fragments": 2,
+                    "no_match_size": 100
+                }
+            }
+        },
+        "_source": ["title", "content"]
+    }
+
+    response = es.search(index=BLOG_INDEX, body=body)
+
+    results = []
+    for hit in response["hits"]["hits"]:
+        highlight = hit.get("highlight", {})
+
+        # 合并所有高亮片段（标题+内容）
+        all_highlights = []
+        if "title" in highlight:
+            all_highlights.extend(highlight["title"])
+        if "content" in highlight:
+            all_highlights.extend(highlight["content"])
+
+        # 优先使用高亮内容，否则截取内容开头
+        displayed_content = " ".join(all_highlights) if all_highlights \
+            else hit["_source"].get("content", "")[:200] + "..."
+
+        results.append(BlogSearch(
+            id=hit["_id"],
+            title=hit["_source"]["title"],
+            content=displayed_content,
+            highlight=" | ".join(all_highlights)  # 合并所有高亮片段
+        ))
+
+    return results