"""
Servicio SOAP compatible con el cliente MIG (Web Reference WSMIG.RibiSoft).
Mantiene los nombres de operacion exactamente como los espera el cliente
generado por la Web Reference original (RegistrarEmpresa, IngresoEmpresa,
ValidarEmpresa, RegistrarLicencia, ValidarLicenciaMensual, InactivarLicencia)
para que el cambio en el cliente sea solo de URL/namespace.
Namespace: http://migsistemas.com/licencias/ (ya no http://localhost/)
Montaje en Flask: register_soap(app) lo expone en /LicenciasService.asmx
sirviendo WSDL en GET ?wsdl y aceptando POST con sobre SOAP.
"""
from __future__ import annotations
import os
import time
import xml.etree.ElementTree as ET
from datetime import datetime
from typing import Optional
from flask import Flask, request
from spyne import Application, ServiceBase, Unicode, rpc
from spyne.protocol.soap import Soap11
from spyne.server.wsgi import WsgiApplication
from werkzeug.wrappers import Response
import crypto
from models import Empresa, LicenciaEmpresa, WebServiceLog, db
SOAP_USER = os.getenv('SOAP_USER', 'admin')
SOAP_PASSWORD = os.getenv('SOAP_PASSWORD', '123*/')
NS = 'http://migsistemas.com/licencias/'
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _flask_app() -> Flask:
"""Devuelve la app Flask actual para usar su contexto desde Spyne."""
from flask import current_app
return current_app._get_current_object() # type: ignore[attr-defined]
def _autenticar(usuario: Optional[str], contrasenia: Optional[str]) -> bool:
return (usuario or '') == SOAP_USER and (contrasenia or '') == SOAP_PASSWORD
def _ip_cliente() -> str:
if request:
return request.headers.get('X-Forwarded-For', request.remote_addr or '')
return ''
def _log(metodo: str, nit: Optional[str], equipo: Optional[str],
resultado: str, mensaje: str, duracion_ms: int) -> None:
try:
log = WebServiceLog(
Fecha=datetime.utcnow(),
Metodo=metodo,
nit=nit,
IpCliente=_ip_cliente(),
nombreequipo=equipo,
Resultado=resultado,
Mensaje=mensaje,
DuracionMs=duracion_ms,
)
db.session.add(log)
db.session.commit()
except Exception:
db.session.rollback()
def _bool_from_str(v: Optional[str]) -> bool:
"""El cliente manda 'SI'/'NO' o '1'/'0'; normalizamos a bool."""
if v is None:
return False
s = str(v).strip().upper()
return s in ('1', 'SI', 'TRUE', 'T', 'Y', 'YES')
def _empresas_a_dataset_xml(empresas) -> str:
"""Replica el formato DataSet.GetXml() de C# para ValidarEmpresa.
El cliente espera un sobre tipo:
"""
root = ET.Element('NewDataSet')
for e in empresas:
t = ET.SubElement(root, 'Table')
for col, val in e.to_dict().items():
if val is None:
continue
sub = ET.SubElement(t, col)
sub.text = str(val)
return ET.tostring(root, encoding='unicode')
def _licencias_a_dataset_xml(licencias, empresa) -> str:
"""Sobre tipo DataSet.GetXml() con UNA fila por licencia.
El cliente MIG (General.cs:1046 ValidarEmpresa) busca su `claveproducto`
en las filas y mira `licenciaactiva`. Si esta False, borra
ControlFechaTrabajo y hace Application.Exit(). Por eso ESTE metodo es
el que activa el bloqueo desde el panel admin.
Cada mezcla datos de la licencia (claveproducto, licenciaactiva,
nombreequipo, etc.) con datos de la empresa (nit, razonsocial,
mensualidad...) para retrocompat con cualquier consumidor que lea esos
campos. licenciaactiva se serializa como 'True'/'False' porque el
cliente hace bool.Parse() sobre el valor.
"""
empresa_extra = {}
if empresa is not None:
empresa_extra = {k: v for k, v in empresa.to_dict().items()
if k not in ('claveproducto', 'nombreequipo', 'osversion',
'EmpresaID', 'fecharegistro', 'fechaactualizacion')}
root = ET.Element('NewDataSet')
for l in licencias:
t = ET.SubElement(root, 'Table')
# Campos base de empresa (nit, razonsocial, mensualidad, regulariza, tipotpv...)
for col, val in empresa_extra.items():
if val is None:
continue
ET.SubElement(t, col).text = str(val)
# Campos de la licencia (sobreescriben/complementan)
d = l.to_dict()
for col in ('LicenciaID', 'claveproducto', 'licenciaactiva', 'bloquearlicencia',
'licenciatemporal', 'nombreequipo', 'osversion', 'fecharegistro',
'responsable', 'correoresponsable', 'telefonoponsable',
'motivolicencia', 'clavemes', 'iddepartamento', 'idciudad'):
val = d.get(col)
if val is None:
continue
# bool -> 'True'/'False' (lo que C# bool.Parse espera)
ET.SubElement(t, col).text = str(val)
return ET.tostring(root, encoding='unicode')
# ---------------------------------------------------------------------------
# Service
# ---------------------------------------------------------------------------
class LicenciasService(ServiceBase):
"""Implementa los 6 metodos que el cliente MIG consume hoy."""
# -------------------------------------------------------------------
# RegistrarEmpresa - el cliente la llama tras instalar el sistema
# -------------------------------------------------------------------
@rpc(Unicode, Unicode, Unicode, Unicode, Unicode, Unicode, Unicode,
Unicode, Unicode, Unicode, Unicode,
_returns=Unicode)
def RegistrarEmpresa(ctx, usuario, contrasenia, nit, razonsocial,
nombrecomercial, direccion, telefono, celular,
mensualidad, regulariza, tipotpv):
t0 = time.time()
flask_app = _flask_app()
try:
with flask_app.app_context():
if not _autenticar(usuario, contrasenia):
_log('RegistrarEmpresa', nit, None, 'AUTH_FAIL', 'credenciales invalidas',
int((time.time() - t0) * 1000))
return 'AUTH_FAIL'
empresa = Empresa.query.filter_by(nit=nit).first()
creada = False
if empresa is None:
empresa = Empresa(nit=nit)
db.session.add(empresa)
creada = True
empresa.razonsocial = razonsocial
empresa.nombrecomercial = nombrecomercial
empresa.direccion = direccion
empresa.telefono = telefono
empresa.celular = celular
empresa.mensualidad = _bool_from_str(mensualidad)
empresa.regulariza = _bool_from_str(regulariza)
empresa.tipotpv = tipotpv
empresa.fechaactualizacion = datetime.utcnow()
if creada:
empresa.fecharegistro = datetime.utcnow()
db.session.commit()
resultado = 'CREADA' if creada else 'ACTUALIZADA'
_log('RegistrarEmpresa', nit, None, 'OK', resultado,
int((time.time() - t0) * 1000))
return resultado
except Exception as ex:
with flask_app.app_context():
db.session.rollback()
_log('RegistrarEmpresa', nit, None, 'ERROR', str(ex)[:1000],
int((time.time() - t0) * 1000))
return f'ERROR: {ex}'
# -------------------------------------------------------------------
# IngresoEmpresa - aviso al iniciar sesion en el POS
# -------------------------------------------------------------------
@rpc(Unicode, Unicode, Unicode, Unicode, Unicode, Unicode, Unicode,
Unicode, Unicode, Unicode, Unicode, Unicode, Unicode, Unicode,
_returns=Unicode)
def IngresoEmpresa(ctx, usuario, contrasenia, nit, razonsocial,
nombrecomercial, direccion, telefono, celular,
mensualidad, regulariza, tipotpv,
claveproducto, nombreequipo, osversion):
t0 = time.time()
flask_app = _flask_app()
try:
with flask_app.app_context():
if not _autenticar(usuario, contrasenia):
_log('IngresoEmpresa', nit, nombreequipo, 'AUTH_FAIL', 'credenciales invalidas',
int((time.time() - t0) * 1000))
return 'AUTH_FAIL'
empresa = Empresa.query.filter_by(nit=nit).first()
creada = empresa is None
if creada:
empresa = Empresa(nit=nit, fecharegistro=datetime.utcnow())
db.session.add(empresa)
empresa.razonsocial = razonsocial
empresa.nombrecomercial = nombrecomercial
empresa.direccion = direccion
empresa.telefono = telefono
empresa.celular = celular
# mensualidad/regulariza son banderas que maneja el PANEL/admin, NO el
# cliente. El build viejo manda "NO" y corrompia el flag en cada arranque
# (Problema C). Solo las fijamos al CREAR la empresa; si ya existe, se
# respeta lo que tenga el servidor (lo decide el admin).
if creada:
empresa.mensualidad = _bool_from_str(mensualidad)
empresa.regulariza = _bool_from_str(regulariza)
empresa.tipotpv = tipotpv
empresa.claveproducto = claveproducto
empresa.nombreequipo = nombreequipo
empresa.osversion = osversion
empresa.fechaactualizacion = datetime.utcnow()
db.session.commit()
_log('IngresoEmpresa', nit, nombreequipo, 'OK', 'registrado',
int((time.time() - t0) * 1000))
return 'OK'
except Exception as ex:
with flask_app.app_context():
db.session.rollback()
_log('IngresoEmpresa', nit, nombreequipo, 'ERROR', str(ex)[:1000],
int((time.time() - t0) * 1000))
return f'ERROR: {ex}'
# -------------------------------------------------------------------
# ValidarEmpresa - cliente consulta info de la empresa
# -------------------------------------------------------------------
@rpc(Unicode, Unicode, Unicode, _returns=Unicode)
def ValidarEmpresa(ctx, usuario, contrasenia, nit):
"""Retorna una fila por licencia distinta (group by claveproducto)
para que el cliente vea licenciaactiva. Sin esto el bloqueo desde
el panel admin no llega al cliente.
"""
from sqlalchemy import func
t0 = time.time()
flask_app = _flask_app()
try:
with flask_app.app_context():
if not _autenticar(usuario, contrasenia):
_log('ValidarEmpresa', nit, None, 'AUTH_FAIL', '',
int((time.time() - t0) * 1000))
return ''
# Subquery: ultima fila por claveproducto para este NIT.
# Asi colapsamos los miles de RegistrarLicencia con motivo
# 'ACTUALIZACION' que se acumulan a una fila por equipo.
sub = (db.session.query(
func.max(LicenciaEmpresa.LicenciaID).label('max_id'))
.filter(LicenciaEmpresa.nit == nit,
LicenciaEmpresa.claveproducto.isnot(None),
LicenciaEmpresa.claveproducto != '')
.group_by(LicenciaEmpresa.claveproducto)
.subquery())
licencias = (LicenciaEmpresa.query
.filter(LicenciaEmpresa.LicenciaID.in_(
db.session.query(sub.c.max_id)))
.all())
empresa = Empresa.query.filter_by(nit=nit).first()
xml = _licencias_a_dataset_xml(licencias, empresa)
_log('ValidarEmpresa', nit, None, 'OK',
f'{len(licencias)} licencia(s)',
int((time.time() - t0) * 1000))
return xml
except Exception as ex:
with flask_app.app_context():
_log('ValidarEmpresa', nit, None, 'ERROR', str(ex)[:1000],
int((time.time() - t0) * 1000))
return ''
# -------------------------------------------------------------------
# RegistrarLicencia - solicitud de licencia (temporal o permanente)
# -------------------------------------------------------------------
@rpc(Unicode, Unicode, Unicode, Unicode, Unicode, Unicode, Unicode,
Unicode, Unicode, Unicode, Unicode, Unicode, Unicode, Unicode,
Unicode, Unicode, Unicode, Unicode, Unicode,
_returns=Unicode)
def RegistrarLicencia(ctx, usuario, contrasenia, nit, razonsocial,
nombrecomercial, direccion, telefono, celular,
departamento, ciudad, licenciatemporal,
responsable, correoresponsable, telefonoponsable,
motivolicencia, claveproducto, clavemes,
nombreequipo, osversion):
t0 = time.time()
flask_app = _flask_app()
try:
with flask_app.app_context():
if not _autenticar(usuario, contrasenia):
_log('RegistrarLicencia', nit, nombreequipo, 'AUTH_FAIL', '',
int((time.time() - t0) * 1000))
return 'AUTH_FAIL'
lic = LicenciaEmpresa(
nit=nit,
razonsocial=razonsocial,
nombrecomercial=nombrecomercial,
direccion=direccion,
telefono=telefono,
celular=celular,
iddepartamento=departamento,
idciudad=ciudad,
licenciatemporal=_bool_from_str(licenciatemporal),
fecharegistro=datetime.utcnow(),
responsable=responsable,
correoresponsable=correoresponsable,
telefonoponsable=telefonoponsable,
motivolicencia=motivolicencia,
claveproducto=claveproducto,
clavemes=clavemes,
nombreequipo=nombreequipo,
osversion=osversion,
licenciaactiva=True,
bloquearlicencia=False,
)
db.session.add(lic)
db.session.commit()
_log('RegistrarLicencia', nit, nombreequipo, 'OK', motivolicencia or '',
int((time.time() - t0) * 1000))
return 'OK'
except Exception as ex:
with flask_app.app_context():
db.session.rollback()
_log('RegistrarLicencia', nit, nombreequipo, 'ERROR', str(ex)[:1000],
int((time.time() - t0) * 1000))
return f'ERROR: {ex}'
# -------------------------------------------------------------------
# ValidarLicenciaMensual - retorna Encrypt(nit + MM) si esta activa
# -------------------------------------------------------------------
@rpc(Unicode, Unicode, Unicode, _returns=Unicode)
def ValidarLicenciaMensual(ctx, usuario, contrasenia, nit):
t0 = time.time()
flask_app = _flask_app()
try:
with flask_app.app_context():
if not _autenticar(usuario, contrasenia):
_log('ValidarLicenciaMensual', nit, None, 'AUTH_FAIL', '',
int((time.time() - t0) * 1000))
return ''
empresa = Empresa.query.filter_by(nit=nit).first()
if not empresa:
_log('ValidarLicenciaMensual', nit, None, 'NO_EMPRESA', '',
int((time.time() - t0) * 1000))
return ''
# AUTO-CORRECCION (Problema C): el cliente SOLO invoca este metodo
# cuando su flag local ControlFechaTrabajo.mensualidad=1 (ver
# General.cs:700). O sea, quien llama ES mensual de verdad. Si el
# flag del servidor quedo en False (lo corrompio el build viejo via
# IngresoEmpresa mandando "NO"), lo auto-corregimos aqui en vez de
# tener que poner mensualidad=SI a mano cliente por cliente.
if not empresa.mensualidad:
empresa.mensualidad = True
db.session.commit()
_log('ValidarLicenciaMensual', nit, None, 'AUTOFIX_MENSUAL',
'mensualidad False->True', int((time.time() - t0) * 1000))
# Bloqueada manualmente por el admin?
bloqueada = LicenciaEmpresa.query.filter_by(
nit=nit, bloquearlicencia=True
).first()
if bloqueada:
_log('ValidarLicenciaMensual', nit, None, 'BLOQUEADA', '',
int((time.time() - t0) * 1000))
return ''
# El mes debe estar ACTIVADO/PAGADO en el panel (boton "Activar
# mes"). periodopagado se compara con el mes actual (YYYY-MM); si
# no coincide, el cliente NO recibe la clave -> aviso dias 5-8 y
# cierre dia 9. Asi se obliga el pago mes a mes.
periodo_actual = datetime.now().strftime('%Y-%m')
if (empresa.periodopagado or '') != periodo_actual:
_log('ValidarLicenciaMensual', nit, None, 'NO_PAGADO',
f'periodopagado={empresa.periodopagado} != {periodo_actual}',
int((time.time() - t0) * 1000))
return ''
# Generar clavemes = Encrypt(nit + MM)
mm = datetime.now().strftime('%m')
clavemes = crypto.encrypt(f'{nit}{mm}')
# Guardar la clavemes generada en la licencia mas reciente
lic = (LicenciaEmpresa.query
.filter_by(nit=nit)
.order_by(LicenciaEmpresa.fecharegistro.desc())
.first())
if lic:
lic.clavemes = clavemes
lic.licenciaactiva = True
db.session.commit()
_log('ValidarLicenciaMensual', nit, None, 'OK', f'mes={mm}',
int((time.time() - t0) * 1000))
return clavemes
except Exception as ex:
with flask_app.app_context():
_log('ValidarLicenciaMensual', nit, None, 'ERROR', str(ex)[:1000],
int((time.time() - t0) * 1000))
return ''
# -------------------------------------------------------------------
# InactivarLicencia - admin/sistema bloquea una licencia
# -------------------------------------------------------------------
@rpc(Unicode, Unicode, Unicode, _returns=Unicode)
def InactivarLicencia(ctx, usuario, contrasenia, nit):
t0 = time.time()
flask_app = _flask_app()
try:
with flask_app.app_context():
if not _autenticar(usuario, contrasenia):
_log('InactivarLicencia', nit, None, 'AUTH_FAIL', '',
int((time.time() - t0) * 1000))
return 'AUTH_FAIL'
actualizadas = (LicenciaEmpresa.query
.filter_by(nit=nit)
.update({'licenciaactiva': False,
'bloquearlicencia': True}))
db.session.commit()
_log('InactivarLicencia', nit, None, 'OK', f'{actualizadas} licencia(s)',
int((time.time() - t0) * 1000))
return 'OK'
except Exception as ex:
with flask_app.app_context():
db.session.rollback()
_log('InactivarLicencia', nit, None, 'ERROR', str(ex)[:1000],
int((time.time() - t0) * 1000))
return f'ERROR: {ex}'
# ---------------------------------------------------------------------------
# Wiring: monta el SOAP en /LicenciasService.asmx
# ---------------------------------------------------------------------------
_spyne_app = Application(
[LicenciasService],
tns=NS,
name='LicenciasService',
in_protocol=Soap11(validator='lxml'),
out_protocol=Soap11(),
)
_wsgi_app = WsgiApplication(_spyne_app)
def register_soap(app: Flask) -> None:
"""Monta el endpoint SOAP en la app Flask."""
@app.route('/LicenciasService.asmx', methods=['GET', 'POST'])
def licencias_service_endpoint():
# Cuando piden ?wsdl Spyne devuelve el WSDL
from werkzeug.wsgi import ClosingIterator
environ = request.environ
response_data = {}
def start_response(status, headers, exc_info=None):
response_data['status'] = status
response_data['headers'] = headers
return lambda x: None
body_iter = _wsgi_app(environ, start_response)
body = b''.join(body_iter) if not isinstance(body_iter, (bytes, bytearray)) else body_iter
if isinstance(body_iter, ClosingIterator):
body_iter.close()
status_code = int(response_data.get('status', '200 OK').split()[0])
return Response(
body,
status=status_code,
headers=response_data.get('headers', []),
)