| """Streamlit tab: PDF/image quotation -> editable structured JSON/CSV with validation.""" |
| from __future__ import annotations |
|
|
| import hashlib |
| import io |
| import json |
| from pathlib import Path |
|
|
| import pandas as pd |
| import streamlit as st |
|
|
| from quotation_pipeline import extract_pages, extract_quote, validate_quote, build_payload |
|
|
| HERE = Path(__file__).resolve().parent |
| DEMO = HERE / 'demos' |
| FIELD_NAMES = [ |
| ('folio', 'Folio'), ('fecha', 'Fecha del documento'), ('cliente', 'Cliente / razón social'), |
| ('rfc', 'RFC del cliente'), ('correo', 'Correo del cliente'), ('proyecto', 'Proyecto / obra'), |
| ('moneda', 'Moneda'), ('subtotal', 'Subtotal declarado'), ('iva', 'IVA declarado'), ('total', 'Total declarado'), |
| ] |
| EDITABLE = ['codigo', 'descripcion', 'cantidad', 'unidad', 'precio_unitario', 'importe_documento'] |
|
|
|
|
| def render_quotation_ui(): |
| st.subheader('Cotizaciones estructuradas') |
| st.caption('Demostración de ingestión y control: PDF o fotografía → texto/OCR → campos → validación → revisión → JSON/CSV.') |
| st.info( |
| 'La extracción de PDF con texto es determinista; las imágenes y PDFs escaneados pasan por RapidOCR neuronal. ' |
| 'Los conceptos se interpretan con un formato de tabla explícito; otros diseños de proveedor pueden requerir ' |
| 'corregir datos o completar partidas. Nunca se inventan precios faltantes.' |
| ) |
|
|
| st.markdown('#### 1. Prueba con dos cotizaciones de ejemplo') |
| a, b = st.columns(2) |
| with a: |
| st.download_button('⬇️ Cotización correcta (PDF)', |
| data=(DEMO / 'cotizacion_demo_correcta.pdf').read_bytes(), |
| file_name='cotizacion_demo_correcta.pdf', mime='application/pdf', key='q_demo_ok') |
| st.caption('3 partidas · Subtotal 9,800 MXN · IVA 1,568 · Total 11,368. Valores ficticios.') |
| with b: |
| st.download_button('⬇️ Cotización con error (PDF)', |
| data=(DEMO / 'cotizacion_demo_error.pdf').read_bytes(), |
| file_name='cotizacion_demo_error.pdf', mime='application/pdf', key='q_demo_bad') |
| st.caption('La primera partida declara 5,990 en lugar de 6,000. Debe generar alertas.') |
|
|
| uploaded = st.file_uploader('Sube una cotización PDF, JPG o PNG (máximo 20 MB)', |
| type=['pdf', 'jpg', 'jpeg', 'png'], key='q_upload') |
| if uploaded is None: |
| st.markdown('**Para demostrarlo:** descarga el PDF con error, súbelo aquí y pulsa **Extraer datos**. ' |
| 'Después corrige las partidas y enseña cómo se desbloquea la exportación validada.') |
| return |
| data = uploaded.getvalue() |
| identity = hashlib.sha256(uploaded.name.encode() + data).hexdigest() |
| if len(data) > 20 * 1024 * 1024: |
| st.error('El archivo supera el máximo de 20 MB.') |
| return |
|
|
| if st.button('🔎 Extraer datos y detectar conceptos', type='primary', key='q_parse'): |
| with st.spinner('Leyendo PDF, identificando campos y comprobando importes...'): |
| try: |
| extracted = extract_pages(data, uploaded.name) |
| parsed = extract_quote(extracted['texto']) |
| initial_check = validate_quote(parsed['campos'], parsed['conceptos']) |
| st.session_state['q_identity'] = identity |
| st.session_state['q_extracted'] = extracted |
| st.session_state['q_original'] = parsed |
| st.session_state['q_initial_alerts'] = initial_check['alertas'] |
| st.session_state['q_approval_sig'] = None |
| st.session_state['q_ack_source'] = False |
| st.session_state['q_reviewer_note'] = '' |
| st.session_state['q_confirm_content'] = False |
| for key, _ in FIELD_NAMES: |
| st.session_state[f'q_field_{key}'] = parsed['campos'].get(key, '') |
| st.session_state['q_editor_items'] = pd.DataFrame(parsed['conceptos'], columns=EDITABLE) |
| st.success(f"Extracción terminada · {extracted['paginas']} página(s) · " |
| f"{len(parsed['conceptos'])} partida(s) detectada(s).") |
| except Exception as exc: |
| st.error(f'No se pudo extraer el documento: {exc}') |
| return |
|
|
| if st.session_state.get('q_identity') != identity: |
| st.warning('Pulsa **Extraer datos** para cargar este nuevo archivo.') |
| return |
|
|
| parsed = st.session_state['q_original'] |
| extracted = st.session_state['q_extracted'] |
| raw_alerts = st.session_state.get('q_initial_alerts', []) |
| motors = ', '.join(dict.fromkeys(extracted['motores'])) |
| st.caption(f'Archivo: **{uploaded.name}** · Extracción: **{motors}** · {extracted["paginas"]} página(s)') |
|
|
| st.markdown('#### 2. Verificar la información del cliente y los importes') |
| st.caption('Puedes corregir cualquier campo. Los valores se revisarán de nuevo después de cada edición.') |
| cols = st.columns(3) |
| for index, (key, label) in enumerate(FIELD_NAMES): |
| with cols[index % 3]: |
| st.text_input(label, key=f'q_field_{key}') |
| fields = {key: st.session_state.get(f'q_field_{key}', '') for key, _ in FIELD_NAMES} |
|
|
| st.markdown('#### 3. Conceptos extraídos — tabla editable') |
| st.caption('Añade filas manualmente si el formato del proveedor no se puede identificar. ' |
| 'Cantidad × precio se recalcula independientemente del importe del documento.') |
| original_frame = st.session_state.get('q_editor_items', pd.DataFrame(columns=EDITABLE)) |
| edit = st.data_editor( |
| original_frame, |
| hide_index=True, |
| num_rows='dynamic', |
| use_container_width=True, |
| key=f'q_data_editor_{identity[:12]}', |
| column_config={ |
| 'codigo': st.column_config.TextColumn('Código'), |
| 'descripcion': st.column_config.TextColumn('Descripción', width='large'), |
| 'cantidad': st.column_config.NumberColumn('Cantidad', format='%.2f'), |
| 'unidad': st.column_config.TextColumn('Unidad'), |
| 'precio_unitario': st.column_config.NumberColumn('Precio unitario', format='%.2f'), |
| 'importe_documento': st.column_config.NumberColumn('Importe declarado', format='%.2f'), |
| }, |
| ) |
| rows = edit[EDITABLE].to_dict(orient='records') |
| checked = validate_quote(fields, rows) |
| counts = st.columns(3) |
| with counts[0]: |
| st.metric('Partidas registradas', len(rows)) |
| with counts[1]: |
| sub = checked['totales']['subtotal_recalculado'] |
| st.metric('Subtotal recalculado', f'{sub:,.2f} {fields.get("moneda", "")}' if sub is not None else 'N/D') |
| with counts[2]: |
| st.metric('Alertas actuales', len(checked['alertas'])) |
|
|
| if checked['alertas']: |
| st.error('**Documento pendiente de revisión:**') |
| for a in checked['alertas']: |
| st.markdown('- ' + a) |
| else: |
| st.success('Las verificaciones **aritméticas y de campos obligatorios** pasaron. ' |
| 'Esto no garantiza autenticidad del cliente, validez fiscal ni precios de mercado.') |
|
|
| if raw_alerts: |
| with st.expander(f'Alertas del documento original ({len(raw_alerts)}) — historial conservado', expanded=True): |
| for a in raw_alerts: |
| st.write('• ' + a) |
| st.warning('Si se corrigió el dato, se guardará como modificación del revisor. ' |
| 'No cambia el PDF original: una discrepancia debe confirmarse con su emisor.') |
|
|
| with st.expander('Auditoría técnica: texto extraído del PDF/OCR'): |
| st.code(extracted['texto'][:18000], language=None) |
| if parsed['lineas_no_estructuradas']: |
| st.caption(f"{len(parsed['lineas_no_estructuradas'])} línea(s) fuera de la tabla estructurada. " |
| 'Revisa que no falten conceptos antes de aprobar.') |
|
|
| st.markdown('#### 4. Revisión y autorización para exportar') |
| note = st.text_area('Nota de revisión (qué comprobaste y con quién)', |
| key='q_reviewer_note', placeholder='Ejemplo: Importes contrastados con el proveedor...') |
| reviewed = st.checkbox('Confirmo que cotejé el cliente, todos los conceptos y cantidades con la cotización.', |
| key='q_confirm_content') |
| acknowledged = True |
| if raw_alerts: |
| acknowledged = st.checkbox( |
| 'Confirmo que cualquier discrepancia del PDF original fue aclarada con el emisor; ' |
| 'el registro conservará el cambio y las alertas originales.', |
| key='q_ack_source' |
| ) |
| if raw_alerts and not note.strip(): |
| st.caption('Con discrepancias de origen, agrega una nota de revisión antes de aprobar.') |
|
|
| |
| fingerprint_source = json.dumps({'fields': fields, 'items': rows, 'ack': acknowledged, |
| 'reviewed': reviewed, 'note': note}, |
| ensure_ascii=False, sort_keys=True, default=str) |
| approval_fingerprint = hashlib.sha256(fingerprint_source.encode('utf-8')).hexdigest() |
| eligible = bool(checked['apto_aprobacion'] and reviewed and acknowledged and (not raw_alerts or note.strip())) |
| if st.button('✅ Aprobar estructura revisada', disabled=not eligible, key='q_approve'): |
| st.session_state['q_approval_sig'] = approval_fingerprint |
| approved = eligible and st.session_state.get('q_approval_sig') == approval_fingerprint |
| if approved: |
| st.success('Estructura **aprobada por el revisor**. Lista para exportación; no se envía automáticamente a ningún ERP.') |
| else: |
| st.info('Estado actual: **REVISIÓN PENDIENTE**. Puedes exportar el borrador, pero no enviarlo como aprobado.') |
|
|
| |
| changes = [] |
| for key, _ in FIELD_NAMES: |
| old = str(parsed['campos'].get(key, '') or '') |
| new = str(fields.get(key, '') or '') |
| if old != new: |
| changes.append({'campo': key, 'original': old, 'revisado': new}) |
| original_items = parsed.get('conceptos', []) |
| for index, row in enumerate(rows): |
| oldrow = original_items[index] if index < len(original_items) else {} |
| for key in EDITABLE: |
| if str(oldrow.get(key, '')) != str(row.get(key, '')): |
| changes.append({'campo': f'concepto_{index+1}.{key}', |
| 'original': oldrow.get(key), 'revisado': row.get(key)}) |
|
|
| payload = build_payload(fields, checked, uploaded.name, extracted['motores'], approved) |
| payload['control']['alertas_de_origen'] = raw_alerts |
| payload['control']['correcciones_registradas'] = changes |
| payload['control']['nota_revisor'] = note.strip() |
| payload['control']['formato_reconocido'] = 'TABLA_DELIMITADA' if parsed['conceptos'] else 'REVISION_MANUAL' |
| payload['control']['advertencia'] = ( |
| 'Prototipo independiente. Estructura generada tras extracción heurística y revisión, ' |
| 'no implica envío, integración ni validación fiscal.' |
| ) |
|
|
| with st.expander('Ver JSON estructurado antes de descargar', expanded=True): |
| st.json(payload, expanded=False) |
| json_bytes = json.dumps(payload, ensure_ascii=False, indent=2, allow_nan=False).encode('utf-8') |
| export_df = pd.DataFrame(checked['conceptos']) |
| export_df.insert(0, 'folio', payload['folio']) |
| export_df.insert(1, 'cliente', payload['cliente']['nombre']) |
| export_df.insert(2, 'proyecto', payload['proyecto']) |
| export_df.insert(3, 'moneda', payload['moneda']) |
| c1, c2 = st.columns(2) |
| with c1: |
| st.download_button( |
| '⬇️ Exportar JSON ' + ('APROBADO' if approved else 'BORRADOR'), |
| data=json_bytes, mime='application/json', |
| file_name=f"cotizacion_{'aprobada' if approved else 'borrador'}.json", key='q_json') |
| with c2: |
| st.download_button( |
| '⬇️ Exportar conceptos CSV ' + ('APROBADO' if approved else 'BORRADOR'), |
| data=export_df.to_csv(index=False).encode('utf-8-sig'), mime='text/csv', |
| file_name=f"conceptos_{'aprobados' if approved else 'borrador'}.csv", key='q_csv') |
| st.caption('La salida queda disponible para una integración futura mediante API o ETL. ' |
| 'Aquí **no se transmite** ningún dato a un CRM, ERP, base de datos o correo.') |
|
|
