ilia
c64a97d147
CI / skip-ci-check (pull_request) Successful in 29s
CI / python-lint (pull_request) Successful in 31s
CI / docker-ci (pull_request) Successful in 32s
CI / secret-scan (pull_request) Successful in 37s
CI / e2e (pull_request) Successful in 4m32s
CI / viewer-unit (pull_request) Successful in 4m42s
CI / admin-unit (pull_request) Successful in 5m6s
Capture optional contact fields on identify/approve, add Select All and Approve Next 10 on User Identified Faces, and email admins when the pending queue grows (ADMIN_NOTIFY_EMAIL via existing noreply SMTP).
294 lines
11 KiB
Python
294 lines
11 KiB
Python
"""SQLAlchemy models for PunimTag Web - matching desktop schema exactly."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from typing import TYPE_CHECKING
|
|
|
|
from sqlalchemy import (
|
|
Boolean,
|
|
CheckConstraint,
|
|
Column,
|
|
Date,
|
|
DateTime,
|
|
ForeignKey,
|
|
Index,
|
|
Integer,
|
|
LargeBinary,
|
|
Numeric,
|
|
Text,
|
|
UniqueConstraint,
|
|
)
|
|
from sqlalchemy.orm import declarative_base, relationship
|
|
|
|
from backend.constants.roles import DEFAULT_USER_ROLE
|
|
|
|
if TYPE_CHECKING:
|
|
pass
|
|
|
|
Base = declarative_base()
|
|
|
|
|
|
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(DateTime, default=datetime.utcnow, nullable=False)
|
|
date_taken = Column(Date, 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"
|
|
# Browser-safe playback (H.264/AAC); see web_video_service / migrations.
|
|
web_playable_path = Column(Text, nullable=True)
|
|
web_transcode_status = Column(Text, default="none", nullable=False)
|
|
web_transcode_error = Column(Text, nullable=True)
|
|
web_transcode_job_id = Column(Text, nullable=True)
|
|
|
|
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)
|
|
email = Column(Text, nullable=True)
|
|
phone = Column(Text, nullable=True)
|
|
created_date = Column(DateTime, 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(DateTime, 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(DateTime, 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(DateTime, 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(DateTime, 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(DateTime, default=datetime.utcnow, nullable=False)
|
|
last_login = Column(DateTime, 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(DateTime, 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"),
|
|
)
|
|
|