225 lines
8.2 KiB
Python
225 lines
8.2 KiB
Python
import mimetypes
|
|
from pathlib import Path
|
|
from typing import Literal
|
|
from urllib.parse import quote
|
|
|
|
from fastapi import HTTPException, UploadFile, status
|
|
from sqlalchemy.orm import Session
|
|
from starlette.requests import Request
|
|
|
|
from app.models.repair import Repair, RepairDocument
|
|
from app.models.user import User
|
|
from app.repositories.repair_repository import RepairRepository
|
|
from app.schemas.repair import RepairDocumentUpdate
|
|
from app.services.audit_service import write_audit_log
|
|
from app.storage import get_storage_service
|
|
from app.storage.exceptions import StorageFileNotFoundError, StorageValidationError
|
|
|
|
ALLOWED_REPAIR_DOCUMENT_EXTENSIONS = {".jpg", ".jpeg", ".png", ".webp", ".pdf"}
|
|
ALLOWED_REPAIR_DOCUMENT_MIME_TYPES = {
|
|
"image/jpeg",
|
|
"image/png",
|
|
"image/webp",
|
|
"application/pdf",
|
|
}
|
|
|
|
|
|
DOCUMENT_TYPE_LABELS = {
|
|
"device_photo": "Gerätefoto",
|
|
"fault_photo": "Fehlerbild",
|
|
"measurement": "Messbild",
|
|
"estimate": "Kostenvoranschlag",
|
|
"repair_report": "Reparaturbericht",
|
|
"shipping": "Versandbeleg",
|
|
"other": "Dokument",
|
|
}
|
|
|
|
|
|
def storage_validation_error(exc: StorageValidationError) -> HTTPException:
|
|
detail = str(exc) or "Ungültige Datei"
|
|
status_code = status.HTTP_413_REQUEST_ENTITY_TOO_LARGE if "groß" in detail else status.HTTP_400_BAD_REQUEST
|
|
return HTTPException(status_code=status_code, detail=detail)
|
|
|
|
|
|
def _repair_label(repair: Repair) -> str:
|
|
return f"{repair.repair_number} · {repair.customer_name}"
|
|
|
|
|
|
def _document_label(document: RepairDocument) -> str:
|
|
return f"{document.title} · {document.original_filename}"
|
|
|
|
|
|
def _audit_document_data(document: RepairDocument) -> dict:
|
|
return {
|
|
"id": document.id,
|
|
"repair_id": document.repair_id,
|
|
"title": document.title,
|
|
"document_type": document.document_type,
|
|
"original_filename": document.original_filename,
|
|
"mime_type": document.mime_type,
|
|
"size_bytes": document.size_bytes,
|
|
"visibility": document.visibility,
|
|
"uploaded_by_user_id": document.uploaded_by_user_id,
|
|
"created_at": document.created_at,
|
|
"updated_at": document.updated_at,
|
|
}
|
|
|
|
|
|
class RepairDocumentService:
|
|
@staticmethod
|
|
async def upload_document(
|
|
db: Session,
|
|
repair: Repair,
|
|
*,
|
|
file: UploadFile,
|
|
title: str,
|
|
document_type: str,
|
|
visibility: str,
|
|
note: str | None,
|
|
actor: User,
|
|
request: Request,
|
|
) -> RepairDocument:
|
|
storage_service = get_storage_service()
|
|
max_bytes = storage_service.max_upload_mb * 1024 * 1024
|
|
content = await file.read(max_bytes + 1)
|
|
original_filename = file.filename or ""
|
|
mime_type = RepairDocumentService._resolve_mime_type(original_filename, file.content_type)
|
|
|
|
RepairDocumentService._validate_repair_document_type(original_filename, mime_type)
|
|
|
|
try:
|
|
metadata = storage_service.save_file(
|
|
namespace=f"repairs/{repair.id}/documents",
|
|
content=content,
|
|
original_filename=original_filename,
|
|
mime_type=mime_type,
|
|
)
|
|
except StorageValidationError as exc:
|
|
raise storage_validation_error(exc) from exc
|
|
|
|
try:
|
|
document = RepairRepository.create_document(
|
|
db,
|
|
repair_id=repair.id,
|
|
title=title,
|
|
document_type=document_type,
|
|
original_filename=metadata.original_filename,
|
|
stored_filename=metadata.stored_filename,
|
|
storage_path=metadata.storage_key,
|
|
mime_type=metadata.mime_type,
|
|
size_bytes=metadata.size,
|
|
checksum_sha256=metadata.checksum_sha256,
|
|
visibility=visibility,
|
|
note=note,
|
|
uploaded_by_user_id=actor.id,
|
|
)
|
|
except Exception:
|
|
storage_service.delete_file(metadata.storage_key)
|
|
raise
|
|
|
|
write_audit_log(
|
|
db,
|
|
action="repairs.documents.upload",
|
|
entity_type="repair_documents",
|
|
entity_id=document.id,
|
|
entity_label=_document_label(document),
|
|
actor=actor,
|
|
request=request,
|
|
metadata={
|
|
"repair_id": repair.id,
|
|
"repair_number": repair.repair_number,
|
|
"document_type": document.document_type,
|
|
"visibility": document.visibility,
|
|
"mime_type": document.mime_type,
|
|
"size_bytes": document.size_bytes,
|
|
},
|
|
)
|
|
return document
|
|
|
|
@staticmethod
|
|
def update_document(
|
|
db: Session,
|
|
repair: Repair,
|
|
document: RepairDocument,
|
|
payload: RepairDocumentUpdate,
|
|
*,
|
|
actor: User,
|
|
request: Request,
|
|
) -> RepairDocument:
|
|
before_data = _audit_document_data(document)
|
|
updated = RepairRepository.update_document(db, document, payload)
|
|
write_audit_log(
|
|
db,
|
|
action="repairs.documents.update",
|
|
entity_type="repair_documents",
|
|
entity_id=updated.id,
|
|
entity_label=_document_label(updated),
|
|
actor=actor,
|
|
request=request,
|
|
before_data=before_data,
|
|
after_data=_audit_document_data(updated),
|
|
metadata={"repair_id": repair.id, "repair_number": repair.repair_number},
|
|
)
|
|
return updated
|
|
|
|
@staticmethod
|
|
def delete_document(
|
|
db: Session,
|
|
repair: Repair,
|
|
document: RepairDocument,
|
|
*,
|
|
actor: User,
|
|
request: Request,
|
|
) -> None:
|
|
before_data = _audit_document_data(document)
|
|
storage_path = document.storage_path
|
|
label = _document_label(document)
|
|
try:
|
|
get_storage_service().delete_file(storage_path)
|
|
except StorageValidationError as exc:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Ungültiger Dateipfad") from exc
|
|
|
|
RepairRepository.delete_document(db, document)
|
|
write_audit_log(
|
|
db,
|
|
action="repairs.documents.delete",
|
|
entity_type="repair_documents",
|
|
entity_id=document.id,
|
|
entity_label=label,
|
|
actor=actor,
|
|
request=request,
|
|
before_data=before_data,
|
|
metadata={"repair_id": repair.id, "repair_number": repair.repair_number},
|
|
)
|
|
|
|
@staticmethod
|
|
def open_document_file(document: RepairDocument) -> Path:
|
|
if not document.storage_path:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Datei nicht gefunden")
|
|
try:
|
|
return get_storage_service().open_file(document.storage_path)
|
|
except StorageFileNotFoundError as exc:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Datei nicht gefunden") from exc
|
|
except StorageValidationError as exc:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Ungültiger Dateipfad") from exc
|
|
|
|
@staticmethod
|
|
def content_disposition(document: RepairDocument, *, mode: Literal["inline", "attachment"]) -> str:
|
|
filename = document.original_filename or document.stored_filename or "reparatur-dokument"
|
|
safe_filename = filename.replace('"', "")
|
|
encoded_filename = quote(filename)
|
|
return f'{mode}; filename="{safe_filename}"; filename*=UTF-8\'\'{encoded_filename}'
|
|
|
|
@staticmethod
|
|
def _resolve_mime_type(original_filename: str, content_type: str | None) -> str:
|
|
if content_type and content_type != "application/octet-stream":
|
|
return content_type
|
|
return mimetypes.guess_type(original_filename)[0] or "application/octet-stream"
|
|
|
|
@staticmethod
|
|
def _validate_repair_document_type(original_filename: str, mime_type: str) -> None:
|
|
extension = Path(original_filename or "").suffix.lower()
|
|
if extension not in ALLOWED_REPAIR_DOCUMENT_EXTENSIONS:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Dateityp ist nicht erlaubt")
|
|
if mime_type not in ALLOWED_REPAIR_DOCUMENT_MIME_TYPES:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="MIME-Type ist nicht erlaubt")
|