""" ORM 模型定义 SQLite 数据库模型,使用 Text 存储 JSON 格式的向量。 """ from __future__ import annotations import json 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 class Base(DeclarativeBase): """所有 ORM 模型的基类""" pass class TimestampMixin: """时间戳 Mixin""" created_at: Mapped[datetime] = mapped_column( DateTime, default=func.now(), comment="创建时间" ) updated_at: Mapped[datetime] = mapped_column( DateTime, default=func.now(), onupdate=func.now(), comment="更新时间" ) def to_dict(self) -> Dict[str, Any]: """将模型转为字典""" result = {} for col in self.__table__.columns: val = getattr(self, col.name) if isinstance(val, datetime): val = val.isoformat() result[col.name] = val 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="来源文件路径") page_number: Mapped[int | None] = mapped_column(Integer, 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="直播日期") metadata_json: Mapped[str | None] = mapped_column(Text, nullable=True, comment="额外元数据(JSON)") embedding: Mapped[str | None] = mapped_column(Text, nullable=True, comment="向量嵌入(JSON)") 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, comment="关联页面 ID") chunk_index: Mapped[int] = mapped_column(Integer, nullable=False, comment="分块索引") content: Mapped[str] = mapped_column(Text, nullable=False, comment="分块内容") embedding: Mapped[str | None] = mapped_column(Text, nullable=True, comment="向量嵌入(JSON)") 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="图片文件路径") original_filename: Mapped[str | None] = mapped_column(String(200), nullable=True, comment="原始文件名") ocr_text: Mapped[str | None] = mapped_column(Text, nullable=True, comment="OCR 识别文本") confidence: Mapped[float | None] = mapped_column(Float, nullable=True, comment="OCR 置信度") provider: Mapped[str | None] = mapped_column(String(50), nullable=True, comment="OCR 提供商") status: Mapped[str] = mapped_column( String(20), default="pending", comment="状态: pending/processing/completed/failed" ) blocks: Mapped[str | None] = mapped_column(Text, nullable=True, comment="OCR 文本块(JSON)") @property def blocks_list(self) -> list[dict]: if self.blocks is None: return [] return json.loads(self.blocks) @blocks_list.setter def blocks_list(self, value: list[dict]): self.blocks = json.dumps(value) if value else None class WebsiteSettings(Base): """网站设置(单例)""" __tablename__ = "website_settings" key: Mapped[str] = mapped_column(String(100), primary_key=True, comment="设置键") value: Mapped[str | None] = mapped_column(Text, nullable=True, comment="设置值(JSON)") updated_at: Mapped[datetime] = mapped_column( DateTime, default=func.now(), onupdate=func.now(), comment="更新时间" ) def to_dict(self) -> Dict[str, Any]: """将模型转为字典""" result = {} for col in self.__table__.columns: val = getattr(self, col.name) if isinstance(val, datetime): val = val.isoformat() result[col.name] = val return result