"""MongoDB 文档模型 — helper class（非 ORM，纯字典封装） 3 个 collection： Book → book BookDocument → book_document DocumentFile → document_file 每个 class 有 to_mongo / from_mongo / to_dict 等方法。 """ import hashlib from datetime import datetime, timezone from typing import Optional def _now() -> datetime: return datetime.now(timezone.utc) # ============================================================ # Book — 教辅 # ============================================================ class Book: """教辅文档。_id = book_id（int）""" COLLECTION = 'book' def __init__(self, **kwargs): self.id: Optional[int] = kwargs.get('id') self.name: str = kwargs['name'] self.subject: Optional[str] = kwargs.get('subject') self.grade: Optional[str] = kwargs.get('grade') self.version: Optional[str] = kwargs.get('version') self.type: Optional[str] = kwargs.get('type') self.zip_path: str = kwargs['zip_path'] self.zip_sha256: str = kwargs['zip_sha256'] self.zip_size: int = kwargs.get('zip_size', 0) self.status: str = kwargs.get('status', 'pending') self.total_count: int = kwargs.get('total_count', 0) self.success_count: int = kwargs.get('success_count', 0) self.failed_count: int = kwargs.get('failed_count', 0) self.error_msg: Optional[str] = kwargs.get('error_msg') self.created_at: datetime = kwargs.get('created_at', _now()) self.updated_at: datetime = kwargs.get('updated_at', _now()) self.created_by: Optional[str] = kwargs.get('created_by') self.is_deleted: int = kwargs.get('is_deleted', 0) def to_mongo(self) -> dict: d = { 'name': self.name, 'subject': self.subject, 'grade': self.grade, 'version': self.version, 'type': self.type, 'zip_path': self.zip_path, 'zip_sha256': self.zip_sha256, 'zip_size': self.zip_size, 'status': self.status, 'total_count': self.total_count, 'success_count': self.success_count, 'failed_count': self.failed_count, 'error_msg': self.error_msg, 'created_at': self.created_at, 'updated_at': self.updated_at, 'created_by': self.created_by, 'is_deleted': self.is_deleted, } if self.id is not None: d['_id'] = self.id return d @classmethod def from_mongo(cls, doc: dict) -> 'Book': if not doc: return None d = dict(doc) d['id'] = d.pop('_id', None) return cls(**d) def to_dict(self) -> dict: return { 'id': self.id, 'name': self.name, 'subject': self.subject, 'grade': self.grade, 'version': self.version, 'type': self.type, 'zip_path': self.zip_path, 'zip_sha256': self.zip_sha256, 'zip_size': self.zip_size, 'status': self.status, 'total_count': self.total_count, 'success_count': self.success_count, 'failed_count': self.failed_count, 'error_msg': self.error_msg, 'created_at': self.created_at.isoformat() if self.created_at else None, 'updated_at': self.updated_at.isoformat() if self.updated_at else None, 'created_by': self.created_by, } # ============================================================ # BookDocument — 教辅文档 # ============================================================ class BookDocument: """教辅文档。_id = doc_id（16 字符 sha1 前缀）""" COLLECTION = 'book_document' DOC_TYPES = ('课时作业', '周末提升', '阶段质量评估', '综合评估', 'unknown') STATUSES = ('pending', 'html_ready', 'json_ready', 'failed') def __init__(self, **kwargs): self.book_id: int = kwargs['book_id'] self.original_filename: str = kwargs['original_filename'] self.original_path: str = kwargs['original_path'] self.doc_id: str = kwargs.get('doc_id') or self.make_doc_id( kwargs['book_id'], kwargs['original_path']) self.title: Optional[str] = kwargs.get('title') self.doc_type: str = kwargs.get('doc_type', 'unknown') self.order_index: int = kwargs.get('order_index', 0) self.status: str = kwargs.get('status', 'pending') self.error_msg: Optional[str] = kwargs.get('error_msg') self.created_at: datetime = kwargs.get('created_at', _now()) self.updated_at: datetime = kwargs.get('updated_at', _now()) @staticmethod def make_doc_id(book_id: int, original_path: str) -> str: """稳定 doc_id：sha1(book_id|path).hexdigest()[:16]""" h = hashlib.sha1(f"{book_id}|{original_path}".encode('utf-8')).hexdigest() return h[:16] def to_mongo(self) -> dict: return { '_id': self.doc_id, 'book_id': self.book_id, 'original_filename': self.original_filename, 'original_path': self.original_path, 'title': self.title, 'doc_type': self.doc_type, 'order_index': self.order_index, 'status': self.status, 'error_msg': self.error_msg, 'created_at': self.created_at, 'updated_at': self.updated_at, } @classmethod def from_mongo(cls, doc: dict) -> 'BookDocument': if not doc: return None d = dict(doc) d['doc_id'] = d.pop('_id') return cls(**d) def to_dict(self) -> dict: return { 'book_id': self.book_id, 'doc_id': self.doc_id, 'original_filename': self.original_filename, 'original_path': self.original_path, 'title': self.title, 'doc_type': self.doc_type, 'order_index': self.order_index, 'status': self.status, 'error_msg': self.error_msg, 'created_at': self.created_at.isoformat() if self.created_at else None, 'updated_at': self.updated_at.isoformat() if self.updated_at else None, } # ============================================================ # DocumentFile — 文档文件 # ============================================================ class DocumentFile: """文档文件。_id = 自增 int（next_id('file')）""" COLLECTION = 'document_file' FILE_TYPES = ('docx', 'html', 'json', 'preview_png', 'error_log') def __init__(self, **kwargs): self.id: Optional[int] = kwargs.get('id') self.doc_id: str = kwargs['doc_id'] self.file_type: str = kwargs['file_type'] self.object_key: str = kwargs['object_key'] self.object_bucket: str = kwargs['object_bucket'] self.size: int = kwargs.get('size', 0) self.checksum: Optional[str] = kwargs.get('checksum') self.url: Optional[str] = kwargs.get('url') self.meta: Optional[dict] = kwargs.get('meta') self.created_at: datetime = kwargs.get('created_at', _now()) def to_mongo(self) -> dict: d = { 'doc_id': self.doc_id, 'file_type': self.file_type, 'object_key': self.object_key, 'object_bucket': self.object_bucket, 'size': self.size, 'checksum': self.checksum, 'url': self.url, 'meta': self.meta, 'created_at': self.created_at, } if self.id is not None: d['_id'] = self.id return d @classmethod def from_mongo(cls, doc: dict) -> 'DocumentFile': if not doc: return None d = dict(doc) d['id'] = d.pop('_id') return cls(**d) def to_dict(self) -> dict: return { 'id': self.id, 'doc_id': self.doc_id, 'file_type': self.file_type, 'object_key': self.object_key, 'object_bucket': self.object_bucket, 'size': self.size, 'checksum': self.checksum, 'url': self.url, 'meta': self.meta, 'created_at': self.created_at.isoformat() if self.created_at else None, }