322 lines
10 KiB
Python
322 lines
10 KiB
Python
"""Consent router — per-video consent toggles with versioned audit trail.
|
|
|
|
Creator endpoints (ownership-gated):
|
|
GET /consent/videos List consent for the current creator's videos
|
|
GET /consent/videos/{video_id} Single video consent status
|
|
PUT /consent/videos/{video_id} Upsert consent (partial update, audit logged)
|
|
GET /consent/videos/{video_id}/history Audit trail for a video
|
|
|
|
Admin endpoint:
|
|
GET /consent/admin/summary Aggregate consent flag counts
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import uuid
|
|
from typing import Annotated
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, Request, status
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
from auth import get_current_user, reject_impersonation, require_role
|
|
from database import get_session
|
|
from models import (
|
|
ConsentAuditLog,
|
|
ConsentField,
|
|
SourceVideo,
|
|
User,
|
|
UserRole,
|
|
VideoConsent,
|
|
)
|
|
from schemas import (
|
|
ConsentAuditEntry,
|
|
ConsentListResponse,
|
|
ConsentSummary,
|
|
VideoConsentRead,
|
|
VideoConsentUpdate,
|
|
)
|
|
|
|
logger = logging.getLogger("chrysopedia.consent")
|
|
|
|
router = APIRouter(prefix="/consent", tags=["consent"])
|
|
|
|
|
|
# ── Helpers ──────────────────────────────────────────────────────────────────
|
|
|
|
|
|
async def _verify_video_ownership(
|
|
video_id: uuid.UUID,
|
|
user: User,
|
|
session: AsyncSession,
|
|
) -> SourceVideo:
|
|
"""Load a SourceVideo and verify the user owns it (or is admin).
|
|
|
|
Returns the SourceVideo on success.
|
|
Raises 403 if user has no creator_id or doesn't own the video.
|
|
Raises 404 if video doesn't exist.
|
|
"""
|
|
result = await session.execute(
|
|
select(SourceVideo).where(SourceVideo.id == video_id)
|
|
)
|
|
video = result.scalar_one_or_none()
|
|
if video is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Video not found",
|
|
)
|
|
|
|
# Admin bypasses ownership check
|
|
if user.role == UserRole.admin:
|
|
return video
|
|
|
|
if user.creator_id is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="User is not linked to a creator profile",
|
|
)
|
|
|
|
if video.creator_id != user.creator_id:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="You do not own this video",
|
|
)
|
|
|
|
return video
|
|
|
|
|
|
def _consent_to_read(consent: VideoConsent, filename: str) -> VideoConsentRead:
|
|
"""Map a VideoConsent ORM instance to the read schema."""
|
|
return VideoConsentRead(
|
|
source_video_id=consent.source_video_id,
|
|
video_filename=filename,
|
|
creator_id=consent.creator_id,
|
|
kb_inclusion=consent.kb_inclusion,
|
|
training_usage=consent.training_usage,
|
|
public_display=consent.public_display,
|
|
updated_at=consent.updated_at,
|
|
)
|
|
|
|
|
|
# ── Endpoints ────────────────────────────────────────────────────────────────
|
|
|
|
|
|
@router.get("/videos", response_model=ConsentListResponse)
|
|
async def list_video_consents(
|
|
current_user: Annotated[User, Depends(get_current_user)],
|
|
session: Annotated[AsyncSession, Depends(get_session)],
|
|
offset: Annotated[int, Query(ge=0)] = 0,
|
|
limit: Annotated[int, Query(ge=1, le=100)] = 50,
|
|
):
|
|
"""List consent records for the current creator's videos."""
|
|
if current_user.creator_id is None and current_user.role != UserRole.admin:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="User is not linked to a creator profile",
|
|
)
|
|
|
|
stmt = (
|
|
select(VideoConsent)
|
|
.join(SourceVideo, VideoConsent.source_video_id == SourceVideo.id)
|
|
.options(selectinload(VideoConsent.source_video))
|
|
)
|
|
|
|
# Non-admin sees only their own videos
|
|
if current_user.role != UserRole.admin:
|
|
stmt = stmt.where(VideoConsent.creator_id == current_user.creator_id)
|
|
|
|
# Count
|
|
count_stmt = select(func.count()).select_from(stmt.subquery())
|
|
total = (await session.execute(count_stmt)).scalar() or 0
|
|
|
|
# Fetch page
|
|
stmt = stmt.order_by(VideoConsent.updated_at.desc())
|
|
stmt = stmt.offset(offset).limit(limit)
|
|
result = await session.execute(stmt)
|
|
consents = result.scalars().all()
|
|
|
|
items = [
|
|
_consent_to_read(c, c.source_video.filename) for c in consents
|
|
]
|
|
return ConsentListResponse(items=items, total=total)
|
|
|
|
|
|
@router.get("/videos/{video_id}", response_model=VideoConsentRead)
|
|
async def get_video_consent(
|
|
video_id: uuid.UUID,
|
|
current_user: Annotated[User, Depends(get_current_user)],
|
|
session: Annotated[AsyncSession, Depends(get_session)],
|
|
):
|
|
"""Get consent status for a single video."""
|
|
video = await _verify_video_ownership(video_id, current_user, session)
|
|
|
|
result = await session.execute(
|
|
select(VideoConsent).where(VideoConsent.source_video_id == video_id)
|
|
)
|
|
consent = result.scalar_one_or_none()
|
|
if consent is None:
|
|
# No consent record yet — return defaults
|
|
return VideoConsentRead(
|
|
source_video_id=video_id,
|
|
video_filename=video.filename,
|
|
creator_id=video.creator_id,
|
|
kb_inclusion=False,
|
|
training_usage=False,
|
|
public_display=True,
|
|
updated_at=video.created_at,
|
|
)
|
|
|
|
return _consent_to_read(consent, video.filename)
|
|
|
|
|
|
@router.put("/videos/{video_id}", response_model=VideoConsentRead)
|
|
async def update_video_consent(
|
|
video_id: uuid.UUID,
|
|
body: VideoConsentUpdate,
|
|
current_user: Annotated[User, Depends(reject_impersonation)],
|
|
session: Annotated[AsyncSession, Depends(get_session)],
|
|
request: Request,
|
|
):
|
|
"""Upsert consent for a video. Only non-None fields are changed.
|
|
|
|
Creates audit log entries for each changed field with incrementing
|
|
version numbers.
|
|
"""
|
|
video = await _verify_video_ownership(video_id, current_user, session)
|
|
|
|
# Load or create consent record
|
|
result = await session.execute(
|
|
select(VideoConsent).where(VideoConsent.source_video_id == video_id)
|
|
)
|
|
consent = result.scalar_one_or_none()
|
|
is_new = consent is None
|
|
|
|
if is_new:
|
|
consent = VideoConsent(
|
|
source_video_id=video_id,
|
|
creator_id=video.creator_id,
|
|
updated_by=current_user.id,
|
|
)
|
|
session.add(consent)
|
|
await session.flush() # get consent.id for audit entries
|
|
|
|
# Determine the next version number
|
|
max_version_result = await session.execute(
|
|
select(func.coalesce(func.max(ConsentAuditLog.version), 0)).where(
|
|
ConsentAuditLog.video_consent_id == consent.id
|
|
)
|
|
)
|
|
next_version = (max_version_result.scalar() or 0) + 1
|
|
|
|
# Collect client IP for audit
|
|
client_ip = request.client.host if request.client else None
|
|
|
|
# Apply changes and build audit entries
|
|
fields_changed: list[str] = []
|
|
update_data = body.model_dump(exclude_none=True)
|
|
|
|
for field_name, new_value in update_data.items():
|
|
# Validate field name against the enum
|
|
try:
|
|
ConsentField(field_name)
|
|
except ValueError:
|
|
continue
|
|
|
|
old_value = getattr(consent, field_name)
|
|
|
|
# Skip if no actual change
|
|
if old_value == new_value:
|
|
continue
|
|
|
|
# Update the consent record
|
|
setattr(consent, field_name, new_value)
|
|
fields_changed.append(field_name)
|
|
|
|
# Create audit entry
|
|
audit_entry = ConsentAuditLog(
|
|
video_consent_id=consent.id,
|
|
version=next_version,
|
|
field_name=field_name,
|
|
old_value=old_value if not is_new else None,
|
|
new_value=new_value,
|
|
changed_by=current_user.id,
|
|
ip_address=client_ip,
|
|
)
|
|
session.add(audit_entry)
|
|
next_version += 1
|
|
|
|
if fields_changed:
|
|
consent.updated_by = current_user.id
|
|
await session.commit()
|
|
await session.refresh(consent)
|
|
|
|
logger.info(
|
|
"Consent updated: video_id=%s fields_changed=%s user=%s",
|
|
video_id,
|
|
fields_changed,
|
|
current_user.id,
|
|
)
|
|
else:
|
|
# No actual changes — still commit if we created a new record
|
|
if is_new:
|
|
await session.commit()
|
|
await session.refresh(consent)
|
|
else:
|
|
# Nothing changed, no audit entries
|
|
pass
|
|
|
|
return _consent_to_read(consent, video.filename)
|
|
|
|
|
|
@router.get("/videos/{video_id}/history", response_model=list[ConsentAuditEntry])
|
|
async def get_consent_history(
|
|
video_id: uuid.UUID,
|
|
current_user: Annotated[User, Depends(get_current_user)],
|
|
session: Annotated[AsyncSession, Depends(get_session)],
|
|
):
|
|
"""Get the audit trail for a video's consent changes."""
|
|
await _verify_video_ownership(video_id, current_user, session)
|
|
|
|
# Find the consent record
|
|
result = await session.execute(
|
|
select(VideoConsent).where(VideoConsent.source_video_id == video_id)
|
|
)
|
|
consent = result.scalar_one_or_none()
|
|
if consent is None:
|
|
return []
|
|
|
|
# Fetch audit entries ordered by version
|
|
audit_result = await session.execute(
|
|
select(ConsentAuditLog)
|
|
.where(ConsentAuditLog.video_consent_id == consent.id)
|
|
.order_by(ConsentAuditLog.version.asc())
|
|
)
|
|
return audit_result.scalars().all()
|
|
|
|
|
|
@router.get(
|
|
"/admin/summary",
|
|
response_model=ConsentSummary,
|
|
dependencies=[Depends(require_role(UserRole.admin))],
|
|
)
|
|
async def consent_admin_summary(
|
|
session: Annotated[AsyncSession, Depends(get_session)],
|
|
):
|
|
"""Aggregate consent flag counts across all videos (admin only)."""
|
|
result = await session.execute(
|
|
select(
|
|
func.count().label("total"),
|
|
func.sum(VideoConsent.kb_inclusion.cast(int)).label("kb"),
|
|
func.sum(VideoConsent.training_usage.cast(int)).label("tu"),
|
|
func.sum(VideoConsent.public_display.cast(int)).label("pd"),
|
|
)
|
|
)
|
|
row = result.one()
|
|
return ConsentSummary(
|
|
total_videos=row.total or 0,
|
|
kb_inclusion_granted=row.kb or 0,
|
|
training_usage_granted=row.tu or 0,
|
|
public_display_granted=row.pd or 0,
|
|
)
|