import sys
sys.path.insert(0, "/var/www/html/metafoliadorv2")
import re, io, os, uuid, threading, tempfile
from app.services.timezone import now as tznow
from collections import Counter
from fastapi import APIRouter, Depends, HTTPException, Query
from fastapi.responses import FileResponse
from sqlalchemy.orm import Session
from openpyxl import Workbook
from openpyxl.styles import Font, PatternFill, Alignment, Border, Side
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
import matplotlib.ticker as mticker
from reportlab.lib.pagesizes import A4, landscape
from reportlab.lib.units import cm
from reportlab.lib import colors
from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer, Image, Table, TableStyle, PageBreak
from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle
from reportlab.lib.enums import TA_CENTER, TA_LEFT
from app.models.database import get_db, SessionLocal, tenant_db_url_ctx
from app.models.alfresco_site import AlfrescoSite
from app.models.alfresco_folder import AlfrescoFolder
from app.models.alfresco_file import AlfrescoFile
from app.models.sync_progress import SyncProgress
from app.config import settings
router = APIRouter(prefix="/api/v1/reportes", tags=["reportes-gen"])
CHART_COLORS = ["#6366F1", "#EC4899", "#F59E0B", "#10B981", "#3B82F6", "#8B5CF6", "#EF4444", "#14B8A6", "#F97316", "#06B6D4", "#84CC16", "#D946EF"]
def _make_session(db_url: str):
from sqlalchemy import create_engine, event
from sqlalchemy.orm import sessionmaker
from app.models.database import _set_session_utc
engine = create_engine(db_url)
event.listen(engine, "connect", _set_session_utc)
return sessionmaker(autocommit=False, autoflush=False, bind=engine)()
def has_tipologia(name):
return bool(re.match(r"^\d+(?:-\d+)*\.\d+\s", name))
def _load_tipologia_keys(db):
"""Carga las claves de tipología desde alfresco_tipologys."""
from app.models.alfresco_tipology import AlfrescoTipology
rows = db.query(AlfrescoTipology).filter(
AlfrescoTipology.deleted_at == None,
AlfrescoTipology.value != None,
AlfrescoTipology.value != "",
).all()
return [t.value for t in rows]
def _find_tipologia_folders(db, folder_ids: list[str], tipologia_keys: list[str]):
"""Retorna un set de folder_ids cuyos archivos tienen propiedades que coinciden con alguna key de tipología."""
if not folder_ids or not tipologia_keys:
return set()
from sqlalchemy import text
result = set()
fids = list(folder_ids)
keys_array = "{" + ",".join(tipologia_keys) + "}"
for i in range(0, len(fids), 2000):
batch = fids[i:i + 2000]
placeholders = ", ".join(f":f{j}" for j in range(len(batch)))
params = {f"f{j}": fid for j, fid in enumerate(batch)}
params["keys"] = keys_array
rows = db.execute(text(f"""
SELECT DISTINCT folder_id
FROM alfresco_files
WHERE folder_id IN ({placeholders})
AND deleted = false
AND name != 'index.xml'
AND properties_json IS NOT NULL
AND properties_json->'properties' ?| :keys
"""), params).fetchall()
for (fid,) in rows:
result.add(fid)
return result
def _build_folder_tree(db, root_folder_ids: list):
"""CTE recursivo: obtiene el árbol completo de carpetas bajo los roots dados.
Retorna (folder_map, root_children) donde folder_map es {id: {name, parent_id, has_index_xml}}
y root_children es la lista de hijos directos de cada root."""
from sqlalchemy import text
placeholders = ", ".join(f":rid{i}" for i in range(len(root_folder_ids)))
params = {f"rid{i}": rid for i, rid in enumerate(root_folder_ids)}
rows = db.execute(text(f"""
WITH RECURSIVE tree AS (
SELECT id, parent_id, name, has_index_xml, 0 AS depth
FROM alfresco_folders
WHERE parent_id IN ({placeholders}) AND deleted = false
UNION ALL
SELECT f.id, f.parent_id, f.name, f.has_index_xml, t.depth + 1
FROM alfresco_folders f
JOIN tree t ON f.parent_id = t.id
WHERE f.deleted = false AND t.depth < 20
)
SELECT id, parent_id, name, has_index_xml FROM tree
"""), params).fetchall()
folder_map = {}
for r in rows:
folder_map[r[0]] = {"parent_id": r[1], "name": r[2], "has_index_xml": r[3]}
return folder_map
def _count_files_by_folder(db, folder_ids: list):
"""Cuenta archivos por carpeta en una sola query."""
from sqlalchemy import text
if not folder_ids:
return {}
placeholders = ", ".join(f":fid{i}" for i in range(len(folder_ids)))
params = {f"fid{i}": fid for i, fid in enumerate(folder_ids)}
rows = db.execute(text(f"""
SELECT folder_id, COUNT(*) AS cnt
FROM alfresco_files
WHERE folder_id IN ({placeholders}) AND deleted = false AND name != 'index.xml'
GROUP BY folder_id
"""), params).fetchall()
return {r[0]: r[1] for r in rows}
def _count_files_recursive(folder_map, folder_id, file_counts):
"""Cuenta archivos recursivamente desde un diccionario en memoria."""
total = file_counts.get(folder_id, 0)
children = [fid for fid, f in folder_map.items() if f["parent_id"] == folder_id]
for ch in children:
total += _count_files_recursive(folder_map, ch, file_counts)
return total
def _count_indices_recursive(folder_map, folder_id):
"""Cuenta índices recursivamente desde memoria."""
total = 1 if folder_map.get(folder_id, {}).get("has_index_xml") else 0
children = [fid for fid, f in folder_map.items() if f["parent_id"] == folder_id]
for ch in children:
total += _count_indices_recursive(folder_map, ch)
return total
def _get_tipologias_recursive_fast(folder_map, file_counts, root_id):
"""Obtiene tipologías recursivamente desde memoria."""
from collections import Counter
tipologias = Counter()
children = [(c, folder_map[c]["name"]) for c in folder_map if folder_map[c]["parent_id"] == root_id]
for ch_id, ch_name in children:
if has_tipologia(ch_name):
tipologias[ch_name] += _count_files_recursive(folder_map, ch_id, file_counts)
else:
sub = _get_tipologias_recursive_fast(folder_map, file_counts, ch_id)
for k, v in sub.items():
tipologias[k] += v
fc = file_counts.get(root_id, 0)
if fc > 0:
tipologias["SIN TIPOLOGIA"] += fc
return tipologias
def _get_extensions_by_folder(db, folder_ids: list):
"""Obtiene extensiones de archivos agrupadas por carpeta en una sola query."""
from sqlalchemy import text
if not folder_ids:
return {}
placeholders = ", ".join(f":fid{i}" for i in range(len(folder_ids)))
params = {f"fid{i}": fid for i, fid in enumerate(folder_ids)}
rows = db.execute(text(f"""
SELECT folder_id, name
FROM alfresco_files
WHERE folder_id IN ({placeholders}) AND deleted = false AND name != 'index.xml'
"""), params).fetchall()
result = {}
for fid, fname in rows:
if fid not in result:
result[fid] = Counter()
if fname and "." in fname:
ext = fname.rsplit(".", 1)[-1].lower()
result[fid][ext] += 1
else:
result[fid]["SIN EXTENSION"] += 1
return result
def _get_extensions_recursive_fast(folder_map, ext_by_folder, root_id):
"""Suma extensiones recursivamente desde memoria."""
from collections import Counter
total = Counter()
for k, v in ext_by_folder.get(root_id, {}).items():
total[k] += v
children = [c for c in folder_map if folder_map[c]["parent_id"] == root_id]
for ch in children:
sub = _get_extensions_recursive_fast(folder_map, ext_by_folder, ch)
for k, v in sub.items():
total[k] += v
return total
def _fig_to_bytes(fig, dpi=120):
buf = io.BytesIO()
fig.savefig(buf, format="png", dpi=dpi, bbox_inches="tight", facecolor="white")
buf.seek(0)
plt.close(fig)
return buf
def _make_bar_chart(data, title, xlabel="", ylabel="Cantidad", max_bars=20):
items = data.most_common(max_bars)
labels = [it[0] for it in items]
values = [it[1] for it in items]
if not values or all(v == 0 for v in values):
return None
fig, ax = plt.subplots(figsize=(8, 4.5))
colors_list = [CHART_COLORS[i % len(CHART_COLORS)] for i in range(len(labels))]
bars = ax.bar(range(len(labels)), values, color=colors_list, edgecolor="white", linewidth=0.5)
ax.set_xticks(range(len(labels)))
ax.set_xticklabels(labels, rotation=45, ha="right", fontsize=8)
ax.set_ylabel(ylabel, fontsize=9)
ax.set_title(title, fontsize=11, fontweight="bold", pad=10)
for bar, val in zip(bars, values):
ax.text(bar.get_x() + bar.get_width() / 2, bar.get_height() + max(values) * 0.01,
str(val), ha="center", va="bottom", fontsize=7)
fig.tight_layout()
return _fig_to_bytes(fig)
def _make_pie_chart(data, title, max_slices=10):
items = data.most_common(max_slices)
labels = [it[0] for it in items]
values = [it[1] for it in items]
total = sum(values)
if total == 0:
return None
fig, ax = plt.subplots(figsize=(8, 4.5))
colors_list = [CHART_COLORS[i % len(CHART_COLORS)] for i in range(len(labels))]
wedges, texts, autotexts = ax.pie(values, labels=None, autopct=lambda p: f"{p:.1f}%" if p > 0 else "",
colors=colors_list, startangle=90,
textprops={"fontsize": 8})
for at in autotexts:
at.set_fontsize(7)
ax.set_title(title, fontsize=11, fontweight="bold", pad=10)
ax.legend(wedges, labels, title="", loc="center left", bbox_to_anchor=(1, 0.5), fontsize=7)
fig.tight_layout()
return _fig_to_bytes(fig)
def _make_line_chart(data, title, xlabel="", ylabel="Cantidad", max_points=20):
items = data.most_common(max_points)
labels = [it[0] for it in items]
values = [it[1] for it in items]
if not values or all(v == 0 for v in values):
return None
fig, ax = plt.subplots(figsize=(8, 4.5))
ax.plot(range(len(labels)), values, marker="o", color=CHART_COLORS[0], linewidth=2, markersize=5)
ax.set_xticks(range(len(labels)))
ax.set_xticklabels(labels, rotation=45, ha="right", fontsize=8)
ax.set_ylabel(ylabel, fontsize=9)
ax.set_title(title, fontsize=11, fontweight="bold", pad=10)
ax.grid(True, linestyle="--", alpha=0.5)
for i, val in enumerate(values):
ax.annotate(str(val), (i, val), textcoords="offset points", xytext=(0, 8),
ha="center", fontsize=7)
fig.tight_layout()
return _fig_to_bytes(fig)
def _make_horizontal_bar_chart(data, title, xlabel="Cantidad", max_bars=20):
items = data.most_common(max_bars)
labels = [it[0] for it in items]
values = [it[1] for it in items]
if not values or all(v == 0 for v in values):
return None
fig, ax = plt.subplots(figsize=(8, 4.5))
colors_list = [CHART_COLORS[i % len(CHART_COLORS)] for i in range(len(labels))]
bars = ax.barh(range(len(labels)), values, color=colors_list, edgecolor="white", linewidth=0.5)
ax.set_yticks(range(len(labels)))
ax.set_yticklabels(labels, fontsize=8)
ax.set_xlabel(xlabel, fontsize=9)
ax.set_title(title, fontsize=11, fontweight="bold", pad=10)
ax.invert_yaxis()
for bar, val in zip(bars, values):
ax.text(bar.get_width() + max(values) * 0.01, bar.get_y() + bar.get_height() / 2,
str(val), va="center", fontsize=7)
fig.tight_layout()
return _fig_to_bytes(fig)
def _make_area_chart(data, title, xlabel="", ylabel="Cantidad", max_points=20):
items = data.most_common(max_points)
labels = [it[0] for it in items]
values = [it[1] for it in items]
if not values or all(v == 0 for v in values):
return None
fig, ax = plt.subplots(figsize=(8, 4.5))
ax.fill_between(range(len(labels)), values, alpha=0.4, color=CHART_COLORS[0])
ax.plot(range(len(labels)), values, marker="o", color=CHART_COLORS[0], linewidth=2, markersize=5)
ax.set_xticks(range(len(labels)))
ax.set_xticklabels(labels, rotation=45, ha="right", fontsize=8)
ax.set_ylabel(ylabel, fontsize=9)
ax.set_title(title, fontsize=11, fontweight="bold", pad=10)
ax.grid(True, linestyle="--", alpha=0.3)
for i, val in enumerate(values):
ax.annotate(str(val), (i, val), textcoords="offset points", xytext=(0, 8),
ha="center", fontsize=7)
fig.tight_layout()
return _fig_to_bytes(fig)
def _make_scatter_chart(data, title, xlabel="", ylabel="Cantidad", max_points=20):
items = data.most_common(max_points)
labels = [it[0] for it in items]
values = [it[1] for it in items]
if not values or all(v == 0 for v in values):
return None
fig, ax = plt.subplots(figsize=(8, 4.5))
x = range(len(labels))
ax.scatter(x, values, c=CHART_COLORS[0], s=80, alpha=0.7, edgecolors="white", linewidth=0.5)
ax.set_xticks(x)
ax.set_xticklabels(labels, rotation=45, ha="right", fontsize=8)
ax.set_ylabel(ylabel, fontsize=9)
ax.set_title(title, fontsize=11, fontweight="bold", pad=10)
ax.grid(True, linestyle="--", alpha=0.3)
for i, val in enumerate(values):
ax.annotate(str(val), (i, val), textcoords="offset points", xytext=(0, 8),
ha="center", fontsize=7)
fig.tight_layout()
return _fig_to_bytes(fig)
def _make_histogram_chart(data, title, xlabel="", ylabel="Frecuencia"):
values = list(data.values())
if not values or all(v == 0 for v in values):
return None
fig, ax = plt.subplots(figsize=(8, 4.5))
ax.hist(values, bins=min(15, len(set(values)) or 1), color=CHART_COLORS[0],
edgecolor="white", alpha=0.8)
ax.set_xlabel(xlabel or "Cantidad", fontsize=9)
ax.set_ylabel(ylabel, fontsize=9)
ax.set_title(title, fontsize=11, fontweight="bold", pad=10)
ax.grid(True, linestyle="--", alpha=0.3)
fig.tight_layout()
return _fig_to_bytes(fig)
CHART_GENERATORS = {
"barras": _make_bar_chart,
"circular": _make_pie_chart,
"lineal": _make_line_chart,
"barras_horizontal": _make_horizontal_bar_chart,
"area": _make_area_chart,
"dispersion": _make_scatter_chart,
"histograma": _make_histogram_chart,
}
CHART_NAMES = {
"barras": "de barras",
"circular": "circular",
"lineal": "lineal",
"barras_horizontal": "de barras horizontales",
"area": "de área",
"dispersion": "de dispersión",
"histograma": "histograma",
}
@router.get("/trd-tipologias/status", summary="Estado de reporte TRD/Tipologías", description="Consulta el estado de generación de un reporte TRD/Tipologías por su job_id.")
async def get_report_status(job_id: str = Query(..., description="ID del trabajo de generación del reporte"), db: Session = Depends(get_db)):
progress = db.query(SyncProgress).filter(SyncProgress.task_id == job_id).first()
if not progress:
return {"status": "not_found"}
return {"status": progress.status, "error": progress.error_message or ""}
@router.get("/trd-tipologias/download", summary="Descargar reporte TRD/Tipologías", description="Descarga el archivo Excel del reporte TRD/Tipologías generado. Elimina el registro de progreso tras la descarga.")
async def download_report(job_id: str = Query(..., description="ID del trabajo de generación del reporte"), db: Session = Depends(get_db)):
progress = db.query(SyncProgress).filter(SyncProgress.task_id == job_id).first()
if not progress or progress.status != "completed":
raise HTTPException(status_code=404, detail="Reporte no disponible")
path = "/tmp/trd_" + job_id + ".xlsx"
if not os.path.exists(path):
raise HTTPException(status_code=404, detail="Archivo no encontrado")
db.delete(progress)
db.commit()
site2 = db.query(AlfrescoSite).filter(AlfrescoSite.id == progress.site_id).first()
ts = tznow().strftime("%Y%m%d_%H%M%S")
fname = "TRD_Tipologias_" + (site2.name if site2 else "") + "_" + ts + ".xlsx"
return FileResponse(path, filename=fname,
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")
@router.get("/sites-with-folders", summary="Sitios con Series, Subseries y Expedientes", description="Lista los sitios activos que tienen Series, Subseries y Expedientes sincronizados, para usar en selectores de reportes.")
async def sites_with_folders(db: Session = Depends(get_db)):
from sqlalchemy import distinct
site_ids = db.query(distinct(AlfrescoFolder.site_id)).filter(AlfrescoFolder.deleted == False).all()
ids = [s[0] for s in site_ids]
sites = db.query(AlfrescoSite).filter(AlfrescoSite.id.in_(ids), AlfrescoSite.active == True).order_by(AlfrescoSite.name).all()
return [{"id": s.id, "name": s.name + " (" + (s.title or s.name) + ")"} for s in sites]
@router.get("/trd-tipologias", summary="Generar reporte TRD/Tipologías", description="Genera de forma asíncrona un reporte Excel con el árbol jerárquico de tipologías documentales y conteo de archivos por Serie, Subserie o Expediente. Devuelve un job_id para consultar estado y descargar.")
async def generate_trd_report(site_id: int = Query(..., description="ID del sitio para el reporte TRD"), max_levels: int = Query(3, description="Niveles máximos de profundidad en el árbol"), db: Session = Depends(get_db)):
site = db.query(AlfrescoSite).filter(AlfrescoSite.id == site_id).first()
if not site:
raise HTTPException(status_code=404, detail="Sitio no encontrado")
job_id = str(uuid.uuid4())
progress = SyncProgress(task_id=job_id, site_id=site_id, status="running", message="Generando reporte...")
db.add(progress)
db.commit()
db_url = tenant_db_url_ctx.get() or settings.database_url
def run_report():
db2 = _make_session(db_url)
try:
_generate_trd(db2, site_id, max_levels, job_id, db_url)
except Exception as e:
db3 = _make_session(db_url)
try:
p = db3.query(SyncProgress).filter(SyncProgress.task_id == job_id).first()
if p:
p.status = "error"
p.error_message = str(e)[:500]
db3.commit()
finally:
db3.close()
finally:
db2.close()
threading.Thread(target=run_report, daemon=True).start()
return {"ok": True, "job_id": job_id}
def _generate_trd(db, site_id, max_levels, job_id, db_url=None):
site = db.query(AlfrescoSite).filter(AlfrescoSite.id == site_id).first()
if not site:
return
roots = db.query(AlfrescoFolder).filter(
AlfrescoFolder.site_id == site_id,
AlfrescoFolder.parent_id == None,
AlfrescoFolder.deleted == False
).order_by(AlfrescoFolder.name).all()
doclib = None
for root in roots:
if root.name == "documentLibrary":
doclib = root
break
if doclib:
root_ids = [doclib.id]
elif roots:
root_ids = [r.id for r in roots]
else:
root_ids = []
folder_map = _build_folder_tree(db, root_ids) if root_ids else {}
all_ids = list(folder_map.keys())
file_counts = _count_files_by_folder(db, all_ids)
tipologia_keys = _load_tipologia_keys(db)
tipologia_folders = _find_tipologia_folders(db, all_ids, tipologia_keys) if tipologia_keys else set()
wb = Workbook()
ws = wb.active
ws.title = "TRD Tipologias"
header_font = Font(bold=True, color="FFFFFF", size=11)
header_fill = PatternFill(start_color="6366F1", end_color="6366F1", fill_type="solid")
thin_border = Border(
left=Side(style="thin"), right=Side(style="thin"),
top=Side(style="thin"), bottom=Side(style="thin")
)
rows = []
folder_count = [0]
top_ids = [fid for fid in folder_map if folder_map[fid]["parent_id"] in root_ids]
def process_branch(folder_id, depth, max_depth, parents):
name = folder_map[folder_id]["name"]
is_tip = folder_id in tipologia_folders
file_count = _count_files_recursive(folder_map, folder_id, file_counts)
row = list(parents) + [name]
last_val = ""
if is_tip:
last_val = str(file_count)
row.append(last_val)
rows.append((row, "tipologia"))
return
elif depth >= max_depth - 1:
last_val = "SIN TIPOLOGIA - " + str(file_count) + " archivos"
row.append(last_val)
rows.append((row, "sin_tip"))
return
children = [fid for fid in folder_map if folder_map[fid]["parent_id"] == folder_id]
children.sort(key=lambda cid: folder_map[cid]["name"])
if not children:
last_val = "SIN TIPOLOGIA - " + str(file_count) + " archivos"
row.append(last_val)
rows.append((row, "sin_tip"))
else:
new_parents = list(parents) + [name]
for ch_id in children:
folder_count[0] += 1
if folder_count[0] % 100 == 0:
try:
db3 = _make_session(db_url)
p = db3.query(SyncProgress).filter(SyncProgress.task_id == job_id).first()
if p:
p.message = f"Generando: {folder_count[0]} Series, Subseries y Expedientes procesados"
db3.commit()
db3.close()
except: pass
process_branch(ch_id, depth + 1, max_depth, new_parents)
for fid in top_ids:
process_branch(fid, 0, max_levels, [])
max_cols = max(len(r[0]) for r in rows) if rows else 2
max_cols = min(max_cols, 50)
headers = []
for i in range(max_cols - 1):
headers.append("Nivel " + str(i + 1))
headers.append("Archivos")
for c, h in enumerate(headers, 1):
cell = ws.cell(row=1, column=c, value=h)
cell.font = header_font
cell.fill = header_fill
cell.alignment = Alignment(horizontal="center")
cell.border = thin_border
for r_idx, (row, marker) in enumerate(rows, 2):
padded_row = list(row) + [""] * (max_cols - len(row))
for c_idx in range(max_cols):
val = padded_row[c_idx]
cell = ws.cell(row=r_idx, column=c_idx + 1, value=val)
cell.border = thin_border
cell.alignment = Alignment(wrap_text=True)
for col in range(1, max_cols + 1):
ws.column_dimensions[ws.cell(row=1, column=col).column_letter].width = 45
tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".xlsx", dir="/tmp")
wb.save(tmp.name)
tmp.close()
final_path = "/tmp/trd_" + job_id + ".xlsx"
os.rename(tmp.name, final_path)
db3 = _make_session(db_url or settings.database_url)
try:
p = db3.query(SyncProgress).filter(SyncProgress.task_id == job_id).first()
if p:
p.status = "completed"
p.message = "Completado: " + str(len(rows)) + " filas"
db3.commit()
finally:
db3.close()
@router.get("/graficos-dependencia/status", summary="Estado de reporte gráficos por dependencia", description="Consulta el estado de generación de un reporte PDF de gráficos por su job_id.")
async def get_graficos_status(job_id: str = Query(..., description="ID del trabajo de generación del reporte"), db: Session = Depends(get_db)):
progress = db.query(SyncProgress).filter(SyncProgress.task_id == job_id).first()
if not progress:
return {"status": "not_found"}
return {"status": progress.status, "error": progress.error_message or ""}
@router.get("/graficos-dependencia/download", summary="Descargar reporte gráficos por dependencia", description="Descarga el archivo PDF del reporte de gráficos por dependencia generado. Elimina el registro de progreso tras la descarga.")
async def download_graficos(job_id: str = Query(..., description="ID del trabajo de generación del reporte"), db: Session = Depends(get_db)):
progress = db.query(SyncProgress).filter(SyncProgress.task_id == job_id).first()
if not progress or progress.status != "completed":
raise HTTPException(status_code=404, detail="Reporte no disponible")
path = "/tmp/graficos_" + job_id + ".pdf"
if not os.path.exists(path):
raise HTTPException(status_code=404, detail="Archivo no encontrado")
db.delete(progress)
db.commit()
site2 = db.query(AlfrescoSite).filter(AlfrescoSite.id == progress.site_id).first()
ts = tznow().strftime("%Y%m%d_%H%M%S")
fname = "Graficos_Dependencia_" + (site2.name if site2 else "") + "_" + ts + ".pdf"
return FileResponse(path, filename=fname, media_type="application/pdf")
@router.get("/graficos-dependencia", summary="Generar reporte gráficos por dependencia", description="Genera de forma asíncrona un reporte PDF con gráficos estadísticos por dependencia: tipologías, extensiones de archivo y series documentales. Soporta múltiples tipos de gráfico (barras, circular, lineal, barras_horizontal, area, dispersion, histograma).")
async def generate_graficos_report(
site_id: int = Query(..., description="ID del sitio para el reporte de gráficos"),
chart_types: str = Query("barras", description="Tipos de gráfico separados por coma: barras, circular, lineal, barras_horizontal, area, dispersion, histograma"),
db: Session = Depends(get_db)
):
site = db.query(AlfrescoSite).filter(AlfrescoSite.id == site_id).first()
if not site:
raise HTTPException(status_code=404, detail="Sitio no encontrado")
requested = [ct.strip() for ct in chart_types.split(",") if ct.strip() in CHART_GENERATORS]
if not requested:
requested = ["barras"]
job_id = str(uuid.uuid4())
progress = SyncProgress(task_id=job_id, site_id=site_id, status="running", message="Generando reporte de gráficos...")
db.add(progress)
db.commit()
db_url = tenant_db_url_ctx.get() or settings.database_url
def run_report():
db2 = _make_session(db_url)
try:
_generate_graficos(db2, site_id, requested, job_id, db_url)
except Exception as e:
db3 = _make_session(db_url)
try:
p = db3.query(SyncProgress).filter(SyncProgress.task_id == job_id).first()
if p:
p.status = "error"
p.error_message = str(e)[:500]
db3.commit()
finally:
db3.close()
finally:
db2.close()
threading.Thread(target=run_report, daemon=True).start()
return {"ok": True, "job_id": job_id}
def _generate_graficos(db, site_id, chart_types, job_id, db_url=None):
site = db.query(AlfrescoSite).filter(AlfrescoSite.id == site_id).first()
if not site:
return
roots = db.query(AlfrescoFolder).filter(
AlfrescoFolder.site_id == site_id,
AlfrescoFolder.parent_id == None,
AlfrescoFolder.deleted == False
).order_by(AlfrescoFolder.name).all()
doclib = None
for root in roots:
if root.name == "documentLibrary":
doclib = root
break
if not doclib:
doclib = roots[0] if roots else None
if not doclib:
_fail_job(job_id, "No se encontró documentLibrary ni Series, Subseries o Expedientes raíz", db_url)
return
dependencias = db.query(AlfrescoFolder).filter(
AlfrescoFolder.parent_id == doclib.id,
AlfrescoFolder.deleted == False
).order_by(AlfrescoFolder.name).all()
if not dependencias:
_fail_job(job_id, "No se encontraron dependencias en el sitio", db_url)
return
dep_ids = [d.id for d in dependencias]
folder_map = _build_folder_tree(db, dep_ids)
all_ids = list(folder_map.keys()) + dep_ids
for d in dependencias:
folder_map[d.id] = {"parent_id": doclib.id, "name": d.name, "has_index_xml": d.has_index_xml}
file_counts = _count_files_by_folder(db, all_ids)
ext_by_folder = _get_extensions_by_folder(db, all_ids)
styles = getSampleStyleSheet()
title_style = ParagraphStyle("CustomTitle", parent=styles["Title"], fontSize=16, spaceAfter=6, alignment=TA_CENTER)
h2_style = ParagraphStyle("CustomH2", parent=styles["Heading2"], fontSize=13, spaceAfter=4, spaceBefore=10)
h3_style = ParagraphStyle("CustomH3", parent=styles["Heading3"], fontSize=11, spaceAfter=3, spaceBefore=8)
normal_style = ParagraphStyle("CustomNormal", parent=styles["Normal"], fontSize=9, leading=13)
pdf_path = "/tmp/graficos_" + job_id + ".pdf"
doc = SimpleDocTemplate(pdf_path, pagesize=A4,
leftMargin=1.5 * cm, rightMargin=1.5 * cm,
topMargin=1.5 * cm, bottomMargin=1.5 * cm)
story = []
story.append(Paragraph("Reporte de Gráficos por Dependencia", title_style))
story.append(Paragraph("Sitio: " + (site.title or site.name), normal_style))
story.append(Paragraph("Fecha: " + tznow().strftime("%Y-%m-%d %H:%M"), normal_style))
story.append(Spacer(1, 0.5 * cm))
dependencia_data = []
for dep in dependencias:
idx_count = _count_indices_recursive(folder_map, dep.id)
dependencia_data.append((dep.name, idx_count))
table_data = [["Dependencia", "Cantidad de Índices"]]
for name, count in dependencia_data:
table_data.append([Paragraph(name, normal_style), str(count)])
tbl = Table(table_data, colWidths=[12 * cm, 4 * cm])
tbl.setStyle(TableStyle([
("BACKGROUND", (0, 0), (-1, 0), colors.HexColor("#6366F1")),
("TEXTCOLOR", (0, 0), (-1, 0), colors.white),
("FONTNAME", (0, 0), (-1, 0), "Helvetica-Bold"),
("FONTSIZE", (0, 0), (-1, 0), 10),
("FONTSIZE", (0, 1), (-1, -1), 9),
("ALIGN", (1, 0), (1, -1), "CENTER"),
("GRID", (0, 0), (-1, -1), 0.5, colors.HexColor("#D1D5DB")),
("ROWBACKGROUNDS", (0, 1), (-1, -1), [colors.white, colors.HexColor("#F3F4F6")]),
("VALIGN", (0, 0), (-1, -1), "MIDDLE"),
("TOPPADDING", (0, 0), (-1, -1), 4),
("BOTTOMPADDING", (0, 0), (-1, -1), 4),
]))
story.append(Paragraph("Resumen de Dependencias", h2_style))
story.append(tbl)
story.append(Spacer(1, 0.5 * cm))
sin_datos_style = ParagraphStyle("SinDatos", parent=normal_style, textColor=colors.HexColor("#9CA3AF"), fontSize=9, leftIndent=10)
def _add_chart_section(label, data_counter, chart_types, dep_name):
story.append(Paragraph(label, h3_style))
if not data_counter or sum(data_counter.values()) == 0:
story.append(Paragraph("sin datos a mostrar", sin_datos_style))
story.append(Spacer(1, 0.3 * cm))
return
any_chart = False
for ct in chart_types:
gen = CHART_GENERATORS[ct]
chart_title = "Gráfico " + CHART_NAMES.get(ct, ct) + " " + label + " - " + dep_name
buf = gen(data_counter, chart_title)
if buf is None:
continue
any_chart = True
img = Image(buf, width=16 * cm, height=9 * cm, kind="proportional")
story.append(img)
story.append(Spacer(1, 0.3 * cm))
if not any_chart:
story.append(Paragraph("sin datos a mostrar", sin_datos_style))
story.append(Spacer(1, 0.3 * cm))
for dep in dependencias:
story.append(PageBreak())
story.append(Paragraph("Dependencia: " + dep.name, h2_style))
story.append(Spacer(1, 0.3 * cm))
tipologias = _get_tipologias_recursive_fast(folder_map, file_counts, dep.id)
_add_chart_section("Gráfico por Tipología Documental", tipologias, chart_types, dep.name)
exts = _get_extensions_recursive_fast(folder_map, ext_by_folder, dep.id)
_add_chart_section("Gráfico por Extensión de Archivo", exts, chart_types, dep.name)
series = [fid for fid in folder_map if folder_map[fid]["parent_id"] == dep.id]
serie_counts = Counter()
for s_id in sorted(series, key=lambda x: folder_map[x]["name"]):
serie_counts[folder_map[s_id]["name"]] = _count_files_recursive(folder_map, s_id, file_counts)
_add_chart_section("Gráfico por Serie Documental (Archivos Totales)", serie_counts, chart_types, dep.name)
doc.build(story)
db3 = _make_session(db_url or settings.database_url)
try:
p = db3.query(SyncProgress).filter(SyncProgress.task_id == job_id).first()
if p:
p.status = "completed"
p.message = "Completado: " + str(len(dependencias)) + " dependencias procesadas"
db3.commit()
finally:
db3.close()
def _fail_job(job_id, message, db_url=None):
db3 = _make_session(db_url or settings.database_url)
try:
p = db3.query(SyncProgress).filter(SyncProgress.task_id == job_id).first()
if p:
p.status = "error"
p.error_message = message
db3.commit()
finally:
db3.close()
# ── FUID (Formato Único de Inventario Documental) ──────────────────────────
def _get_descendant_folders(db, parent_id):
"""Recursively get all descendant folders of a given folder."""
result = []
children = db.query(AlfrescoFolder).filter(
AlfrescoFolder.parent_id == parent_id, AlfrescoFolder.deleted == False
).all()
for ch in children:
result.append(ch)
result.extend(_get_descendant_folders(db, ch.id))
return result
def _extract_trd_code_and_name(name):
"""Extrae código y nombre de carpeta con formato TRD: '3-1110-1-06 CERTIFICACIONES Y CONSTANCIAS' → ('3-1110-1-06', 'CERTIFICACIONES Y CONSTANCIAS').
También soporta códigos simples como '540000 Oficina...' → ('540000', 'Oficina...')."""
if not name:
return "", ""
m = re.match(r'^([\d]+(?:-[\d]+)*(?:\.[\d]+)?)\s+(.+)', name)
if m:
return m.group(1).strip(), m.group(2).strip()
return "", name
def _get_folder_ancestors_to_dep(db, folder, dependency_id):
"""Devuelve la lista de ancestros desde dependency_id (exclusive) hasta folder (exclusive), ordenados de dependency hacia folder."""
ancestors = []
current_id = folder.parent_id
visited = set()
while current_id and current_id != dependency_id:
if current_id in visited:
break
visited.add(current_id)
parent = db.query(AlfrescoFolder).filter(
AlfrescoFolder.id == current_id, AlfrescoFolder.deleted == False
).first()
if not parent:
break
ancestors.append(parent)
current_id = parent.parent_id
ancestors.reverse()
return ancestors
def _get_all_files_under(db, folder_id):
"""Obtiene recursivamente todos los archivos no eliminados bajo una carpeta."""
files = db.query(AlfrescoFile).filter(
AlfrescoFile.folder_id == folder_id, AlfrescoFile.deleted == False
).all()
result = list(files)
children = db.query(AlfrescoFolder).filter(
AlfrescoFolder.parent_id == folder_id, AlfrescoFolder.deleted == False
).all()
for ch in children:
result.extend(_get_all_files_under(db, ch.id))
return result
@router.get("/fuid/dependencias", summary="FUID - listar dependencias", description="Lista las dependencias (primer nivel bajo documentLibrary) de un sitio para el reporte FUID.")
async def fuid_dependencias(site_id: int = Query(..., description="ID del sitio Alfresco"), db: Session = Depends(get_db)):
site = db.query(AlfrescoSite).filter(AlfrescoSite.id == site_id).first()
if not site:
raise HTTPException(status_code=404, detail="Sitio no encontrado")
roots = db.query(AlfrescoFolder).filter(
AlfrescoFolder.site_id == site_id,
AlfrescoFolder.parent_id == None,
AlfrescoFolder.deleted == False
).all()
doclib = None
for r in roots:
if r.name == "documentLibrary":
doclib = r
break
if not doclib:
doclib = roots[0] if roots else None
if not doclib:
return []
deps = db.query(AlfrescoFolder).filter(
AlfrescoFolder.parent_id == doclib.id,
AlfrescoFolder.deleted == False
).order_by(AlfrescoFolder.name).all()
return [{"id": d.id, "name": d.name, "node_path": d.node_path} for d in deps]
@router.get("/fuid", summary="Generar reporte FUID", description="Genera de forma asíncrona el Formato Único de Inventario Documental (FUID). Si no se especifica dependencia, genera para todas las dependencias del sitio (una hoja por dependencia en Excel, una página por dependencia en PDF).")
async def generate_fuid_report(
site_id: int = Query(..., description="ID del sitio Alfresco"),
dependency_id: str = Query(None, description="Node ID de la dependencia (opcional). Si se omite, genera para todas."),
format: str = Query("excel", description="Formato de salida: excel o pdf"),
db: Session = Depends(get_db)
):
site = db.query(AlfrescoSite).filter(AlfrescoSite.id == site_id).first()
if not site:
raise HTTPException(status_code=404, detail="Sitio no encontrado")
if format not in ("excel", "pdf"):
raise HTTPException(status_code=400, detail="Formato inválido. Use 'excel' o 'pdf'")
deps_to_process = []
if dependency_id:
dep = db.query(AlfrescoFolder).filter(
AlfrescoFolder.id == dependency_id,
AlfrescoFolder.site_id == site_id,
AlfrescoFolder.deleted == False
).first()
if not dep:
raise HTTPException(status_code=404, detail="Dependencia no encontrada")
deps_to_process = [dep]
else:
roots = db.query(AlfrescoFolder).filter(
AlfrescoFolder.site_id == site_id,
AlfrescoFolder.parent_id == None,
AlfrescoFolder.deleted == False
).all()
doclib = None
for r in roots:
if r.name == "documentLibrary":
doclib = r
break
if not doclib:
doclib = roots[0] if roots else None
if doclib:
deps_to_process = db.query(AlfrescoFolder).filter(
AlfrescoFolder.parent_id == doclib.id,
AlfrescoFolder.deleted == False
).order_by(AlfrescoFolder.name).all()
if not deps_to_process:
raise HTTPException(status_code=404, detail="No se encontraron dependencias en el sitio")
job_id = str(uuid.uuid4())
dep_names = ", ".join(d.name for d in deps_to_process[:3])
if len(deps_to_process) > 3:
dep_names += f" y {len(deps_to_process) - 3} más"
progress = SyncProgress(
task_id=job_id, site_id=site_id, status="running",
message=f"Generando FUID {format.upper()} para {dep_names}..."
)
db.add(progress)
db.commit()
db_url = tenant_db_url_ctx.get() or settings.database_url
dep_ids = [d.id for d in deps_to_process]
def run_report():
db2 = _make_session(db_url)
try:
if len(dep_ids) == 1:
_generate_fuid(db2, site_id, dep_ids[0], format, job_id, db_url)
else:
_generate_fuid_multi(db2, site_id, dep_ids, format, job_id, db_url)
except Exception as e:
db3 = _make_session(db_url)
try:
p = db3.query(SyncProgress).filter(SyncProgress.task_id == job_id).first()
if p:
p.status = "error"
p.error_message = str(e)[:500]
db3.commit()
finally:
db3.close()
finally:
db2.close()
threading.Thread(target=run_report, daemon=True).start()
return {"ok": True, "job_id": job_id}
@router.get("/fuid/status", summary="FUID - estado de generación", description="Consulta el estado de generación de un reporte FUID por su job_id.")
async def fuid_status(job_id: str = Query(..., description="ID del trabajo"), db: Session = Depends(get_db)):
progress = db.query(SyncProgress).filter(SyncProgress.task_id == job_id).first()
if not progress:
return {"status": "not_found"}
return {"status": progress.status, "error": progress.error_message or ""}
@router.get("/fuid/download", summary="FUID - descargar reporte", description="Descarga el archivo FUID generado en formato Excel o PDF.")
async def fuid_download(
job_id: str = Query(..., description="ID del trabajo"),
format: str = Query("excel", description="Formato del archivo generado"),
db: Session = Depends(get_db)
):
progress = db.query(SyncProgress).filter(SyncProgress.task_id == job_id).first()
if not progress or progress.status != "completed":
raise HTTPException(status_code=404, detail="Reporte no disponible")
ext = ".xlsx" if format == "excel" else ".pdf"
content_type = (
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
if format == "excel" else "application/pdf"
)
path = "/tmp/fuid_" + job_id + ext
if not os.path.exists(path):
raise HTTPException(status_code=404, detail="Archivo no encontrado")
db.delete(progress)
db.commit()
site2 = db.query(AlfrescoSite).filter(AlfrescoSite.id == progress.site_id).first()
ts = tznow().strftime("%Y%m%d_%H%M%S")
fname = "FUID_" + (site2.name if site2 else "") + "_" + ts + ext
return FileResponse(path, filename=fname, media_type=content_type)
def _build_fuid_rows(db, dependency_id):
"""Extrae y agrupa los datos FUID para una dependencia. Retorna (rows, dep_nombre, dep_codigo).
Optimizado: 1 consulta recursiva para carpetas + 1 consulta agrupada para archivos."""
dep = db.query(AlfrescoFolder).filter(AlfrescoFolder.id == dependency_id).first()
if not dep:
return [], "", ""
dep_codigo, dep_nombre = _extract_trd_code_and_name(dep.name)
if not dep_nombre:
dep_nombre = dep.name
from sqlalchemy import text
desc_tree = db.execute(text("""
WITH RECURSIVE tree AS (
SELECT id, parent_id, name, 0 AS depth
FROM alfresco_folders
WHERE id = :dep_id AND deleted = false
UNION ALL
SELECT f.id, f.parent_id, f.name, t.depth + 1
FROM alfresco_folders f
JOIN tree t ON f.parent_id = t.id
WHERE f.deleted = false AND t.depth < 20
)
SELECT id, parent_id, name, depth FROM tree
"""), {"dep_id": dependency_id}).fetchall()
folder_map = {}
all_folder_ids = []
for r in desc_tree:
fid = r[0]
folder_map[fid] = {"parent_id": r[1], "name": r[2], "depth": r[3]}
all_folder_ids.append(fid)
def _resolve_ancestors_to_dep(folder_id):
ancestors = []
cur = folder_map.get(folder_id, {}).get("parent_id")
visited = set()
while cur and cur != dependency_id:
if cur in visited or cur not in folder_map:
break
visited.add(cur)
ancestors.append(folder_map[cur])
cur = folder_map[cur].get("parent_id")
ancestors.reverse()
return ancestors
if not all_folder_ids:
return [], dep_nombre, dep_codigo
file_rows = db.execute(text("""
SELECT folder_id, MIN(alfresco_created_at), MAX(alfresco_created_at), SUM(COALESCE(folios, 0))
FROM alfresco_files
WHERE folder_id = ANY(:fids) AND deleted = false AND name != 'index.xml'
GROUP BY folder_id
"""), {"fids": all_folder_ids}).fetchall()
file_data = {}
for r in file_rows:
file_data[r[0]] = {
"fecha_inicial": r[1],
"fecha_final": r[2],
"folios": r[3] or 0,
}
rows = []
for folder_id, finfo in folder_map.items():
if folder_id == dependency_id:
serie_codigo = ""
serie_nombre = ""
subserie_codigo = ""
subserie_nombre = ""
else:
ancestors = _resolve_ancestors_to_dep(folder_id)
if len(ancestors) == 0:
serie_codigo, serie_nombre = _extract_trd_code_and_name(finfo["name"])
if not serie_nombre:
serie_nombre = finfo["name"]
subserie_codigo = ""
subserie_nombre = ""
else:
serie_codigo, serie_nombre = _extract_trd_code_and_name(ancestors[0]["name"])
if not serie_nombre:
serie_nombre = ancestors[0]["name"]
if len(ancestors) >= 2:
subserie_codigo, subserie_nombre = _extract_trd_code_and_name(ancestors[1]["name"])
if not subserie_nombre:
subserie_nombre = ancestors[1]["name"]
elif len(ancestors) == 1:
subserie_codigo, subserie_nombre = _extract_trd_code_and_name(finfo["name"])
if not subserie_nombre:
subserie_nombre = finfo["name"]
else:
subserie_codigo = ""
subserie_nombre = ""
fd = file_data.get(folder_id, {})
rows.append({
"dep_codigo": dep_codigo,
"dep_nombre": dep_nombre,
"serie_codigo": serie_codigo,
"serie_nombre": serie_nombre,
"subserie_codigo": subserie_codigo,
"subserie_nombre": subserie_nombre,
"fecha_inicial": fd.get("fecha_inicial"),
"fecha_final": fd.get("fecha_final"),
"folios": fd.get("folios", 0),
})
grouped = {}
for r in rows:
key = (r["serie_codigo"], r["serie_nombre"], r["subserie_codigo"], r["subserie_nombre"])
if key not in grouped:
grouped[key] = dict(r)
else:
g = grouped[key]
g["folios"] += r["folios"]
if r["fecha_inicial"] and (not g["fecha_inicial"] or r["fecha_inicial"] < g["fecha_inicial"]):
g["fecha_inicial"] = r["fecha_inicial"]
if r["fecha_final"] and (not g["fecha_final"] or r["fecha_final"] > g["fecha_final"]):
g["fecha_final"] = r["fecha_final"]
rows = list(grouped.values())
rows.sort(key=lambda r: (
r["serie_codigo"] or "ZZZZZ",
r["subserie_codigo"] or "ZZZZZ",
))
return rows, dep_nombre, dep_codigo
def _generate_fuid(db, site_id, dependency_id, format, job_id, db_url):
site = db.query(AlfrescoSite).filter(AlfrescoSite.id == site_id).first()
if not site:
return
rows, dep_nombre, dep_codigo = _build_fuid_rows(db, dependency_id)
if format == "excel":
_generate_fuid_excel(rows, job_id, site, dep_nombre, db_url)
else:
_generate_fuid_pdf(rows, job_id, site, dep_nombre, db_url)
def _generate_fuid_multi(db, site_id, dep_ids, format, job_id, db_url):
site = db.query(AlfrescoSite).filter(AlfrescoSite.id == site_id).first()
if not site:
return
all_data = []
for did in dep_ids:
rows, dep_nombre, dep_codigo = _build_fuid_rows(db, did)
all_data.append((dep_nombre, rows))
if format == "excel":
_generate_fuid_excel_multi(all_data, job_id, site, db_url)
else:
_generate_fuid_pdf_multi(all_data, job_id, site, db_url)
def _generate_fuid_excel(rows, job_id, site, dep_name, db_url):
wb = Workbook()
ws = wb.active
ws.title = "FUID"
header_font = Font(bold=True, color="FFFFFF", size=9, name="Arial")
header_fill = PatternFill(start_color="1F4E79", end_color="1F4E79", fill_type="solid")
header_align = Alignment(horizontal="center", vertical="center", wrap_text=True)
cell_align = Alignment(vertical="top", wrap_text=True)
thin_border = Border(
left=Side(style="thin"), right=Side(style="thin"),
top=Side(style="thin"), bottom=Side(style="thin")
)
bold_label = Font(bold=True, size=10, name="Arial")
normal_label = Font(size=10, name="Arial")
title_font = Font(bold=True, size=14, name="Arial", color="1F4E79")
center_align = Alignment(horizontal="center", vertical="center")
left_align = Alignment(horizontal="left", vertical="center")
right_align = Alignment(horizontal="right", vertical="center")
headers = [
"No.",
"CÓDIGO\nDEPENDENCIA",
"DEPENDENCIA",
"CÓDIGO\nSERIE",
"SERIE DOCUMENTAL",
"CÓDIGO\nSUBSERIE",
"SUBSERIE DOCUMENTAL",
"FECHA INICIAL",
"FECHA FINAL",
"NÚMERO\nDE FOLIOS",
"SOPORTE",
"NOTAS",
]
ncols = len(headers)
# ── Encabezado AGN ──
row = 1
left_cols = 8
right_cols = ncols - left_cols
# Bloque izquierdo: ENTIDAD REMITENTE, ENTIDAD PRODUCTORA, UNIDAD ADMINISTRATIVA, OFICINA PRODUCTORA, OBJETO
left_items = [
("ENTIDAD REMITENTE:", site.title or site.name),
("ENTIDAD PRODUCTORA:", site.title or site.name),
("UNIDAD ADMINISTRATIVA:", dep_name),
("OFICINA PRODUCTORA:", dep_name),
("OBJETO:", "Inventario documental - Transferencia documental"),
]
for label, value in left_items:
ws.merge_cells(start_row=row, start_column=1, end_row=row, end_column=left_cols)
c = ws.cell(row=row, column=1, value=f"{label} {value}")
c.font = Font(bold=True, size=9, name="Arial") if label != "OBJETO:" else bold_label
c.alignment = left_align
row += 1
# Bloque derecho: HOJA N° y REGISTRO DE ENTRADA
hoja_row = 1
ws.merge_cells(start_row=hoja_row, start_column=left_cols + 1, end_row=hoja_row, end_column=ncols)
c = ws.cell(row=hoja_row, column=left_cols + 1, value="HOJA N° _____ DE _____")
c.font = bold_label
c.alignment = right_align
hoja_row += 1
ws.merge_cells(start_row=hoja_row, start_column=left_cols + 1, end_row=hoja_row, end_column=ncols)
c = ws.cell(row=hoja_row, column=left_cols + 1, value="REGISTRO DE ENTRADA")
c.font = Font(bold=True, size=9, name="Arial")
c.alignment = center_align
hoja_row += 1
reg_headers = ["AÑO", "MES", "DÍA", "Nº T:"]
for ci, rh in enumerate(reg_headers):
c = ws.cell(row=hoja_row, column=left_cols + 1 + ci, value=rh)
c.font = Font(bold=True, size=8, name="Arial")
c.alignment = center_align
c.border = thin_border
hoja_row += 1
for ci in range(len(reg_headers)):
c = ws.cell(row=hoja_row, column=left_cols + 1 + ci, value="")
c.border = thin_border
c.alignment = center_align
# Título centrado
row = max(row, hoja_row + 1)
row += 1
ws.merge_cells(start_row=row, start_column=1, end_row=row, end_column=ncols)
c = ws.cell(row=row, column=1, value="FORMATO ÚNICO DE INVENTARIO DOCUMENTAL")
c.font = title_font
c.alignment = center_align
row += 1
row += 1
table_start = row
for c_idx, h in enumerate(headers, 1):
cell = ws.cell(row=row, column=c_idx, value=h)
cell.font = header_font
cell.fill = header_fill
cell.alignment = header_align
cell.border = thin_border
row += 1
data_start = row
for r_idx, r in enumerate(rows):
values = [
r_idx + 1,
r["dep_codigo"],
r["dep_nombre"],
r["serie_codigo"],
r["serie_nombre"],
r["subserie_codigo"],
r["subserie_nombre"],
r["fecha_inicial"].strftime("%d/%m/%Y") if r["fecha_inicial"] else "",
r["fecha_final"].strftime("%d/%m/%Y") if r["fecha_final"] else "",
r["folios"],
"Digital",
"",
]
for c_idx, v in enumerate(values, 1):
cell = ws.cell(row=row, column=c_idx, value=v)
cell.border = thin_border
cell.alignment = cell_align
cell.font = Font(size=9, name="Arial")
row += 1
data_end = row - 1
# ── Pie de página AGN: ELABORADO POR | ENTREGADO POR | RECIBIDO POR ──
row += 1
thirds = ncols // 3
cols1 = thirds
cols2 = thirds * 2
# Títulos
ws.merge_cells(start_row=row, start_column=1, end_row=row, end_column=cols1)
c = ws.cell(row=row, column=1, value="ELABORADO POR:")
c.font = bold_label
ws.merge_cells(start_row=row, start_column=cols1 + 1, end_row=row, end_column=cols2)
c = ws.cell(row=row, column=cols1 + 1, value="ENTREGADO POR:")
c.font = bold_label
ws.merge_cells(start_row=row, start_column=cols2 + 1, end_row=row, end_column=ncols)
c = ws.cell(row=row, column=cols2 + 1, value="RECIBIDO POR:")
c.font = bold_label
row += 1
for label in ["Cargo:", "Firma:", "Lugar y Fecha:"]:
for col_start, col_end in [(1, cols1), (cols1 + 1, cols2), (cols2 + 1, ncols)]:
ws.merge_cells(start_row=row, start_column=col_start, end_row=row, end_column=col_end)
c = ws.cell(row=row, column=col_start,
value=f"{label} ________________________________")
c.font = normal_label
row += 1
col_widths = [5, 16, 26, 16, 30, 16, 30, 14, 14, 12, 10, 26]
for c_idx, w in enumerate(col_widths, 1):
ws.column_dimensions[ws.cell(row=table_start, column=c_idx).column_letter].width = w
ws.page_setup.orientation = "landscape"
ws.page_setup.paperSize = ws.PAPERSIZE_LETTER
ws.page_margins.left = 0.3
ws.page_margins.right = 0.3
ws.auto_filter.ref = f"A{table_start}:L{data_end}"
tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".xlsx", dir="/tmp")
wb.save(tmp.name)
tmp.close()
final_path = "/tmp/fuid_" + job_id + ".xlsx"
os.rename(tmp.name, final_path)
if db_url:
db3 = _make_session(db_url)
try:
p = db3.query(SyncProgress).filter(SyncProgress.task_id == job_id).first()
if p:
p.status = "completed"
p.message = f"Completado: {len(rows)} registros"
db3.commit()
finally:
db3.close()
def _generate_fuid_excel_multi(all_data, job_id, site, db_url):
"""Genera un Excel con una hoja por dependencia."""
wb = Workbook()
wb.remove(wb.active)
header_font = Font(bold=True, color="FFFFFF", size=9, name="Arial")
header_fill = PatternFill(start_color="1F4E79", end_color="1F4E79", fill_type="solid")
header_align = Alignment(horizontal="center", vertical="center", wrap_text=True)
cell_align = Alignment(vertical="top", wrap_text=True)
thin_border = Border(
left=Side(style="thin"), right=Side(style="thin"),
top=Side(style="thin"), bottom=Side(style="thin")
)
bold_label = Font(bold=True, size=10, name="Arial")
normal_label = Font(size=10, name="Arial")
title_font = Font(bold=True, size=14, name="Arial", color="1F4E79")
center_align = Alignment(horizontal="center", vertical="center")
left_align = Alignment(horizontal="left", vertical="center")
right_align = Alignment(horizontal="right", vertical="center")
headers = [
"No.", "CÓDIGO\nDEPENDENCIA", "DEPENDENCIA", "CÓDIGO\nSERIE", "SERIE DOCUMENTAL",
"CÓDIGO\nSUBSERIE", "SUBSERIE DOCUMENTAL", "FECHA INICIAL", "FECHA FINAL",
"NÚMERO\nDE FOLIOS", "SOPORTE", "NOTAS",
]
ncols = len(headers)
left_cols = 8
total_rows = 0
for dep_name, rows in all_data:
safe_name = dep_name[:31].replace(":", "").replace("\\", "").replace("/", "").replace("?", "").replace("*", "").replace("[", "").replace("]", "")
ws = wb.create_sheet(title=safe_name)
# Encabezado AGN
entidad = site.title or site.name
row = 1
left_items = [
("ENTIDAD REMITENTE:", entidad),
("ENTIDAD PRODUCTORA:", entidad),
("UNIDAD ADMINISTRATIVA:", dep_name),
("OFICINA PRODUCTORA:", dep_name),
("OBJETO:", "Inventario documental - Transferencia documental"),
]
for label, value in left_items:
ws.merge_cells(start_row=row, start_column=1, end_row=row, end_column=left_cols)
ws.cell(row=row, column=1, value=f"{label} {value}").font = bold_label
ws.cell(row=row, column=1).alignment = left_align
row += 1
hoja_row = 1
ws.merge_cells(start_row=hoja_row, start_column=left_cols+1, end_row=hoja_row, end_column=ncols)
ws.cell(row=hoja_row, column=left_cols+1, value="HOJA N° _____ DE _____").font = bold_label
ws.cell(row=hoja_row, column=left_cols+1).alignment = right_align
hoja_row += 1
ws.merge_cells(start_row=hoja_row, start_column=left_cols+1, end_row=hoja_row, end_column=ncols)
ws.cell(row=hoja_row, column=left_cols+1, value="REGISTRO DE ENTRADA").font = Font(bold=True, size=9, name="Arial")
ws.cell(row=hoja_row, column=left_cols+1).alignment = center_align
hoja_row += 1
for ci, rh in enumerate(["AÑO", "MES", "DÍA", "Nº T:"]):
c = ws.cell(row=hoja_row, column=left_cols+1+ci, value=rh)
c.font = Font(bold=True, size=8, name="Arial")
c.alignment = center_align
c.border = thin_border
hoja_row += 1
for ci in range(4):
c = ws.cell(row=hoja_row, column=left_cols+1+ci, value="")
c.border = thin_border
c.alignment = center_align
row = max(row, hoja_row + 1) + 1
ws.merge_cells(start_row=row, start_column=1, end_row=row, end_column=ncols)
ws.cell(row=row, column=1, value="FORMATO ÚNICO DE INVENTARIO DOCUMENTAL").font = title_font
ws.cell(row=row, column=1).alignment = center_align
row += 2
table_start = row
for c_idx, h in enumerate(headers, 1):
cell = ws.cell(row=row, column=c_idx, value=h)
cell.font = header_font
cell.fill = header_fill
cell.alignment = header_align
cell.border = thin_border
row += 1
data_start = row
for r_idx, r in enumerate(rows):
values = [
r_idx + 1, r["dep_codigo"], r["dep_nombre"],
r["serie_codigo"], r["serie_nombre"],
r["subserie_codigo"], r["subserie_nombre"],
r["fecha_inicial"].strftime("%d/%m/%Y") if r["fecha_inicial"] else "",
r["fecha_final"].strftime("%d/%m/%Y") if r["fecha_final"] else "",
r["folios"], "Digital", "",
]
for c_idx, v in enumerate(values, 1):
cell = ws.cell(row=row, column=c_idx, value=v)
cell.border = thin_border
cell.alignment = cell_align
cell.font = Font(size=9, name="Arial")
row += 1
data_end = row - 1
row += 1
thirds = ncols // 3
cols1, cols2 = thirds, thirds * 2
ws.merge_cells(start_row=row, start_column=1, end_row=row, end_column=cols1)
ws.cell(row=row, column=1, value="ELABORADO POR:").font = bold_label
ws.merge_cells(start_row=row, start_column=cols1+1, end_row=row, end_column=cols2)
ws.cell(row=row, column=cols1+1, value="ENTREGADO POR:").font = bold_label
ws.merge_cells(start_row=row, start_column=cols2+1, end_row=row, end_column=ncols)
ws.cell(row=row, column=cols2+1, value="RECIBIDO POR:").font = bold_label
row += 1
for label in ["Cargo:", "Firma:", "Lugar y Fecha:"]:
for col_start, col_end in [(1, cols1), (cols1+1, cols2), (cols2+1, ncols)]:
ws.merge_cells(start_row=row, start_column=col_start, end_row=row, end_column=col_end)
ws.cell(row=row, column=col_start, value=f"{label} ________________________________").font = normal_label
row += 1
col_widths = [5, 16, 26, 16, 30, 16, 30, 14, 14, 12, 10, 26]
for c_idx, w in enumerate(col_widths, 1):
ws.column_dimensions[ws.cell(row=table_start, column=c_idx).column_letter].width = w
ws.page_setup.orientation = "landscape"
ws.page_setup.paperSize = ws.PAPERSIZE_LETTER
ws.page_margins.left = 0.3
ws.page_margins.right = 0.3
ws.auto_filter.ref = f"A{table_start}:L{data_end}"
total_rows += len(rows)
tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".xlsx", dir="/tmp")
wb.save(tmp.name)
tmp.close()
final_path = "/tmp/fuid_" + job_id + ".xlsx"
os.rename(tmp.name, final_path)
if db_url:
db3 = _make_session(db_url)
try:
p = db3.query(SyncProgress).filter(SyncProgress.task_id == job_id).first()
if p:
p.status = "completed"
p.message = f"Completado: {total_rows} registros en {len(all_data)} dependencias"
db3.commit()
finally:
db3.close()
def _generate_fuid_pdf_multi(all_data, job_id, site, db_url):
"""Genera un PDF con una página por dependencia."""
from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer, Table, TableStyle, PageBreak
from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle
from reportlab.lib.enums import TA_CENTER
styles = getSampleStyleSheet()
title_style = ParagraphStyle("FuidTitle", parent=styles["Title"],
fontSize=11, spaceAfter=6, alignment=TA_CENTER,
textColor=colors.HexColor("#1F4E79"))
normal_style = ParagraphStyle("FuidNormal", parent=styles["Normal"],
fontSize=7.5, leading=10)
label_style = ParagraphStyle("FuidLabel", parent=styles["Normal"],
fontSize=7.5, leading=10, fontName="Helvetica-Bold")
header_style = ParagraphStyle("FuidHeader", parent=styles["Normal"],
fontSize=6.5, leading=8, textColor=colors.white,
alignment=TA_CENTER, fontName="Helvetica-Bold")
cell_style_multi = ParagraphStyle("FuidCellM", parent=styles["Normal"],
fontSize=6.5, leading=8)
pdf_path = "/tmp/fuid_" + job_id + ".pdf"
doc = SimpleDocTemplate(pdf_path, pagesize=landscape(A4),
leftMargin=0.3*cm, rightMargin=0.3*cm,
topMargin=0.5*cm, bottomMargin=0.5*cm)
story = []
total_rows = 0
entidad = site.title or site.name
for idx, (dep_name, rows) in enumerate(all_data):
if idx > 0:
story.append(PageBreak())
# Encabezado
left_block = [
f"ENTIDAD REMITENTE: {entidad}",
f"ENTIDAD PRODUCTORA: {entidad}",
f"UNIDAD ADMINISTRATIVA: {dep_name}",
f"OFICINA PRODUCTORA: {dep_name}",
f"OBJETO: Inventario documental - Transferencia documental",
]
right_block = (
"HOJA N° _____ DE _____
"
"REGISTRO DE ENTRADA
"
''
'| AÑO | '
'MES | '
'DÍA | '
'Nº T: |
'
' | | | |
'
)
header_tbl = Table([
[Paragraph("
".join(left_block), normal_style), Paragraph(right_block, normal_style)]
], colWidths=[18*cm, 10*cm])
header_tbl.setStyle(TableStyle([
("VALIGN", (0, 0), (-1, -1), "TOP"),
("TOPPADDING", (0, 0), (-1, -1), 2),
("BOTTOMPADDING", (0, 0), (-1, -1), 2),
("LINEBELOW", (0, 0), (-1, -1), 0.5, colors.HexColor("#1F4E79")),
]))
story.append(header_tbl)
story.append(Spacer(1, 0.5*cm))
story.append(Paragraph("FORMATO ÚNICO DE INVENTARIO DOCUMENTAL", title_style))
story.append(Spacer(1, 0.2*cm))
# Tabla
simplified_headers = [
"No.", "CÓD. DEP.", "DEPENDENCIA", "CÓD. SERIE", "SERIE DOCUMENTAL",
"CÓD. SUBS.", "SUBSERIE DOC.", "F. INICIAL", "F. FINAL",
"FOLIOS", "SOP.", "NOTAS",
]
header_row = [Paragraph(h, header_style) for h in simplified_headers]
table_data = [header_row]
for r_idx, r in enumerate(rows, 1):
table_data.append([
Paragraph(str(r_idx), cell_style_multi),
Paragraph(r["dep_codigo"], cell_style_multi),
Paragraph(r["dep_nombre"][:26], cell_style_multi),
Paragraph(r["serie_codigo"], cell_style_multi),
Paragraph(r["serie_nombre"][:24], cell_style_multi),
Paragraph(r["subserie_codigo"], cell_style_multi),
Paragraph(r["subserie_nombre"][:24], cell_style_multi),
Paragraph(r["fecha_inicial"].strftime("%d/%m/%Y") if r["fecha_inicial"] else "", cell_style_multi),
Paragraph(r["fecha_final"].strftime("%d/%m/%Y") if r["fecha_final"] else "", cell_style_multi),
Paragraph(str(r["folios"]), cell_style_multi),
Paragraph("Digital", cell_style_multi),
Paragraph("", cell_style_multi),
])
col_widths_pdf = [0.7*cm, 2.2*cm, 3.2*cm, 2.2*cm, 3.8*cm,
2.2*cm, 3.8*cm, 2.2*cm, 2.2*cm, 1.2*cm, 1*cm, 3.2*cm]
tbl = Table(table_data, colWidths=col_widths_pdf, repeatRows=1)
tbl.setStyle(TableStyle([
("BACKGROUND", (0, 0), (-1, 0), colors.HexColor("#1F4E79")),
("TEXTCOLOR", (0, 0), (-1, 0), colors.white),
("FONTNAME", (0, 0), (-1, -1), "Helvetica"),
("FONTSIZE", (0, 1), (-1, -1), 6.5),
("ALIGN", (0, 0), (-1, 0), "CENTER"),
("ALIGN", (0, 1), (0, -1), "CENTER"),
("ALIGN", (7, 1), (9, -1), "CENTER"),
("GRID", (0, 0), (-1, -1), 0.4, colors.HexColor("#B0B8C1")),
("ROWBACKGROUNDS", (0, 1), (-1, -1), [colors.white, colors.HexColor("#F5F7FA")]),
("VALIGN", (0, 0), (-1, -1), "MIDDLE"),
("TOPPADDING", (0, 0), (-1, -1), 2),
("BOTTOMPADDING", (0, 0), (-1, -1), 2),
("LEFTPADDING", (0, 0), (-1, -1), 2),
("RIGHTPADDING", (0, 0), (-1, -1), 2),
]))
story.append(tbl)
story.append(Spacer(1, 0.4*cm))
# Pie (KeepTogether para no recortar al final de página)
third = 9
footer_rows = []
# Línea de títulos
footer_rows.append(Table([[
Paragraph("ELABORADO POR:", normal_style),
Paragraph("ENTREGADO POR:", normal_style),
Paragraph("RECIBIDO POR:", normal_style),
]], colWidths=[third*cm, third*cm, third*cm]))
# Cargo, Firma, Lugar y Fecha
for label in ["Cargo:", "Firma:", "Lugar y Fecha:"]:
footer_rows.append(Table([[
Paragraph(f"{label} ______________________________", normal_style),
Paragraph(f"{label} ______________________________", normal_style),
Paragraph(f"{label} ______________________________", normal_style),
]], colWidths=[third*cm, third*cm, third*cm]))
for fr in footer_rows:
fr.setStyle(TableStyle([
("VALIGN", (0, 0), (-1, -1), "MIDDLE"),
("TOPPADDING", (0, 0), (-1, -1), 2),
("BOTTOMPADDING", (0, 0), (-1, -1), 2),
("LEFTPADDING", (0, 0), (0, 0), 0),
("LINEBEFORE", (1, 0), (1, 0), 0.5, colors.HexColor("#1F4E79")),
("LINEBEFORE", (2, 0), (2, 0), 0.5, colors.HexColor("#1F4E79")),
]))
from reportlab.platypus import KeepTogether
story.append(KeepTogether(footer_rows))
total_rows += len(rows)
try:
doc.build(story)
except Exception:
pass
if db_url:
db3 = _make_session(db_url)
try:
p = db3.query(SyncProgress).filter(SyncProgress.task_id == job_id).first()
if p:
p.status = "completed"
p.message = f"Completado: {total_rows} registros en {len(all_data)} dependencias"
db3.commit()
finally:
db3.close()
def _generate_fuid_pdf(rows, job_id, site, dep_name, db_url):
from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer, Table, TableStyle
from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle
from reportlab.lib.enums import TA_CENTER, TA_LEFT, TA_RIGHT
styles = getSampleStyleSheet()
title_style = ParagraphStyle("FuidTitle", parent=styles["Title"],
fontSize=11, spaceAfter=6, alignment=TA_CENTER,
textColor=colors.HexColor("#1F4E79"))
label_style = ParagraphStyle("FuidLabel", parent=styles["Normal"],
fontSize=7.5, leading=10, fontName="Helvetica-Bold")
normal_style = ParagraphStyle("FuidNormal", parent=styles["Normal"],
fontSize=7.5, leading=10)
header_style = ParagraphStyle("FuidHeader", parent=styles["Normal"],
fontSize=6.5, leading=8, textColor=colors.white,
alignment=TA_CENTER, fontName="Helvetica-Bold")
cell_style_pdf = ParagraphStyle("FuidCellS", parent=styles["Normal"],
fontSize=6.5, leading=8)
pdf_path = "/tmp/fuid_" + job_id + ".pdf"
doc = SimpleDocTemplate(pdf_path, pagesize=landscape(A4),
leftMargin=0.3*cm, rightMargin=0.3*cm,
topMargin=0.5*cm, bottomMargin=0.5*cm)
story = []
# ── Encabezado AGN: izquierda + derecha ──
left_block = [
f"ENTIDAD REMITENTE: {entidad}",
f"ENTIDAD PRODUCTORA: {entidad}",
f"UNIDAD ADMINISTRATIVA: {dep_name}",
f"OFICINA PRODUCTORA: {dep_name}",
f"OBJETO: Inventario documental - Transferencia documental",
]
left_text = "
".join(left_block)
right_top = "HOJA N° _____ DE _____"
right_mid = "REGISTRO DE ENTRADA"
right_bottom = (
''
'| AÑO | '
'MES | '
'DÍA | '
'Nº T: |
'
' | | | |
'
'
'
)
right_text = f"{right_top}
{right_mid}
{right_bottom}"
header_table_data = [
[
Paragraph(left_text, normal_style),
Paragraph(right_text, normal_style),
],
]
header_tbl = Table(header_table_data, colWidths=[18*cm, 10*cm])
header_tbl.setStyle(TableStyle([
("VALIGN", (0, 0), (-1, -1), "TOP"),
("TOPPADDING", (0, 0), (-1, -1), 2),
("BOTTOMPADDING", (0, 0), (-1, -1), 2),
("LINEBELOW", (0, 0), (-1, -1), 0.5, colors.HexColor("#1F4E79")),
]))
story.append(header_tbl)
story.append(Spacer(1, 0.5*cm))
story.append(Paragraph("FORMATO ÚNICO DE INVENTARIO DOCUMENTAL", title_style))
story.append(Spacer(1, 0.2*cm))
# ── Tabla de datos ──
simplified_headers = [
"No.", "CÓD. DEP.", "DEPENDENCIA", "CÓD. SERIE", "SERIE DOCUMENTAL",
"CÓD. SUBS.", "SUBSERIE DOC.", "F. INICIAL", "F. FINAL",
"FOLIOS", "SOP.", "NOTAS",
]
header_row = [Paragraph(h, header_style) for h in simplified_headers]
table_data = [header_row]
row_count = 0
for r in rows:
row_count += 1
table_data.append([
Paragraph(str(row_count), cell_style_pdf),
Paragraph(r["dep_codigo"], cell_style_pdf),
Paragraph(r["dep_nombre"][:26], cell_style_pdf),
Paragraph(r["serie_codigo"], cell_style_pdf),
Paragraph(r["serie_nombre"][:24], cell_style_pdf),
Paragraph(r["subserie_codigo"], cell_style_pdf),
Paragraph(r["subserie_nombre"][:24], cell_style_pdf),
Paragraph(r["fecha_inicial"].strftime("%d/%m/%Y") if r["fecha_inicial"] else "", cell_style_pdf),
Paragraph(r["fecha_final"].strftime("%d/%m/%Y") if r["fecha_final"] else "", cell_style_pdf),
Paragraph(str(r["folios"]), cell_style_pdf),
Paragraph("Digital", cell_style_pdf),
Paragraph("", cell_style_pdf),
])
col_widths_pdf = [
0.7*cm, 2.2*cm, 3.2*cm, 2.2*cm, 3.8*cm,
2.2*cm, 3.8*cm, 2.2*cm, 2.2*cm,
1.2*cm, 1*cm, 3.2*cm,
]
tbl = Table(table_data, colWidths=col_widths_pdf, repeatRows=1)
tbl.setStyle(TableStyle([
("BACKGROUND", (0, 0), (-1, 0), colors.HexColor("#1F4E79")),
("TEXTCOLOR", (0, 0), (-1, 0), colors.white),
("FONTNAME", (0, 0), (-1, -1), "Helvetica"),
("FONTSIZE", (0, 1), (-1, -1), 6.5),
("ALIGN", (0, 0), (-1, 0), "CENTER"),
("ALIGN", (0, 1), (0, -1), "CENTER"),
("ALIGN", (7, 1), (9, -1), "CENTER"),
("GRID", (0, 0), (-1, -1), 0.4, colors.HexColor("#B0B8C1")),
("ROWBACKGROUNDS", (0, 1), (-1, -1), [colors.white, colors.HexColor("#F5F7FA")]),
("VALIGN", (0, 0), (-1, -1), "MIDDLE"),
("TOPPADDING", (0, 0), (-1, -1), 2),
("BOTTOMPADDING", (0, 0), (-1, -1), 2),
("LEFTPADDING", (0, 0), (-1, -1), 2),
("RIGHTPADDING", (0, 0), (-1, -1), 2),
]))
story.append(tbl)
story.append(Spacer(1, 0.4*cm))
# ── Pie de página: ELABORADO POR | ENTREGADO POR | RECIBIDO POR ──
third = 5
footer_rows = []
footer_rows.append(Table([[
Paragraph("ELABORADO POR:", normal_style),
Paragraph("ENTREGADO POR:", normal_style),
Paragraph("RECIBIDO POR:", normal_style),
]], colWidths=[third*cm, third*cm, third*cm]))
for label in ["Cargo:", "Firma:", "Lugar y Fecha:"]:
footer_rows.append(Table([[
Paragraph(f"{label} ______________________________", normal_style),
Paragraph(f"{label} ______________________________", normal_style),
Paragraph(f"{label} ______________________________", normal_style),
]], colWidths=[third*cm, third*cm, third*cm]))
for fr in footer_rows:
fr.setStyle(TableStyle([
("VALIGN", (0, 0), (-1, -1), "MIDDLE"),
("TOPPADDING", (0, 0), (-1, -1), 2),
("BOTTOMPADDING", (0, 0), (-1, -1), 2),
("LEFTPADDING", (0, 0), (0, 0), 0),
("LINEBEFORE", (1, 0), (1, 0), 0.5, colors.HexColor("#1F4E79")),
("LINEBEFORE", (2, 0), (2, 0), 0.5, colors.HexColor("#1F4E79")),
]))
from reportlab.platypus import KeepTogether
story.append(KeepTogether(footer_rows))
try:
doc.build(story)
except Exception:
pass
if db_url:
db3 = _make_session(db_url)
try:
p = db3.query(SyncProgress).filter(SyncProgress.task_id == job_id).first()
if p:
p.status = "completed"
p.message = f"Completado: {len(rows)} registros"
db3.commit()
finally:
db3.close()