Tanya
ff47c87e41
Some checks failed
CI / skip-ci-check (pull_request) Successful in 1m4s
CI / lint-and-type-check (pull_request) Has been cancelled
CI / python-lint (pull_request) Has been cancelled
CI / test-backend (pull_request) Has been cancelled
CI / build (pull_request) Has been cancelled
CI / secret-scanning (pull_request) Has been cancelled
CI / dependency-scan (pull_request) Has been cancelled
CI / sast-scan (pull_request) Has been cancelled
CI / workflow-summary (pull_request) Has been cancelled
Add on-demand H.264/AAC web playback (RQ, ffmpeg) with API routes and Next.js proxies; extend admin UI with WebPlaybackVideo and shared hooks. Store transcode cache beside pending-photos (WEB_VIDEO_CACHE_DIR / UPLOAD_DIR) and ignore data/web_videos. Centralize FastAPI URL helpers, optional Vite and Next base paths for subfolder deploy, and fix modal reopen by using router.replace when closing the home photo viewer. Include migration, install scripts, deployment doc updates, and CI admin build env tweak. Made-with: Cursor
292 lines
11 KiB
Python
292 lines
11 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,
|
|
ForeignKey,
|
|
Index,
|
|
Integer,
|
|
LargeBinary,
|
|
Numeric,
|
|
Text,
|
|
UniqueConstraint,
|
|
CheckConstraint,
|
|
)
|
|
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)
|
|
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"),
|
|
)
|
|
|