53 lines
1.5 KiB
Python
53 lines
1.5 KiB
Python
"""Qdrant 向量存储数据模型 (M6)。"""
|
|
|
|
from typing import Any
|
|
|
|
from pydantic import BaseModel, Field
|
|
|
|
|
|
class SearchFilter(BaseModel):
|
|
"""可选检索过滤条件，全部为 AND 关系。"""
|
|
|
|
source_id: str | None = None
|
|
source_ids: list[str] | None = None
|
|
stock_codes: list[str] | None = Field(default=None, description="match any")
|
|
sentiment: str | None = None # positive / neutral / negative
|
|
importance_min: int | None = None # >= N
|
|
event_types: list[str] | None = None # match any
|
|
publish_date_from: str | None = None # YYYY-MM-DD
|
|
publish_date_to: str | None = None # YYYY-MM-DD
|
|
|
|
|
|
class SearchResult(BaseModel):
|
|
"""单条检索结果（含双语信息）。"""
|
|
|
|
url_hash: str
|
|
score: float
|
|
title: str = ""
|
|
title_zh: str = ""
|
|
url: str = ""
|
|
source_id: str = ""
|
|
publish_time: str = ""
|
|
events: list[dict[str, Any]] = Field(default_factory=list)
|
|
content_zh_preview: str = "" # 中文正文前 300 字
|
|
|
|
def short_summary(self) -> str:
|
|
codes = set()
|
|
for ev in self.events:
|
|
codes.update(ev.get("stock_codes", []))
|
|
codes_str = ",".join(sorted(codes)[:5]) or "-"
|
|
return (
|
|
f"[{self.source_id}] score={self.score:.4f} "
|
|
f"《{self.title_zh[:40] or self.title[:40]}》 {codes_str}"
|
|
)
|
|
|
|
|
|
class CollectionInfo(BaseModel):
|
|
"""Collection 概览信息。"""
|
|
|
|
name: str
|
|
exists: bool
|
|
vectors_count: int = 0
|
|
indexed_vectors_count: int | None = None
|
|
segments_count: int | None = None
|