import os import csv import io from datetime import datetime, date from typing import List, Dict, Any, Optional from src.models.invoice import Invoice from src.models.company import Company from src.models.user import db class Report606Service: """Servicio para generar reportes en formato 606 de la DGII""" def __init__(self): self.field_separator = "|" self.line_separator = "\n" def generate_606_report(self, company_id: int, year: int, month: int, include_pending: bool = False) -> Dict[str, Any]: """ Genera el reporte 606 para una empresa en un período específico Args: company_id: ID de la empresa year: Año del reporte month: Mes del reporte (1-12) include_pending: Si incluir facturas pendientes de aprobación Returns: Dict con el contenido del reporte y metadatos """ try: # Obtener empresa company = Company.query.get(company_id) if not company: raise ValueError("Empresa no encontrada") # Obtener facturas del período invoices = self._get_invoices_for_period(company_id, year, month, include_pending) if not invoices: return { 'success': False, 'message': 'No se encontraron facturas para el período especificado', 'total_invoices': 0, 'total_amount': 0, 'total_itbis': 0 } # Generar contenido del archivo 606 report_content = self._generate_606_content(company, invoices, year, month) # Calcular totales totals = self._calculate_totals(invoices) # Generar nombre del archivo filename = f"606_{company.rnc}_{year}{month:02d}.txt" return { 'success': True, 'filename': filename, 'content': report_content, 'company': company.to_dict(), 'period': f"{month:02d}/{year}", 'total_invoices': len(invoices), 'total_amount': totals['total_amount'], 'total_itbis': totals['total_itbis'], 'invoices_summary': [invoice.to_dict() for invoice in invoices] } except Exception as e: return { 'success': False, 'error': str(e), 'total_invoices': 0, 'total_amount': 0, 'total_itbis': 0 } def _get_invoices_for_period(self, company_id: int, year: int, month: int, include_pending: bool) -> List[Invoice]: """Obtiene las facturas de una empresa para un período específico""" query = Invoice.query.filter( Invoice.company_id == company_id, db.extract('year', Invoice.invoice_date) == year, db.extract('month', Invoice.invoice_date) == month ) if not include_pending: query = query.filter(Invoice.status == 'approved') else: query = query.filter(Invoice.status.in_(['approved', 'pending'])) return query.order_by(Invoice.invoice_date, Invoice.invoice_number).all() def _generate_606_content(self, company: Company, invoices: List[Invoice], year: int, month: int) -> str: """Genera el contenido del archivo 606 según especificaciones DGII""" lines = [] # Línea de encabezado (606) header = self._generate_header_line(company, year, month, len(invoices)) lines.append(header) # Líneas de detalle (607) for invoice in invoices: detail_line = self._generate_detail_line(invoice) lines.append(detail_line) return self.line_separator.join(lines) def _generate_header_line(self, company: Company, year: int, month: int, total_records: int) -> str: """Genera la línea de encabezado del formato 606""" # Formato: 606|RNC|PERIODO|CANTIDAD_REGISTROS period = f"{year}{month:02d}" fields = [ "606", # Tipo de registro company.rnc, # RNC de la empresa period, # Período (YYYYMM) str(total_records) # Cantidad de registros de detalle ] return self.field_separator.join(fields) def _generate_detail_line(self, invoice: Invoice) -> str: """Genera una línea de detalle del formato 607""" # Mapear tipo de identificación a código numérico id_type_map = { 'RNC': '1', 'CEDULA': '2', 'PASAPORTE': '3' } id_type_code = id_type_map.get(invoice.supplier_id_type, '1') # Formato según especificaciones DGII para el formato 607 fields = [ "607", # Tipo de registro invoice.supplier_rnc or "", # RNC del proveedor id_type_code, # Tipo de identificación (1=RNC, 2=Cédula) invoice.ncf or "", # NCF invoice.ncf_modified or "", # NCF modificado self._format_date(invoice.invoice_date), # Fecha de comprobante self._format_date(invoice.payment_date), # Fecha de pago self._format_amount(invoice.invoiced_amount), # Monto facturado self._format_amount(invoice.itbis_invoiced), # ITBIS facturado self._format_amount(invoice.itbis_retained), # ITBIS retenido self._format_amount(invoice.itbis_perceived), # ITBIS percibido self._format_amount(invoice.income_retention), # Retención renta self._format_amount(invoice.isr_perceived), # ISR percibido self._format_amount(invoice.selective_consumption_tax), # Impuesto selectivo al consumo self._format_amount(invoice.other_taxes), # Otros impuestos/tasas self._format_amount(invoice.legal_tip), # Propina legal invoice.expense_type or "06" # Tipo de gasto ] return self.field_separator.join(fields) def _format_date(self, date_value: Optional[date]) -> str: """Formatea una fecha para el reporte 606 (YYYYMMDD)""" if date_value is None: return "" if isinstance(date_value, str): try: date_value = datetime.strptime(date_value, '%Y-%m-%d').date() except ValueError: return "" return date_value.strftime('%Y%m%d') def _format_amount(self, amount: Optional[float]) -> str: """Formatea un monto para el reporte 606 (sin decimales, sin separadores)""" if amount is None or amount == 0: return "0" # Convertir a centavos (multiplicar por 100) y formatear sin decimales amount_cents = int(round(amount * 100)) return str(amount_cents) def _calculate_totals(self, invoices: List[Invoice]) -> Dict[str, float]: """Calcula los totales de las facturas""" total_amount = sum(invoice.invoiced_amount or 0 for invoice in invoices) total_itbis = sum(invoice.itbis_invoiced or 0 for invoice in invoices) return { 'total_amount': total_amount, 'total_itbis': total_itbis } def generate_csv_report(self, company_id: int, year: int, month: int, include_pending: bool = False) -> Dict[str, Any]: """Genera un reporte en formato CSV para análisis""" try: company = Company.query.get(company_id) if not company: raise ValueError("Empresa no encontrada") invoices = self._get_invoices_for_period(company_id, year, month, include_pending) if not invoices: return { 'success': False, 'message': 'No se encontraron facturas para el período especificado' } # Crear CSV en memoria output = io.StringIO() writer = csv.writer(output) # Escribir encabezados headers = [ 'Número Factura', 'NCF', 'Proveedor', 'RNC Proveedor', 'Fecha Factura', 'Fecha Pago', 'Monto Facturado', 'ITBIS Facturado', 'ITBIS Retenido', 'Retención Renta', 'Tipo Gasto', 'Descripción Gasto', 'Estado' ] writer.writerow(headers) # Escribir datos for invoice in invoices: row = [ invoice.invoice_number or '', invoice.ncf or '', invoice.supplier_name or '', invoice.supplier_rnc or '', invoice.invoice_date.strftime('%Y-%m-%d') if invoice.invoice_date else '', invoice.payment_date.strftime('%Y-%m-%d') if invoice.payment_date else '', f"{invoice.invoiced_amount:.2f}" if invoice.invoiced_amount else '0.00', f"{invoice.itbis_invoiced:.2f}" if invoice.itbis_invoiced else '0.00', f"{invoice.itbis_retained:.2f}" if invoice.itbis_retained else '0.00', f"{invoice.income_retention:.2f}" if invoice.income_retention else '0.00', invoice.expense_type or '', invoice.expense_description or '', invoice.status or '' ] writer.writerow(row) csv_content = output.getvalue() output.close() filename = f"facturas_{company.rnc}_{year}{month:02d}.csv" return { 'success': True, 'filename': filename, 'content': csv_content, 'total_invoices': len(invoices) } except Exception as e: return { 'success': False, 'error': str(e) } def get_period_summary(self, company_id: int, year: int, month: int) -> Dict[str, Any]: """Obtiene un resumen del período para el dashboard""" try: invoices = self._get_invoices_for_period(company_id, year, month, include_pending=True) # Separar por estado approved_invoices = [inv for inv in invoices if inv.status == 'approved'] pending_invoices = [inv for inv in invoices if inv.status == 'pending'] rejected_invoices = [inv for inv in invoices if inv.status == 'rejected'] # Calcular totales por estado approved_totals = self._calculate_totals(approved_invoices) pending_totals = self._calculate_totals(pending_invoices) # Resumen por tipo de gasto expense_summary = self._get_expense_type_summary(approved_invoices) # Resumen por proveedor supplier_summary = self._get_supplier_summary(approved_invoices) return { 'success': True, 'period': f"{month:02d}/{year}", 'totals': { 'approved': { 'count': len(approved_invoices), 'amount': approved_totals['total_amount'], 'itbis': approved_totals['total_itbis'] }, 'pending': { 'count': len(pending_invoices), 'amount': pending_totals['total_amount'], 'itbis': pending_totals['total_itbis'] }, 'rejected': { 'count': len(rejected_invoices) } }, 'expense_summary': expense_summary, 'supplier_summary': supplier_summary[:10] # Top 10 proveedores } except Exception as e: return { 'success': False, 'error': str(e) } def _get_expense_type_summary(self, invoices: List[Invoice]) -> List[Dict[str, Any]]: """Genera resumen por tipo de gasto""" expense_totals = {} for invoice in invoices: expense_type = invoice.expense_type or '06' if expense_type not in expense_totals: expense_totals[expense_type] = { 'type': expense_type, 'description': invoice.expense_description or 'Sin descripción', 'count': 0, 'amount': 0 } expense_totals[expense_type]['count'] += 1 expense_totals[expense_type]['amount'] += invoice.invoiced_amount or 0 # Ordenar por monto descendente return sorted(expense_totals.values(), key=lambda x: x['amount'], reverse=True) def _get_supplier_summary(self, invoices: List[Invoice]) -> List[Dict[str, Any]]: """Genera resumen por proveedor""" supplier_totals = {} for invoice in invoices: supplier_key = f"{invoice.supplier_rnc}_{invoice.supplier_name}" if supplier_key not in supplier_totals: supplier_totals[supplier_key] = { 'supplier_name': invoice.supplier_name or 'Sin nombre', 'supplier_rnc': invoice.supplier_rnc or 'Sin RNC', 'count': 0, 'amount': 0 } supplier_totals[supplier_key]['count'] += 1 supplier_totals[supplier_key]['amount'] += invoice.invoiced_amount or 0 # Ordenar por monto descendente return sorted(supplier_totals.values(), key=lambda x: x['amount'], reverse=True)