367 lines
12 KiB
Python
367 lines
12 KiB
Python
import json
|
|
import time
|
|
from datetime import datetime, timedelta
|
|
from dotenv import load_dotenv
|
|
|
|
import logger_utilities
|
|
import sunat_manager
|
|
import hana_manager
|
|
import sapb1_manager
|
|
from constants import ESTADOS_COMPROBANTE
|
|
from hana_manager import actualizar_documento_sap
|
|
|
|
# ============================================================================
|
|
|
|
config = {}
|
|
|
|
|
|
# ============================================================================
|
|
def cargar_programacion() -> dict:
|
|
global config
|
|
with open("scheduler.json", "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
data["HORARIOS"] = [tuple(h) for h in data["HORARIOS"]]
|
|
config = data
|
|
return config
|
|
|
|
|
|
def obtener_horarios() -> list:
|
|
return config.get("HORARIOS", [])
|
|
|
|
|
|
def obtener_dias_activos() -> list:
|
|
return config.get("DIAS_ACTIVOS", [])
|
|
|
|
|
|
# ============================================================================
|
|
def obtener_intervalo(dt: datetime) -> tuple[int, str] | None:
|
|
minutos = dt.hour * 60 + dt.minute
|
|
|
|
for hi, mi, hf, mf, intervalo, modo in obtener_horarios():
|
|
inicio = hi * 60 + mi
|
|
fin = hf * 60 + mf
|
|
|
|
if inicio <= minutos < fin:
|
|
return (intervalo, modo)
|
|
|
|
return None
|
|
|
|
# ============================================================================
|
|
def segundos_hasta_proximo_inicio(dt: datetime) -> int:
|
|
horarios = obtener_horarios()
|
|
if horarios:
|
|
hi, mi = horarios[0][0], horarios[0][1]
|
|
else:
|
|
hi, mi = 6, 30
|
|
|
|
inicio = dt.replace(hour=hi, minute=mi, second=0, microsecond=0)
|
|
|
|
if dt >= inicio:
|
|
inicio += timedelta(days=1)
|
|
|
|
return int((inicio - dt).total_seconds())
|
|
|
|
# ============================================================================
|
|
def segundos_hasta_proximo_dia_activo(dt: datetime) -> int:
|
|
horarios = obtener_horarios()
|
|
hi, mi = (horarios[0][0], horarios[0][1]) if horarios else (6, 30)
|
|
dias_activos = obtener_dias_activos()
|
|
|
|
for dias in range(1, 8):
|
|
prox = dt + timedelta(days=dias)
|
|
if prox.weekday() in dias_activos:
|
|
target = prox.replace(hour=hi, minute=mi, second=0, microsecond=0)
|
|
return int((target - dt).total_seconds())
|
|
return 86400
|
|
|
|
# ============================================================================
|
|
def cargar_empresas():
|
|
with open("empresas.json", "r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
|
|
# ============================================================================
|
|
def enviar_mensaje_sl(usercode, doc, resultado, schema):
|
|
estado_cp = resultado.get("data", {}).get("estadoCp", "")
|
|
estado_texto = ESTADOS_COMPROBANTE.get(estado_cp, "DESCONOCIDO")
|
|
|
|
subject = f"❌ SUNAT - {doc['Indicator']}-{doc['FolioPref']}-{doc['FolioNum']}"
|
|
observaciones = resultado.get("data", {}).get("observaciones", [""])[0] or ""
|
|
|
|
text = (
|
|
f"El documento {doc['Indicator']}-{doc['FolioPref']}-{doc['FolioNum']} "
|
|
f"de {schema} entrego el siguiente estado por SUNAT.\n"
|
|
f"Estado: {estado_texto} (código: {estado_cp})\n"
|
|
f"RUC Proveedor: {doc.get('LicTradNum', 'N/A')}\n"
|
|
f"Razón Social: {doc.get('CardName', 'N/A')}\n"
|
|
f"Mensaje SUNAT: {resultado.get('message', 'N/A')}\n"
|
|
)
|
|
|
|
if observaciones:
|
|
text += f"Observaciones SUNAT: {observaciones}\n"
|
|
|
|
payload = {
|
|
"MessageDataColumns": [
|
|
{
|
|
"ColumnName": "Numero",
|
|
"Link": "tNO",
|
|
"MessageDataLines": [
|
|
{
|
|
"Object":"",
|
|
"ObjectKey":"",
|
|
"Value":f"{doc['Indicator']}-{doc['FolioPref']}-{doc['FolioNum']} "
|
|
}
|
|
]
|
|
},
|
|
{
|
|
"ColumnName": "Documento",
|
|
"Link": "tYES",
|
|
"MessageDataLines": [
|
|
{
|
|
"Object": str(doc.get("ObjType", "18")),
|
|
"ObjectKey": str(doc["DocEntry"]),
|
|
"Value": str(doc["DocNum"])
|
|
}
|
|
]
|
|
},
|
|
{
|
|
"ColumnName": "Razón Social / Nombre",
|
|
"Link": "tNO",
|
|
"MessageDataLines": [
|
|
{
|
|
"Object": "",
|
|
"ObjectKey": "",
|
|
"Value": f"{doc.get('CardName', 'N/A')}"
|
|
}
|
|
]
|
|
}
|
|
],
|
|
"RecipientCollection": [
|
|
{
|
|
"SendInternal": "tYES",
|
|
"UserCode": usercode
|
|
}
|
|
],
|
|
"Subject": subject,
|
|
"Text": text
|
|
}
|
|
|
|
|
|
|
|
print(f" ┌─ 📩 Enviando Messages a {usercode} ───────────")
|
|
# for line in json.dumps(payload, indent=2).splitlines():
|
|
# print(f" │ {line}")
|
|
sapb1_manager.send_message(payload)
|
|
print(f" │ ✅ Mensaje enviado a {usercode}")
|
|
print(f" └───────────────────────────────────────────────")
|
|
|
|
# ============================================================================
|
|
def procesar_empresa(schema, datos_empresa, modo):
|
|
print(f"\n{'='*60}")
|
|
print(f"🏢 Procesando empresa: {schema} [{modo}]")
|
|
print(f"{'='*60}")
|
|
|
|
if modo == "rapido":
|
|
documentos = hana_manager.obtener_documentos_creados_actualizados_hace_60_minutos(schema, 60)
|
|
else:
|
|
documentos = hana_manager.obtener_documentos_fecha_emision_ultimos_8_dias(schema)
|
|
if not documentos:
|
|
print(f"⚠️ No se encontraron documentos para {schema}")
|
|
return 0, 0
|
|
|
|
## documentos = hana_manager.obtener_documentos_creados_hace_x_dias(schema, 8)
|
|
|
|
print(f"📄 Documentos encontrados: {len(documentos)}")
|
|
|
|
print(f"🔑 Solicitando token SUNAT...")
|
|
token = sunat_manager.token(datos_empresa)
|
|
if not token:
|
|
print(f"❌ No se obtuvo token para {schema}")
|
|
return 0, len(documentos)
|
|
|
|
print(f"✅ Token obtenido correctamente")
|
|
|
|
sapb1_manager.login()
|
|
print(f"✅ Sesión SL establecida para {schema}")
|
|
|
|
ruc_empresa = datos_empresa["Ruc"]
|
|
procesados = 0
|
|
errores = 0
|
|
|
|
try:
|
|
for idx, doc in enumerate(documentos, start=1):
|
|
print(f"\n--- [{idx}/{len(documentos)}] {doc['FolioPref']}-{doc['FolioNum']} ---")
|
|
|
|
try:
|
|
payload = {
|
|
"numRuc": doc["LicTradNum"],
|
|
"codComp": "R1" if doc["Indicator"] == "02" else doc["Indicator"],
|
|
"numeroSerie": doc["FolioPref"],
|
|
"numero": doc["FolioNum"],
|
|
"fechaEmision": doc["TaxDate"],
|
|
}
|
|
|
|
if doc["FolioPref"] and doc["FolioPref"][0].isalpha():
|
|
payload["monto"] = str(doc["DocTotal"])
|
|
|
|
print(f" ┌─ 📤 Enviando a SUNAT {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}─────────────────────")
|
|
for line in json.dumps(payload, indent=2).splitlines():
|
|
print(f" │ {line}")
|
|
print(f" └────────────────────────────────────────────")
|
|
resultado = sunat_manager.validar_comprobante(token, ruc_empresa, payload)
|
|
print(f" ┌─ 📥 Respuesta de SUNAT {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}───────────────────")
|
|
for line in json.dumps(resultado, indent=2).splitlines():
|
|
print(f" │ {line}")
|
|
print(f" └────────────────────────────────────────────")
|
|
|
|
registro = {
|
|
"objtype": str(doc["ObjType"]),
|
|
"companydb": schema,
|
|
"fieldvalue": doc["DocEntry"],
|
|
}
|
|
|
|
actualizar_documento_sap(schema, registro, resultado)
|
|
|
|
if resultado.get("success"):
|
|
estado_cp = resultado.get("data", {}).get("estadoCp", "")
|
|
print(f"✅ OK - Estado CP: {estado_cp}")
|
|
|
|
if estado_cp not in ["1","3"] and doc.get("UserCode"):
|
|
enviar_mensaje_sl(doc["UserCode"], doc, resultado, schema)
|
|
|
|
else:
|
|
print(f"⚠️ Falló - {resultado.get('message', 'Error desconocido')}")
|
|
|
|
procesados += 1
|
|
|
|
if idx < len(documentos):
|
|
time.sleep(2)
|
|
|
|
except Exception as e:
|
|
print(f"❌ Error procesando documento: {e}")
|
|
errores += 1
|
|
finally:
|
|
sapb1_manager.logout()
|
|
print(f"🔒 Sesión SL cerrada para {schema}")
|
|
|
|
return procesados, errores
|
|
|
|
# ============================================================================
|
|
def procesar_todas_empresas(inicio_ciclo, modo):
|
|
|
|
empresas = cargar_empresas()
|
|
print(f"📋 Empresas cargadas: {len(empresas)}")
|
|
|
|
total_procesados = 0
|
|
total_errores = 0
|
|
|
|
for empresa in empresas:
|
|
schema = empresa["CompanyDB"]
|
|
|
|
if empresa.get("Locked") == "Y":
|
|
print(f"\n⏭️ Empresa {schema} está bloqueada (LOCKED=Y), saltando...")
|
|
continue
|
|
|
|
sapb1_manager._sap_credentials = {
|
|
"company": empresa["CompanyDB"],
|
|
"username": empresa["UserName"],
|
|
"password": empresa["Password"],
|
|
}
|
|
|
|
try:
|
|
p, e = procesar_empresa(schema, empresa, modo)
|
|
total_procesados += p
|
|
total_errores += e
|
|
except Exception as ex:
|
|
print(f"❌ Error procesando empresa {schema}: {ex}")
|
|
total_errores += 1
|
|
|
|
duracion_ciclo = datetime.now() - inicio_ciclo
|
|
print(f"\n📊 CICLO COMPLETADO en {str(duracion_ciclo).split('.')[0]}")
|
|
print(f" ✅ {total_procesados} procesados | ❌ {total_errores} errores")
|
|
return total_procesados, total_errores
|
|
|
|
# ============================================================================
|
|
def main():
|
|
|
|
log_path = logger_utilities.setup_logger()
|
|
_ultima_limpieza = None
|
|
|
|
load_dotenv(dotenv_path=".env")
|
|
|
|
cargar_programacion()
|
|
|
|
print("=" * 60)
|
|
print("🚀 Demonio iniciado")
|
|
print(f"📄 Log guardado en: {log_path}")
|
|
print("Servicio iniciado")
|
|
print("=" * 60)
|
|
|
|
while True:
|
|
|
|
inicio_ciclo = datetime.now()
|
|
|
|
cargar_programacion()
|
|
|
|
logger_utilities.rotar_log()
|
|
|
|
if _ultima_limpieza != inicio_ciclo.date():
|
|
logger_utilities.limpiar_logs_antiguos(dias=7)
|
|
_ultima_limpieza = inicio_ciclo.date()
|
|
|
|
if inicio_ciclo.weekday() not in obtener_dias_activos():
|
|
dormir = segundos_hasta_proximo_dia_activo(inicio_ciclo)
|
|
print(
|
|
f"\n[{inicio_ciclo:%Y-%m-%d %H:%M:%S}] "
|
|
f"Día no activo (weekday={inicio_ciclo.weekday()}).\n"
|
|
f"Próxima ejecución: "
|
|
f"{(inicio_ciclo + timedelta(seconds=dormir)):%Y-%m-%d %H:%M:%S}"
|
|
)
|
|
time.sleep(dormir)
|
|
continue
|
|
|
|
resultado = obtener_intervalo(inicio_ciclo)
|
|
|
|
if resultado is not None:
|
|
|
|
intervalo, modo = resultado
|
|
|
|
print(
|
|
f"\n[{inicio_ciclo:%Y-%m-%d %H:%M:%S}] "
|
|
f"Iniciando ciclo [{modo}] (cada {intervalo}s)"
|
|
)
|
|
|
|
procesar_todas_empresas(inicio_ciclo, modo)
|
|
|
|
duracion = (datetime.now() - inicio_ciclo).total_seconds()
|
|
|
|
if duracion > intervalo:
|
|
print(
|
|
f"⚠️ El ciclo demoró {duracion:.1f}s "
|
|
f"(intervalo configurado {intervalo}s)"
|
|
)
|
|
|
|
dormir = max(0, intervalo - duracion)
|
|
|
|
print(f"💤 Próximo ciclo [{modo}] en {dormir:.0f}s...")
|
|
|
|
else:
|
|
|
|
dormir = segundos_hasta_proximo_inicio(inicio_ciclo)
|
|
|
|
despertar = inicio_ciclo + timedelta(seconds=dormir)
|
|
|
|
print(
|
|
f"\n[{inicio_ciclo:%Y-%m-%d %H:%M:%S}] "
|
|
f"Fuera del horario.\n"
|
|
f"Próxima ejecución: {despertar:%Y-%m-%d %H:%M:%S}"
|
|
)
|
|
|
|
time.sleep(dormir)
|
|
|
|
# ============================================================================
|
|
if __name__ == "__main__":
|
|
try:
|
|
main()
|
|
except KeyboardInterrupt:
|
|
print("\n\n🛑 Demonio detenido por el usuario.")
|