import hashlib import hmac import secrets from datetime import UTC, datetime from fastapi import HTTPException, status from sqlalchemy.orm import Session from starlette.requests import Request from app.core.config import settings from app.models.repair import Repair, RepairPublicAccessToken from app.models.user import User from app.repositories.repair_repository import RepairRepository from app.repositories.repair_estimate_repository import RepairEstimateRepository from app.schemas.repair import ( RepairPublicLinkCreateResponse, RepairPublicLinkResponse, RepairPublicStatusHistoryItem, RepairPublicStatusResponse, ) from app.services.audit_service import write_audit_log from app.services.repair_status_labels import STATUS_LABELS from app.services.system_settings_service import SystemSettingsService def normalize_token(token: str) -> str: return token.strip() def repair_label(repair: Repair) -> str: return f"{repair.repair_number} ยท {repair.customer_name}" def public_status_path(db: Session, token: str) -> str: normalized_token = normalize_token(token) path = f"/status/{normalized_token}" base_url = SystemSettingsService.get_public_repair_status_base_url(db) if not base_url: return path return f"{base_url}/{normalized_token}" class RepairPublicLinkService: @staticmethod def hash_token(token: str) -> str: normalized_token = normalize_token(token) return hmac.new(settings.secret_key.encode("utf-8"), normalized_token.encode("utf-8"), hashlib.sha256).hexdigest() @staticmethod def create_token() -> str: return secrets.token_urlsafe(48) @staticmethod def response_from_link(public_link: RepairPublicAccessToken | None) -> RepairPublicLinkResponse: if public_link is None: return RepairPublicLinkResponse(is_active=False) return RepairPublicLinkResponse( is_active=public_link.is_active, token_hint=public_link.token_hint, expires_at=public_link.expires_at, last_used_at=public_link.last_used_at, created_at=public_link.created_at, revoked_at=public_link.revoked_at, ) @staticmethod def get(db: Session, repair: Repair) -> RepairPublicLinkResponse: return RepairPublicLinkService.response_from_link(RepairRepository.get_active_public_link(db, repair.id)) @staticmethod def create(db: Session, repair: Repair, *, actor: User, request: Request) -> RepairPublicLinkCreateResponse: return RepairPublicLinkService.create_with_audit( db, repair, actor=actor, request=request, audit_action="repairs.public_link.create", ) @staticmethod def create_with_audit( db: Session, repair: Repair, *, actor: User | None, request: Request | None, audit_action: str, ) -> RepairPublicLinkCreateResponse: token = RepairPublicLinkService.create_token() public_link = RepairRepository.create_public_link( db, repair_id=repair.id, token_hash=RepairPublicLinkService.hash_token(token), token_hint=token[-6:], ) write_audit_log( db, action=audit_action, entity_type="repairs", entity_id=repair.id, entity_label=repair_label(repair), actor=actor, request=request, metadata={"token_hint": public_link.token_hint}, ) return RepairPublicLinkCreateResponse( is_active=True, token=token, token_hint=public_link.token_hint, expires_at=public_link.expires_at, last_used_at=public_link.last_used_at, created_at=public_link.created_at, revoked_at=public_link.revoked_at, public_status_path=public_status_path(db, token), ) @staticmethod def revoke(db: Session, repair: Repair, *, actor: User, request: Request) -> RepairPublicLinkResponse: active_link = RepairRepository.get_active_public_link(db, repair.id) RepairRepository.revoke_public_links(db, repair_id=repair.id) write_audit_log( db, action="repairs.public_link.revoke", entity_type="repairs", entity_id=repair.id, entity_label=repair_label(repair), actor=actor, request=request, metadata={"token_hint": active_link.token_hint if active_link else None}, ) return RepairPublicLinkResponse(is_active=False, token_hint=active_link.token_hint if active_link else None, revoked_at=datetime.now(UTC)) @staticmethod def public_status(db: Session, token: str) -> RepairPublicStatusResponse: public_link = RepairPublicLinkService.get_public_link_or_404(db, token) repair = public_link.repair history = RepairRepository.get_history_public(db, repair.id) estimate = RepairEstimateRepository.get_latest_public(db, repair.id) from app.services.repair_estimate_service import RepairEstimateService response = RepairPublicStatusResponse( repair_number=repair.repair_number, public_status_label=STATUS_LABELS.get(repair.status, repair.status), device_manufacturer=repair.device_manufacturer, device_model=repair.device_model, status_history_public=[ RepairPublicStatusHistoryItem( status=item.new_status, status_label=STATUS_LABELS.get(item.new_status, item.new_status), created_at=item.created_at, ) for item in history ], updated_at=repair.updated_at, estimate=RepairEstimateService.public_response(estimate), ) RepairRepository.mark_public_link_used(db, public_link) return response @staticmethod def get_public_link_or_404(db: Session, token: str) -> RepairPublicAccessToken: normalized_token = normalize_token(token) if not normalized_token: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Reparaturstatus nicht gefunden") public_link = RepairRepository.get_public_link_by_hash(db, RepairPublicLinkService.hash_token(normalized_token)) if public_link is None or public_link.repair is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Reparaturstatus nicht gefunden") return public_link