"""Internal worklog service — CRUD with integrity hashing.""" import hashlib import logging from datetime import datetime from typing import Optional from uuid import UUID from sqlalchemy.orm import Session from app.models.worklog import Worklog logger = logging.getLogger(__name__) def create_worklog( db: Session, *, entity_type: str, entity_id: UUID, user_id: UUID, activity_type: str, started_at: datetime, duration_seconds: int, ended_at: Optional[datetime] = None, description: Optional[str] = None, ) -> Worklog: """Create a worklog with an auto-computed integrity hash.""" wl = Worklog( entity_type=entity_type, entity_id=entity_id, user_id=user_id, activity_type=activity_type, started_at=started_at, ended_at=ended_at, duration_seconds=duration_seconds, description=description, ) wl.integrity_hash = _compute_hash(wl) db.add(wl) db.commit() db.refresh(wl) return wl def list_worklogs( db: Session, *, entity_type: Optional[str] = None, entity_id: Optional[UUID] = None, user_id: Optional[UUID] = None, ) -> list[Worklog]: """List worklogs with optional filters.""" query = db.query(Worklog) if entity_type: query = query.filter(Worklog.entity_type == entity_type) if entity_id: query = query.filter(Worklog.entity_id == entity_id) if user_id: query = query.filter(Worklog.user_id == user_id) return query.order_by(Worklog.started_at.desc()).all() def verify_worklog_integrity(wl: Worklog) -> bool: """Return True if the worklog has not been tampered with.""" return wl.integrity_hash == _compute_hash(wl) def _compute_hash(wl: Worklog) -> str: """SHA-256 of the immutable fields for audit integrity.""" data = ( f"{wl.entity_type}:{wl.entity_id}:{wl.user_id}:" f"{wl.activity_type}:{wl.started_at}:{wl.duration_seconds}" ) return hashlib.sha256(data.encode()).hexdigest()