BFLOWS/gepafin-rendicontazione-api
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
83bb0a29ec0808e866530a458afa4689047e5992
gepafin-rendicontazione-api/app/routers/practices.py
BFLOWS 83bb0a29ec feat(auth): autorizza ROLE_CONFIDI come proprietario pratica (parallelo BENEFICIARY) 
Risoluzione 403 segnalato da Rinaldo Bonazzo su upload fattura con utente
ROLE_CONFIDI (confidi4@test.test). Pattern allineato al BE Gepafin che
in DashboardDao, CompanyDocumentDao e FaqDao raggruppa BENEFICIARY+CONFIDI
con stessi diritti operativi sulla pratica.

==RAZIONALE==
Sui bandi con call.confidi=true il confidi sottomette la application
per conto dell'azienda e diventa user_id della application. Lato
microservizio rendicontazione la pratica viene ereditata con stesso
user_id, quindi il confidi e proprietario della pratica e deve poter
fare upload/download/delete come il beneficiario.

==MODIFICHE==

app/auth.py:
- Aggiunto AuthUser.is_confidi() — controlla ROLE_CONFIDI
- Aggiunto AuthUser.is_owner_role() — True per BENEFICIARY o CONFIDI
- Aggiornato docstring header con ROLE_CONFIDI
- Manteno is_beneficiary() per backward compat (non rimosso, non chiamato)

Sostituzione is_beneficiary() -> is_owner_role() in 11 punti dove la
semantica era 'proprietario pratica':
- app/routers/files.py: 3 (_can_upload, _can_download, _can_delete)
- app/routers/instructor.py: 2 (respond-beneficiary, ack-amendment)
- app/routers/practices.py: 3 (visibilita, create, schema gating)
- app/routers/custom_checks.py: 3 (declared, gate)

==COMPORTAMENTO==

Per ROLE_CONFIDI vale ora la stessa regola di BENEFICIARY:
- upload/download/delete: solo se practice.user_id == user.user_id
  AND practice.status IN ('DRAFT','AWAITING_AMENDMENT')
- respond-beneficiary: solo se proprietario pratica
- visualizzazione: solo proprie pratiche
- creazione: solo se schema PUBLISHED

Confidi su pratica di altri o su pratica non editabile -> 403 come prima.

==TEST E2E (4 step verdi)==
/tmp/test_confidi_upload.py:
1. CONFIDI proprietario DRAFT upload Invoice_zapier2024.pdf -> 200 (era 403)
2. CONFIDI NON proprietario -> 403 (scoping)
3. CONFIDI proprietario ma SUBMITTED -> 403 (stato)
4. BENEFICIARY proprietario DRAFT (regressione) -> 200
2026-04-27 09:06:10 +02:00

722 lines
31 KiB
Python
Raw Blame History

"""
Endpoint pratiche di rendicontazione (lato beneficiario).
"""
import copy
from datetime import datetime, timezone
from decimal import Decimal
from typing import List, Optional
from uuid import UUID
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from sqlalchemy import text, func
from ..db import get_db
from ..auth import AuthUser, get_current_user
from ..models import (
CallRemissionSchema, RemissionPractice,
RemissionInvoice, RemissionUlaEmployee, RemissionDocument
)
from ..schemas import (
PracticeStartRequest, PracticeUpdate, PracticeOut, PracticeListItem,
InvoiceCreate, InvoiceOut,
UlaEmployeeCreate, UlaEmployeeOut,
DocumentUpsert, DocumentOut,
GateCheckResult,
ApplicationTranchesSummary, CopyUlaOption,
ApiResponse
)
from ..templates import upgrade_schema_to_v2
router = APIRouter(prefix="/api/remission-practices", tags=["remission-practices"])
# ---------- helpers ----------
def _get_practice_or_404(db: Session, practice_id: UUID, user: AuthUser) -> RemissionPractice:
"""Recupera la pratica validando ownership beneficiario (o admin)."""
practice = db.query(RemissionPractice).filter(RemissionPractice.id == practice_id).first()
if not practice:
raise HTTPException(status_code=404, detail=f"Pratica {practice_id} non trovata")
# Solo il beneficiario owner o un superadmin può accedere
if user.is_owner_role() and practice.user_id != user.user_id:
raise HTTPException(status_code=403, detail="Accesso negato a questa pratica")
return practice
def _ensure_editable(practice: RemissionPractice):
if practice.status != "DRAFT":
raise HTTPException(
status_code=409,
detail=f"Pratica in stato {practice.status}: non modificabile"
)
def _compute_gate_check(db: Session, practice: RemissionPractice) -> GateCheckResult:
"""Valuta le gate_rules dello schema snapshot contro il contenuto della pratica.
Calcola:
- per_category_declared: totali dichiarati dal beneficiario (sempre)
- per_category_verified: totali verificati dall istruttore (solo AMMESSA/PARZIALE)
- grand_total: totale dichiarato (rif. beneficiario)
- grand_total_verified: totale verificato (rif. remissione finale)
"""
rules = practice.schema_snapshot.get("gate_rules", {}) or {}
sections = practice.schema_snapshot.get("sections", []) or []
# Base di calcolo ammissibile configurata per il bando:
# - "imponibile_always" -> sempre imponibile
# - "imponibile_only_ordinario" -> imponibile se regime ordinario, totale altrimenti
# - "totale_always" -> sempre totale
# Default: imponibile_only_ordinario (comportamento foglio excel originale)
amount_basis = rules.get("amount_basis")
if not amount_basis:
# backward compat: se c'era il vecchio flag booleano
if rules.get("iva_ordinario_imponibile_only", True):
amount_basis = "imponibile_only_ordinario"
else:
amount_basis = "totale_always"
if amount_basis == "imponibile_always":
use_taxable_only = True
elif amount_basis == "totale_always":
use_taxable_only = False
else: # imponibile_only_ordinario
use_taxable_only = (practice.iva_regime == "ORDINARIO")
per_category_declared = {}
per_category_verified = {}
grand_total = Decimal("0")
grand_total_verified = Decimal("0")
any_verified = False
all_verified = len(practice.invoices) > 0
for inv in practice.invoices:
# Dichiarato (sempre)
amt_decl = inv.taxable if use_taxable_only else inv.total
per_category_declared[inv.category_code] = per_category_declared.get(inv.category_code, Decimal("0")) + amt_decl
grand_total += amt_decl
# Verificato (solo se stato AMMESSA o PARZIALE)
if inv.verification_status in ("AMMESSA", "PARZIALE"):
any_verified = True
# se PARZIALE usa i valori verified; se AMMESSA ma verified sono null, usa dichiarato
if inv.verification_status == "PARZIALE":
tax_v = inv.taxable_verified if inv.taxable_verified is not None else inv.taxable
tot_v = inv.total_verified if inv.total_verified is not None else inv.total
else: # AMMESSA
tax_v = inv.taxable_verified if inv.taxable_verified is not None else inv.taxable
tot_v = inv.total_verified if inv.total_verified is not None else inv.total
amt_ver = tax_v if use_taxable_only else tot_v
per_category_verified[inv.category_code] = per_category_verified.get(inv.category_code, Decimal("0")) + amt_ver
grand_total_verified += amt_ver
elif inv.verification_status == "PENDING":
all_verified = False
# RESPINTA: non contribuisce ai verified
# cap remissione
amt_erogato = practice.amount_erogato
cap_pct = Decimal(str(rules.get("cap_pct_erogato", 0.5)))
cap_abs = Decimal(str(rules.get("cap_absolute", 12500)))
# Cap assoluto per l'application (somma di tutte le tranche ammissibili)
max_remission_global = min(cap_pct * amt_erogato, cap_abs)
# Cumulativo multi-tranche v2: sommo remission approvate delle tranche precedenti
# della stessa application per calcolare il residuo disponibile.
already_approved = db.query(
func.coalesce(func.sum(RemissionPractice.approved_remission), 0)
).filter(
RemissionPractice.application_id == practice.application_id,
RemissionPractice.sequence_number < practice.sequence_number,
RemissionPractice.status == 'APPROVED'
).scalar() or 0
already_approved = Decimal(str(already_approved))
max_remission_this_tranche = max(Decimal("0"), max_remission_global - already_approved)
# Legacy: max_remission = questo tranche (usato dai check sotto).
max_remission = max_remission_this_tranche
# 5 VOCI CECILIA:
# (1) max_remission_global
# (2) pre_check_admissible = min(grand_total_declared, max_remission_this_tranche)
# (3) remission_due = min(effective_total, max_remission_this_tranche)
# (4) amount_erogato
# (5) residuo_da_restituire = amt_erogato - SUM(approvata) (post-controllo su tutte le tranche)
pre_check_admissible = min(grand_total, max_remission_this_tranche)
# Se almeno 1 verifica fatta -> uso grand_total_verified per remission_due
# altrimenti uso grand_total (dichiarato) per preview pre-istruttoria
effective_total = grand_total_verified if any_verified else grand_total
remission_due = min(effective_total, max_remission_this_tranche)
# Conteggio tranche totali per questa application (per info UI/PDF)
tranches_count = db.query(RemissionPractice).filter(
RemissionPractice.application_id == practice.application_id
).count()
tranches_max = int(rules.get("max_tranches", 1))
# Per compatibilità: per_category e grand_total restano "dichiarato"
per_category = per_category_declared
checks = []
# Check 1: regime IVA scelto
if not practice.iva_regime:
checks.append({"id": "iva_selected", "label": "Regime IVA dichiarato",
"passed": False, "detail": "Seleziona il regime IVA"})
else:
checks.append({"id": "iva_selected", "label": "Regime IVA dichiarato",
"passed": True, "detail": practice.iva_regime})
# Check 2: importo totale > 0
checks.append({
"id": "has_invoices",
"label": "Almeno una fattura caricata",
"passed": len(practice.invoices) > 0,
"detail": f"{len(practice.invoices)} fatture per un totale di {grand_total} EUR"
})
# Check 3: una fattura per ogni categoria (se richiesto)
if rules.get("require_at_least_one_invoice_per_nonzero_category", False):
expenses = next((s for s in sections if s.get("type") == "category_grid"), {})
cats = expenses.get("categories", []) or []
missing = [c["code"] for c in cats if c.get("code") and c["code"] not in per_category]
checks.append({
"id": "invoice_per_category",
"label": "Fattura per ogni categoria",
"passed": len(missing) == 0,
"detail": f"Categorie senza fatture: {', '.join(missing)}" if missing else "Tutte le categorie coperte"
})
# Check 4: ULA (se richiesto)
ula_section = next((s for s in sections if s.get("type") == "ula_block"), {})
if ula_section.get("enabled") and rules.get("require_ula_above_threshold", False):
threshold = Decimal(str(ula_section.get("threshold", 1.0)))
total_fte = sum((e.fte_pct for e in practice.ula_employees), Decimal("0"))
checks.append({
"id": "ula_threshold",
"label": f"Incremento ULA >= {threshold}",
"passed": total_fte >= threshold,
"detail": f"Totale FTE: {total_fte} (soglia: {threshold})"
})
# Check 5: documenti richiesti (se richiesto)
if rules.get("require_all_documents_resolved", False):
docs_section = next((s for s in sections if s.get("type") == "document_checklist"), {})
required = docs_section.get("required_types", []) or []
uploaded_codes = {d.doc_code for d in practice.documents if d.filename}
required_codes = [r["code"] if isinstance(r, dict) else r for r in required]
missing_docs = [c for c in required_codes if c not in uploaded_codes]
checks.append({
"id": "docs_resolved",
"label": "Tutti i documenti richiesti caricati",
"passed": len(missing_docs) == 0,
"detail": f"Mancanti: {', '.join(missing_docs)}" if missing_docs else "Tutti presenti"
})
# Check 5b: documenti non scaduti (gate hard su EXPIRED)
# 2026-04-20: documento EXPIRED blocca la submit. Status letto live via JOIN sul
# BE Gepafin per doc collegati dal repository; per upload diretto PC controlla expires_at.
from datetime import date as _date_today_cls
today = _date_today_cls.today()
expired_docs = []
for doc in practice.documents:
if doc.source_company_document_id:
cd_status = db.execute(text("""
SELECT status FROM gepafin_schema.company_document
WHERE id = :cid AND is_deleted = false
"""), {"cid": doc.source_company_document_id}).scalar()
if cd_status == 'EXPIRED':
expired_docs.append(doc.doc_code)
elif doc.expires_at is not None and doc.expires_at < today:
expired_docs.append(doc.doc_code)
checks.append({
"id": "documents_not_expired",
"label": "Nessun documento scaduto",
"passed": len(expired_docs) == 0,
"detail": f"Scaduti: {', '.join(expired_docs)}" if expired_docs else "Tutti validi"
})
# Check 6: importo range (cap erogato)
amt_range = rules.get("amount_range", {})
min_e = Decimal(str(amt_range.get("min", 0)))
max_e = Decimal(str(amt_range.get("max", 999999999)))
checks.append({
"id": "erogato_in_range",
"label": f"Importo erogato entro range ({min_e}-{max_e})",
"passed": min_e <= amt_erogato <= max_e,
"detail": f"Erogato: {amt_erogato} EUR"
})
return GateCheckResult(
passed=all(c["passed"] for c in checks),
checks=checks,
totals={
"per_category": {k: float(v) for k, v in per_category.items()},
"per_category_declared": {k: float(v) for k, v in per_category_declared.items()},
"per_category_verified": {k: float(v) for k, v in per_category_verified.items()},
"grand_total": float(grand_total),
"grand_total_declared": float(grand_total),
"grand_total_verified": float(grand_total_verified),
"max_remission": float(max_remission),
"remission_due": float(remission_due),
"amount_erogato": float(amt_erogato),
"any_verified": any_verified,
"all_verified": all_verified,
"residuo_da_restituire": float(max(amt_erogato - already_approved - Decimal(str(remission_due)), Decimal("0"))),
"amount_basis": amount_basis,
"use_taxable_only": use_taxable_only,
# multi-tranche v2
"max_remission_global": float(max_remission_global),
"already_approved_previous_tranches": float(already_approved),
"max_remission_this_tranche": float(max_remission_this_tranche),
"pre_check_admissible": float(pre_check_admissible),
"sequence_number": practice.sequence_number,
"tranches_count": tranches_count,
"tranches_max": tranches_max
}
)
def _enrich_list_item(db: Session, p: RemissionPractice) -> PracticeListItem:
# Nome call e company dal DB Gepafin
q = db.execute(text("""
SELECT c.name as call_name, comp.company_name as company_name
FROM gepafin_schema.call c
JOIN gepafin_schema.company comp ON comp.id = :cid
WHERE c.id = :call_id
"""), {"call_id": p.call_id, "cid": p.company_id}).first()
item = PracticeListItem.model_validate(p)
if q:
item.call_name = q[0]
item.company_name = q[1]
item.invoice_count = len(p.invoices)
item.ula_count = len(p.ula_employees)
item.document_count = len([d for d in p.documents if d.filename])
return item
def _read_original_instructor(db: Session, application_id: int) -> Optional[int]:
"""Legge l'istruttore originariamente assegnato alla domanda nel BE Gepafin.
Restituisce user_id solo se l'utente e ancora attivo con ruolo PRE_INSTRUCTOR o INSTRUCTOR_MANAGER.
Altrimenti None (finira in coda 'da assegnare' per il manager).
"""
row = db.execute(text("""
SELECT aa.user_id, r.role_type, u.is_deleted
FROM gepafin_schema.assigned_applications aa
JOIN gepafin_schema.gepafin_user u ON u.id = aa.user_id
JOIN gepafin_schema.role r ON r.id = u.role_id
WHERE aa.application_id = :aid
AND aa.is_deleted = false
AND u.is_deleted = false
ORDER BY aa.assigned_at DESC
LIMIT 1
"""), {"aid": application_id}).mappings().first()
if not row:
return None
if row["role_type"] not in ("ROLE_PRE_INSTRUCTOR", "ROLE_INSTRUCTOR_MANAGER"):
return None
return row["user_id"]
def _get_schema_published(db: Session, call_id: int) -> Optional[CallRemissionSchema]:
return db.query(CallRemissionSchema).filter(CallRemissionSchema.call_id == call_id).first()
# ---------- endpoints ----------
@router.get("/mine", response_model=ApiResponse)
def list_my_practices(db: Session = Depends(get_db), user: AuthUser = Depends(get_current_user)):
"""Lista pratiche del beneficiario raggruppate per application_id (v2 multi-tranche).
Ogni application ha il riepilogo cumulativo + elenco tranche esistenti + stato apertura nuova tranche.
"""
# Tutte le pratiche del beneficiario ordinate per application+sequence
practices = db.query(RemissionPractice).filter(
RemissionPractice.user_id == user.user_id
).order_by(
RemissionPractice.application_id, RemissionPractice.sequence_number
).all()
# Raggruppo per application_id
by_app = {}
for p in practices:
by_app.setdefault(p.application_id, []).append(p)
# Applications CONTRACT_SIGNED del beneficiario
rows = db.execute(text("""
SELECT a.id as application_id, a.call_id, a.company_id, a.amount_accepted, a.status,
c.name as call_name, comp.company_name as company_name
FROM gepafin_schema.application a
JOIN gepafin_schema.call c ON c.id = a.call_id
LEFT JOIN gepafin_schema.company comp ON comp.id = a.company_id
WHERE a.user_id = :uid AND a.status = 'CONTRACT_SIGNED' AND a.is_deleted = false
ORDER BY a.id
"""), {"uid": user.user_id}).mappings().all()
applications = []
for r in rows:
app_id = r["application_id"]
trs = by_app.get(app_id, [])
# leggo schema del bando per max_tranches e cap
schema = _get_schema_published(db, r["call_id"])
rules = (schema.schema_json.get("gate_rules", {}) if schema else {}) or {}
max_tranches = int(rules.get("max_tranches", 1))
cap_pct = Decimal(str(rules.get("cap_pct_erogato", 0.5)))
cap_abs = Decimal(str(rules.get("cap_absolute", 12500)))
amt_erogato = Decimal(str(r["amount_accepted"] or 0))
max_remission_global = min(cap_pct * amt_erogato, cap_abs)
already_approved_sum = sum(
(t.approved_remission or Decimal("0")) for t in trs if t.status == "APPROVED"
)
max_remission_next = max(Decimal("0"), max_remission_global - already_approved_sum)
# Stato apertura nuova tranche
can_start = True
reason = None
if len(trs) >= max_tranches:
can_start = False
reason = f"Limite tranches raggiunto ({max_tranches})"
elif len(trs) > 0 and trs[-1].status not in ("APPROVED", "REJECTED"):
can_start = False
reason = "Completa prima la rendicontazione in corso"
elif max_remission_next <= 0:
can_start = False
reason = f"Remissione massima gia raggiunta (euro {float(already_approved_sum):.2f})"
# Summary tranche (serialize with enriched fields)
tranche_items = []
for t in trs:
item = _enrich_list_item(db, t).model_dump(mode="json")
tranche_items.append(item)
applications.append({
"application_id": app_id,
"call_id": r["call_id"],
"call_name": r["call_name"],
"company_id": r["company_id"],
"company_name": r["company_name"],
"amount_erogato": float(amt_erogato),
"max_tranches": max_tranches,
"tranches": tranche_items,
"can_start_new": can_start,
"start_blocked_reason": reason,
"already_approved_sum": float(already_approved_sum),
"max_remission_global": float(max_remission_global),
"max_remission_next_tranche": float(max_remission_next),
})
return ApiResponse(data={"applications": applications})
@router.post("/start", response_model=ApiResponse)
def start_practice(body: PracticeStartRequest, db: Session = Depends(get_db),
user: AuthUser = Depends(get_current_user)):
"""Avvia una nuova pratica o tranche N+1 per una application CONTRACT_SIGNED.
Validazioni server-side v2:
- count(tranches) < max_tranches
- last tranche in {APPROVED, REJECTED} oppure count==0
- max_remission_this_tranche > 0
Se sequence_number > 1 e copy_ula_from_previous=True: bulk copy ULA dalla tranche N-1
con reset verification_*.
"""
# Verifica application
app_row = db.execute(text("""
SELECT id, call_id, company_id, user_id, status, amount_accepted
FROM gepafin_schema.application
WHERE id = :aid AND is_deleted = false
"""), {"aid": body.application_id}).mappings().first()
if not app_row:
raise HTTPException(status_code=404, detail=f"Application {body.application_id} non trovata")
if app_row["status"] != "CONTRACT_SIGNED":
raise HTTPException(status_code=409,
detail=f"Application in stato {app_row['status']}, richiesto CONTRACT_SIGNED")
if user.is_owner_role() and app_row["user_id"] != user.user_id:
raise HTTPException(status_code=403, detail="Application non di tua proprieta")
# Schema del bando
schema = _get_schema_published(db, app_row["call_id"])
if not schema:
raise HTTPException(status_code=409,
detail="Nessuno schema di rendicontazione configurato per questo bando.")
if schema.status != "PUBLISHED" and user.is_owner_role():
raise HTTPException(status_code=409,
detail="Lo schema di rendicontazione non e ancora stato pubblicato.")
# Tranche esistenti
existing_tranches = db.query(RemissionPractice).filter(
RemissionPractice.application_id == body.application_id
).order_by(RemissionPractice.sequence_number).all()
rules = (schema.schema_json.get("gate_rules", {}) or {})
max_tranches = int(rules.get("max_tranches", 1))
cap_pct = Decimal(str(rules.get("cap_pct_erogato", 0.5)))
cap_abs = Decimal(str(rules.get("cap_absolute", 12500)))
amt_erogato = Decimal(str(app_row["amount_accepted"] or 0))
max_remission_global = min(cap_pct * amt_erogato, cap_abs)
# VALIDAZIONI v2
if len(existing_tranches) >= max_tranches:
raise HTTPException(status_code=400,
detail=f"Limite tranches raggiunto (max {max_tranches})")
if existing_tranches:
last = existing_tranches[-1]
if last.status not in ("APPROVED", "REJECTED"):
raise HTTPException(status_code=400,
detail="Completa prima la rendicontazione in corso")
already_approved = sum(
(t.approved_remission or Decimal("0")) for t in existing_tranches if t.status == "APPROVED"
)
max_remission_this = max(Decimal("0"), max_remission_global - already_approved)
if max_remission_this <= 0:
raise HTTPException(status_code=400,
detail=f"Remissione massima gia raggiunta (euro {float(already_approved):.2f})")
# Nuovo sequence_number
next_seq = (existing_tranches[-1].sequence_number + 1) if existing_tranches else 1
# suggested_instructor: solo alla tranche 1 leggo da assigned_applications
suggested_instructor_id = None
assigned_instructor_id = None
if next_seq == 1:
suggested_instructor_id = _read_original_instructor(db, body.application_id)
assigned_instructor_id = suggested_instructor_id
else:
# tranche successiva: eredita suggested dalla tranche 1, assegnato ricomincia NULL
first = existing_tranches[0]
suggested_instructor_id = first.suggested_instructor_id
# Snapshot schema aggiornato a v2 se schema_version < 2
snapshot = copy.deepcopy(schema.schema_json)
snapshot = upgrade_schema_to_v2(snapshot)
practice = RemissionPractice(
call_id=app_row["call_id"],
application_id=body.application_id,
company_id=app_row["company_id"],
user_id=app_row["user_id"],
status="DRAFT",
schema_snapshot=snapshot,
amount_erogato=amt_erogato,
sequence_number=next_seq,
period_label=body.period_label,
suggested_instructor_id=suggested_instructor_id,
assigned_instructor_id=assigned_instructor_id,
)
db.add(practice)
db.flush()
# Copy ULA da tranche precedente
if next_seq > 1 and body.copy_ula_from_previous:
prev = existing_tranches[-1]
for prev_emp in prev.ula_employees:
new_emp = RemissionUlaEmployee(
practice_id=practice.id,
codice_fiscale=prev_emp.codice_fiscale,
full_name=prev_emp.full_name,
contract_type=prev_emp.contract_type,
role_description=prev_emp.role_description,
fte_pct=prev_emp.fte_pct,
period_start_date=prev_emp.period_start_date,
period_end_date=prev_emp.period_end_date,
supporting_doc_type=prev_emp.supporting_doc_type,
# reset verification: non copiare status/notes/verified_by/verified_at
verification_status="PENDING",
)
db.add(new_emp)
db.commit()
db.refresh(practice)
return ApiResponse(
message=f"Tranche {next_seq}/{max_tranches} avviata",
data=PracticeOut.model_validate(practice).model_dump(mode="json")
)
@router.get("/{practice_id}/copy-ula-options", response_model=ApiResponse)
def copy_ula_options(practice_id: UUID, db: Session = Depends(get_db),
user: AuthUser = Depends(get_current_user)):
"""Preview dei dipendenti ULA della tranche N-1 copiabili in questa tranche N.
Usato dal FE al click su "+Nuova rendicontazione" per mostrare il pre-fill."""
p = _get_practice_or_404(db, practice_id, user)
if p.sequence_number <= 1:
return ApiResponse(data={"options": [], "previous_sequence": None})
prev = db.query(RemissionPractice).filter(
RemissionPractice.application_id == p.application_id,
RemissionPractice.sequence_number == p.sequence_number - 1
).first()
if not prev:
return ApiResponse(data={"options": [], "previous_sequence": None})
options = [CopyUlaOption(
codice_fiscale=e.codice_fiscale,
full_name=e.full_name,
contract_type=e.contract_type,
role_description=e.role_description,
fte_pct=float(e.fte_pct),
period_start_date=e.period_start_date,
period_end_date=e.period_end_date,
supporting_doc_type=e.supporting_doc_type,
).model_dump(mode="json") for e in prev.ula_employees]
return ApiResponse(data={"options": options, "previous_sequence": prev.sequence_number,
"previous_id": str(prev.id)})
@router.get("/{practice_id}", response_model=ApiResponse)
def get_practice(practice_id: UUID, db: Session = Depends(get_db), user: AuthUser = Depends(get_current_user)):
p = _get_practice_or_404(db, practice_id, user)
from ..schemas import AmendmentRequestOut
payload = PracticeOut.model_validate(p).model_dump(mode="json")
payload["amendments"] = [AmendmentRequestOut.model_validate(a).model_dump(mode="json") for a in p.amendment_requests]
return ApiResponse(data=payload)
@router.put("/{practice_id}", response_model=ApiResponse)
def update_practice(practice_id: UUID, body: PracticeUpdate,
db: Session = Depends(get_db), user: AuthUser = Depends(get_current_user)):
p = _get_practice_or_404(db, practice_id, user)
_ensure_editable(p)
if body.iva_regime is not None:
p.iva_regime = body.iva_regime
if body.notes_beneficiario is not None:
p.notes_beneficiario = body.notes_beneficiario
db.commit()
db.refresh(p)
return ApiResponse(message="Pratica aggiornata",
data=PracticeOut.model_validate(p).model_dump(mode="json"))
# ---------- Invoices ----------
@router.post("/{practice_id}/invoices", response_model=ApiResponse)
def add_invoice(practice_id: UUID, body: InvoiceCreate,
db: Session = Depends(get_db), user: AuthUser = Depends(get_current_user)):
p = _get_practice_or_404(db, practice_id, user)
_ensure_editable(p)
inv = RemissionInvoice(practice_id=p.id, **body.model_dump())
db.add(inv)
db.commit()
db.refresh(inv)
return ApiResponse(message="Fattura aggiunta",
data=InvoiceOut.model_validate(inv).model_dump(mode="json"))
@router.delete("/{practice_id}/invoices/{invoice_id}", response_model=ApiResponse)
def delete_invoice(practice_id: UUID, invoice_id: UUID,
db: Session = Depends(get_db), user: AuthUser = Depends(get_current_user)):
p = _get_practice_or_404(db, practice_id, user)
_ensure_editable(p)
inv = db.query(RemissionInvoice).filter(
RemissionInvoice.id == invoice_id, RemissionInvoice.practice_id == practice_id
).first()
if not inv:
raise HTTPException(status_code=404, detail="Fattura non trovata")
db.delete(inv)
db.commit()
return ApiResponse(message="Fattura rimossa")
# ---------- ULA Employees ----------
@router.post("/{practice_id}/ula-employees", response_model=ApiResponse)
def add_ula_employee(practice_id: UUID, body: UlaEmployeeCreate,
db: Session = Depends(get_db), user: AuthUser = Depends(get_current_user)):
p = _get_practice_or_404(db, practice_id, user)
_ensure_editable(p)
e = RemissionUlaEmployee(practice_id=p.id, **body.model_dump())
db.add(e)
db.commit()
db.refresh(e)
return ApiResponse(message="Dipendente aggiunto",
data=UlaEmployeeOut.model_validate(e).model_dump(mode="json"))
@router.delete("/{practice_id}/ula-employees/{employee_id}", response_model=ApiResponse)
def delete_ula_employee(practice_id: UUID, employee_id: UUID,
db: Session = Depends(get_db), user: AuthUser = Depends(get_current_user)):
p = _get_practice_or_404(db, practice_id, user)
_ensure_editable(p)
e = db.query(RemissionUlaEmployee).filter(
RemissionUlaEmployee.id == employee_id, RemissionUlaEmployee.practice_id == practice_id
).first()
if not e:
raise HTTPException(status_code=404, detail="Dipendente non trovato")
db.delete(e)
db.commit()
return ApiResponse(message="Dipendente rimosso")
# ---------- Documents ----------
@router.put("/{practice_id}/documents/{doc_code}", response_model=ApiResponse)
def upsert_document(practice_id: UUID, doc_code: str, body: DocumentUpsert,
db: Session = Depends(get_db), user: AuthUser = Depends(get_current_user)):
p = _get_practice_or_404(db, practice_id, user)
_ensure_editable(p)
d = db.query(RemissionDocument).filter(
RemissionDocument.practice_id == practice_id,
RemissionDocument.doc_code == doc_code
).first()
if not d:
d = RemissionDocument(practice_id=p.id, doc_code=doc_code)
db.add(d)
d.filename = body.filename
d.uploaded_at = body.uploaded_at or (datetime.now(timezone.utc) if body.filename else None)
d.expires_at = body.expires_at
d.notes = body.notes
db.commit()
db.refresh(d)
return ApiResponse(message="Documento aggiornato",
data=DocumentOut.model_validate(d).model_dump(mode="json"))
@router.delete("/{practice_id}/documents/{doc_code}", response_model=ApiResponse)
def clear_document(practice_id: UUID, doc_code: str,
db: Session = Depends(get_db), user: AuthUser = Depends(get_current_user)):
p = _get_practice_or_404(db, practice_id, user)
_ensure_editable(p)
d = db.query(RemissionDocument).filter(
RemissionDocument.practice_id == practice_id,
RemissionDocument.doc_code == doc_code
).first()
if d:
db.delete(d)
db.commit()
return ApiResponse(message="Documento rimosso")
# ---------- Gate check + submit ----------
@router.get("/{practice_id}/gate-check", response_model=ApiResponse)
def gate_check(practice_id: UUID, db: Session = Depends(get_db),
user: AuthUser = Depends(get_current_user)):
p = _get_practice_or_404(db, practice_id, user)
result = _compute_gate_check(db, p)
return ApiResponse(data=result.model_dump(mode="json"))
@router.post("/{practice_id}/submit", response_model=ApiResponse)
def submit_practice(practice_id: UUID, db: Session = Depends(get_db),
user: AuthUser = Depends(get_current_user)):
p = _get_practice_or_404(db, practice_id, user)
_ensure_editable(p)
check = _compute_gate_check(db, p)
if not check.passed:
raise HTTPException(status_code=422, detail={
"message": "Gate rules non soddisfatte",
"checks": check.checks
})
p.status = "SUBMITTED"
p.submitted_at = datetime.now(timezone.utc)
db.commit()
db.refresh(p)
return ApiResponse(message="Pratica inviata con successo",
data=PracticeOut.model_validate(p).model_dump(mode="json"))
Reference in New Issue View Git Blame Copy Permalink