from __future__ import annotations import csv import json import zipfile from dataclasses import dataclass from decimal import Decimal from io import BytesIO, StringIO from django.core.exceptions import ObjectDoesNotExist from django.template.loader import get_template from django.urls import reverse from xhtml2pdf import pisa from apps.boletas.models import Boleta from apps.boletas.pdf import build_boleta_pdf_bytes from apps.expediente.models import ExpedienteLog from apps.muestreo_laboratorio.models import ResultadoLaboratorio, TomaMuestra from apps.recepciones.models import PesoRecepcion, Recepcion def _build_abs_url(request, url: str | None) -> str | None: if not url: return None if request: return request.build_absolute_uri(url) return url def _username_from_user(u) -> str | None: if not u: return None username = getattr(u, "username", "") if username: return username if hasattr(u, "get_full_name"): full_name = (u.get_full_name() or "").strip() if full_name: return full_name return str(u) def _peso(recepcion: Recepcion, tipo: str): p = next((x for x in recepcion.pesos.all() if x.tipo == tipo), None) return p.valor if p else None def compute_pesos(recepcion: Recepcion | None) -> dict: if not recepcion: return { "lleno": None, "vacio": None, "bruto": None, "deducciones": [], "bonificaciones": [], "neto": None, } lleno = _peso(recepcion, PesoRecepcion.TIPO_LLENO) vacio = _peso(recepcion, PesoRecepcion.TIPO_VACIO) bruto = None if lleno is not None and vacio is not None: bruto = Decimal(lleno) - Decimal(vacio) deducciones = [] for d in recepcion.deducciones.select_related("parametro").all(): deducciones.append( { "parametro": getattr(getattr(d, "parametro", None), "nombre", None) or str(d.parametro_id), "valor_deducido": d.valor_deducido, "tipo": d.tipo_deduccion, } ) bonificaciones = [] for b in recepcion.bonificaciones.select_related("parametro").all(): bonificaciones.append( { "parametro": getattr(getattr(b, "parametro", None), "nombre", None) or str(b.parametro_id), "monto_total": b.monto_total, "monto_por_tonelada": b.monto_por_tonelada, } ) neto = bruto if bruto is not None: total = Decimal("0") for d in recepcion.deducciones.all(): if d.tipo_deduccion == "fijo": total += Decimal(d.valor_deducido) elif d.tipo_deduccion == "porcentaje": total += (bruto * Decimal(d.valor_deducido)) / Decimal("100") neto = bruto - total return { "lleno": lleno, "vacio": vacio, "bruto": bruto, "deducciones": deducciones, "bonificaciones": bonificaciones, "neto": neto, } def build_expediente_payload(*, boleta: Boleta, request=None) -> dict: try: recepcion = boleta.recepcion except ObjectDoesNotExist: recepcion = None pesos = compute_pesos(recepcion) fotos_recepcion = [] muestreo_dict = None toma_dict = None resultado_dict = None if recepcion is not None: for f in recepcion.fotos.all().order_by("id"): url = reverse("recepcion-foto-download", kwargs={"pk": f.pk}) fotos_recepcion.append( { "tipo": f.tipo, "url": _build_abs_url(request, url), "fecha": getattr(f, "created_at", None), } ) muestreo = getattr(recepcion, "muestreo", None) if recepcion is not None else None resultado_model = None if muestreo is not None: tecnico = getattr(muestreo, "tecnico_asignado", None) muestreo_dict = { "tipo": muestreo.tipo_muestreo, "tecnico": (tecnico.get_full_name() or tecnico.username) if tecnico else None, "fecha_toma": None, "fotos": [], } toma = getattr(muestreo, "toma_muestra", None) if isinstance(toma, TomaMuestra): muestreo_dict["fecha_toma"] = toma.fecha_toma for f in toma.fotos.all().order_by("id"): url = reverse("muestreo-laboratorio-foto-muestra-download", kwargs={"pk": f.pk}) muestreo_dict["fotos"].append( { "tipo": f.tipo, "url": _build_abs_url(request, url), "fecha": getattr(f, "created_at", None), } ) resultado = getattr(muestreo, "resultado_laboratorio", None) if isinstance(resultado, ResultadoLaboratorio): resultado_model = resultado params = [] for p in resultado.parametros.select_related("parametro").all().order_by("id"): parametro = getattr(p, "parametro", None) params.append( { "nombre": getattr(parametro, "nombre", None) or str(p.parametro_id), "valor": p.valor, "unidad": p.unidad, "fuera_de_rango": p.fuera_de_rango, } ) fotos_res = [] for f in resultado.fotos.all().order_by("id"): url = reverse("muestreo-laboratorio-foto-resultado-download", kwargs={"pk": f.pk}) fotos_res.append( { "tipo": f.tipo, "url": _build_abs_url(request, url), "fecha": getattr(f, "created_at", None), } ) informe_url = None if resultado.informe: url = reverse("muestreo-laboratorio-informe-download", kwargs={"pk": resultado.pk}) informe_url = _build_abs_url(request, url) resultado_dict = { "fecha_analisis": resultado.fecha_publicacion or resultado.created_at, "estado": resultado.estado, "parametros": params, "informe_externo": informe_url, "fotos": fotos_res, } pdf_boleta_url = _build_abs_url( request, reverse("boleta-pdf", kwargs={"pk": boleta.pk}), ) documentos = { "pdf_boleta": pdf_boleta_url, "firmas": { "receptor": (recepcion.firmas or {}).get("firma_receptor") if recepcion else None, "chofer": (recepcion.firmas or {}).get("firma_chofer") if recepcion else None, }, "notas": (recepcion.observaciones if recepcion else "") or "", } auditoria = [] auditoria.append( { "usuario": _username_from_user(getattr(boleta, "created_by", None)), "accion": "boleta_creada", "fecha": getattr(boleta, "created_at", None), "detalle": {}, } ) if recepcion is not None: auditoria.append( { "usuario": _username_from_user(getattr(recepcion, "created_by", None)), "accion": "recepcion_creada", "fecha": getattr(recepcion, "created_at", None), "detalle": {"estado": recepcion.estado}, } ) if recepcion.fecha_cierre: auditoria.append( { "usuario": _username_from_user(getattr(recepcion, "updated_by", None)), "accion": "recepcion_cerrada", "fecha": recepcion.fecha_cierre, "detalle": {"estado": recepcion.estado}, } ) if resultado_dict is not None: if resultado_dict.get("estado") == "publicado" and resultado_dict.get("fecha_analisis"): auditoria.append( { "usuario": _username_from_user(getattr(resultado_model, "publicado_por", None)) if resultado_model is not None else None, "accion": "laboratorio_publicado", "fecha": resultado_dict.get("fecha_analisis"), "detalle": {}, } ) logs = ( ExpedienteLog.objects.select_related("usuario") .filter(boleta=boleta) .order_by("fecha", "id") ) for l in logs: u = getattr(l, "usuario", None) username = _username_from_user(u) auditoria.append( { "usuario": username, "accion": l.accion, "fecha": l.fecha, "detalle": l.detalle, } ) if recepcion is None: mensaje = "La boleta aún no tiene recepción registrada." else: mensaje = "" variedad = getattr(boleta, "variedad", None) bodega = getattr(recepcion, "bodega_destino", None) if recepcion is not None else None boleta_dict = { "id": boleta.id, "folio": boleta.folio, "qr": boleta.qr_texto, "estado": boleta.estado, "bodega_destino": { "id": getattr(bodega, "id", None), "codigo": getattr(bodega, "codigo", None), "nombre": getattr(bodega, "nombre", None), } if bodega is not None else None, "productor": { "id": boleta.productor_id, "nombre": getattr(getattr(boleta, "productor", None), "nombre", None), }, "contrato": { "numero": getattr(getattr(boleta, "contrato", None), "numero_contrato", None), "variedad": getattr(variedad, "nombre", None), }, "zafra": {"nombre": getattr(getattr(boleta, "zafra", None), "nombre", None)}, "fecha_ingreso": boleta.fecha_emision, } camion = { "placas": boleta.placas_camion, "chofer": boleta.chofer_nombre, "telefono": boleta.chofer_telefono, "fotos": fotos_recepcion, } recepcion_out = None if recepcion is not None: recepcion_out = { "id": recepcion.id, "estado": recepcion.estado, "rechazo_calidad": getattr(recepcion, "rechazo_calidad", False), "rechazos_calidad": getattr(recepcion, "rechazos_calidad", []) or [], "fecha_cierre": recepcion.fecha_cierre, "firmas": recepcion.firmas or {}, } out = { "boleta": boleta_dict, "camion": camion, "recepcion": recepcion_out, "pesos": pesos, "muestreo": muestreo_dict, "resultados_laboratorio": resultado_dict, "documentos": documentos, "auditoria": auditoria, } if mensaje: out["mensaje"] = mensaje return out def build_expediente_pdf_bytes(*, boleta: Boleta, request=None) -> bytes: payload = build_expediente_payload(boleta=boleta, request=request) template = get_template("expediente/expediente_pdf.html") html = template.render(payload) out = BytesIO() result = pisa.CreatePDF(src=html, dest=out, encoding="utf-8") if result.err: raise ValueError("No se pudo generar el PDF del expediente.") return out.getvalue() def build_expediente_zip_bytes(*, boleta: Boleta, request=None) -> bytes: payload = build_expediente_payload(boleta=boleta, request=request) try: recepcion = boleta.recepcion except ObjectDoesNotExist: recepcion = None pdf_boleta_bytes = None if boleta.pdf_generado: try: with boleta.pdf_generado.open("rb") as f: pdf_boleta_bytes = f.read() except FileNotFoundError: pdf_boleta_bytes = None if pdf_boleta_bytes is None: pdf_boleta_bytes = build_boleta_pdf_bytes(boleta) zip_buf = BytesIO() with zipfile.ZipFile(zip_buf, mode="w", compression=zipfile.ZIP_DEFLATED) as zf: zf.writestr(f"boleta_{boleta.folio}.pdf", pdf_boleta_bytes) expediente_pdf = build_expediente_pdf_bytes(boleta=boleta, request=request) zf.writestr(f"expediente_{boleta.folio}.pdf", expediente_pdf) zf.writestr("expediente.json", json.dumps(payload, default=str, ensure_ascii=False, indent=2)) zf.writestr("data/pesos.json", json.dumps(payload.get("pesos"), default=str, ensure_ascii=False, indent=2)) zf.writestr( "data/resultados_laboratorio.json", json.dumps(payload.get("resultados_laboratorio"), default=str, ensure_ascii=False, indent=2), ) pesos_csv = StringIO() writer = csv.writer(pesos_csv) writer.writerow(["campo", "valor"]) pesos = payload.get("pesos") or {} writer.writerow(["lleno", pesos.get("lleno")]) writer.writerow(["vacio", pesos.get("vacio")]) writer.writerow(["bruto", pesos.get("bruto")]) writer.writerow(["neto", pesos.get("neto")]) zf.writestr("data/pesos.csv", pesos_csv.getvalue()) ded_csv = StringIO() writer = csv.writer(ded_csv) writer.writerow(["parametro", "valor_deducido", "tipo"]) for d in (pesos.get("deducciones") or []): writer.writerow([d.get("parametro"), d.get("valor_deducido"), d.get("tipo")]) zf.writestr("data/deducciones.csv", ded_csv.getvalue()) lab = payload.get("resultados_laboratorio") or {} params_csv = StringIO() writer = csv.writer(params_csv) writer.writerow(["parametro", "valor", "unidad", "fuera_de_rango"]) for p in (lab.get("parametros") or []): writer.writerow([p.get("nombre"), p.get("valor"), p.get("unidad"), p.get("fuera_de_rango")]) zf.writestr("data/lab_parametros.csv", params_csv.getvalue()) if recepcion is not None: for f in recepcion.fotos.all().order_by("id"): if not f.imagen: continue try: with f.imagen.open("rb") as fh: name = (getattr(f.imagen, "name", "") or "").split("/")[-1] or f"{f.tipo}_{f.pk}" zf.writestr(f"fotos/recepcion/{name}", fh.read()) except FileNotFoundError: continue muestreo = getattr(recepcion, "muestreo", None) toma = getattr(muestreo, "toma_muestra", None) if muestreo is not None else None if isinstance(toma, TomaMuestra): for f in toma.fotos.all().order_by("id"): if not f.imagen: continue try: with f.imagen.open("rb") as fh: name = (getattr(f.imagen, "name", "") or "").split("/")[-1] or f"{f.tipo}_{f.pk}" zf.writestr(f"fotos/muestreo/{name}", fh.read()) except FileNotFoundError: continue resultado = getattr(muestreo, "resultado_laboratorio", None) if muestreo is not None else None if isinstance(resultado, ResultadoLaboratorio): for f in resultado.fotos.all().order_by("id"): if not f.imagen: continue try: with f.imagen.open("rb") as fh: name = (getattr(f.imagen, "name", "") or "").split("/")[-1] or f"{f.tipo}_{f.pk}" zf.writestr(f"fotos/laboratorio/{name}", fh.read()) except FileNotFoundError: continue if resultado.informe: try: with resultado.informe.open("rb") as fh: name = (getattr(resultado.informe, "name", "") or "").split("/")[-1] or f"informe_{resultado.pk}.pdf" zf.writestr(f"laboratorio/{name}", fh.read()) except FileNotFoundError: pass return zip_buf.getvalue() @dataclass class ErpSendResult: success: bool mensaje: str def simular_envio_erp(*, boleta: Boleta) -> ErpSendResult: return ErpSendResult(success=True, mensaje="Envío a ERP registrado (simulado)")