from flask import Flask, request, jsonify, send_file, render_template, send_from_directory from flask_cors import CORS from flask_sqlalchemy import SQLAlchemy from sqlalchemy import text from datetime import datetime, timedelta import calendar import logging import os from pathlib import Path import io import pandas as pd from typing import Dict, List, Optional, Any import json # Configuración de logging para producción log_dir = '/var/www/html/TransmisionesMig/logs' os.makedirs(log_dir, exist_ok=True) logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler(os.path.join(log_dir, 'reporte_mensual.log')), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) # Crear la aplicación Flask app = Flask(__name__) # CONFIGURACIÓN DE BASE DE DATOS DB_SERVER = 'localhost' DB_PORT = '1433' DB_DATABASE = 'TRANSMISIONESMIG' DB_USERNAME = 'sa' DB_PASSWORD = 'Sistemas123*/' DB_DRIVER = 'ODBC Driver 17 for SQL Server' # Construir la cadena de conexión app.config['SQLALCHEMY_DATABASE_URI'] = ( f"mssql+pyodbc://{DB_USERNAME}:{DB_PASSWORD}@{DB_SERVER}:{DB_PORT}/{DB_DATABASE}" f"?driver={DB_DRIVER.replace(' ', '+')}&TrustServerCertificate=yes&Encrypt=yes" ) app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False app.config['SECRET_KEY'] = 'K8m2#D9v$L4x@R7n*W5q&F1s!P6j%T3y^B8e@N7k$H4m*V9r&L2x#Q5w' app.config['DEBUG'] = False # Habilitar CORS CORS(app) # Inicializar SQLAlchemy db = SQLAlchemy(app) # MODELO DE EMPRESAS CLIENTES class EmpresaCliente(db.Model): __tablename__ = 'Empresas_Clientes' ID = db.Column(db.Integer, primary_key=True) RazonSocial = db.Column(db.String(255), nullable=False) NombreComercial = db.Column(db.String(255), nullable=False) NIT = db.Column(db.String(20), nullable=False) Email = db.Column(db.String(255)) Telefono = db.Column(db.String(50)) Direccion = db.Column(db.String(500)) Activo = db.Column(db.Boolean, default=True) FechaCreacion = db.Column(db.DateTime, default=datetime.utcnow) FechaModificacion = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) # CAMBIO: Constraint unique por NIT + NombreComercial __table_args__ = (db.UniqueConstraint('NIT', 'NombreComercial', name='UQ_NIT_NombreComercial'),) reportes = db.relationship('ControlReportes', backref='empresa_cliente', lazy='dynamic') def to_dict(self): return { 'id': self.ID, 'razon_social': self.RazonSocial or '', 'nombre_comercial': self.NombreComercial or '', 'nit': self.NIT or '', 'email': self.Email or '', 'telefono': self.Telefono or '', 'direccion': self.Direccion or '', 'activo': bool(self.Activo) if self.Activo is not None else True, 'fecha_creacion': self.FechaCreacion.isoformat() if self.FechaCreacion else None, 'fecha_modificacion': self.FechaModificacion.isoformat() if self.FechaModificacion else None } # MODELO MODIFICADO PARA REPORTES (CON CLIENTE) class ControlReportes(db.Model): __tablename__ = 'Control_Reportes' ID = db.Column(db.Integer, primary_key=True) Periodo = db.Column(db.String(7), nullable=False) ClienteID = db.Column(db.Integer, db.ForeignKey('Empresas_Clientes.ID'), nullable=False) FechaEnvio = db.Column(db.DateTime, nullable=False, default=datetime.utcnow) Enviado = db.Column(db.Boolean, default=True) Observaciones = db.Column(db.String(500)) TotalDocumentos = db.Column(db.Integer, default=0) DocumentosExitosos = db.Column(db.Integer, default=0) ValorTotal = db.Column(db.Numeric(18, 2), default=0) TasaExito = db.Column(db.Numeric(5, 2), default=0) ContenidoCompleto = db.Column(db.Text) DetalleFacturas = db.Column(db.Text) DetalleGastos = db.Column(db.Text) # NUEVA COLUMNA PARA MARCAR SI YA FUE FACTURADO Facturado = db.Column(db.Boolean, default=False) FechaFacturacion = db.Column(db.DateTime, nullable=True) NumeroFactura = db.Column(db.String(50), nullable=True) # Índice único por período y cliente __table_args__ = (db.UniqueConstraint('Periodo', 'ClienteID', name='_periodo_cliente_uc'),) def to_dict(self): cliente_info = None if self.empresa_cliente: cliente_info = { 'id': self.empresa_cliente.ID, 'razon_social': self.empresa_cliente.RazonSocial, 'nombre_comercial': self.empresa_cliente.NombreComercial, 'nit': self.empresa_cliente.NIT, 'email': self.empresa_cliente.Email, 'telefono': self.empresa_cliente.Telefono, 'direccion': self.empresa_cliente.Direccion, 'activo': self.empresa_cliente.Activo } # Parsear detalles guardados detalle_facturas_raw = None detalle_gastos_raw = None try: if self.DetalleFacturas: detalle_facturas_raw = json.loads(self.DetalleFacturas) except: pass try: if self.DetalleGastos: detalle_gastos_raw = json.loads(self.DetalleGastos) except: pass # CORREGIR: Si los datos están mal, usar los totales del reporte # El problema es que DetalleFacturas/DetalleGastos pueden tener datos incorrectos # Mejor confiar en TotalDocumentos y DocumentosExitosos que son los correctos total_docs = int(self.TotalDocumentos or 0) total_exitosos = int(self.DocumentosExitosos or 0) # Intentar obtener detalle de facturas del JSON facturas_total = 0 facturas_exitosos = 0 gastos_total = 0 gastos_exitosos = 0 if detalle_facturas_raw: facturas_total = int(detalle_facturas_raw.get('total_documentos', 0)) facturas_exitosos = int(detalle_facturas_raw.get('documentos_exitosos', 0)) if detalle_gastos_raw: gastos_total = int(detalle_gastos_raw.get('total_documentos', 0)) gastos_exitosos = int(detalle_gastos_raw.get('documentos_exitosos', 0)) # VALIDACIÓN: Si la suma no coincide con el total, hay un problema suma_total = facturas_total + gastos_total suma_exitosos = facturas_exitosos + gastos_exitosos # Si los datos no coinciden, recalcular proporcionalmente if suma_total != total_docs or suma_exitosos != total_exitosos: logger.warning("Discrepancia en datos del reporte ID {0}: Total guardado={1}, Suma detalles={2}".format( self.ID, total_docs, suma_total )) # Si hay facturas en el detalle pero no gastos, asignar todo a facturas if facturas_total > 0 and gastos_total == 0: facturas_total = total_docs facturas_exitosos = total_exitosos # Si hay gastos pero no facturas, asignar todo a gastos elif gastos_total > 0 and facturas_total == 0: gastos_total = total_docs gastos_exitosos = total_exitosos # Si no hay ningún detalle, asignar todo a facturas por defecto elif facturas_total == 0 and gastos_total == 0: facturas_total = total_docs facturas_exitosos = total_exitosos # Construir detalle_facturas corregido detalle_facturas = { 'total_documentos': facturas_total, 'documentos_exitosos': facturas_exitosos, 'documentos_recibidos': detalle_facturas_raw.get('documentos_recibidos', 0) if detalle_facturas_raw else 0, 'documentos_aceptados': detalle_facturas_raw.get('documentos_aceptados', 0) if detalle_facturas_raw else 0, 'documentos_rechazados': detalle_facturas_raw.get('documentos_rechazados', 0) if detalle_facturas_raw else 0, 'total_subtotal': float(detalle_facturas_raw.get('total_subtotal', 0)) if detalle_facturas_raw else 0, 'total_iva': float(detalle_facturas_raw.get('total_iva', 0)) if detalle_facturas_raw else 0, 'total_valor': float(detalle_facturas_raw.get('total_valor', 0)) if detalle_facturas_raw else 0, 'tasa_exito': round((facturas_exitosos / facturas_total * 100) if facturas_total > 0 else 0, 2) } # Construir detalle_gastos corregido detalle_gastos = { 'total_documentos': gastos_total, 'documentos_exitosos': gastos_exitosos, 'documentos_recibidos': 0, 'documentos_aceptados': 0, 'documentos_rechazados': 0, 'total_subtotal': float(detalle_gastos_raw.get('total_subtotal', 0)) if detalle_gastos_raw else 0, 'total_iva': float(detalle_gastos_raw.get('total_iva', 0)) if detalle_gastos_raw else 0, 'total_valor': float(detalle_gastos_raw.get('total_valor', 0)) if detalle_gastos_raw else 0, 'tasa_exito': round((gastos_exitosos / gastos_total * 100) if gastos_total > 0 else 0, 2) } return { 'id': self.ID, 'periodo': self.Periodo, 'cliente_id': self.ClienteID, 'cliente': cliente_info, 'fecha_envio': self.FechaEnvio.isoformat() if self.FechaEnvio else None, 'total_documentos': total_docs, 'documentos_exitosos': total_exitosos, 'valor_total': float(self.ValorTotal or 0), 'tasa_exito': float(self.TasaExito or 0), 'observaciones': self.Observaciones, 'status': 'completed' if self.Enviado else 'pending', 'contenido_completo': self.ContenidoCompleto, 'detalle_facturas': detalle_facturas, 'detalle_gastos': detalle_gastos, # NUEVA INFORMACIÓN DE FACTURACIÓN 'facturado': bool(self.Facturado), 'fecha_facturacion': self.FechaFacturacion.isoformat() if self.FechaFacturacion else None, 'numero_factura': self.NumeroFactura, 'estado_facturacion': 'Facturado' if self.Facturado else 'Pendiente', # INFORMACIÓN PARA CARTERA 'cliente_info_resumida': "{0} ({1})".format(cliente_info['nombre_comercial'], cliente_info['nit']) if cliente_info else "Cliente no encontrado", 'cliente_razon_social': cliente_info['razon_social'] if cliente_info else None, 'cliente_nombre_comercial': cliente_info['nombre_comercial'] if cliente_info else None, 'cliente_nit': cliente_info['nit'] if cliente_info else None } @app.route('/api/facturacion-por-cliente', methods=['GET']) def get_facturacion_por_cliente(): """Obtiene resumen de facturación agrupado por cliente - útil para cartera""" try: year = request.args.get('year', datetime.now().year) month = request.args.get('month') # Construir filtros filters = [] if year: if month: period_filter = f"{year}-{str(month).zfill(2)}" filters.append(ControlReportes.Periodo == period_filter) else: filters.append(ControlReportes.Periodo.like(f"{year}-%")) # Consultar reportes con información de cliente query = db.session.query(ControlReportes, EmpresaCliente).join( EmpresaCliente, ControlReportes.ClienteID == EmpresaCliente.ID ).filter(ControlReportes.Enviado == True) if filters: for filter_condition in filters: query = query.filter(filter_condition) reportes = query.order_by(ControlReportes.Periodo.desc()).all() # Agrupar por cliente clientes_facturacion = {} for reporte, cliente in reportes: cliente_key = cliente.ID if cliente_key not in clientes_facturacion: clientes_facturacion[cliente_key] = { 'cliente': { 'id': cliente.ID, 'razon_social': cliente.RazonSocial, 'nombre_comercial': cliente.NombreComercial, 'nit': cliente.NIT, 'email': cliente.Email, 'telefono': cliente.Telefono, 'activo': cliente.Activo }, 'reportes': [], 'total_documentos': 0, 'total_valor': 0.0, 'periodos_reportados': 0, 'tasa_exito_promedio': 0.0 } # Agregar reporte clientes_facturacion[cliente_key]['reportes'].append({ 'periodo': reporte.Periodo, 'total_documentos': reporte.TotalDocumentos, 'documentos_exitosos': reporte.DocumentosExitosos, 'valor_total': float(reporte.ValorTotal or 0), 'tasa_exito': float(reporte.TasaExito or 0), 'fecha_envio': reporte.FechaEnvio.isoformat() if reporte.FechaEnvio else None }) # Acumular totales clientes_facturacion[cliente_key]['total_documentos'] += reporte.TotalDocumentos or 0 clientes_facturacion[cliente_key]['total_valor'] += float(reporte.ValorTotal or 0) clientes_facturacion[cliente_key]['periodos_reportados'] += 1 # Calcular promedios for cliente_data in clientes_facturacion.values(): if cliente_data['periodos_reportados'] > 0: tasa_total = sum(r['tasa_exito'] for r in cliente_data['reportes']) cliente_data['tasa_exito_promedio'] = round(tasa_total / cliente_data['periodos_reportados'], 2) return jsonify({ 'success': True, 'year': year, 'month': month, 'total_clientes': len(clientes_facturacion), 'clientes_facturacion': list(clientes_facturacion.values()) }) except Exception as e: logger.error(f"Error obteniendo facturación por cliente: {str(e)}") return jsonify({ 'success': False, 'error': str(e) }), 500 # CLASE MODIFICADA PARA MANEJO DE REPORTES CON CLIENTES class ReportManager: """Maneja la generación y gestión de reportes mensuales por cliente""" def __init__(self): pass def generate_monthly_report(self, cliente_id: int, target_date: datetime = None) -> Dict[str, Any]: """Genera el reporte mensual para un cliente específico""" if target_date is None: today = datetime.now() target_date = datetime(today.year, today.month, 1) - timedelta(days=1) period = target_date.strftime("%Y-%m") start_date = datetime(target_date.year, target_date.month, 1) end_date = datetime(target_date.year, target_date.month, calendar.monthrange(target_date.year, target_date.month)[1], 23, 59, 59) # Obtener información del cliente cliente = EmpresaCliente.query.get(cliente_id) if not cliente: return { 'success': False, 'message': f'Cliente con ID {cliente_id} no encontrado', 'period': period } logger.info(f"Generando reporte para cliente {cliente.RazonSocial} - período {period} ({start_date} - {end_date})") try: # Verificar si ya existe un reporte para este período y cliente existing_report = ControlReportes.query.filter_by(Periodo=period, ClienteID=cliente_id, Enviado=True).first() if existing_report: return { 'success': False, 'message': f'Ya existe un reporte para el cliente {cliente.RazonSocial} en el período {period}', 'period': period } # Obtener estadísticas de facturas electrónicas para este cliente invoice_stats = self._get_invoice_statistics_by_client(cliente_id, start_date, end_date) # Obtener estadísticas de gastos para este cliente expense_stats = self._get_expense_statistics_by_client(cliente_id, start_date, end_date) # Combinar estadísticas combined_stats = self._combine_statistics(invoice_stats, expense_stats) # Generar contenido del reporte report_content = self._generate_report_content( period, start_date, end_date, cliente, combined_stats, invoice_stats, expense_stats ) # Guardar reporte en la base de datos report_id = self._save_report(period, cliente_id, combined_stats, report_content, invoice_stats, expense_stats) logger.info(f"Reporte generado exitosamente para cliente {cliente.RazonSocial} - período {period}") return { 'success': True, 'message': f'Reporte generado exitosamente para {cliente.RazonSocial} - {period}', 'period': period, 'cliente': cliente.to_dict(), 'report_id': report_id, 'statistics': combined_stats } except Exception as e: logger.error(f"Error generando reporte mensual para cliente {cliente_id}: {str(e)}") return { 'success': False, 'message': f'Error generando reporte: {str(e)}', 'period': period } def _empty_stats(self) -> Dict[str, Any]: """Retorna estadísticas vacías""" return { 'total_documentos': 0, 'documentos_exitosos': 0, 'documentos_recibidos': 0, 'documentos_aceptados': 0, 'documentos_rechazados': 0, 'total_subtotal': 0.0, 'total_iva': 0.0, 'total_valor': 0.0, 'tasa_exito': 0.0 } def _combine_statistics(self, invoice_stats: Dict[str, Any], expense_stats: Dict[str, Any]) -> Dict[str, Any]: """Combina las estadísticas de facturas y gastos""" combined = {} for key in invoice_stats.keys(): combined[key] = invoice_stats[key] + expense_stats[key] # Recalcular tasa de éxito total = combined['total_documentos'] exitosos = combined['documentos_exitosos'] combined['tasa_exito'] = round((exitosos / total * 100) if total > 0 else 0, 2) return combined def _generate_report_content(self, period: str, start_date: datetime, end_date: datetime, cliente: EmpresaCliente, combined_stats: Dict[str, Any], invoice_stats: Dict[str, Any], expense_stats: Dict[str, Any]) -> str: """Genera el contenido textual del reporte para un cliente específico""" month_names = [ 'Enero', 'Febrero', 'Marzo', 'Abril', 'Mayo', 'Junio', 'Julio', 'Agosto', 'Septiembre', 'Octubre', 'Noviembre', 'Diciembre' ] month_name = month_names[start_date.month - 1] year = start_date.year content = [] content.append("REPORTE MENSUAL CONSOLIDADO DE FACTURACIÓN ELECTRÓNICA") content.append("=" * 65) content.append(f"Fecha de generación: {datetime.now().strftime('%d/%m/%Y %H:%M:%S')}") content.append(f"Período analizado: {month_name} {year}") content.append("") # INFORMACIÓN DEL PROVEEDOR (TU EMPRESA) content.append("INFORMACIÓN DEL PROVEEDOR:") content.append("-" * 30) content.append("Razón Social: TransmisionesMig SAS") content.append("Nombre Comercial: TransmisionesMig") content.append("Servicio: Facturación Electrónica") content.append("") # INFORMACIÓN DEL CLIENTE (LA EMPRESA QUE PAGA) content.append("INFORMACIÓN DEL CLIENTE:") content.append("-" * 30) content.append(f"Razón Social: {cliente.RazonSocial}") content.append(f"Nombre Comercial: {cliente.NombreComercial}") content.append(f"NIT: {cliente.NIT}") if cliente.Email: content.append(f"Email: {cliente.Email}") if cliente.Telefono: content.append(f"Teléfono: {cliente.Telefono}") if cliente.Direccion: content.append(f"Dirección: {cliente.Direccion}") content.append("") content.append("=" * 65) content.append("") # RESUMEN EJECUTIVO content.append("RESUMEN EJECUTIVO") content.append("-" * 20) content.append(f"Cliente: {cliente.NombreComercial}") content.append(f"Período de facturación: {start_date.strftime('%d/%m/%Y')} - {end_date.strftime('%d/%m/%Y')}") if combined_stats['total_documentos'] > 0: content.append(f"Total documentos procesados: {combined_stats['total_documentos']:,}") content.append(f"Documentos transmitidos exitosamente: {combined_stats['documentos_exitosos']:,}") content.append(f"Tasa de éxito: {combined_stats['tasa_exito']:.2f}%") content.append(f"Valor total facturado: ${combined_stats['total_valor']:,.2f}") else: content.append(f"No se encontraron documentos para {cliente.RazonSocial} en este período.") content.append("") # DETALLE TÉCNICO (resto del contenido existente...) if combined_stats['total_documentos'] > 0: content.append("DETALLE DE FACTURACIÓN ELECTRÓNICA") content.append("=" * 40) content.append("") content.append("Estadísticas de Documentos (Facturas y Gastos):") content.append(f"- Total Documentos: {combined_stats['total_documentos']:,}") content.append(f"- Documentos Transmitidos Exitosamente: {combined_stats['documentos_exitosos']:,}") content.append(f"- Documentos NO Transmitidos: {combined_stats['total_documentos'] - combined_stats['documentos_exitosos']:,}") content.append("") content.append("Indicadores de Efectividad:") content.append(f"- Tasa de Transmisión Exitosa: {combined_stats['tasa_exito']:.2f}%") content.append(f"- Tasa de Fallos en Transmisión: {100 - combined_stats['tasa_exito']:.2f}%") content.append("") content.append("Resumen Financiero:") content.append(f"- Subtotal: ${combined_stats['total_subtotal']:,.2f}") content.append(f"- IVA: ${combined_stats['total_iva']:,.2f}") content.append(f"- Total: ${combined_stats['total_valor']:,.2f}") content.append("") # Detalle separado por tipo de documento content.append("DETALLE POR TIPO DE DOCUMENTO") content.append("=" * 35) content.append("") # Detalle de Facturas if invoice_stats['total_documentos'] > 0: tasa_exito_facturas = (invoice_stats['documentos_exitosos'] / invoice_stats['total_documentos'] * 100) if invoice_stats['total_documentos'] > 0 else 0 content.append("Facturas Electrónicas:") content.append(f"- Total Documentos: {invoice_stats['total_documentos']:,}") content.append(f"- Transmitidos Exitosamente: {invoice_stats['documentos_exitosos']:,} ({tasa_exito_facturas:.2f}%)") content.append(f"- NO Transmitidos: {invoice_stats['total_documentos'] - invoice_stats['documentos_exitosos']:,} ({(100 - tasa_exito_facturas):.2f}%)") content.append(f"- Valor Total: ${invoice_stats['total_valor']:,.2f}") content.append("") # Detalle de Gastos if expense_stats['total_documentos'] > 0: tasa_exito_gastos = (expense_stats['documentos_exitosos'] / expense_stats['total_documentos'] * 100) if expense_stats['total_documentos'] > 0 else 0 content.append("Documentos de Gastos:") content.append(f"- Total Documentos: {expense_stats['total_documentos']:,}") content.append(f"- Transmitidos Exitosamente: {expense_stats['documentos_exitosos']:,} ({tasa_exito_gastos:.2f}%)") content.append(f"- NO Transmitidos: {expense_stats['total_documentos'] - expense_stats['documentos_exitosos']:,} ({(100 - tasa_exito_gastos):.2f}%)") content.append(f"- Valor Total: ${expense_stats['total_valor']:,.2f}") content.append("") content.append("=" * 65) content.append("NOTA IMPORTANTE:") content.append("Este reporte incluye documentos de las tablas fe_documentos y Gastos1") content.append("procesados durante el período indicado.") content.append("Solo se evalúa el estado de transmisión (transmitido/no transmitido) para ambas tablas.") content.append("") content.append("Para cualquier consulta sobre este reporte, contacte a:") content.append("TransmisionesMig SAS - Servicios de Facturación Electrónica") content.append("=" * 65) return "\n".join(content) def _save_report(self, period: str, cliente_id: int, combined_stats: Dict[str, Any], content: str, invoice_stats: Dict[str, Any], expense_stats: Dict[str, Any]) -> int: """Guarda el reporte en la base de datos con información del cliente""" try: report = ControlReportes( Periodo=period, ClienteID=cliente_id, FechaEnvio=datetime.utcnow(), Enviado=True, Observaciones='Reporte generado automáticamente', TotalDocumentos=combined_stats['total_documentos'], DocumentosExitosos=combined_stats['documentos_exitosos'], ValorTotal=combined_stats['total_valor'], TasaExito=combined_stats['tasa_exito'], ContenidoCompleto=content, DetalleFacturas=json.dumps(invoice_stats), DetalleGastos=json.dumps(expense_stats) ) db.session.add(report) db.session.commit() return report.ID except Exception as e: db.session.rollback() logger.error(f"Error guardando reporte: {str(e)}") raise def get_all_reports(self) -> List[Dict[str, Any]]: """Obtiene todos los reportes con información del cliente""" try: reports = ControlReportes.query.order_by(ControlReportes.Periodo.desc()).all() return [report.to_dict() for report in reports] except Exception as e: logger.error(f"Error obteniendo reportes: {str(e)}") return [] def get_report_detail(self, report_id: int) -> Optional[Dict[str, Any]]: """Obtiene el detalle completo de un reporte""" try: report = ControlReportes.query.get(report_id) return report.to_dict() if report else None except Exception as e: logger.error(f"Error obteniendo detalle del reporte {report_id}: {str(e)}") return None def get_reports_by_client(self, cliente_id: int) -> List[Dict[str, Any]]: """Obtiene todos los reportes de un cliente específico""" try: reports = ControlReportes.query.filter_by(ClienteID=cliente_id).order_by(ControlReportes.Periodo.desc()).all() return [report.to_dict() for report in reports] except Exception as e: logger.error(f"Error obteniendo reportes del cliente {cliente_id}: {str(e)}") return [] # FUNCIÓN PARA CREAR TABLAS Y DATOS INICIALES def crear_tablas(): """Crear todas las tablas automáticamente""" try: # Crear todas las tablas db.create_all() logger.info("✓ Tablas creadas/verificadas exitosamente") db.session.commit() logger.info("✓ Clientes de ejemplo creados") except Exception as e: logger.error(f"✗ Error creando tablas: {e}") db.session.rollback() return False # Inicializar el report manager report_manager = ReportManager() # RUTAS DE LA APLICACIÓN @app.route('/') def index(): """Página principal""" try: return render_template('index.html') except Exception as e: logger.error(f"Error cargando página principal: {str(e)}") return f"Error cargando la aplicación: {str(e)}", 500 @app.route('/static/') def static_files(filename): """Servir archivos estáticos""" try: return send_from_directory('/var/www/html/TransmisionesMig/static', filename) except Exception as e: logger.error(f"Error sirviendo archivo estático {filename}: {str(e)}") return "Archivo no encontrado", 404 @app.route('/test') def test(): """Endpoint de prueba""" return jsonify({ 'status': 'ok', 'message': 'Servidor funcionando correctamente', 'timestamp': datetime.now().isoformat() }) @app.route('/test-db') def test_db(): """Probar conexión a base de datos""" try: db.session.execute(text('SELECT 1')) clientes_count = EmpresaCliente.query.count() reportes_count = ControlReportes.query.count() return jsonify({ 'status': 'ok', 'message': 'Base de datos conectada', 'clientes_count': clientes_count, 'reportes_count': reportes_count, 'db_uri': app.config['SQLALCHEMY_DATABASE_URI'].split('@')[1] }) except Exception as e: logger.error(f"Error en test-db: {str(e)}") return jsonify({ 'status': 'error', 'message': str(e) }), 500 @app.route('/init-db') def init_database(): """Inicializar base de datos""" try: resultado = crear_tablas() if resultado: return jsonify({ 'success': True, 'message': 'Base de datos inicializada correctamente' }) else: return jsonify({ 'success': False, 'message': 'Error al inicializar la base de datos' }), 500 except Exception as e: logger.error(f"Error en init-db: {str(e)}") return jsonify({ 'success': False, 'error': str(e) }), 500 # RUTAS PARA GESTIÓN DE CLIENTES @app.route('/api/clientes', methods=['GET']) def get_clientes(): """Obtiene la lista de clientes""" try: clientes = EmpresaCliente.query.filter_by(Activo=True).order_by(EmpresaCliente.RazonSocial).all() return jsonify([cliente.to_dict() for cliente in clientes]) except Exception as e: logger.error(f"Error obteniendo clientes: {str(e)}") return jsonify({'error': str(e)}), 500 @app.route('/api/clientes', methods=['POST']) def create_cliente(): """Crea un nuevo cliente""" try: data = request.get_json() # Validar campos obligatorios required_fields = ['razon_social', 'nombre_comercial', 'nit'] for field in required_fields: if not data.get(field): return jsonify({ 'success': False, 'error': f'El campo {field} es obligatorio' }), 400 # Verificar si ya existe un cliente con ese NIT existing = EmpresaCliente.query.filter_by(NIT=data['nit']).first() if existing: return jsonify({ 'success': False, 'error': f'Ya existe un cliente con NIT {data["nit"]}' }), 400 # Crear nuevo cliente cliente = EmpresaCliente( RazonSocial=data['razon_social'], NombreComercial=data['nombre_comercial'], NIT=data['nit'], Email=data.get('email', ''), Telefono=data.get('telefono', ''), Direccion=data.get('direccion', ''), Activo=data.get('activo', True) ) db.session.add(cliente) db.session.commit() logger.info(f"Cliente creado: {cliente.RazonSocial} (NIT: {cliente.NIT})") return jsonify({ 'success': True, 'message': 'Cliente creado exitosamente', 'cliente': cliente.to_dict() }), 201 except Exception as e: logger.error(f"Error creando cliente: {str(e)}") db.session.rollback() return jsonify({ 'success': False, 'error': str(e) }), 500 @app.route('/api/clientes/', methods=['PUT']) def update_cliente(cliente_id): """Actualiza un cliente""" try: cliente = EmpresaCliente.query.get(cliente_id) if not cliente: return jsonify({ 'success': False, 'error': 'Cliente no encontrado' }), 404 data = request.get_json() # Actualizar campos if 'razon_social' in data: cliente.RazonSocial = data['razon_social'] if 'nombre_comercial' in data: cliente.NombreComercial = data['nombre_comercial'] if 'email' in data: cliente.Email = data['email'] if 'telefono' in data: cliente.Telefono = data['telefono'] if 'direccion' in data: cliente.Direccion = data['direccion'] if 'activo' in data: cliente.Activo = data['activo'] cliente.FechaModificacion = datetime.utcnow() db.session.commit() logger.info(f"Cliente actualizado: {cliente.RazonSocial} (ID: {cliente.ID})") return jsonify({ 'success': True, 'message': 'Cliente actualizado exitosamente', 'cliente': cliente.to_dict() }) except Exception as e: logger.error(f"Error actualizando cliente {cliente_id}: {str(e)}") db.session.rollback() return jsonify({ 'success': False, 'error': str(e) }), 500 @app.route('/api/clientes/', methods=['DELETE']) def delete_cliente(cliente_id): """Desactiva un cliente (soft delete)""" try: cliente = EmpresaCliente.query.get(cliente_id) if not cliente: return jsonify({ 'success': False, 'error': 'Cliente no encontrado' }), 404 # Verificar si tiene reportes asociados reportes_count = ControlReportes.query.filter_by(ClienteID=cliente_id).count() if reportes_count > 0: # Soft delete: solo desactivar cliente.Activo = False cliente.FechaModificacion = datetime.utcnow() db.session.commit() message = f'Cliente {cliente.RazonSocial} desactivado (tiene {reportes_count} reportes asociados)' else: # Hard delete: eliminar completamente razon_social = cliente.RazonSocial db.session.delete(cliente) db.session.commit() message = f'Cliente {razon_social} eliminado completamente' logger.info(f"Cliente eliminado/desactivado: {cliente.RazonSocial} (ID: {cliente_id})") return jsonify({ 'success': True, 'message': message }) except Exception as e: logger.error(f"Error eliminando cliente {cliente_id}: {str(e)}") db.session.rollback() return jsonify({ 'success': False, 'error': str(e) }), 500 @app.route('/api/clientes//reportes', methods=['GET']) def get_reportes_by_cliente(cliente_id): """Obtiene todos los reportes de un cliente específico""" try: cliente = EmpresaCliente.query.get(cliente_id) if not cliente: return jsonify({ 'success': False, 'error': 'Cliente no encontrado' }), 404 reportes = report_manager.get_reports_by_client(cliente_id) return jsonify({ 'success': True, 'cliente': cliente.to_dict(), 'reportes': reportes }) except Exception as e: logger.error(f"Error obteniendo reportes del cliente {cliente_id}: {str(e)}") return jsonify({ 'success': False, 'error': str(e) }), 500 @app.route('/api/test-pos-connection', methods=['GET']) def test_pos_connection(): """Endpoint para que el POS pruebe la conectividad""" try: return jsonify({ 'status': 'ok', 'message': 'Conexión desde POS exitosa', 'timestamp': datetime.now().isoformat(), 'server': 'Reportes Web MIG', 'version': '1.0', 'cliente_sistema': 'TransmisionesMig POS' }) except Exception as e: logger.error(f"Error en test-pos-connection: {str(e)}") return jsonify({ 'status': 'error', 'message': str(e) }), 500 @app.route('/api/reportes-desde-pos', methods=['POST']) def recibir_reporte_desde_pos(): """Recibe reportes enviados desde el sistema POS""" try: data = request.get_json() logger.info(f"📥 Recibiendo reporte desde POS - Período: {data.get('periodo')}") # Validar datos requeridos required_fields = ['periodo', 'fecha_envio', 'empresa'] for field in required_fields: if not data.get(field): return jsonify({ 'success': False, 'error': f'Campo requerido faltante: {field}' }), 400 # Verificar o crear cliente basado en NIT empresa_data = data['empresa'] cliente = verificar_o_crear_cliente_desde_pos(empresa_data) if not cliente: return jsonify({ 'success': False, 'error': 'No se pudo crear/verificar el cliente' }), 400 # Verificar si ya existe un reporte para este período y cliente period = data['periodo'] existing_report = ControlReportes.query.filter_by( Periodo=period, ClienteID=cliente.ID, Enviado=True ).first() if existing_report: # Actualizar reporte existente si viene del POS actualizar_reporte_existente(existing_report, data) mensaje = f'Reporte {period} actualizado para {cliente.RazonSocial}' else: # Crear nuevo reporte report_id = crear_reporte_desde_pos(cliente.ID, data) mensaje = f'Reporte {period} creado para {cliente.RazonSocial}' logger.info(f"✅ {mensaje}") return jsonify({ 'success': True, 'message': mensaje, 'cliente': { 'id': cliente.ID, 'razon_social': cliente.RazonSocial, 'nombre_comercial': cliente.NombreComercial, 'nit': cliente.NIT }, 'periodo': period }) except Exception as e: logger.error(f"❌ Error recibiendo reporte desde POS: {str(e)}") return jsonify({ 'success': False, 'error': f'Error interno: {str(e)}' }), 500 def verificar_o_crear_cliente_desde_pos(empresa_data): try: nit = (empresa_data.get('nit') or '').strip() nombre_comercial = (empresa_data.get('nombre_comercial') or 'Sin Definir').strip() if not nit: logger.warning("⚠️ NIT vacío") return None # CAMBIO: Buscar por NIT + NombreComercial cliente = EmpresaCliente.query.filter_by( NIT=nit, NombreComercial=nombre_comercial ).first() if cliente: # Actualizar si cambió razón social o contacto actualizado = False razon_social = empresa_data.get('razon_social') if razon_social and cliente.RazonSocial != razon_social: cliente.RazonSocial = razon_social actualizado = True email = empresa_data.get('email') if email and cliente.Email != email: cliente.Email = email actualizado = True telefono = empresa_data.get('telefono') if telefono and cliente.Telefono != telefono: cliente.Telefono = telefono actualizado = True if actualizado: cliente.FechaModificacion = datetime.utcnow() db.session.commit() logger.info(f"Cliente actualizado: {cliente.NombreComercial}") return cliente else: # Crear nuevo cliente nuevo_cliente = EmpresaCliente( RazonSocial=empresa_data.get('razon_social') or 'Sin Definir', NombreComercial=nombre_comercial, NIT=nit, Email=empresa_data.get('email') or '', Telefono=empresa_data.get('telefono') or '', Direccion='', Activo=True ) db.session.add(nuevo_cliente) db.session.commit() logger.info(f"Nuevo cliente: {nuevo_cliente.NombreComercial} ({nit})") return nuevo_cliente except Exception as ex: logger.error(f"Error en verificar_o_crear_cliente_desde_pos: {str(ex)}") db.session.rollback() return None def actualizar_reporte_existente(reporte, data): """Actualiza un reporte existente con datos del POS""" try: # Actualizar campos del reporte if data.get('total_documentos'): reporte.TotalDocumentos = data['total_documentos'] if data.get('documentos_exitosos'): reporte.DocumentosExitosos = data['documentos_exitosos'] if data.get('valor_total'): reporte.ValorTotal = data['valor_total'] if data.get('tasa_exito'): reporte.TasaExito = data['tasa_exito'] if data.get('contenido_completo'): reporte.ContenidoCompleto = data['contenido_completo'] # Actualizar observaciones reporte.Observaciones = f"Actualizado desde POS - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}" reporte.FechaEnvio = datetime.utcnow() db.session.commit() except Exception as e: logger.error(f"Error actualizando reporte existente: {str(e)}") db.session.rollback() def crear_reporte_desde_pos(cliente_id, data): """Crea un nuevo reporte desde datos del POS""" try: nuevo_reporte = ControlReportes( Periodo=data['periodo'], ClienteID=cliente_id, FechaEnvio=datetime.utcnow(), Enviado=True, Observaciones=f"Creado desde POS - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}", TotalDocumentos=data.get('total_documentos', 0), DocumentosExitosos=data.get('documentos_exitosos', 0), ValorTotal=data.get('valor_total', 0), TasaExito=data.get('tasa_exito', 0), ContenidoCompleto=data.get('contenido_completo', ''), DetalleFacturas=json.dumps(data.get('detalle_facturas', {})), DetalleGastos=json.dumps(data.get('detalle_gastos', {})) ) db.session.add(nuevo_reporte) db.session.commit() return nuevo_reporte.ID except Exception as e: logger.error(f"Error creando reporte desde POS: {str(e)}") db.session.rollback() return None # RUTAS MODIFICADAS PARA REPORTES CON CLIENTES @app.route('/api/status', methods=['GET']) def get_status(): """Obtiene el estado del sistema con información de clientes""" try: # Probar conexión a la base de datos try: db.session.execute(text('SELECT 1')) db_connected = True except: db_connected = False # Obtener información del último reporte last_report = None if db_connected: try: ultimo = ControlReportes.query.order_by(ControlReportes.FechaEnvio.desc()).first() if ultimo: last_report = { 'periodo': ultimo.Periodo, 'fecha_envio': ultimo.FechaEnvio.isoformat(), 'cliente': ultimo.empresa_cliente.to_dict() if ultimo.empresa_cliente else None } except Exception as e: logger.warning(f"Error obteniendo último reporte: {str(e)}") # Estadísticas generales clientes_activos = EmpresaCliente.query.filter_by(Activo=True).count() if db_connected else 0 total_reportes = ControlReportes.query.count() if db_connected else 0 # Calcular próxima fecha de reporte now = datetime.now() next_month = datetime(now.year, now.month + 1, 1) if now.month < 12 else datetime(now.year + 1, 1, 1) next_report_date = datetime(next_month.year, next_month.month, 5) return jsonify({ 'db_connected': db_connected, 'last_report': last_report, 'next_report_date': next_report_date.isoformat(), 'current_time': datetime.now().isoformat(), 'clientes_activos': clientes_activos, 'total_reportes': total_reportes }) except Exception as e: logger.error(f"Error obteniendo estado: {str(e)}") return jsonify({ 'db_connected': False, 'error': str(e), 'current_time': datetime.now().isoformat() }), 500 @app.route('/api/reports/summary', methods=['GET']) def get_reports_summary(): """Obtiene solo un resumen ligero de reportes (sin detalles pesados)""" try: # Query optimizada: solo campos necesarios para la lista reportes = db.session.query( ControlReportes.ID, ControlReportes.Periodo, ControlReportes.ClienteID, ControlReportes.FechaEnvio, ControlReportes.TotalDocumentos, ControlReportes.DocumentosExitosos, ControlReportes.ValorTotal, ControlReportes.TasaExito, ControlReportes.Enviado, EmpresaCliente.RazonSocial, EmpresaCliente.NombreComercial, EmpresaCliente.NIT ).join( EmpresaCliente, ControlReportes.ClienteID == EmpresaCliente.ID ).filter( ControlReportes.Enviado == True ).order_by( ControlReportes.Periodo.desc() ).all() # Formato ligero result = [] for r in reportes: result.append({ 'id': r.ID, 'periodo': r.Periodo, 'cliente_id': r.ClienteID, 'cliente': { 'razon_social': r.RazonSocial, 'nombre_comercial': r.NombreComercial, 'nit': r.NIT }, 'fecha_envio': r.FechaEnvio.isoformat() if r.FechaEnvio else None, 'total_documentos': r.TotalDocumentos or 0, 'documentos_exitosos': r.DocumentosExitosos or 0, 'valor_total': float(r.ValorTotal or 0), 'tasa_exito': float(r.TasaExito or 0), 'status': 'completed' if r.Enviado else 'pending' }) return jsonify(result) except Exception as e: logger.error(f"Error obteniendo resumen de reportes: {str(e)}") return jsonify({'error': str(e)}), 500 @app.route('/api/reports', methods=['GET']) def get_reports(): """Obtiene la lista de reportes con información de clientes""" try: # Permitir filtrar por cliente cliente_id = request.args.get('cliente_id') if cliente_id: reports = report_manager.get_reports_by_client(int(cliente_id)) else: reports = report_manager.get_all_reports() return jsonify(reports) except Exception as e: logger.error(f"Error obteniendo reportes: {str(e)}") return jsonify({'error': str(e)}), 500 @app.route('/api/reports/', methods=['GET']) def get_report_detail(report_id): """Obtiene el detalle de un reporte específico""" try: report = report_manager.get_report_detail(report_id) if report: return jsonify(report) else: return jsonify({'error': 'Reporte no encontrado'}), 404 except Exception as e: logger.error(f"Error obteniendo detalle del reporte {report_id}: {str(e)}") return jsonify({'error': str(e)}), 500 @app.route('/api/generate-report', methods=['POST']) def generate_report(): """Genera un nuevo reporte mensual para un cliente específico""" try: data = request.get_json() or {} # Cliente es obligatorio cliente_id = data.get('cliente_id') if not cliente_id: return jsonify({ 'success': False, 'message': 'El cliente_id es obligatorio' }), 400 target_date = None if 'year' in data and 'month' in data: target_date = datetime(int(data['year']), int(data['month']), 1) result = report_manager.generate_monthly_report(cliente_id, target_date) if result['success']: return jsonify(result) else: return jsonify(result), 400 except Exception as e: logger.error(f"Error generando reporte: {str(e)}") return jsonify({ 'success': False, 'message': f'Error interno del servidor: {str(e)}' }), 500 @app.route('/api/generate-all-reports', methods=['POST']) def generate_all_reports(): """Genera reportes para todos los clientes activos""" try: data = request.get_json() or {} target_date = None if 'year' in data and 'month' in data: target_date = datetime(int(data['year']), int(data['month']), 1) # Obtener todos los clientes activos clientes = EmpresaCliente.query.filter_by(Activo=True).all() results = [] generated_count = 0 for cliente in clientes: result = report_manager.generate_monthly_report(cliente.ID, target_date) results.append({ 'cliente': cliente.to_dict(), 'result': result }) if result['success']: generated_count += 1 return jsonify({ 'success': True, 'message': f'Se generaron {generated_count} reportes de {len(clientes)} clientes', 'generated_count': generated_count, 'total_clients': len(clientes), 'results': results }) except Exception as e: logger.error(f"Error generando reportes para todos los clientes: {str(e)}") return jsonify({ 'success': False, 'message': f'Error interno del servidor: {str(e)}' }), 500 @app.route('/api/verify-missing', methods=['POST']) def verify_missing(): """Verifica y genera reportes faltantes para todos los clientes""" try: # Obtener todos los clientes activos clientes = EmpresaCliente.query.filter_by(Activo=True).all() current_date = datetime.now() current_year = current_date.year current_month = current_date.month previous_year = current_year - 1 total_generated = 0 results_by_client = [] for cliente in clientes: generated_count = 0 generated_periods = [] # Verificar año anterior completo (12 meses) for month in range(1, 13): period = f"{previous_year}-{month:02d}" existing = ControlReportes.query.filter_by(Periodo=period, ClienteID=cliente.ID, Enviado=True).first() if not existing: target_date = datetime(previous_year, month, calendar.monthrange(previous_year, month)[1]) result = report_manager.generate_monthly_report(cliente.ID, target_date) if result['success']: generated_count += 1 generated_periods.append(period) # Verificar año actual hasta el mes anterior for month in range(1, current_month): period = f"{current_year}-{month:02d}" existing = ControlReportes.query.filter_by(Periodo=period, ClienteID=cliente.ID, Enviado=True).first() if not existing: target_date = datetime(current_year, month, calendar.monthrange(current_year, month)[1]) result = report_manager.generate_monthly_report(cliente.ID, target_date) if result['success']: generated_count += 1 generated_periods.append(period) total_generated += generated_count results_by_client.append({ 'cliente': cliente.to_dict(), 'generated_count': generated_count, 'generated_periods': generated_periods }) logger.info(f"Verificación completada. Se generaron {total_generated} reportes faltantes para {len(clientes)} clientes.") return jsonify({ 'success': True, 'generated_count': total_generated, 'total_clients': len(clientes), 'message': f'Se generaron {total_generated} reportes faltantes para {len(clientes)} clientes', 'results_by_client': results_by_client }) except Exception as e: logger.error(f"Error verificando reportes faltantes: {str(e)}") return jsonify({ 'success': False, 'generated_count': 0, 'message': f'Error interno del servidor: {str(e)}' }), 500 @app.route('/api/reports//download', methods=['GET']) def download_report(report_id): """Descarga un reporte en formato texto""" try: report = report_manager.get_report_detail(report_id) if not report: return jsonify({'error': 'Reporte no encontrado'}), 404 content = report.get('contenido_completo', '') if not content: return jsonify({'error': 'Contenido del reporte no disponible'}), 404 # Crear archivo en memoria bytes_buffer = io.BytesIO() bytes_buffer.write(content.encode('utf-8')) bytes_buffer.seek(0) # Nombre del archivo incluye información del cliente cliente_info = report.get('cliente', {}) cliente_name = cliente_info.get('nombre_comercial', 'Cliente').replace(' ', '_') filename = f"Reporte_{cliente_name}_{report['periodo']}.txt" return send_file( bytes_buffer, as_attachment=True, download_name=filename, mimetype='text/plain' ) except Exception as e: logger.error(f"Error descargando reporte {report_id}: {str(e)}") return jsonify({'error': str(e)}), 500 @app.route('/api/reports/export', methods=['GET']) def export_reports(): """Exporta reportes filtrados a Excel con formato de detalle de documentos por organización""" try: # Obtener parámetros de filtro cliente_id = request.args.get('cliente_id') year = request.args.get('year') month = request.args.get('month') # Construir query base query = db.session.query(ControlReportes, EmpresaCliente).join( EmpresaCliente, ControlReportes.ClienteID == EmpresaCliente.ID ).filter(ControlReportes.Enviado == True) # Aplicar filtros if cliente_id: query = query.filter(ControlReportes.ClienteID == int(cliente_id)) if year: if month: period_filter = "{0}-{1}".format(year, str(month).zfill(2)) query = query.filter(ControlReportes.Periodo == period_filter) else: query = query.filter(ControlReportes.Periodo.like("{0}-%".format(year))) reportes = query.order_by(EmpresaCliente.RazonSocial, ControlReportes.Periodo).all() if not reportes: return jsonify({'error': 'No hay reportes para exportar con los filtros seleccionados'}), 404 # Preparar datos en el nuevo formato excel_data = [] for reporte, cliente in reportes: try: # Extraer año y mes del período periodo_parts = reporte.Periodo.split('-') anio = int(periodo_parts[0]) mes = int(periodo_parts[1]) # USAR to_dict() que ya tiene la lógica de corrección reporte_dict = reporte.to_dict() detalle_facturas = reporte_dict.get('detalle_facturas', {}) detalle_gastos = reporte_dict.get('detalle_gastos', {}) # Agregar registros de Facturas - SOLO EXITOSOS exitosos_facturas = int(detalle_facturas.get('documentos_exitosos', 0)) if exitosos_facturas > 0: excel_data.append({ 'Organización': cliente.RazonSocial, 'Nombre comercial': cliente.NombreComercial, 'NIT Emisor': cliente.NIT, 'Tipo de documento': 'Factura', 'Año': anio, 'Mes': mes, 'Cantidad': exitosos_facturas }) # Agregar registros de Gastos (DocSopAdq) - SOLO EXITOSOS exitosos_gastos = int(detalle_gastos.get('documentos_exitosos', 0)) if exitosos_gastos > 0: excel_data.append({ 'Organización': cliente.RazonSocial, 'Nombre comercial': cliente.NombreComercial, 'NIT Emisor': cliente.NIT, 'Tipo de documento': 'DocSopAdq', 'Año': anio, 'Mes': mes, 'Cantidad': exitosos_gastos }) logger.info("Exportando - Cliente: {0}, Periodo: {1}, Facturas: {2}, Gastos: {3}".format( cliente.NombreComercial, reporte.Periodo, exitosos_facturas, exitosos_gastos )) except Exception as e: logger.error("Error procesando reporte {0}: {1}".format(reporte.ID, str(e))) import traceback logger.error(traceback.format_exc()) continue if not excel_data: return jsonify({'error': 'No se pudieron procesar los reportes'}), 500 logger.info("Total de filas exportadas: {0}".format(len(excel_data))) # Crear DataFrame con el orden específico de columnas df = pd.DataFrame(excel_data, columns=[ 'Organización', 'Nombre comercial', 'NIT Emisor', 'Tipo de documento', 'Año', 'Mes', 'Cantidad' ]) # Ordenar por Organización, Año, Mes df = df.sort_values(by=['Organización', 'Año', 'Mes'], ascending=True) # Crear archivo Excel en memoria excel_buffer = io.BytesIO() try: with pd.ExcelWriter(excel_buffer, engine='openpyxl') as writer: df.to_excel(writer, sheet_name='Reportes', index=False) # Ajustar ancho de columnas worksheet = writer.sheets['Reportes'] column_widths = { 'A': 40, # Organización 'B': 40, # Nombre comercial 'C': 15, # NIT Emisor 'D': 18, # Tipo de documento 'E': 8, # Año 'F': 8, # Mes 'G': 12 # Cantidad } for col, width in column_widths.items(): worksheet.column_dimensions[col].width = width except Exception as e: logger.error("Error creando Excel: {0}".format(str(e))) return jsonify({'error': 'Error al crear archivo Excel: {0}'.format(str(e))}), 500 excel_buffer.seek(0) # Nombre del archivo con información de filtros filename_parts = ['Reportes'] if cliente_id: cliente_obj = EmpresaCliente.query.get(int(cliente_id)) if cliente_obj: filename_parts.append(cliente_obj.NombreComercial.replace(' ', '_')) if year: filename_parts.append(str(year)) if month: filename_parts.append(str(month).zfill(2)) filename = "{0}_{1}.xlsx".format('_'.join(filename_parts), datetime.now().strftime('%Y%m%d_%H%M%S')) return send_file( excel_buffer, as_attachment=True, download_name=filename, mimetype='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet' ) except Exception as e: logger.error("Error exportando reportes: {0}".format(str(e))) import traceback logger.error(traceback.format_exc()) return jsonify({'error': 'Error interno al exportar: {0}'.format(str(e))}), 500 @app.route('/api/actualizar-estado-facturacion', methods=['POST']) def actualizar_estado_facturacion(): """Actualiza el estado de facturación de un reporte desde el POS""" try: data = request.get_json() logger.info(f"📋 Actualizando estado de facturación - Período: {data.get('periodo')}") # Validar datos requeridos required_fields = ['periodo', 'numero_factura', 'cliente_nit', 'fecha_facturacion'] for field in required_fields: if not data.get(field): return jsonify({ 'success': False, 'error': f'Campo requerido faltante: {field}' }), 400 periodo = data['periodo'] numero_factura = data['numero_factura'] cliente_nit = data['cliente_nit'] fecha_facturacion_str = data['fecha_facturacion'] # Convertir fecha try: fecha_facturacion = datetime.fromisoformat(fecha_facturacion_str.replace('T', ' ')) except: fecha_facturacion = datetime.now() # Buscar el cliente por NIT cliente = EmpresaCliente.query.filter_by(NIT=cliente_nit).first() if not cliente: return jsonify({ 'success': False, 'error': f'Cliente con NIT {cliente_nit} no encontrado' }), 404 # Buscar el reporte específico reporte = ControlReportes.query.filter_by( Periodo=periodo, ClienteID=cliente.ID, Enviado=True ).first() if not reporte: return jsonify({ 'success': False, 'error': f'Reporte para período {periodo} y cliente {cliente.RazonSocial} no encontrado' }), 404 # Actualizar estado de facturación reporte.Facturado = True reporte.FechaFacturacion = fecha_facturacion reporte.NumeroFactura = numero_factura # Actualizar observaciones observaciones_facturacion = f"Facturado desde POS - Factura #{numero_factura} - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}" if reporte.Observaciones: reporte.Observaciones += f" | {observaciones_facturacion}" else: reporte.Observaciones = observaciones_facturacion db.session.commit() logger.info(f"✅ Estado de facturación actualizado - Cliente: {cliente.RazonSocial}, Período: {periodo}, Factura: {numero_factura}") return jsonify({ 'success': True, 'message': f'Estado de facturación actualizado exitosamente', 'cliente': { 'id': cliente.ID, 'razon_social': cliente.RazonSocial, 'nombre_comercial': cliente.NombreComercial, 'nit': cliente.NIT }, 'reporte': { 'id': reporte.ID, 'periodo': reporte.Periodo, 'facturado': reporte.Facturado, 'numero_factura': reporte.NumeroFactura, 'fecha_facturacion': reporte.FechaFacturacion.isoformat() if reporte.FechaFacturacion else None } }) except Exception as e: logger.error(f"❌ Error actualizando estado de facturación: {str(e)}") db.session.rollback() return jsonify({ 'success': False, 'error': f'Error interno: {str(e)}' }), 500 @app.route('/api/reportes-pendientes-facturacion', methods=['GET']) def get_reportes_pendientes_facturacion(): """Obtiene reportes que están pendientes de facturación""" try: # Filtros opcionales year = request.args.get('year') month = request.args.get('month') cliente_id = request.args.get('cliente_id') # Consulta base: reportes enviados pero no facturados query = db.session.query(ControlReportes, EmpresaCliente).join( EmpresaCliente, ControlReportes.ClienteID == EmpresaCliente.ID ).filter( ControlReportes.Enviado == True, ControlReportes.Facturado == False # Solo los no facturados ) # Aplicar filtros if year: if month: period_filter = f"{year}-{str(month).zfill(2)}" query = query.filter(ControlReportes.Periodo == period_filter) else: query = query.filter(ControlReportes.Periodo.like(f"{year}-%")) if cliente_id: query = query.filter(ControlReportes.ClienteID == cliente_id) reportes_pendientes = query.order_by(ControlReportes.Periodo.desc()).all() # Formatear respuesta resultado = [] for reporte, cliente in reportes_pendientes: resultado.append({ 'reporte_id': reporte.ID, 'periodo': reporte.Periodo, 'cliente_nit': cliente.NIT, 'cliente_razon_social': cliente.RazonSocial, 'cliente_nombre_comercial': cliente.NombreComercial, 'valor_total': float(reporte.ValorTotal or 0), 'total_documentos': reporte.TotalDocumentos or 0, 'documentos_exitosos': reporte.DocumentosExitosos or 0, 'fecha_envio': reporte.FechaEnvio.isoformat() if reporte.FechaEnvio else None, 'facturado': reporte.Facturado or False }) logger.info(f"📋 Obtenidos {len(resultado)} reportes pendientes de facturación") return jsonify(resultado) except Exception as e: logger.error(f"❌ Error obteniendo reportes pendientes: {str(e)}") return jsonify({ 'success': False, 'error': str(e) }), 500 @app.route('/api/reportes-facturados', methods=['GET']) def get_reportes_facturados(): """Obtiene reportes que ya han sido facturados""" try: # Filtros opcionales year = request.args.get('year') month = request.args.get('month') cliente_id = request.args.get('cliente_id') # Consulta base: reportes facturados query = db.session.query(ControlReportes, EmpresaCliente).join( EmpresaCliente, ControlReportes.ClienteID == EmpresaCliente.ID ).filter( ControlReportes.Enviado == True, ControlReportes.Facturado == True # Solo los facturados ) # Aplicar filtros if year: if month: period_filter = f"{year}-{str(month).zfill(2)}" query = query.filter(ControlReportes.Periodo == period_filter) else: query = query.filter(ControlReportes.Periodo.like(f"{year}-%")) if cliente_id: query = query.filter(ControlReportes.ClienteID == cliente_id) reportes_facturados = query.order_by(ControlReportes.FechaFacturacion.desc()).all() # Formatear respuesta resultado = [] for reporte, cliente in reportes_facturados: resultado.append({ 'reporte_id': reporte.ID, 'periodo': reporte.Periodo, 'cliente_nit': cliente.NIT, 'cliente_razon_social': cliente.RazonSocial, 'cliente_nombre_comercial': cliente.NombreComercial, 'valor_total': float(reporte.ValorTotal or 0), 'total_documentos': reporte.TotalDocumentos or 0, 'documentos_exitosos': reporte.DocumentosExitosos or 0, 'fecha_envio': reporte.FechaEnvio.isoformat() if reporte.FechaEnvio else None, 'facturado': reporte.Facturado, 'numero_factura': reporte.NumeroFactura, 'fecha_facturacion': reporte.FechaFacturacion.isoformat() if reporte.FechaFacturacion else None }) logger.info(f"💰 Obtenidos {len(resultado)} reportes facturados") return jsonify(resultado) except Exception as e: logger.error(f"❌ Error obteniendo reportes facturados: {str(e)}") return jsonify({ 'success': False, 'error': str(e) }), 500 @app.route('/api/estadisticas-facturacion', methods=['GET']) def get_estadisticas_facturacion(): """Obtiene estadísticas generales de facturación""" try: year = request.args.get('year', datetime.now().year) # Consultar reportes del año reportes_query = ControlReportes.query.filter( ControlReportes.Periodo.like(f"{year}-%"), ControlReportes.Enviado == True ) total_reportes = reportes_query.count() reportes_facturados = reportes_query.filter(ControlReportes.Facturado == True).count() reportes_pendientes = total_reportes - reportes_facturados # Calcular valores valor_total_facturado = db.session.query( db.func.sum(ControlReportes.ValorTotal) ).filter( ControlReportes.Periodo.like(f"{year}-%"), ControlReportes.Enviado == True, ControlReportes.Facturado == True ).scalar() or 0 valor_total_pendiente = db.session.query( db.func.sum(ControlReportes.ValorTotal) ).filter( ControlReportes.Periodo.like(f"{year}-%"), ControlReportes.Enviado == True, ControlReportes.Facturado == False ).scalar() or 0 # Calcular porcentajes porcentaje_facturado = round((reportes_facturados / total_reportes * 100) if total_reportes > 0 else 0, 2) porcentaje_pendiente = round((reportes_pendientes / total_reportes * 100) if total_reportes > 0 else 0, 2) return jsonify({ 'year': year, 'total_reportes': total_reportes, 'reportes_facturados': reportes_facturados, 'reportes_pendientes': reportes_pendientes, 'porcentaje_facturado': porcentaje_facturado, 'porcentaje_pendiente': porcentaje_pendiente, 'valor_total_facturado': float(valor_total_facturado), 'valor_total_pendiente': float(valor_total_pendiente), 'valor_total_general': float(valor_total_facturado + valor_total_pendiente) }) except Exception as e: logger.error(f"❌ Error obteniendo estadísticas: {str(e)}") return jsonify({ 'success': False, 'error': str(e) }), 500 # AGREGAR este nuevo endpoint ANTES de get_reports() @app.route('/api/reports/paginated', methods=['GET']) def get_reports_paginated(): """Obtiene reportes con paginación""" try: # Parámetros de paginación page = int(request.args.get('page', 1)) per_page = int(request.args.get('per_page', 100)) # Validar límites if per_page not in [100, 200, 500, 1000]: per_page = 100 if page < 1: page = 1 # Filtros opcionales cliente_id = request.args.get('cliente_id') year = request.args.get('year') month = request.args.get('month') status = request.args.get('status') search = request.args.get('search', '').strip().lower() # Query base optimizado query = db.session.query( ControlReportes.ID, ControlReportes.Periodo, ControlReportes.ClienteID, ControlReportes.FechaEnvio, ControlReportes.TotalDocumentos, ControlReportes.DocumentosExitosos, ControlReportes.ValorTotal, ControlReportes.TasaExito, ControlReportes.Enviado, EmpresaCliente.RazonSocial, EmpresaCliente.NombreComercial, EmpresaCliente.NIT ).join( EmpresaCliente, ControlReportes.ClienteID == EmpresaCliente.ID ).filter( ControlReportes.Enviado == True ) # Aplicar filtros if cliente_id: query = query.filter(ControlReportes.ClienteID == int(cliente_id)) if year: if month: period_filter = f"{year}-{str(month).zfill(2)}" query = query.filter(ControlReportes.Periodo == period_filter) else: query = query.filter(ControlReportes.Periodo.like(f"{year}-%")) if status: if status == 'completed': query = query.filter(ControlReportes.Enviado == True) elif status == 'pending': query = query.filter(ControlReportes.Enviado == False) # Búsqueda por texto if search: query = query.filter( db.or_( EmpresaCliente.NombreComercial.ilike(f'%{search}%'), EmpresaCliente.RazonSocial.ilike(f'%{search}%'), EmpresaCliente.NIT.ilike(f'%{search}%'), ControlReportes.Periodo.ilike(f'%{search}%') ) ) # Ordenar alfabéticamente por nombre comercial query = query.order_by(EmpresaCliente.NombreComercial.asc(), ControlReportes.Periodo.desc()) # Contar total total = query.count() # Calcular páginas total_pages = (total + per_page - 1) // per_page # Aplicar paginación offset = (page - 1) * per_page reportes = query.offset(offset).limit(per_page).all() # Formato de respuesta result = [] for r in reportes: result.append({ 'id': r.ID, 'periodo': r.Periodo, 'cliente_id': r.ClienteID, 'cliente': { 'razon_social': r.RazonSocial, 'nombre_comercial': r.NombreComercial, 'nit': r.NIT }, 'fecha_envio': r.FechaEnvio.isoformat() if r.FechaEnvio else None, 'total_documentos': r.TotalDocumentos or 0, 'documentos_exitosos': r.DocumentosExitosos or 0, 'valor_total': float(r.ValorTotal or 0), 'tasa_exito': float(r.TasaExito or 0), 'status': 'completed' if r.Enviado else 'pending' }) return jsonify({ 'success': True, 'data': result, 'pagination': { 'page': page, 'per_page': per_page, 'total': total, 'total_pages': total_pages, 'has_prev': page > 1, 'has_next': page < total_pages } }) except Exception as e: logger.error(f"Error obteniendo reportes paginados: {str(e)}") return jsonify({ 'success': False, 'error': str(e) }), 500 # Manejadores de errores (mantenemos los existentes) @app.errorhandler(404) def not_found(error): logger.warning(f"404 error: {request.url}") if request.path.startswith('/api/'): return jsonify({'error': 'Endpoint no encontrado'}), 404 if '/static/' in request.path: return 'Archivo no encontrado', 404 return render_template('404.html') if os.path.exists('/var/www/html/TransmisionesMig/templates/404.html') else ('Página no encontrada', 404) @app.errorhandler(500) def internal_error(error): logger.error(f"500 error: {str(error)} - URL: {request.url}") db.session.rollback() if request.path.startswith('/api/'): return jsonify({'error': 'Error interno del servidor'}), 500 return render_template('500.html') if os.path.exists('/var/www/html/TransmisionesMig/templates/500.html') else ('Error interno del servidor', 500) @app.errorhandler(Exception) def handle_exception(e): """Maneja excepciones no capturadas""" logger.error(f"Excepción no capturada: {str(e)} - URL: {request.url}") db.session.rollback() return jsonify({'error': 'Error interno del servidor'}), 500 # Función de inicialización (solo para Apache/WSGI) def initialize_app(): """Inicialización de la aplicación para producción""" logger.info("Aplicación cargada para producción con Apache - Versión con Clientes") try: with app.app_context(): db.session.execute(text('SELECT 1')) logger.info("✅ Conexión a base de datos exitosa en producción") crear_tablas() except Exception as e: logger.error(f"❌ Error de conexión a base de datos en producción: {str(e)}") # Solo ejecutar inicialización si NO estamos en modo de desarrollo if __name__ != "__main__": with app.app_context(): initialize_app() # Función principal SOLO para desarrollo if __name__ == "__main__": print("=" * 50) print("🚨 MODO DESARROLLO - SISTEMA CON CLIENTES") print("🚨 Esta aplicación maneja múltiples empresas cliente") print("=" * 50) logger.info("Iniciando aplicación con gestión de clientes...") try: with app.app_context(): db.session.execute(text('SELECT 1')) logger.info("✅ Conexión a base de datos exitosa") crear_tablas() except Exception as e: logger.warning(f"❌ No se pudo conectar a la base de datos: {str(e)}") print("\n" + "=" * 50) print("URLs de prueba:") print("- http://localhost:8000/test") print("- http://localhost:8000/test-db") print("- http://localhost:8000/init-db") print("- http://localhost:8000/api/status") print("- http://localhost:8000/api/clientes") print("=" * 50 + "\n") app.run( host='0.0.0.0', port=8000, debug=True, threaded=True ) # Variable para WSGI (Apache) application = app