BFLOWS/gepafin-rendicontazione-api
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
6c089fb7b23fddba978f85eaca58fedef02019fb
gepafin-rendicontazione-api/scripts/seed_sandbox.py
BFLOWS 6c089fb7b2 feat(seed): script sandbox riproducibile con PDF fixture reali 
- scripts/seed_sandbox.py --reset --scenario=napoli-sas --advance=draft|submitted|under_review
- Reset chirurgico (TRUNCATE CASCADE) tabelle remission_* + cleanup /var/uploads
- Ensure schema RE-START pubblicato (idempotente)
- Scenario napoli-sas: 5 fatture (2 B1 Dell/HP, 2 B2 Netcomm/Romano, 1 B3 CertQuality),
  2 ULA (Rossi T_IND FTE 1.0, Bianchi T_DET FTE 0.5), 4 documenti (DURC, VISURA, BILANCIO, ALTRO)
- Tutti gli 11 record hanno PDF fixture generati via weasyprint e caricati nello storage
  tramite lo stesso adapter usato dagli endpoint (no bypass)
- Advance opzionale: lascia la pratica DRAFT / la invia / la fa prendere in carico dall'istruttore
- Documentato nel docstring con esempi di invocazione
- scripts/fixtures/pdf/ directory predisposta per futuri PDF custom

Test: seed --reset --advance=under_review -> 11 PDF reali 196KB totali, practice UNDER_REVIEW
pronta per demo istruttore Cecilia.
2026-04-18 16:54:48 +02:00

431 lines
17 KiB
Python
Raw Blame History

#!/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"<tr><td>{i+1}</td><td>{ln}</td></tr>" for i, ln in enumerate(lines))
html = f"""
<html><head><meta charset="utf-8"><style>
@page {{ size: A4; margin: 2cm; }}
body {{ font-family: "DejaVu Sans", sans-serif; color: #1a202c; }}
h1 {{ color: #1a365d; font-size: 18pt; margin: 0 0 4pt 0; }}
h2 {{ color: #4a5568; font-size: 11pt; margin: 0 0 18pt 0; font-weight: normal; }}
table {{ width: 100%; border-collapse: collapse; margin-top: 12pt; }}
th, td {{ border: 0.5pt solid #cbd5e0; padding: 6pt 10pt; text-align: left; font-size: 10pt; }}
th {{ background: #2d3748; color: white; }}
.stamp {{ margin-top: 40pt; padding: 10pt; border: 1pt solid #4299e1;
background: #ebf4ff; color: #2a4365; font-size: 9pt; }}
</style></head><body>
<h1>{title}</h1>
<h2>{subtitle}</h2>
<table>
<thead><tr><th style="width:50pt">#</th><th>Voce</th></tr></thead>
<tbody>{rows}</tbody>
</table>
<div class="stamp">
<strong>FIXTURE DEMO</strong> — 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.
</div>
</body></html>
"""
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()
Reference in New Issue View Git Blame Copy Permalink