116 lines
4.7 KiB
Python
116 lines
4.7 KiB
Python
"""
|
||
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）")
|
||
error_message: Mapped[str | None] = mapped_column(Text, nullable=True, comment="错误信息")
|
||
|
||
@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
|