import requests import json import mysql.connector import re import os import sys import time from multiprocessing import Process import random # ---------------- CONSTANTES ---------------- LOCKFILE = "/tmp/procesar_factura.lock" MAX_PROCESOS = 10 DELAY_ENTRE_PROCESOS = 10 # segundos mensajes = [] errores = [] facturas_procesadas = [] # ---------------- LOCKFILE ---------------- def crear_lockfile(): if os.path.exists(LOCKFILE): print("❌ Lockfile ya existe. Proceso en ejecución.") sys.exit(0) with open(LOCKFILE, "w") as f: f.write(str(os.getpid())) print("🔒 Lockfile creado.") def eliminar_lockfile(): if os.path.exists(LOCKFILE): os.remove(LOCKFILE) print("✅ Lockfile eliminado.") # ---------------- PROCESO HIJO ---------------- def procesar_factura_individual(factura): try: config = { "host": "localhost", "database": "iohanes_14", "user": "iohanes_14", "password": "bSh4_(+L*)sY" } conn = mysql.connector.connect(**config) cursor = conn.cursor(dictionary=True) factura_id = factura["id"] empresa_emite = factura["razon"] id_clave_corta = factura["id_clave_corta"] total_iva = factura["total_iva"] print(f"\n📄 Procesando Factura ID {factura_id} - Razon {empresa_emite} - Total IVA {total_iva:.2f}") empresa_compra = obtener_empresa_compra(cursor, id_clave_corta) if not empresa_compra: print("⚠️ Saltando factura (empresa_compra no encontrada).") return descripcion, conceptos = obtener_datos_factura(cursor, empresa_emite, empresa_compra) respuesta = enviar_a_deepseek(descripcion, conceptos, total_iva) datos = extraer_json(respuesta) if datos and "escenario" in datos: guardar_escenario_factura(cursor, conn, factura_id, datos["escenario"]) # if datos: # datos = aleatorizar_precios_factura(datos, total_iva) if datos and "conceptos" in datos: insertar_conceptos(cursor, conn, datos["conceptos"], factura_id) marcar_factura_como_procesada(cursor, conn, factura_id) else: print(f"❌ Respuesta inválida para Factura ID {factura_id}. No procesada.") conn.close() except mysql.connector.Error as err: print(f"❌ Error de base de datos en Factura ID {factura['id']}: {err}") # ---------------- FUNCIONES AUXILIARES ---------------- def obtener_facturas_pendientes(): config = { "host": "localhost", "database": "iohanes_14", "user": "iohanes_14", "password": "bSh4_(+L*)sY" } conn = mysql.connector.connect(**config) cursor = conn.cursor(dictionary=True) #hoy = time.strftime("%Y-%m-%d") hoy = '2025-08-08' cursor.execute( "SELECT id, razon, id_clave_corta, total_iva " "FROM facturas " "WHERE (cuadrado IS NULL or cuadrado = '') AND nombre = 13 AND fecha >= %s " "ORDER BY id ASC LIMIT %s", (hoy, MAX_PROCESOS,) ) facturas = cursor.fetchall() conn.close() return facturas def obtener_empresa_compra(cursor, id_clave_corta): cursor.execute("SELECT rfc FROM claves_cortas WHERE id = %s", (id_clave_corta,)) rfc_data = cursor.fetchone() if not rfc_data: print(f"❌ No RFC para id_clave_corta {id_clave_corta}") return None rfc = rfc_data["rfc"] cursor.execute("SELECT id FROM empresas WHERE rfc = %s", (rfc,)) empresa_data = cursor.fetchone() if not empresa_data: print(f"❌ No empresa con RFC {rfc}") return None return empresa_data["id"] def obtener_datos_factura(cursor, empresa_emite, empresa_compra): cursor.execute("SELECT descripcion FROM empresas WHERE id = %s", (empresa_compra,)) desc_data = cursor.fetchone() descripcion = desc_data["descripcion"] if desc_data else "" cursor.execute( "SELECT id, concepto, clave_sat, precio_min, precio_max " "FROM conceptos_depu cd " "WHERE razon = %s " "AND cd.clave_sat IN (SELECT clave_sat FROM conceptos_compra cc WHERE cc.id_empresa = %s) LIMIT 8", (empresa_emite, empresa_compra) ) conceptos = cursor.fetchall() conceptos_cadena = ", ".join([ f"{{'id': {c['id']}, 'nombre': '{c['concepto']}', 'clave_sat': '{c['clave_sat']}', " f"'precio_min': {c['precio_min']}, 'precio_max': {c['precio_max']}}}" for c in conceptos ]) return descripcion, conceptos_cadena def extraer_json(respuesta): try: contenido = respuesta["choices"][0]["message"]["content"] contenido = re.sub(r"^```json\n|\n```$", "", contenido.strip()) return json.loads(contenido) except (KeyError, IndexError, json.JSONDecodeError) as e: print(f"❌ Error al extraer JSON: {e}") print(f"Respuesta completa: {respuesta}") return None def guardar_escenario_factura(cursor, conn, factura_id, escenario): try: cursor.execute( "UPDATE facturas SET ecenario = %s WHERE id = %s", (escenario, factura_id) ) conn.commit() print(f"📝 Escenario guardado en factura ID {factura_id}: {escenario}") facturas_procesadas.append(factura_id) except mysql.connector.Error as e: print(f"❌ Error al guardar escenario en factura ID {factura_id}: {e}") conn.rollback() def enviar_a_deepseek(descripcion, conceptos, total_iva): url = "https://api.deepseek.com/chat/completions" estructura = { "conceptos": [ { "id": 1, "nombre": "Concepto 1", "clave_sat": "123", "cantidad": 10, "precio_unitario": 2, "precio_unitario_iva": 2.32, "total": 20, "total_iva": 23.20, "importe": 20 } ], "subtotal": 20, "iva": 16, "total_iva": 23.20, "total": 20, "moneda": "MXN", "escenario": "Una construcción de inmuebles", } prompt = ( f"Tomando en cuenta los conceptos {conceptos} con sus precios máximos y mínimos, los máximos y mínimos se refieren al precio contando el IVA, es decir para el precio sin IVA tienes que restar el IVA. en caso de ser servicios, intenta generar un precio cerrado, que simule un servicio, en caso de ser productos puedes usar cualquier precio que se encuentre en los limites, usa los productos que no son servicios para rellenar los montos faltantes para completar la factura." f"Altera un poco el nombre de los conceptos para ser más específico sin cambiar la naturaleza. Jamas uses cantidades menores a 1., en caso de que no se logre completar el total de la factura puedes modificar los mínimos y máximos para lograrlo, pero solo en caso de que no se pueda con la información proporcionada" f"No uses marcas o modelos, solo descripciones genéricas. Usa la clave sat especifica proporcionada unicamente, no alteres para nada ese dato." f"Asegúrate que la factura sea consistente y complementaria. usa las claves saras proporcionadas, no inventes claves sat ni conceptos. si no te son proporcionados los conceptos suficientes retorna una factura en 0" f"El escenario es el la situación ficticia en la que pensaste para crear la factura que sea acorde a los conceptos que decidiste usar. siempre llega al total, no uses aproximados ni redondeos, el total debe ser exacto." f"El total con IVA debe ser de {total_iva:.2f} pesos mexicanos exactos. " f"Usa la estructura {json.dumps(estructura, indent=2)} para mostrar la respuesta sin texto adicional." ) payload = json.dumps({ "model": "deepseek-reasoner", "messages": [ {"role": "system", "content": "Eres un sistema de facturación"}, {"role": "user", "content": prompt} ], "stream": False }) headers = { 'Content-Type': 'application/json', 'Authorization': 'Bearer sk-0bd6efe132a846adbb2b00c9cfeb7e2b' # <- PON TU API KEY AQUÍ } response = requests.post(url, headers=headers, data=payload) if response.status_code == 200: return response.json() else: print(f"❌ Error API DeepSeek: {response.status_code} - {response.text}") return None def insertar_conceptos(cursor, conn, conceptos, factura_id): for c in conceptos: try: cursor.execute( """ INSERT INTO facturas_pre (concepto, clave_sat, cantidad, precio_unitario, precio_unitario_iva, total, total_iva, id_factura) VALUES (%s, %s, %s, %s, %s, %s, %s, %s) """, ( c["nombre"], c["clave_sat"], c["cantidad"], c["precio_unitario"], c["precio_unitario_iva"], c["total"], c["total_iva"], factura_id ) ) conn.commit() print(f"✅ Concepto insertado: {c['nombre']}") except mysql.connector.Error as e: print(f"❌ Error al insertar concepto: {e}") conn.rollback() def marcar_factura_como_procesada(cursor, conn, factura_id): cursor.execute("UPDATE facturas SET cuadrado = 1 WHERE id = %s", (factura_id,)) conn.commit() print(f"🟢 Factura ID {factura_id} marcada como procesada.") def aleatorizar_precios_factura(json_data, total_iva_objetivo): if not json_data or "conceptos" not in json_data: print("⚠️ JSON inválido para aleatorizar precios.") return json_data # Retornar sin cambios si no hay datos conceptos = json_data["conceptos"] subtotal = 0 iva_total = 0 total_final = 0 for c in conceptos: # Generar precio unitario con IVA aleatorio (±10% del original) precio_unitario_iva = round( random.uniform(c["precio_unitario_iva"] * 0.9, c["precio_unitario_iva"] * 1.1), 2 ) # Calcular precio sin IVA precio_unitario = round(precio_unitario_iva / 1.16, 2) cantidad = c.get("cantidad", 1) total_concepto = round(precio_unitario * cantidad, 2) total_iva_concepto = round(precio_unitario_iva * cantidad, 2) # Actualizar valores en el concepto c["precio_unitario"] = precio_unitario c["precio_unitario_iva"] = precio_unitario_iva c["total"] = total_concepto c["total_iva"] = total_iva_concepto c["importe"] = total_concepto subtotal += total_concepto iva_total += round(total_concepto * 0.16, 2) total_final += total_iva_concepto # Ajustar el total final para que coincida exactamente con total_iva_objetivo diferencia = round(total_iva_objetivo - total_final, 2) if abs(diferencia) > 0.01: print(f"⚖️ Ajustando diferencia de {diferencia:.2f} al primer concepto.") conceptos[0]["total_iva"] += diferencia conceptos[0]["total"] = round(conceptos[0]["total_iva"] / 1.16, 2) # Actualizar totales en el JSON subtotal = round(sum(c["total"] for c in conceptos), 2) total_final = round(sum(c["total_iva"] for c in conceptos), 2) iva_total = round(total_final - subtotal, 2) json_data["subtotal"] = subtotal json_data["iva"] = iva_total json_data["total_iva"] = total_final json_data["total"] = total_final print(f"🎲 Precios aleatorizados. Total IVA ajustado a {total_final:.2f}") return json_data # ---------------- MAIN ---------------- if __name__ == "__main__": crear_lockfile() procesos = [] try: facturas_pendientes = obtener_facturas_pendientes() if not facturas_pendientes: print("✅ No hay facturas pendientes.") else: for i, factura in enumerate(facturas_pendientes): p = Process(target=procesar_factura_individual, args=(factura,)) p.start() procesos.append(p) print(f"🚀 Proceso {i+1} iniciado para Factura ID {factura['id']}") time.sleep(DELAY_ENTRE_PROCESOS) for p in procesos: p.join() print("✅ Proceso hijo terminado.") finally: #imprimir como json eliminar_lockfile()