73 lines
2.3 KiB
Python
73 lines
2.3 KiB
Python
import os
|
|
import shutil
|
|
import uuid
|
|
import datetime
|
|
from sqlalchemy.orm import Session
|
|
from typing import Optional
|
|
from fastapi import HTTPException, UploadFile
|
|
from app.models.non_conformity import NonConformity
|
|
from app.models.evidence import Evidence
|
|
from app.schemas.non_conformity import NonConformityUpdate
|
|
|
|
UPLOAD_DIR = "uploads"
|
|
|
|
class GuestService:
|
|
def __init__(self, db: Session):
|
|
self._db = db
|
|
|
|
def get_nc_by_hash(self, access_hash: str) -> NonConformity:
|
|
db_nc = self._db.query(NonConformity).filter(NonConformity.access_hash == access_hash).first()
|
|
if not db_nc:
|
|
raise HTTPException(status_code=404, detail="Non-Conformity not found or invalid link")
|
|
return db_nc
|
|
|
|
def update_guest_nc(self, access_hash: str, nc_update: NonConformityUpdate) -> NonConformity:
|
|
db_nc = self.get_nc_by_hash(access_hash)
|
|
|
|
if nc_update.guest_actions is not None:
|
|
db_nc.guest_actions = nc_update.guest_actions
|
|
|
|
if nc_update.closure_description is not None:
|
|
db_nc.closure_description = nc_update.closure_description
|
|
|
|
if nc_update.status is not None:
|
|
db_nc.status = nc_update.status
|
|
|
|
self._db.commit()
|
|
self._db.refresh(db_nc)
|
|
return db_nc
|
|
|
|
def upload_guest_evidence(
|
|
self,
|
|
access_hash: str,
|
|
file: UploadFile,
|
|
description: Optional[str] = None
|
|
) -> Evidence:
|
|
db_nc = self.get_nc_by_hash(access_hash)
|
|
|
|
# Generate unique filename
|
|
file_ext = os.path.splitext(file.filename)[1]
|
|
unique_filename = f"guest_nc_{uuid.uuid4()}{file_ext}"
|
|
file_path = os.path.join(UPLOAD_DIR, unique_filename)
|
|
|
|
# Save file
|
|
if not os.path.exists(UPLOAD_DIR):
|
|
os.makedirs(UPLOAD_DIR)
|
|
|
|
with open(file_path, "wb") as buffer:
|
|
shutil.copyfileobj(file.file, buffer)
|
|
|
|
# Save to database
|
|
db_evidence = Evidence(
|
|
non_conformity_id=db_nc.id,
|
|
file_path=file_path,
|
|
media_type=file.content_type,
|
|
description=description,
|
|
captured_at=datetime.datetime.utcnow()
|
|
)
|
|
self._db.add(db_evidence)
|
|
self._db.commit()
|
|
self._db.refresh(db_evidence)
|
|
|
|
return db_evidence
|