#!/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}
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()