#!/usr/bin/env python3 """ Seed sandbox riproducibile per Gepafin rendicontazione. Usage: python /app/scripts/seed_sandbox.py --reset --scenario=napoli-sas python /app/scripts/seed_sandbox.py --reset --scenario=napoli-sas --advance=under_review Scenari disponibili: napoli-sas : pratica completa con 5 fatture (B1/B2/B3), 2 ULA, 4 documenti, PDF fixture generati e caricati nello storage /var/uploads. Advance (opzionale, default = draft): draft : lascia la pratica in DRAFT (stato beneficiario) submitted : pratica inviata, pronta a essere presa in carico dall'istruttore under_review : istruttore l'ha presa in carico, pronta per verifica """ import argparse import io import os import shutil import sys import uuid from datetime import date, datetime, timezone from decimal import Decimal from pathlib import Path sys.path.insert(0, '/app') from sqlalchemy import text from app.db import engine, SessionLocal from app.models import ( CallRemissionSchema, RemissionPractice, RemissionInvoice, RemissionUlaEmployee, RemissionDocument, RemissionAmendmentRequest, RemissionCustomCheckValue, ) from app.storage import save_upload, BASE_PATH from app.templates import RESTART_TEMPLATE CALL_ID = 1 COMPANY_ID = 1 BENEFICIARY_USER_ID = 9 # beneficiario@sandbox.local INSTRUCTOR_USER_ID = 10 # istruttore@sandbox.local MANAGER_USER_ID = 11 # manager@sandbox.local APPLICATION_ID = 1 # --------------------------------------------------------------------------- # PDF fixture generator (weasyprint) # --------------------------------------------------------------------------- def make_pdf_bytes(title: str, subtitle: str, lines: list[str]) -> bytes: from weasyprint import HTML rows = "".join(f"{i+1}{ln}" for i, ln in enumerate(lines)) html = f"""

{title}

{subtitle}

{rows}
#Voce
FIXTURE DEMO — documento generato automaticamente da seed_sandbox.py il {datetime.now().strftime('%d/%m/%Y %H:%M')}. Questo PDF è un placeholder a scopo dimostrativo per la sandbox Gepafin.
""" return HTML(string=html).write_pdf() def attach_pdf(db, entity, entity_type: str, application_id: int, filename: str, title: str, subtitle: str, lines: list[str], uploader_id: int): """Genera PDF e lo carica tramite lo storage adapter, aggiorna entity.""" pdf = make_pdf_bytes(title, subtitle, lines) rel_path, size, digest, mime, safe_name = save_upload( application_id=application_id, entity_type=entity_type, entity_id=entity.id, file_obj=io.BytesIO(pdf), original_filename=filename, content_type='application/pdf', ) entity.storage_path = rel_path entity.mime = mime entity.size_bytes = size entity.sha256 = digest entity.uploaded_by = uploader_id if hasattr(entity, 'uploaded_at'): entity.uploaded_at = datetime.now(timezone.utc) # filename originale specifico per tipo if entity_type == 'invoice': entity.pdf_filename = safe_name elif entity_type == 'ula': entity.supporting_doc_filename = safe_name elif entity_type == 'document': entity.filename = safe_name entity.uploaded_at = datetime.now(timezone.utc) def ensure_assigned_application(db): """Popola gepafin_schema.assigned_applications per abilitare suggested_instructor_id alla creazione della prima tranche. Idempotente.""" from sqlalchemy import text existing = db.execute(text(""" SELECT id FROM gepafin_schema.assigned_applications WHERE application_id = :aid AND user_id = :uid AND is_deleted = false """), {"aid": APPLICATION_ID, "uid": INSTRUCTOR_USER_ID}).scalar() if existing: print(f"[assigned_applications] gia presente (id={existing})") return db.execute(text(""" INSERT INTO gepafin_schema.assigned_applications (user_id, assigned_by, application_id, status, is_deleted, assigned_at, created_date, updated_date) VALUES (:uid, :admin, :aid, 'ASSIGNED', false, NOW(), NOW(), NOW()) """), {"uid": INSTRUCTOR_USER_ID, "admin": 8, "aid": APPLICATION_ID}) db.commit() print(f"[assigned_applications] user={INSTRUCTOR_USER_ID} assegnato a application={APPLICATION_ID}") # --------------------------------------------------------------------------- # Reset # --------------------------------------------------------------------------- def do_reset(scope: str): """TRUNCATE tabelle remission_* e pulizia storage. scope: 'all' | 'practices'.""" print(f"[reset] scope={scope}") with engine.begin() as conn: if scope == 'all': conn.execute(text(""" TRUNCATE gepafin_rendic.remission_custom_check_value, gepafin_rendic.remission_amendment_request, gepafin_rendic.remission_document, gepafin_rendic.remission_ula_employee, gepafin_rendic.remission_invoice, gepafin_rendic.remission_practice RESTART IDENTITY CASCADE """)) print("[reset] tabelle remission_* truncate-ate") else: conn.execute(text(""" DELETE FROM gepafin_rendic.remission_amendment_request; DELETE FROM gepafin_rendic.remission_document; DELETE FROM gepafin_rendic.remission_ula_employee; DELETE FROM gepafin_rendic.remission_invoice; DELETE FROM gepafin_rendic.remission_practice; """)) # Pulizia storage if BASE_PATH.exists(): for p in BASE_PATH.iterdir(): if p.is_dir(): shutil.rmtree(p) elif p.is_file() and not p.name.startswith('.'): p.unlink() print(f"[reset] {BASE_PATH} pulito") # --------------------------------------------------------------------------- # Schema RE-START pubblicato # --------------------------------------------------------------------------- def ensure_schema_published(db): schema_row = db.query(CallRemissionSchema).filter( CallRemissionSchema.call_id == CALL_ID ).first() if not schema_row: schema_row = CallRemissionSchema( call_id=CALL_ID, schema_version=1, status='PUBLISHED', schema_json=RESTART_TEMPLATE, created_by=1, published_at=datetime.now(timezone.utc), published_by=1, ) db.add(schema_row) db.flush() print("[schema] creato e pubblicato schema RE-START") elif schema_row.status != 'PUBLISHED': schema_row.status = 'PUBLISHED' schema_row.published_at = datetime.now(timezone.utc) schema_row.published_by = 1 print("[schema] schema RE-START promosso a PUBLISHED") return schema_row # --------------------------------------------------------------------------- # Scenario: NAPOLI SAS # --------------------------------------------------------------------------- def scenario_napoli_sas(db, advance='draft'): """ Pratica NAPOLI SAS completa. 5 fatture (2 B1 / 2 B2 / 1 B3), 2 ULA, 4 documenti. Tutti con PDF fixture allegato. """ schema_row = ensure_schema_published(db) practice = RemissionPractice( call_id=CALL_ID, application_id=APPLICATION_ID, company_id=COMPANY_ID, user_id=BENEFICIARY_USER_ID, status='DRAFT', schema_snapshot=schema_row.schema_json, iva_regime='ORDINARIO', amount_erogato=Decimal('17000'), notes_beneficiario='Pratica di rendicontazione bando RE-START — investimento digitale 2021.', ) db.add(practice) db.flush() print(f"[practice] creata id={practice.id}") invoices_data = [ dict(category_code='B1', invoice_number='2021/487', invoice_date=date(2021, 3, 15), payment_date=date(2021, 3, 31), supplier_name='Dell Italia S.r.l.', supplier_vat='IT12345678901', description='Fornitura 5 workstation Precision 3660 per sviluppo software', taxable=Decimal('3000'), vat=Decimal('660'), total=Decimal('3660'), fname='ft_2021_487_dell.pdf'), dict(category_code='B1', invoice_number='2020/988', invoice_date=date(2020, 11, 20), payment_date=date(2020, 12, 10), supplier_name='HP Italia S.p.A.', supplier_vat='IT98765432109', description='2 laptop EliteBook 840 G7 + 3 monitor DreamColor', taxable=Decimal('2000'), vat=Decimal('440'), total=Decimal('2440'), fname='ft_2020_988_hp.pdf'), dict(category_code='B2', invoice_number='2021/58', invoice_date=date(2021, 5, 8), payment_date=date(2021, 5, 30), supplier_name='Netcomm S.c. a r.l.', supplier_vat='IT07008690961', description='Canone annuale Osservatorio eCommerce B2c 2021 + assicurazione accessoria', taxable=Decimal('2500'), vat=Decimal('550'), total=Decimal('3050'), fname='ft_2021_58_netcomm.pdf'), dict(category_code='B2', invoice_number='2021/115', invoice_date=date(2021, 7, 2), payment_date=date(2021, 7, 20), supplier_name='Studio Romano & Associati', supplier_vat='IT03214569876', description='Consulenza strategica digital transformation — 20 ore senior partner', taxable=Decimal('1500'), vat=Decimal('330'), total=Decimal('1830'), fname='ft_2021_115_romano.pdf'), dict(category_code='B3', invoice_number='2021/221', invoice_date=date(2021, 9, 15), payment_date=date(2021, 9, 30), supplier_name='CertQuality S.r.l.', supplier_vat='IT11223344556', description='Certificazione ISO 27001:2013 — audit + rilascio certificato triennale', taxable=Decimal('2400'), vat=Decimal('528'), total=Decimal('2928'), fname='ft_2021_221_certquality.pdf'), ] for i, d in enumerate(invoices_data): inv = RemissionInvoice( practice_id=practice.id, category_code=d['category_code'], invoice_number=d['invoice_number'], invoice_date=d['invoice_date'], payment_date=d['payment_date'], supplier_name=d['supplier_name'], supplier_vat=d['supplier_vat'], description=d['description'], taxable=d['taxable'], vat=d['vat'], total=d['total'], ) db.add(inv) db.flush() attach_pdf( db, inv, 'invoice', APPLICATION_ID, filename=d['fname'], title=f"Fattura n. {d['invoice_number']}", subtitle=f"{d['supplier_name']} — {d['invoice_date'].strftime('%d/%m/%Y')}", lines=[ f"Fornitore: {d['supplier_name']} — P.IVA {d['supplier_vat']}", f"Descrizione: {d['description']}", f"Imponibile: € {d['taxable']}", f"IVA 22%: € {d['vat']}", f"Totale: € {d['total']}", f"Data fattura: {d['invoice_date'].strftime('%d/%m/%Y')}", f"Data pagamento: {d['payment_date'].strftime('%d/%m/%Y')}", ], uploader_id=BENEFICIARY_USER_ID, ) print(f"[invoice] {i+1}/5 {d['category_code']} {d['invoice_number']} + PDF {inv.size_bytes}b") ula_data = [ dict(cf='RSSMRA85T10H501Z', name='Mario Rossi', ctype='T_IND', role='Sviluppatore senior', fte=Decimal('1.0000'), start=date(2021, 1, 27), end=date(2021, 12, 31), doc_type='LUL', fname='lul_rossi_2021.pdf'), dict(cf='BNCLRA90A41F205D', name='Laura Bianchi', ctype='T_DET', role='Digital marketing specialist', fte=Decimal('0.5000'), start=date(2021, 4, 1), end=date(2021, 12, 31), doc_type='LUL', fname='lul_bianchi_2021.pdf'), ] for i, d in enumerate(ula_data): emp = RemissionUlaEmployee( practice_id=practice.id, codice_fiscale=d['cf'], full_name=d['name'], contract_type=d['ctype'], role_description=d['role'], fte_pct=d['fte'], period_start_date=d['start'], period_end_date=d['end'], supporting_doc_type=d['doc_type'], ) db.add(emp) db.flush() attach_pdf( db, emp, 'ula', APPLICATION_ID, filename=d['fname'], title=f"Libro Unico del Lavoro — {d['name']}", subtitle=f"Periodo {d['start'].strftime('%d/%m/%Y')} — {d['end'].strftime('%d/%m/%Y')}", lines=[ f"Codice fiscale: {d['cf']}", f"Nome e cognome: {d['name']}", f"Tipo contratto: {d['ctype']}", f"Mansione: {d['role']}", f"FTE: {float(d['fte']):.2f}", f"Inizio rapporto: {d['start'].strftime('%d/%m/%Y')}", f"Fine rapporto: {d['end'].strftime('%d/%m/%Y')}", f"Tipo documento: {d['doc_type']}", ], uploader_id=BENEFICIARY_USER_ID, ) print(f"[ula] {i+1}/2 {d['name']} FTE={d['fte']} + PDF {emp.size_bytes}b") docs_data = [ dict(code='DURC', label='DURC', fname='durc_napoli.pdf', title='DURC — Documento Unico di Regolarità Contributiva', subtitle='NAPOLI SAS — regolarità contributiva verificata', lines=[ 'Impresa: NAPOLI SAS Sandbox', 'P.IVA: 03517010546-SBX', 'Stato: REGOLARE', 'Data emissione: 15/11/2021', 'Scadenza: 15/03/2022', 'Ente erogatore: INPS', ]), dict(code='VISURA_CAMERALE', label='Visura camerale', fname='visura_napoli.pdf', title='Visura camerale ordinaria', subtitle='NAPOLI SAS Sandbox — estratto CCIAA Perugia', lines=[ 'Denominazione: NAPOLI SAS Sandbox', 'Forma giuridica: Società in accomandita semplice', 'Sede legale: Perugia', 'Iscrizione REA: PG-123456', 'Attività prevalente: 62.02 Consulenza informatica', 'Stato: ATTIVA', ]), dict(code='BILANCIO', label='Bilancio', fname='bilancio_napoli_2021.pdf', title='Bilancio di esercizio 2021', subtitle='NAPOLI SAS Sandbox — stato patrimoniale e conto economico', lines=[ 'Anno: 2021', 'Ricavi: € 85.200', 'Costi: € 68.300', 'Utile ante imposte: € 16.900', 'Patrimonio netto: € 45.600', 'Immobilizzazioni materiali: € 12.400', ]), dict(code='ALTRO', label='Altra documentazione', fname='quietanze_napoli.pdf', title='Quietanze di pagamento', subtitle='Raccolta quietanze fatture rendicontate', lines=[ 'Bonifico Dell € 3.660 il 31/03/2021', 'Bonifico HP € 2.440 il 10/12/2020', 'Bonifico Netcomm € 3.050 il 30/05/2021', 'Bonifico Romano € 1.830 il 20/07/2021', 'Bonifico CertQuality € 2.928 il 30/09/2021', ]), ] for i, d in enumerate(docs_data): doc = RemissionDocument( practice_id=practice.id, doc_code=d['code'], notes=None, ) db.add(doc) db.flush() attach_pdf( db, doc, 'document', APPLICATION_ID, filename=d['fname'], title=d['title'], subtitle=d['subtitle'], lines=d['lines'], uploader_id=BENEFICIARY_USER_ID, ) print(f"[doc] {i+1}/4 {d['code']} + PDF {doc.size_bytes}b") db.commit() # ----- advance state se richiesto ----- if advance == 'draft': print(f"[advance] rimasta in DRAFT") return practice.id # submit practice.status = 'SUBMITTED' practice.submitted_at = datetime.now(timezone.utc) db.commit() print(f"[advance] pratica SUBMITTED") if advance == 'submitted': return practice.id # claim by instructor practice.status = 'UNDER_REVIEW' practice.assigned_instructor_id = INSTRUCTOR_USER_ID db.commit() print(f"[advance] pratica UNDER_REVIEW (claimed by user {INSTRUCTOR_USER_ID})") return practice.id def scenario_napoli_sas_multi(db): """Scenario multi-tranche: - tranche 1 APPROVED con 1 fattura B3 524.50€, rettifica 57.36€ assicurazione, ammesso 467.14€ (caso reale pratica 5888 di Cecilia) - tranche 2 DRAFT vuota, pronta per la demo Popola anche: - assigned_applications (istruttore originariamente assegnato) - 2 custom_checks dichiarati + polizza con PDF allegato su tranche 1 """ schema_row = ensure_schema_published(db) ensure_assigned_application(db) # ---------- Tranche 1 APPROVED ---------- practice1 = RemissionPractice( call_id=CALL_ID, application_id=APPLICATION_ID, company_id=COMPANY_ID, user_id=BENEFICIARY_USER_ID, status="APPROVED", schema_snapshot=schema_row.schema_json, iva_regime="ORDINARIO", amount_erogato=Decimal("17000"), sequence_number=1, period_label="I fase 2021", suggested_instructor_id=INSTRUCTOR_USER_ID, assigned_instructor_id=INSTRUCTOR_USER_ID, approved_remission=Decimal("467.14"), reviewed_at=datetime.now(timezone.utc), reviewed_by=INSTRUCTOR_USER_ID, instructor_final_notes="Pratica tranche I: ammessa 1 fattura B3 con rettifica quota assicurativa.", submitted_at=datetime.now(timezone.utc), ) db.add(practice1) db.flush() print(f"[practice] tranche 1 APPROVED id={practice1.id}") # 1 fattura B3 con PARZIALE (storno 57.36) inv1 = RemissionInvoice( practice_id=practice1.id, category_code="B3", invoice_number="2021/042", invoice_date=date(2021, 4, 15), payment_date=date(2021, 4, 30), supplier_name="Formazione Digitale S.r.l.", supplier_vat="IT03521460542", description="Corso di formazione digitale 40h + quota assicurazione partecipanti", taxable=Decimal("524.50"), vat=Decimal("115.39"), total=Decimal("639.89"), taxable_verified=Decimal("467.14"), vat_verified=Decimal("102.77"), total_verified=Decimal("569.91"), verification_status="PARZIALE", verification_notes="Storno di 57.36 EUR per quota assicurazione partecipanti non ammissibile (non rientra nelle spese formative dirette).", verified_by=INSTRUCTOR_USER_ID, verified_at=datetime.now(timezone.utc), ) db.add(inv1) db.flush() attach_pdf( db, inv1, "invoice", APPLICATION_ID, filename="ft_2021_042_formazione.pdf", title=f"Fattura n. {inv1.invoice_number}", subtitle=f"{inv1.supplier_name}", lines=[ f"Fornitore: {inv1.supplier_name} P.IVA {inv1.supplier_vat}", f"Descrizione: {inv1.description}", f"Imponibile: EUR {inv1.taxable}", f"IVA 22%: EUR {inv1.vat}", f"Totale: EUR {inv1.total}", ], uploader_id=BENEFICIARY_USER_ID, ) print(f"[invoice T1] B3 2021/042 PARZIALE + PDF {inv1.size_bytes}b") # 1 ULA T_IND 1.0 AMMESSA emp1 = RemissionUlaEmployee( practice_id=practice1.id, codice_fiscale="RSSMRA85T10H501Z", full_name="Mario Rossi", contract_type="T_IND", role_description="Sviluppatore senior", fte_pct=Decimal("1.0000"), fte_pct_verified=Decimal("1.0000"), period_start_date=date(2021, 1, 27), period_end_date=date(2021, 12, 31), supporting_doc_type="LUL", verification_status="AMMESSA", verified_by=INSTRUCTOR_USER_ID, verified_at=datetime.now(timezone.utc), ) db.add(emp1) db.flush() attach_pdf( db, emp1, "ula", APPLICATION_ID, filename="lul_rossi_2021_t1.pdf", title=f"LUL {emp1.full_name}", subtitle=f"{emp1.period_start_date} to {emp1.period_end_date}", lines=[f"CF: {emp1.codice_fiscale}", f"FTE: 1.00", "Contratto: T_IND"], uploader_id=BENEFICIARY_USER_ID, ) # Documenti validati for code, label in [("DURC", "DURC"), ("VISURA_CAMERALE", "Visura"), ("BILANCIO", "Bilancio 2021"), ("ANTIRICICLAGGIO", "Antiriciclaggio")]: doc = RemissionDocument( practice_id=practice1.id, doc_code=code, verification_status="VALIDO", verified_by=INSTRUCTOR_USER_ID, verified_at=datetime.now(timezone.utc), ) db.add(doc) db.flush() attach_pdf( db, doc, "document", APPLICATION_ID, filename=f"{code.lower()}_napoli_t1.pdf", title=label, subtitle="Tranche I — NAPOLI SAS", lines=["Documento valido", "Approvato dall istruttore"], uploader_id=BENEFICIARY_USER_ID, ) # Custom checks tranche 1: antiriciclaggio dichiarato + polizza con PDF cc_antir = RemissionCustomCheckValue( practice_id=practice1.id, check_code="antiriciclaggio", beneficiary_declared=True, declared_at=datetime.now(timezone.utc), verification_status="VALIDO", verified_by=INSTRUCTOR_USER_ID, verified_at=datetime.now(timezone.utc), ) db.add(cc_antir) cc_polizza = RemissionCustomCheckValue( practice_id=practice1.id, check_code="polizza_fidejussoria", beneficiary_declared=True, declared_at=datetime.now(timezone.utc), verification_status="VALIDO", verified_by=INSTRUCTOR_USER_ID, verified_at=datetime.now(timezone.utc), ) db.add(cc_polizza) db.flush() # Genero PDF polizza e lo salvo direttamente in custom_checks/ from pathlib import Path as _P pdf = make_pdf_bytes( "Polizza fidejussoria tranche I", "NAPOLI SAS Sandbox — garanzia bando RE-START", [ "Compagnia: Generali Assicurazioni", "Importo garantito: EUR 17.000", "Data emissione: 15/01/2021", "Scadenza: 31/12/2022", "N. polizza: FID-2021-NS-0042", ], ) import hashlib digest = hashlib.sha256(pdf).hexdigest() target_dir = BASE_PATH / "custom_checks" / str(practice1.id) / "polizza_fidejussoria" target_dir.mkdir(parents=True, exist_ok=True) target_file = target_dir / f"{digest[:12]}-polizza_fidejussoria.pdf" target_file.write_bytes(pdf) cc_polizza.storage_path = str(target_file.relative_to(BASE_PATH)) cc_polizza.mime = "application/pdf" cc_polizza.size_bytes = len(pdf) cc_polizza.sha256 = digest cc_polizza.document_uploaded_at = datetime.now(timezone.utc) cc_polizza.uploaded_by = BENEFICIARY_USER_ID db.commit() print(f"[custom_checks T1] antiriciclaggio VALIDO, polizza VALIDO + PDF {len(pdf)}b") # ---------- Tranche 2 DRAFT vuota ---------- practice2 = RemissionPractice( call_id=CALL_ID, application_id=APPLICATION_ID, company_id=COMPANY_ID, user_id=BENEFICIARY_USER_ID, status="DRAFT", schema_snapshot=schema_row.schema_json, iva_regime="ORDINARIO", amount_erogato=Decimal("17000"), sequence_number=2, period_label="II fase 2021", suggested_instructor_id=INSTRUCTOR_USER_ID, assigned_instructor_id=None, # non ancora assegnata (simulo workflow capo) ) db.add(practice2) db.flush() print(f"[practice] tranche 2 DRAFT id={practice2.id} (vuota, pronta demo)") db.commit() return practice1.id, practice2.id # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- def main(): ap = argparse.ArgumentParser(description="Seed sandbox Gepafin rendicontazione") ap.add_argument('--reset', action='store_true', help='Cancella tutti i dati remission_* e pulisci storage prima del seed') ap.add_argument('--scenario', choices=['napoli-sas', 'napoli-sas-multi', 'full'], default='napoli-sas') ap.add_argument('--advance', choices=['draft', 'submitted', 'under_review'], default='under_review', help='Stato finale della pratica dopo il seed') args = ap.parse_args() if args.reset: do_reset('all') db = SessionLocal() try: if args.scenario == 'napoli-sas': pid = scenario_napoli_sas(db, advance=args.advance) print(f"\n✓ Scenario napoli-sas completato. practice_id={pid}") print(f" Accedi a: http://78.46.41.91:18072/istruttoria/{pid}") elif args.scenario == 'napoli-sas-multi': pid1, pid2 = scenario_napoli_sas_multi(db) print(f"\n✓ Scenario napoli-sas-multi completato") print(f" tranche 1 APPROVED id={pid1}") print(f" tranche 2 DRAFT id={pid2}") print(f" Istruttoria T1: http://78.46.41.91:18072/istruttoria/{pid1}") print(f" Rendicontazione T2: http://78.46.41.91:18072/rendicontazioni/{pid2}") elif args.scenario == 'full': pid1, pid2 = scenario_napoli_sas_multi(db) print(f"\n✓ Scenario full = napoli-sas-multi (solo questo disponibile).") print(f" tranche 1={pid1} tranche 2={pid2}") finally: db.close() if __name__ == '__main__': main()