""" Service layer per la gestione delle firme automatiche. """ import uuid from typing import Sequence from sqlalchemy import select, func from sqlalchemy.ext.asyncio import AsyncSession from app.core.exceptions import NotFoundError, ValidationError from app.models.signature import Signature, SignatureAssignment from app.schemas.signature import SignatureCreate, SignatureUpdate, SignatureAssignmentCreate class SignatureService: def __init__(self, db: AsyncSession) -> None: self.db = db # ─── Firme ──────────────────────────────────────────────────────────────── async def list_signatures( self, tenant_id: uuid.UUID, q: str | None = None ) -> tuple[Sequence[Signature], int]: stmt = select(Signature).where(Signature.tenant_id == tenant_id) if q: stmt = stmt.where(Signature.name.ilike(f"%{q}%")) stmt = stmt.order_by(Signature.name) count_stmt = select(func.count()).select_from(stmt.subquery()) total = (await self.db.execute(count_stmt)).scalar_one() items = (await self.db.execute(stmt)).scalars().all() return items, total async def get_signature( self, tenant_id: uuid.UUID, signature_id: uuid.UUID ) -> Signature: stmt = select(Signature).where( Signature.id == signature_id, Signature.tenant_id == tenant_id, ) sig = (await self.db.execute(stmt)).scalar_one_or_none() if sig is None: raise NotFoundError("Firma non trovata") return sig async def create_signature( self, tenant_id: uuid.UUID, data: SignatureCreate, created_by: uuid.UUID | None = None, ) -> Signature: sig = Signature( tenant_id=tenant_id, name=data.name, description=data.description, body_html=data.body_html, body_text=data.body_text, created_by=created_by, ) self.db.add(sig) await self.db.commit() await self.db.refresh(sig) return sig async def update_signature( self, tenant_id: uuid.UUID, signature_id: uuid.UUID, data: SignatureUpdate, ) -> Signature: sig = await self.get_signature(tenant_id, signature_id) if data.name is not None: sig.name = data.name if data.description is not None: sig.description = data.description if data.body_html is not None: sig.body_html = data.body_html if data.body_text is not None: sig.body_text = data.body_text await self.db.commit() await self.db.refresh(sig) return sig async def delete_signature( self, tenant_id: uuid.UUID, signature_id: uuid.UUID ) -> None: sig = await self.get_signature(tenant_id, signature_id) await self.db.delete(sig) await self.db.commit() # ─── Assegnazioni ───────────────────────────────────────────────────────── async def list_assignments( self, tenant_id: uuid.UUID, mailbox_id: uuid.UUID | None = None, virtual_box_id: uuid.UUID | None = None, ) -> tuple[list[dict], int]: """ Restituisce le assegnazioni con il nome della firma incluso. Filtro opzionale per casella o virtual box. """ stmt = ( select(SignatureAssignment, Signature.name.label("signature_name")) .join(Signature, Signature.id == SignatureAssignment.signature_id) .where(SignatureAssignment.tenant_id == tenant_id) ) if mailbox_id: stmt = stmt.where(SignatureAssignment.mailbox_id == mailbox_id) if virtual_box_id: stmt = stmt.where(SignatureAssignment.virtual_box_id == virtual_box_id) stmt = stmt.order_by(SignatureAssignment.created_at) rows = (await self.db.execute(stmt)).all() items = [] for row in rows: assignment: SignatureAssignment = row[0] sig_name: str = row[1] items.append({ "id": assignment.id, "tenant_id": assignment.tenant_id, "signature_id": assignment.signature_id, "mailbox_id": assignment.mailbox_id, "virtual_box_id": assignment.virtual_box_id, "context": assignment.context, "created_at": assignment.created_at, "signature_name": sig_name, }) return items, len(items) async def create_assignment( self, tenant_id: uuid.UUID, data: SignatureAssignmentCreate, ) -> dict: # Validazione: esattamente uno tra mailbox_id e virtual_box_id if bool(data.mailbox_id) == bool(data.virtual_box_id): raise ValidationError( "Specificare esattamente uno tra mailbox_id e virtual_box_id" ) # Verifica che la firma esista nel tenant await self.get_signature(tenant_id, data.signature_id) # Rimuovi eventuale assegnazione precedente per la stessa coppia (target, context) existing_stmt = select(SignatureAssignment).where( SignatureAssignment.tenant_id == tenant_id, SignatureAssignment.context == data.context, ) if data.mailbox_id: existing_stmt = existing_stmt.where( SignatureAssignment.mailbox_id == data.mailbox_id ) else: existing_stmt = existing_stmt.where( SignatureAssignment.virtual_box_id == data.virtual_box_id ) existing = (await self.db.execute(existing_stmt)).scalar_one_or_none() if existing: await self.db.delete(existing) await self.db.flush() # Flush il DELETE prima dell'INSERT per evitare UniqueViolationError assignment = SignatureAssignment( tenant_id=tenant_id, signature_id=data.signature_id, mailbox_id=data.mailbox_id, virtual_box_id=data.virtual_box_id, context=data.context, ) self.db.add(assignment) await self.db.commit() await self.db.refresh(assignment) # Carica il nome della firma per la risposta sig = await self.get_signature(tenant_id, data.signature_id) return { "id": assignment.id, "tenant_id": assignment.tenant_id, "signature_id": assignment.signature_id, "mailbox_id": assignment.mailbox_id, "virtual_box_id": assignment.virtual_box_id, "context": assignment.context, "created_at": assignment.created_at, "signature_name": sig.name, } async def delete_assignment( self, tenant_id: uuid.UUID, assignment_id: uuid.UUID ) -> None: stmt = select(SignatureAssignment).where( SignatureAssignment.id == assignment_id, SignatureAssignment.tenant_id == tenant_id, ) assignment = (await self.db.execute(stmt)).scalar_one_or_none() if assignment is None: raise NotFoundError("Assegnazione firma non trovata") await self.db.delete(assignment) await self.db.commit() async def resolve_signature( self, tenant_id: uuid.UUID, context: str, mailbox_id: uuid.UUID | None = None, virtual_box_id: uuid.UUID | None = None, ) -> Signature | None: """ Restituisce la firma assegnata per casella/vbox nel contesto specificato. Cerca prima un'assegnazione con context == context, poi context == 'both'. """ if not mailbox_id and not virtual_box_id: return None stmt = ( select(SignatureAssignment) .where( SignatureAssignment.tenant_id == tenant_id, SignatureAssignment.context.in_([context, "both"]), ) ) if mailbox_id: stmt = stmt.where(SignatureAssignment.mailbox_id == mailbox_id) else: stmt = stmt.where(SignatureAssignment.virtual_box_id == virtual_box_id) assignments = (await self.db.execute(stmt)).scalars().all() if not assignments: return None # Preferisce il match esatto sul contesto rispetto a 'both' exact = next((a for a in assignments if a.context == context), None) assignment = exact or assignments[0] return await self.get_signature(tenant_id, assignment.signature_id) async def delete_assignment_by_target( self, tenant_id: uuid.UUID, context: str, mailbox_id: uuid.UUID | None = None, virtual_box_id: uuid.UUID | None = None, ) -> None: """Rimuove l'assegnazione per una specifica casella/vbox+contesto (se presente).""" stmt = select(SignatureAssignment).where( SignatureAssignment.tenant_id == tenant_id, SignatureAssignment.context == context, ) if mailbox_id: stmt = stmt.where(SignatureAssignment.mailbox_id == mailbox_id) elif virtual_box_id: stmt = stmt.where(SignatureAssignment.virtual_box_id == virtual_box_id) else: return assignment = (await self.db.execute(stmt)).scalar_one_or_none() if assignment: await self.db.delete(assignment) await self.db.commit()