ilia/punimtag
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Add identification report and clear denied records functionality

Browse Source 
This commit introduces new API endpoints for generating identification reports and clearing denied records from the database. The frontend has been updated to include a report button that fetches user identification statistics, allowing admins to view how many faces each user identified over a specified date range. Additionally, a clear denied records button has been added to permanently remove all denied identifications. The necessary data models and response structures have been implemented to support these features. Documentation has been updated to reflect these changes.
This commit is contained in:
tanyar09
2025-11-24 14:58:11 -05:00
parent dbffaef298
commit a036169b0f
10 changed files with 673 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/web/api/auth.py
+44
View File
@@ -74,6 +74,50 @@ def get_current_user(
)
def get_current_user_with_id(
current_user: Annotated[dict, Depends(get_current_user)],
db: Session = Depends(get_db),
) -> dict:
"""Get current user with ID from main database.
Looks up the user in the main database and returns username and user_id.
If user doesn't exist, creates them (for bootstrap scenarios).
"""
username = current_user["username"]
# Check if user exists in main database
user = db.query(User).filter(User.username == username).first()
# If user doesn't exist, create them (for bootstrap scenarios)
if not user:
from src.web.utils.password import hash_password
# Generate unique email to avoid conflicts
base_email = f"{username}@example.com"
email = base_email
counter = 1
# Ensure email is unique
while db.query(User).filter(User.email == email).first():
email = f"{username}+{counter}@example.com"
counter += 1
# Create user (they should change password)
default_password_hash = hash_password("changeme")
user = User(
username=username,
password_hash=default_password_hash,
email=email,
full_name=username,
is_active=True,
is_admin=False,
)
db.add(user)
db.commit()
db.refresh(user)
return {"username": username, "user_id": user.id}
@router.post("/login", response_model=TokenResponse)
def login(credentials: LoginRequest, db: Session = Depends(get_db)) -> TokenResponse:
"""Authenticate user and return tokens.
src/web/api/faces.py
+7
View File
@@ -8,8 +8,10 @@ from rq import Queue
from redis import Redis
from sqlalchemy import func
from sqlalchemy.orm import Session
from typing import Annotated
from src.web.db.session import get_db
from src.web.api.auth import get_current_user_with_id
from src.web.schemas.faces import (
ProcessFacesRequest,
ProcessFacesResponse,
@@ -260,12 +262,16 @@ def get_batch_similarities(
def identify_face(
face_id: int,
request: IdentifyFaceRequest,
current_user: Annotated[dict, Depends(get_current_user_with_id)],
db: Session = Depends(get_db),
) -> IdentifyFaceResponse:
"""Assign a face (and optional batch) to a person, creating if needed.
Also inserts into person_encodings for each identified face as desktop does.
Tracks which user identified the face.
"""
user_id = current_user["user_id"]
# Validate target face
face = db.query(Face).filter(Face.id == face_id).first()
if not face:
@@ -307,6 +313,7 @@ def identify_face(
if not f:
continue
f.person_id = person.id
f.identified_by_user_id = user_id
db.add(f)
# Insert person_encoding
pe = PersonEncoding(
src/web/api/pending_identifications.py
+215 -3
View File
@@ -5,18 +5,53 @@ from __future__ import annotations
from datetime import date, datetime
from typing import Annotated, Optional
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi import APIRouter, Depends, HTTPException, Query, status
from pydantic import BaseModel, ConfigDict
from sqlalchemy import text
from sqlalchemy import text, func
from sqlalchemy.orm import Session
from src.web.db.session import get_auth_db, get_db
from src.web.db.models import Face, Person, PersonEncoding
from src.web.db.models import Face, Person, PersonEncoding, User
from src.web.api.users import get_current_admin_user
from src.web.utils.password import hash_password
router = APIRouter(prefix="/pending-identifications", tags=["pending-identifications"])
def get_or_create_frontend_user(db: Session) -> User:
"""Get or create the special 'FrontEndUser' system user.
This user represents identifications made through the frontend approval UI,
distinguishing them from direct user identifications.
"""
FRONTEND_USERNAME = "FrontEndUser"
# Try to get existing user
user = db.query(User).filter(User.username == FRONTEND_USERNAME).first()
if user:
return user
# Create the system user if it doesn't exist
# Use a non-loginable password hash (random, won't be used for login)
default_password_hash = hash_password("system_user_not_for_login")
user = User(
username=FRONTEND_USERNAME,
password_hash=default_password_hash,
email="frontend@punimtag.system",
full_name="Frontend System User",
is_active=False, # Not an active user, just a system marker
is_admin=False,
password_change_required=False,
)
db.add(user)
db.commit()
db.refresh(user)
return user
class PendingIdentificationResponse(BaseModel):
"""Pending identification DTO returned from API."""
@@ -74,6 +109,39 @@ class ApproveDenyResponse(BaseModel):
errors: list[str]
class UserIdentificationStats(BaseModel):
"""Statistics for a single user's identifications."""
model_config = ConfigDict(protected_namespaces=())
user_id: int
username: str
full_name: str
email: str
face_count: int
first_identification_date: Optional[datetime] = None
last_identification_date: Optional[datetime] = None
class IdentificationReportResponse(BaseModel):
"""Response containing identification statistics grouped by user."""
model_config = ConfigDict(protected_namespaces=())
items: list[UserIdentificationStats]
total_faces: int
total_users: int
class ClearDatabaseResponse(BaseModel):
"""Response from clearing denied records."""
model_config = ConfigDict(protected_namespaces=())
deleted_records: int
errors: list[str]
@router.get("", response_model=PendingIdentificationsListResponse)
def list_pending_identifications(
current_admin: Annotated[dict, Depends(get_current_admin_user)],
@@ -277,7 +345,10 @@ def approve_deny_pending_identifications(
created_person = True
# Link face to person
# Use FrontEndUser to indicate this was approved through the frontend UI
frontend_user = get_or_create_frontend_user(main_db)
face.person_id = person.id
face.identified_by_user_id = frontend_user.id
main_db.add(face)
# Insert person_encoding
@@ -327,3 +398,144 @@ def approve_deny_pending_identifications(
errors=errors
)
@router.get("/report", response_model=IdentificationReportResponse)
def get_identification_report(
current_admin: Annotated[dict, Depends(get_current_admin_user)],
date_from: Optional[str] = Query(None, description="Filter by identification date (from) - YYYY-MM-DD"),
date_to: Optional[str] = Query(None, description="Filter by identification date (to) - YYYY-MM-DD"),
main_db: Session = Depends(get_db),
) -> IdentificationReportResponse:
"""Get identification statistics grouped by user.
Shows how many faces each user identified and when.
Can be filtered by date range using PersonEncoding.created_date.
"""
# Query faces that have been identified (have person_id and identified_by_user_id)
# Join with PersonEncoding to get created_date (when face was identified)
# Join with User to get user information
# Use distinct count to avoid counting the same face multiple times
# (in case person encodings were updated, creating multiple PersonEncoding records)
query = (
main_db.query(
User.id.label('user_id'),
User.username,
User.full_name,
User.email,
func.count(func.distinct(Face.id)).label('face_count'),
func.min(PersonEncoding.created_date).label('first_date'),
func.max(PersonEncoding.created_date).label('last_date')
)
.join(Face, User.id == Face.identified_by_user_id)
.join(PersonEncoding, Face.id == PersonEncoding.face_id)
.filter(Face.person_id.isnot(None))
.filter(Face.identified_by_user_id.isnot(None))
.group_by(User.id, User.username, User.full_name, User.email)
)
# Apply date filtering if provided (filter before grouping)
if date_from:
try:
date_from_obj = datetime.strptime(date_from, "%Y-%m-%d").date()
query = query.filter(func.date(PersonEncoding.created_date) >= date_from_obj)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Invalid date_from format. Use YYYY-MM-DD"
)
if date_to:
try:
date_to_obj = datetime.strptime(date_to, "%Y-%m-%d").date()
query = query.filter(func.date(PersonEncoding.created_date) <= date_to_obj)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Invalid date_to format. Use YYYY-MM-DD"
)
# Execute query and get results
results = query.order_by(func.count(Face.id).desc(), User.username.asc()).all()
# Convert to response model
items = []
total_faces = 0
for row in results:
total_faces += row.face_count
items.append(
UserIdentificationStats(
user_id=row.user_id,
username=row.username,
full_name=row.full_name or row.username,
email=row.email,
face_count=row.face_count,
first_identification_date=row.first_date,
last_identification_date=row.last_date,
)
)
return IdentificationReportResponse(
items=items,
total_faces=total_faces,
total_users=len(items)
)
@router.post("/clear-denied", response_model=ClearDatabaseResponse)
def clear_denied_identifications(
current_admin: Annotated[dict, Depends(get_current_admin_user)],
auth_db: Session = Depends(get_auth_db),
) -> ClearDatabaseResponse:
"""Delete all denied pending identifications from the database.
This permanently removes all records with status='denied' from the
pending_identifications table in the auth database.
"""
deleted_records = 0
errors = []
try:
# First check if there are any denied records
check_result = auth_db.execute(text("""
SELECT COUNT(*) as count FROM pending_identifications
WHERE status = 'denied'
"""))
denied_count = check_result.fetchone().count if check_result else 0
if denied_count == 0:
# No denied records to delete
return ClearDatabaseResponse(
deleted_records=0,
errors=[]
)
# Delete all denied records
result = auth_db.execute(text("""
DELETE FROM pending_identifications
WHERE status = 'denied'
"""))
deleted_records = result.rowcount if hasattr(result, 'rowcount') else 0
auth_db.commit()
if deleted_records == 0 and denied_count > 0:
errors.append("No records were deleted despite finding denied records")
except Exception as e:
auth_db.rollback()
error_msg = str(e)
errors.append(f"Error deleting denied records: {error_msg}")
# Check if it's a permission error
if "permission denied" in error_msg.lower() or "insufficient privilege" in error_msg.lower():
errors.append(
"Database permission error. Please run this command manually:\n"
"sudo -u postgres psql -d punimtag_auth -c \"GRANT DELETE ON TABLE pending_identifications TO punimtag;\""
)
return ClearDatabaseResponse(
deleted_records=deleted_records,
errors=errors
)
src/web/api/people.py
+9 -1
View File
@@ -2,12 +2,15 @@
from __future__ import annotations
from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy import func
from sqlalchemy.orm import Session
from src.web.db.session import get_db
from src.web.db.models import Person, Face
from src.web.api.auth import get_current_user_with_id
from src.web.schemas.people import (
PeopleListResponse,
PersonCreateRequest,
@@ -185,6 +188,7 @@ def get_person_faces(person_id: int, db: Session = Depends(get_db)) -> PersonFac
def accept_matches(
person_id: int,
request: AcceptMatchesRequest,
current_user: Annotated[dict, Depends(get_current_user_with_id)],
db: Session = Depends(get_db),
) -> IdentifyFaceResponse:
"""Accept auto-match matches for a person.
@@ -193,9 +197,13 @@ def accept_matches(
1. Identifies selected faces with this person
2. Inserts person_encodings for each identified face
3. Updates person encodings (removes old, adds current)
Tracks which user identified the faces.
"""
from src.web.api.auth import get_current_user_with_id
user_id = current_user["user_id"]
identified_count, updated_count = accept_auto_match_matches(
db, person_id, request.face_ids
db, person_id, request.face_ids, user_id=user_id
)
return IdentifyFaceResponse(
src/web/app.py
+42
View File
@@ -221,6 +221,47 @@ def ensure_user_email_unique_constraint(inspector) -> None:
print("ℹ️ SQLite: Unique constraint on email will be enforced by model definition for new tables")
def ensure_face_identified_by_user_id_column(inspector) -> None:
"""Ensure faces table contains identified_by_user_id column."""
if "faces" not in inspector.get_table_names():
return
columns = {column["name"] for column in inspector.get_columns("faces")}
if "identified_by_user_id" in columns:
print("ℹ️ identified_by_user_id column already exists in faces table")
return
print("🔄 Adding identified_by_user_id column to faces table...")
dialect = engine.dialect.name
with engine.connect() as connection:
with connection.begin():
if dialect == "postgresql":
connection.execute(
text("ALTER TABLE faces ADD COLUMN IF NOT EXISTS identified_by_user_id INTEGER REFERENCES users(id)")
)
# Add index
try:
connection.execute(
text("CREATE INDEX IF NOT EXISTS idx_faces_identified_by ON faces(identified_by_user_id)")
)
except Exception:
pass # Index might already exist
else:
# SQLite
connection.execute(
text("ALTER TABLE faces ADD COLUMN identified_by_user_id INTEGER REFERENCES users(id)")
)
# SQLite doesn't support IF NOT EXISTS for indexes, so we'll try to create it
try:
connection.execute(
text("CREATE INDEX idx_faces_identified_by ON faces(identified_by_user_id)")
)
except Exception:
pass # Index might already exist
print("✅ Added identified_by_user_id column to faces table")
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Lifespan context manager for startup and shutdown events."""
@@ -255,6 +296,7 @@ async def lifespan(app: FastAPI):
ensure_user_password_hash_column(inspector)
ensure_user_password_change_required_column(inspector)
ensure_user_email_unique_constraint(inspector)
ensure_face_identified_by_user_id_column(inspector)
except Exception as exc:
print(f"❌ Database initialization failed: {exc}")
raise
src/web/db/models.py
+2
View File
@@ -102,6 +102,7 @@ class Face(Base):
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)
photo = relationship("Photo", back_populates="faces")
person = relationship("Person", back_populates="faces")
@@ -114,6 +115,7 @@ class Face(Base):
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"),
)
src/web/services/face_service.py
+9
View File
@@ -1988,6 +1988,7 @@ def accept_auto_match_matches(
db: Session,
person_id: int,
face_ids: List[int],
user_id: int | None = None,
) -> Tuple[int, int]:
"""Accept auto-match matches for a person, matching desktop logic exactly.
@@ -1996,6 +1997,12 @@ def accept_auto_match_matches(
2. Insert person_encodings for each identified face
3. Update person encodings (remove old, add current)
Args:
db: Database session
person_id: Person ID to identify faces with
face_ids: List of face IDs to identify
user_id: User ID who is performing the identification (optional)
Returns:
(identified_count, updated_count) tuple
"""
@@ -2017,6 +2024,8 @@ def accept_auto_match_matches(
for face in faces:
# Set person_id on face
face.person_id = person_id
if user_id is not None:
face.identified_by_user_id = user_id
db.add(face)
# Insert person_encoding (matching desktop)