"""SQLAlchemy models for PunimTag Web - matching desktop schema exactly.""" from __future__ import annotations from datetime import datetime, date from typing import TYPE_CHECKING from sqlalchemy import ( Boolean, Column, Date, DateTime, String, ForeignKey, Index, Integer, LargeBinary, Numeric, Text, UniqueConstraint, CheckConstraint, TypeDecorator, ) from sqlalchemy.orm import declarative_base, relationship from backend.constants.roles import DEFAULT_USER_ROLE if TYPE_CHECKING: pass Base = declarative_base() class PrismaCompatibleDateTime(TypeDecorator): """ DateTime type that stores in a format compatible with Prisma's SQLite driver. Prisma's SQLite driver has issues with microseconds in datetime strings. This type ensures datetimes are stored in ISO format without microseconds: 'YYYY-MM-DD HH:MM:SS' instead of 'YYYY-MM-DD HH:MM:SS.ffffff' Uses String as the underlying type for SQLite to have full control over the format. """ impl = String cache_ok = True def process_bind_param(self, value, dialect): """Convert Python datetime to SQL string format without microseconds.""" if value is None: return None if isinstance(value, datetime): # Strip microseconds and format as ISO string without microseconds # This ensures Prisma can read it correctly return value.replace(microsecond=0).strftime('%Y-%m-%d %H:%M:%S') # If it's already a string, ensure it doesn't have microseconds if isinstance(value, str): try: # Parse and reformat to remove microseconds if '.' in value: # Has microseconds or timezone info - strip them dt = datetime.strptime(value.split('.')[0], '%Y-%m-%d %H:%M:%S') elif 'T' in value: # ISO format with T dt = datetime.fromisoformat(value.replace('Z', '+00:00').split('.')[0]) else: # Already in correct format return value return dt.strftime('%Y-%m-%d %H:%M:%S') except (ValueError, TypeError): # If parsing fails, return as-is return value return value def process_result_value(self, value, dialect): """Convert SQL string back to Python datetime.""" if value is None: return None if isinstance(value, str): # Parse ISO format string try: # Try parsing with microseconds first (for existing data) if '.' in value: return datetime.strptime(value.split('.')[0], '%Y-%m-%d %H:%M:%S') else: return datetime.strptime(value, '%Y-%m-%d %H:%M:%S') except ValueError: # Fallback to ISO format parser return datetime.fromisoformat(value.replace('Z', '+00:00')) return value class PrismaCompatibleDate(TypeDecorator): """ Date type that stores in DateTime format for Prisma compatibility. Prisma's SQLite driver expects DateTime format (YYYY-MM-DD HH:MM:SS) even for dates. This type stores dates with a time component (00:00:00) so Prisma can read them correctly, while still using Python's date type in the application. Uses String as the underlying type for SQLite to have full control over the format. """ impl = String cache_ok = True def process_bind_param(self, value, dialect): """Convert Python date to space-separated DateTime format for Prisma compatibility.""" if value is None: return None if isinstance(value, date): # Store date in space-separated format: YYYY-MM-DD HH:MM:SS (matching date_added format) return value.strftime('%Y-%m-%d 00:00:00') if isinstance(value, datetime): # If datetime is passed, extract date and format with time component return value.date().strftime('%Y-%m-%d 00:00:00') if isinstance(value, str): # If it's already a string, ensure it's in space-separated format try: # Try to parse and convert to space-separated format if 'T' in value: # ISO format with T - convert to space-separated date_part, time_part = value.split('T', 1) time_part = time_part.split('+')[0].split('-')[0].split('Z')[0].split('.')[0] if len(time_part.split(':')) == 3: return f"{date_part} {time_part}" else: return f"{date_part} 00:00:00" elif ' ' in value: # Already space-separated - ensure it has time component parts = value.split(' ', 1) if len(parts) == 2: date_part, time_part = parts time_part = time_part.split('.')[0] # Remove microseconds if present if len(time_part.split(':')) == 3: return f"{date_part} {time_part}" # Missing time component - add it return f"{parts[0]} 00:00:00" else: # Just date (YYYY-MM-DD) - add time component d = datetime.strptime(value, '%Y-%m-%d').date() return d.strftime('%Y-%m-%d 00:00:00') except (ValueError, TypeError): # If parsing fails, return as-is return value return value def process_result_value(self, value, dialect): """Convert SQL string back to Python date.""" if value is None: return None if isinstance(value, str): # Extract date part from ISO 8601 or space-separated DateTime string try: if 'T' in value: # ISO format with T return datetime.fromisoformat(value.split('T')[0]).date() elif ' ' in value: # Space-separated format - extract date part return datetime.strptime(value.split()[0], '%Y-%m-%d').date() else: # Just date (YYYY-MM-DD) return datetime.strptime(value, '%Y-%m-%d').date() except ValueError: # Fallback to ISO format parser try: return datetime.fromisoformat(value.split('T')[0]).date() except: return datetime.strptime(value.split()[0], '%Y-%m-%d').date() if isinstance(value, (date, datetime)): if isinstance(value, datetime): return value.date() return value return value class Photo(Base): """Photo model - matches desktop schema exactly.""" __tablename__ = "photos" id = Column(Integer, primary_key=True, autoincrement=True, index=True) path = Column(Text, unique=True, nullable=False, index=True) filename = Column(Text, nullable=False) date_added = Column(PrismaCompatibleDateTime, default=datetime.utcnow, nullable=False) date_taken = Column(PrismaCompatibleDate, nullable=True, index=True) processed = Column(Boolean, default=False, nullable=False, index=True) file_hash = Column(Text, nullable=True, index=True) # Nullable to support existing photos without hashes media_type = Column(Text, default="image", nullable=False, index=True) # "image" or "video" faces = relationship("Face", back_populates="photo", cascade="all, delete-orphan") photo_tags = relationship( "PhotoTagLinkage", back_populates="photo", cascade="all, delete-orphan" ) favorites = relationship("PhotoFavorite", back_populates="photo", cascade="all, delete-orphan") video_people = relationship( "PhotoPersonLinkage", back_populates="photo", cascade="all, delete-orphan" ) __table_args__ = ( Index("idx_photos_processed", "processed"), Index("idx_photos_date_taken", "date_taken"), Index("idx_photos_date_added", "date_added"), Index("idx_photos_file_hash", "file_hash"), ) class Person(Base): """Person model - matches desktop schema exactly.""" __tablename__ = "people" id = Column(Integer, primary_key=True, autoincrement=True, index=True) first_name = Column(Text, nullable=False) last_name = Column(Text, nullable=False) middle_name = Column(Text, nullable=True) maiden_name = Column(Text, nullable=True) date_of_birth = Column(Date, nullable=True) created_date = Column(PrismaCompatibleDateTime, default=datetime.utcnow, nullable=False) faces = relationship("Face", back_populates="person") person_encodings = relationship( "PersonEncoding", back_populates="person", cascade="all, delete-orphan" ) video_photos = relationship( "PhotoPersonLinkage", back_populates="person", cascade="all, delete-orphan" ) __table_args__ = ( UniqueConstraint( "first_name", "last_name", "middle_name", "maiden_name", "date_of_birth", name="uq_people_names_dob" ), ) class Face(Base): """Face detection model - matches desktop schema exactly.""" __tablename__ = "faces" id = Column(Integer, primary_key=True, autoincrement=True, index=True) photo_id = Column(Integer, ForeignKey("photos.id"), nullable=False, index=True) person_id = Column(Integer, ForeignKey("people.id"), nullable=True, index=True) encoding = Column(LargeBinary, nullable=False) location = Column(Text, nullable=False) confidence = Column(Numeric, default=0.0, nullable=False) quality_score = Column(Numeric, default=0.0, nullable=False, index=True) is_primary_encoding = Column(Boolean, default=False, nullable=False) detector_backend = Column(Text, default="retinaface", nullable=False) model_name = Column(Text, default="ArcFace", nullable=False) face_confidence = Column(Numeric, default=0.0, nullable=False) exif_orientation = Column(Integer, nullable=True) pose_mode = Column(Text, default="frontal", nullable=False, index=True) yaw_angle = Column(Numeric, nullable=True) pitch_angle = Column(Numeric, nullable=True) roll_angle = Column(Numeric, nullable=True) landmarks = Column(Text, nullable=True) # JSON string of facial landmarks identified_by_user_id = Column(Integer, ForeignKey("users.id"), nullable=True, index=True) excluded = Column(Boolean, default=False, nullable=False, index=True) # Exclude from identification photo = relationship("Photo", back_populates="faces") person = relationship("Person", back_populates="faces") person_encodings = relationship( "PersonEncoding", back_populates="face", cascade="all, delete-orphan" ) __table_args__ = ( Index("idx_faces_person_id", "person_id"), Index("idx_faces_photo_id", "photo_id"), Index("idx_faces_quality", "quality_score"), Index("idx_faces_pose_mode", "pose_mode"), Index("idx_faces_identified_by", "identified_by_user_id"), Index("idx_faces_excluded", "excluded"), ) class PersonEncoding(Base): """Person encoding model - matches desktop schema exactly (was person_encodings).""" __tablename__ = "person_encodings" id = Column(Integer, primary_key=True, autoincrement=True, index=True) person_id = Column(Integer, ForeignKey("people.id"), nullable=False, index=True) face_id = Column(Integer, ForeignKey("faces.id"), nullable=False, index=True) encoding = Column(LargeBinary, nullable=False) quality_score = Column(Numeric, default=0.0, nullable=False, index=True) detector_backend = Column(Text, default="retinaface", nullable=False) model_name = Column(Text, default="ArcFace", nullable=False) created_date = Column(PrismaCompatibleDateTime, default=datetime.utcnow, nullable=False) person = relationship("Person", back_populates="person_encodings") face = relationship("Face", back_populates="person_encodings") __table_args__ = ( Index("idx_person_encodings_person_id", "person_id"), Index("idx_person_encodings_quality", "quality_score"), ) class Tag(Base): """Tag model - matches desktop schema exactly.""" __tablename__ = "tags" id = Column(Integer, primary_key=True, autoincrement=True, index=True) tag_name = Column(Text, unique=True, nullable=False, index=True) created_date = Column(PrismaCompatibleDateTime, default=datetime.utcnow, nullable=False) photo_tags = relationship( "PhotoTagLinkage", back_populates="tag", cascade="all, delete-orphan" ) class PhotoTagLinkage(Base): """Photo-Tag linkage model - matches desktop schema exactly (was phototaglinkage).""" __tablename__ = "phototaglinkage" linkage_id = Column(Integer, primary_key=True, autoincrement=True) photo_id = Column(Integer, ForeignKey("photos.id"), nullable=False, index=True) tag_id = Column(Integer, ForeignKey("tags.id"), nullable=False, index=True) linkage_type = Column( Integer, default=0, nullable=False, server_default="0" ) created_date = Column(PrismaCompatibleDateTime, default=datetime.utcnow, nullable=False) photo = relationship("Photo", back_populates="photo_tags") tag = relationship("Tag", back_populates="photo_tags") __table_args__ = ( UniqueConstraint("photo_id", "tag_id", name="uq_photo_tag"), CheckConstraint("linkage_type IN (0, 1)", name="ck_linkage_type"), Index("idx_photo_tags_tag", "tag_id"), Index("idx_photo_tags_photo", "photo_id"), ) class PhotoFavorite(Base): """Photo favorites model - user-specific favorites.""" __tablename__ = "photo_favorites" id = Column(Integer, primary_key=True, autoincrement=True) username = Column(Text, nullable=False, index=True) photo_id = Column(Integer, ForeignKey("photos.id"), nullable=False, index=True) created_date = Column(PrismaCompatibleDateTime, default=datetime.utcnow, nullable=False) photo = relationship("Photo", back_populates="favorites") __table_args__ = ( UniqueConstraint("username", "photo_id", name="uq_user_photo_favorite"), Index("idx_favorites_username", "username"), Index("idx_favorites_photo", "photo_id"), ) class User(Base): """User model for main database - separate from auth database users.""" __tablename__ = "users" id = Column(Integer, primary_key=True, autoincrement=True, index=True) username = Column(Text, unique=True, nullable=False, index=True) password_hash = Column(Text, nullable=False) # Hashed password email = Column(Text, unique=True, nullable=False, index=True) full_name = Column(Text, nullable=False) is_active = Column(Boolean, default=True, nullable=False) is_admin = Column(Boolean, default=False, nullable=False, index=True) role = Column( Text, nullable=False, default=DEFAULT_USER_ROLE, server_default=DEFAULT_USER_ROLE, index=True, ) password_change_required = Column(Boolean, default=True, nullable=False, index=True) created_date = Column(PrismaCompatibleDateTime, default=datetime.utcnow, nullable=False) last_login = Column(PrismaCompatibleDateTime, nullable=True) __table_args__ = ( Index("idx_users_username", "username"), Index("idx_users_email", "email"), Index("idx_users_is_admin", "is_admin"), Index("idx_users_password_change_required", "password_change_required"), Index("idx_users_role", "role"), ) class PhotoPersonLinkage(Base): """Direct linkage between Video (Photo with media_type='video') and Person. This allows identifying people in videos without requiring face detection. Only used for videos, not photos (photos use Face model for identification). """ __tablename__ = "photo_person_linkage" id = Column(Integer, primary_key=True, autoincrement=True) photo_id = Column(Integer, ForeignKey("photos.id"), nullable=False, index=True) person_id = Column(Integer, ForeignKey("people.id"), nullable=False, index=True) identified_by_user_id = Column(Integer, ForeignKey("users.id"), nullable=True, index=True) created_date = Column(PrismaCompatibleDateTime, default=datetime.utcnow, nullable=False) photo = relationship("Photo", back_populates="video_people") person = relationship("Person", back_populates="video_photos") __table_args__ = ( UniqueConstraint("photo_id", "person_id", name="uq_photo_person"), Index("idx_photo_person_photo", "photo_id"), Index("idx_photo_person_person", "person_id"), Index("idx_photo_person_user", "identified_by_user_id"), ) class RolePermission(Base): """Role-to-feature permission matrix.""" __tablename__ = "role_permissions" id = Column(Integer, primary_key=True, autoincrement=True) role = Column(Text, nullable=False, index=True) feature_key = Column(Text, nullable=False, index=True) allowed = Column(Boolean, nullable=False, default=False, server_default="0") __table_args__ = ( UniqueConstraint("role", "feature_key", name="uq_role_feature"), Index("idx_role_permissions_role_feature", "role", "feature_key"), )