from __future__ import annotations from datetime import date, datetime from typing import Optional from sqlalchemy import BigInteger, Boolean, Date, DateTime, Float, ForeignKey, Index, Integer, LargeBinary, String, Text, UniqueConstraint from sqlalchemy.orm import Mapped, deferred, mapped_column, relationship from .db import Base def now() -> datetime: return datetime.now().astimezone() class User(Base): __tablename__ = "users" id: Mapped[int] = mapped_column(primary_key=True) username: Mapped[str] = mapped_column(String(64), unique=True, index=True) display_name: Mapped[str] = mapped_column(String(100)) password_hash: Mapped[str] = mapped_column(String(300)) role: Mapped[str] = mapped_column(String(20), default="uploader", index=True) is_active: Mapped[bool] = mapped_column(Boolean, default=True) must_change_password: Mapped[bool] = mapped_column(Boolean, default=True) entry_key_hash: Mapped[Optional[str]] = mapped_column(String(64), unique=True, nullable=True, index=True) entry_key_updated_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=now) updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=now, onupdate=now) class LoginSession(Base): __tablename__ = "login_sessions" id: Mapped[int] = mapped_column(primary_key=True) token_hash: Mapped[str] = mapped_column(String(64), unique=True, index=True) user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"), index=True) expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), index=True) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=now) ip_address: Mapped[Optional[str]] = mapped_column(String(64), nullable=True) user_agent: Mapped[Optional[str]] = mapped_column(String(300), nullable=True) user: Mapped[User] = relationship() class EntrySession(Base): __tablename__ = "entry_sessions" id: Mapped[int] = mapped_column(primary_key=True) token_hash: Mapped[str] = mapped_column(String(64), unique=True, index=True) user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"), index=True) expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), index=True) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=now) ip_address: Mapped[Optional[str]] = mapped_column(String(64), nullable=True) user: Mapped[User] = relationship() class EntryBatch(Base): __tablename__ = "entry_batches" id: Mapped[int] = mapped_column(primary_key=True) entry_no: Mapped[str] = mapped_column(String(40), unique=True, index=True) record_count: Mapped[int] = mapped_column(Integer, default=1) status: Mapped[str] = mapped_column(String(20), default="success", index=True) created_by: Mapped[int] = mapped_column(ForeignKey("users.id"), index=True) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=now, index=True) creator: Mapped[User] = relationship() class MaterialRecord(Base): __tablename__ = "material_records" __table_args__ = ( UniqueConstraint("entry_batch_id", "experiment_id", name="uq_entry_experiment"), Index("ix_record_material_date", "material_category", "test_date"), ) id: Mapped[int] = mapped_column(primary_key=True) entry_batch_id: Mapped[int] = mapped_column(ForeignKey("entry_batches.id", ondelete="CASCADE"), index=True) experiment_id: Mapped[str] = mapped_column(String(100), unique=True, index=True) test_date: Mapped[date] = mapped_column(Date, index=True) material_category: Mapped[str] = mapped_column(String(100), index=True) material_name: Mapped[str] = mapped_column(String(200), index=True) apparent_density: Mapped[Optional[float]] = mapped_column(Float) uv_absorption_355nm: Mapped[Optional[float]] = mapped_column(Float) material_thickness: Mapped[float] = mapped_column(Float) moisture_content: Mapped[Optional[float]] = mapped_column(Float) thermal_conductivity: Mapped[Optional[float]] = mapped_column(Float) initial_decomposition_temp: Mapped[Optional[float]] = mapped_column(Float) melting_vaporization_temp: Mapped[Optional[float]] = mapped_column(Float) specific_heat_capacity: Mapped[Optional[float]] = mapped_column(Float) carbon_residue_rate: Mapped[Optional[float]] = mapped_column(Float) surface_roughness: Mapped[Optional[float]] = mapped_column(Float) hardness: Mapped[Optional[float]] = mapped_column(Float) actual_output_power: Mapped[float] = mapped_column(Float) display_current: Mapped[float] = mapped_column(Float) scanning_speed: Mapped[float] = mapped_column(Float) pulse_frequency: Mapped[float] = mapped_column(Float) pulse_width: Mapped[float] = mapped_column(Float) defocus_amount: Mapped[float] = mapped_column(Float) scan_line_spacing: Mapped[float] = mapped_column(Float) filling_method: Mapped[str] = mapped_column(String(100)) processing_size: Mapped[str] = mapped_column(String(100)) is_cut_through: Mapped[bool] = mapped_column(Boolean) carbonized_edge_width: Mapped[float] = mapped_column(Float) etching_depth: Mapped[float] = mapped_column(Float) is_fire_smolder: Mapped[bool] = mapped_column(Boolean) pattern_clarity_score: Mapped[float] = mapped_column(Float) presentation_balance_score: Mapped[float] = mapped_column(Float) finished_image_filename: Mapped[str] = mapped_column(String(255)) remarks: Mapped[str] = mapped_column(Text) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=now) entry_batch: Mapped[EntryBatch] = relationship() custom_values: Mapped[list["CustomFieldValue"]] = relationship(back_populates="record", cascade="all, delete-orphan") images: Mapped[list["RecordImage"]] = relationship(back_populates="record", cascade="all, delete-orphan", passive_deletes=True, order_by="RecordImage.sort_order, RecordImage.id") class RecordImage(Base): """实验成品图。二进制直接入库,现有数据库备份即可覆盖图片,无需额外挂载卷。""" __tablename__ = "record_images" id: Mapped[int] = mapped_column(primary_key=True) record_id: Mapped[Optional[int]] = mapped_column(ForeignKey("material_records.id", ondelete="CASCADE"), nullable=True, index=True) token: Mapped[str] = mapped_column(String(64), unique=True, index=True) content_type: Mapped[str] = mapped_column(String(40)) original_name: Mapped[str] = mapped_column(String(255)) size_bytes: Mapped[int] = mapped_column(Integer) width: Mapped[int] = mapped_column(Integer) height: Mapped[int] = mapped_column(Integer) sha256: Mapped[str] = mapped_column(String(64), index=True) # deferred:详情接口只需要元信息,默认加载会把每条记录的几 MB 图片全读进内存, # 而公开详情是匿名可访问的,等于把看板变成内存放大器。 data: Mapped[bytes] = deferred(mapped_column(LargeBinary)) thumbnail: Mapped[Optional[bytes]] = deferred(mapped_column(LargeBinary, nullable=True)) thumbnail_type: Mapped[Optional[str]] = mapped_column(String(40), nullable=True) sort_order: Mapped[int] = mapped_column(Integer, default=0) created_by: Mapped[int] = mapped_column(ForeignKey("users.id"), index=True) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=now, index=True) record: Mapped[Optional[MaterialRecord]] = relationship(back_populates="images") creator: Mapped[User] = relationship() class CustomFieldDefinition(Base): __tablename__ = "custom_field_definitions" id: Mapped[int] = mapped_column(primary_key=True) field_key: Mapped[str] = mapped_column(String(50), unique=True, index=True) label: Mapped[str] = mapped_column(String(100)) input_type: Mapped[str] = mapped_column(String(20), index=True) is_required: Mapped[bool] = mapped_column(Boolean, default=False) is_active: Mapped[bool] = mapped_column(Boolean, default=True, index=True) is_public: Mapped[bool] = mapped_column(Boolean, default=True) unit: Mapped[Optional[str]] = mapped_column(String(50), nullable=True) options_json: Mapped[Optional[str]] = mapped_column(Text, nullable=True) sort_order: Mapped[int] = mapped_column(Integer, default=0, index=True) created_by: Mapped[int] = mapped_column(ForeignKey("users.id"), index=True) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=now) updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=now, onupdate=now) creator: Mapped[User] = relationship() values: Mapped[list["CustomFieldValue"]] = relationship(back_populates="definition") class CustomFieldValue(Base): __tablename__ = "custom_field_values" __table_args__ = (UniqueConstraint("record_id", "field_id", name="uq_record_custom_field"),) id: Mapped[int] = mapped_column(primary_key=True) record_id: Mapped[int] = mapped_column(ForeignKey("material_records.id", ondelete="CASCADE"), index=True) field_id: Mapped[int] = mapped_column(ForeignKey("custom_field_definitions.id", ondelete="RESTRICT"), index=True) raw_value: Mapped[Optional[str]] = mapped_column(Text, nullable=True) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=now) updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=now, onupdate=now) record: Mapped[MaterialRecord] = relationship(back_populates="custom_values") definition: Mapped[CustomFieldDefinition] = relationship(back_populates="values") class AuditLog(Base): __tablename__ = "audit_logs" id: Mapped[int] = mapped_column(primary_key=True) user_id: Mapped[Optional[int]] = mapped_column(ForeignKey("users.id"), nullable=True, index=True) action: Mapped[str] = mapped_column(String(80), index=True) resource_type: Mapped[str] = mapped_column(String(80), index=True) resource_id: Mapped[Optional[str]] = mapped_column(String(100), nullable=True) detail: Mapped[Optional[str]] = mapped_column(Text, nullable=True) ip_address: Mapped[Optional[str]] = mapped_column(String(64), nullable=True) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=now, index=True) user: Mapped[Optional[User]] = relationship() class BackupRecord(Base): __tablename__ = "backup_records" id: Mapped[int] = mapped_column(primary_key=True) filename: Mapped[str] = mapped_column(String(255), unique=True) database_type: Mapped[str] = mapped_column(String(30)) # BigInteger:图片入库后备份很容易超过 2 GiB,PostgreSQL 的 integer 到那里就溢出了 size_bytes: Mapped[int] = mapped_column(BigInteger) sha256: Mapped[str] = mapped_column(String(64)) created_by: Mapped[Optional[int]] = mapped_column(ForeignKey("users.id"), nullable=True) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=now, index=True) creator: Mapped[Optional[User]] = relationship()