diff --git a/scripts/seed_sandbox.py b/scripts/seed_sandbox.py new file mode 100644 index 0000000..199be4d --- /dev/null +++ b/scripts/seed_sandbox.py @@ -0,0 +1,430 @@ +#!/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, +) +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 +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) + + +# --------------------------------------------------------------------------- +# 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_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 + + +# --------------------------------------------------------------------------- +# 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', '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 == 'full': + pid = scenario_napoli_sas(db, advance=args.advance) + # placeholder per futuri scenari ROMA-SRL / BOLOGNA-SPA + print(f"\n✓ Scenario 'full' eseguito (solo napoli-sas disponibile).") + print(f" practice_id={pid}") + finally: + db.close() + + +if __name__ == '__main__': + main()