Tanya
fe01ff51b8
This commit introduces a new `PrismaCompatibleDate` type to ensure compatibility with Prisma's SQLite driver by storing dates in a DateTime format. Additionally, the `extract_exif_date`, `extract_video_date`, and `extract_photo_date` functions are updated to include validation checks that reject future dates and dates prior to 1900, enhancing data integrity during photo and video metadata extraction.
430 lines
17 KiB
Python
430 lines
17 KiB
Python
"""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"),
|
|
)
|
|
|