b17786b57b
主要功能: - 图片上传时 OCR 内容去重（3个上传端点统一使用公共函数 _check_ocr_duplicate） - 图片管理 Tab：展示所有图片、手动删除、一键去重 - 搜索结果详情弹窗增加删除按钮（带确认弹窗） - 图片管理卡片点击查看详情（复用 showOcrDetailModal） - 搜索限制和 LLM 批量判断数量可通过网站设置 - MiniMax API 调用添加 reasoning_split=True - 企业微信机器人：WebSocket 长连接、图片搜索、配置化搜索数量 - 版本号升级至 1.2.0
124 lines
5.1 KiB
Python
124 lines
5.1 KiB
Python
"""
|
||
SQLAlchemy 基础模型 + 通用 Mixin
|
||
提供 DeclarativeBase、时间戳 Mixin 和公共序列化方法
|
||
兼容 PostgreSQL（pgvector）和 SQLite（本地开发）
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from datetime import datetime
|
||
from typing import Any, Dict
|
||
|
||
from sqlalchemy import DateTime, Float, Integer, String, Text, func
|
||
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
|
||
|
||
from app.database import IS_SQLITE
|
||
|
||
# SQLite 下不使用 pgvector，用 Text 存储向量（JSON 格式）
|
||
if IS_SQLITE:
|
||
VectorType = Text
|
||
else:
|
||
from pgvector.sqlalchemy import Vector
|
||
VectorType = Vector
|
||
|
||
|
||
class Base(DeclarativeBase):
|
||
"""所有 ORM 模型的基类"""
|
||
pass
|
||
|
||
|
||
class TimestampMixin:
|
||
"""
|
||
时间戳 Mixin，为模型自动添加 created_at / updated_at 字段。
|
||
"""
|
||
created_at: Mapped[datetime] = mapped_column(
|
||
DateTime(timezone=True),
|
||
server_default=func.now(),
|
||
comment="创建时间",
|
||
)
|
||
updated_at: Mapped[datetime] = mapped_column(
|
||
DateTime(timezone=True),
|
||
server_default=func.now(),
|
||
onupdate=func.now(),
|
||
comment="更新时间",
|
||
)
|
||
|
||
def to_dict(self) -> Dict[str, Any]:
|
||
"""将模型实例转换为字典"""
|
||
result: Dict[str, Any] = {}
|
||
for column in self.__table__.columns: # type: ignore[attr-defined]
|
||
value = getattr(self, column.name, None)
|
||
if isinstance(value, datetime):
|
||
value = value.isoformat()
|
||
elif hasattr(value, "tolist"):
|
||
value = value.tolist()
|
||
result[column.name] = value
|
||
return result
|
||
|
||
|
||
# ──────────────────────────── 业务模型 ────────────────────────────
|
||
|
||
class KnowledgePage(Base, TimestampMixin):
|
||
"""知识页面模型"""
|
||
__tablename__ = "knowledge_pages"
|
||
|
||
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
||
title: Mapped[str] = mapped_column(String(500), nullable=False, comment="页面标题")
|
||
content: Mapped[str] = mapped_column(Text, nullable=False, comment="页面正文内容")
|
||
source_file: Mapped[str | None] = mapped_column(String(500), nullable=True, comment="来源文件名")
|
||
course_name: Mapped[str | None] = mapped_column(String(200), nullable=True, comment="课程名称")
|
||
teacher_name: Mapped[str | None] = mapped_column(String(100), nullable=True, comment="讲师名称")
|
||
live_date: Mapped[str | None] = mapped_column(String(20), nullable=True, comment="直播日期")
|
||
page_number: Mapped[int | None] = mapped_column(Integer, nullable=True, comment="原始页码")
|
||
metadata_json: Mapped[str | None] = mapped_column(Text, nullable=True, comment="额外元数据")
|
||
|
||
|
||
class KnowledgeChunk(Base, TimestampMixin):
|
||
"""知识分块模型"""
|
||
__tablename__ = "knowledge_chunks"
|
||
|
||
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
||
page_id: Mapped[int] = mapped_column(Integer, nullable=False, index=True, comment="关联页面 ID")
|
||
chunk_index: Mapped[int] = mapped_column(Integer, nullable=False, comment="分块序号")
|
||
content: Mapped[str] = mapped_column(Text, nullable=False, comment="分块文本内容")
|
||
embedding: Mapped[list | None] = mapped_column(
|
||
VectorType(768) if not IS_SQLITE else Text,
|
||
nullable=True,
|
||
comment="嵌入向量",
|
||
)
|
||
search_vector: Mapped[Any | None] = mapped_column(
|
||
Text, nullable=True, comment="全文搜索向量（仅 PG）",
|
||
)
|
||
|
||
def to_dict(self) -> Dict[str, Any]:
|
||
result = super().to_dict()
|
||
result.pop("embedding", None)
|
||
result.pop("search_vector", None)
|
||
return result
|
||
|
||
|
||
class OCRImage(Base, TimestampMixin):
|
||
"""OCR 图片记录模型"""
|
||
__tablename__ = "ocr_images"
|
||
|
||
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
||
file_path: Mapped[str] = mapped_column(String(500), nullable=False, comment="图片文件路径")
|
||
ocr_text: Mapped[str | None] = mapped_column(Text, nullable=True, comment="OCR 识别全文")
|
||
confidence: Mapped[float | None] = mapped_column(Float, nullable=True, comment="平均置信度")
|
||
provider: Mapped[str | None] = mapped_column(String(50), nullable=True, comment="OCR 提供商")
|
||
status: Mapped[str] = mapped_column(String(20), default="pending", comment="处理状态")
|
||
error_message: Mapped[str | None] = mapped_column(Text, nullable=True, comment="错误信息")
|
||
tags: Mapped[str | None] = mapped_column(Text, nullable=True, comment="LLM 提取的标签，JSON 数组格式")
|
||
story_summary: Mapped[str | None] = mapped_column(Text, nullable=True, comment="Qwen3-8B 提炼的故事摘要")
|
||
|
||
def to_dict(self) -> Dict[str, Any]:
|
||
result = super().to_dict()
|
||
# 将 tags 从 JSON 字符串解析为列表返回
|
||
if result.get("tags") and isinstance(result["tags"], str):
|
||
try:
|
||
import json
|
||
result["tags"] = json.loads(result["tags"])
|
||
except (json.JSONDecodeError, TypeError):
|
||
pass
|
||
return result
|