Permalink
Find file Copy path
Fetching contributors…
Cannot retrieve contributors at this time
4555 lines (4149 sloc) 211 KB
<
#!/usr/bin/python
# -*- coding: utf8 -*-
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by the
# Free Software Foundation; either version 3, or (at your option) any later
# version.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTIBILITY
# or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
# for more details.
"""Módulo para obtener código de operación electrónico (COE) para
Liquidación Primaria Electrónica de Granos del web service WSLPG de AFIP
"""
__author__ = "Mariano Reingart <reingart@gmail.com>"
__copyright__ = "Copyright (C) 2013-2018 Mariano Reingart"
__license__ = "GPL 3.0"
__version__ = "1.31b"
LICENCIA = """
wslpg.py: Interfaz para generar Código de Operación Electrónica para
Liquidación Primaria de Granos (LpgService)
Copyright (C) 2013-2018 Mariano Reingart reingart@gmail.com
http://www.sistemasagiles.com.ar/trac/wiki/LiquidacionPrimariaGranos
Este progarma es software libre, se entrega ABSOLUTAMENTE SIN GARANTIA
y es bienvenido a redistribuirlo respetando la licencia GPLv3.
Para información adicional sobre garantía, soporte técnico comercial
e incorporación/distribución en programas propietarios ver PyAfipWs:
http://www.sistemasagiles.com.ar/trac/wiki/PyAfipWs
"""
AYUDA="""
Opciones:
--ayuda: este mensaje
--debug: modo depuración (detalla y confirma las operaciones)
--formato: muestra el formato de los archivos de entrada/salida
--prueba: genera y autoriza una liquidación de prueba (no usar en producción!)
--xml: almacena los requerimientos y respuestas XML (depuración)
--dbf: utilizar tablas DBF (xBase) para los archivos de intercambio
--json: utilizar formato json para el archivo de intercambio
--dummy: consulta estado de servidores
--autorizar: Autorizar Liquidación Primaria de Granos (liquidacionAutorizar)
--ajustar: Ajustar Liquidación Primaria de Granos (liquidacionAjustar)
--anular: Anular una Liquidación Primaria de Granos (liquidacionAnular)
--autorizar-anticipo: Autoriza un Anticipo (lpgAutorizarAnticipo)
--consultar: Consulta una liquidación (parámetros: nro de orden, COE, pdf)
--cancelar-anticipo: anteponer para anticipos (lpgCancelarAnticipo)
--ult: Consulta el último número de orden registrado en AFIP
(liquidacionUltimoNroOrdenConsultar)
--pdf: genera el formulario C 1116 B en formato PDF
--mostrar: muestra el documento PDF generado (usar con --pdf)
--imprimir: imprime el documento PDF generado (usar con --mostrar y --pdf)
--autorizar-lsg: Autoriza una Liquidación Secundaria de Granos (lsgAutorizar)
--lsg --anular: Anula una LSG (lsgAnular)
--lsg --consular: Consulta una LSG por pto_emision, nro_orden o COE
--lsg --ult: Consulta el último Nº LSG emitida (lsgConsultarUltimoNroOrden)
--lsg --asociar: Asocia una liq. sec. a un contrato (lsgAsociarAContrato)
--ajustar-lsg: Ajusta una liquidación secundaria (lsgAjustar por COE/Contrato)
--autorizar-cg: Autorizar Certificación de Granos (cgAutorizar)
--cg --anular: Solicita anulación de un CG (cgSolicitarAnulacion)
--cg --consultar: Consulta una CG por pto_emision, nro_orden o COE
--cg --ult: Consulta el último Nº LSG emitida (cgConsultarUltimoNroOrden)
--informar-calidad: Informa la calidad de una CG (cgInformarCalidad)
--buscar-ctg: devuelve los datos de la CTG a certificar
espera tipo_certificado, cuit_depositante, nro_planta, cod_grano, campania
--buscar-cert-con-saldo-disp: CG disponible para liquidar/retirar/transferir
espera cuit_depositante, cod_grano, campania, coe fecha_emision_des/has
--provincias: obtiene el listado de provincias
--localidades: obtiene el listado de localidades por provincia
--tipograno: obtiene el listado de los tipos de granos disponibles
--campanias: obtiene el listado de las campañas
--gradoref: obtiene el listado de los grados de referencias
--gradoent: obtiene el listado de los grados y valores entregados
--certdeposito: obtiene el listado de los tipos de certificados de depósito
--deducciones: obtiene el listado de los tipos de deducciones
--retenciones: obtiene el listado de los tipos de retenciones
--puertos: obtiene el listado de los puertos habilitados
--actividades: obtiene el listado de las actividades habilitados
--actividadesrep: devuelve las actividades en las que emisor/representado
se encuentra inscripto en RUOCA
--operaciones: obtiene el listado de las operaciones para el representado
Ver wslpg.ini para parámetros de configuración (URL, certificados, etc.)"
"""
import os, sys, shelve
import decimal, datetime
import traceback
import pprint
import warnings
from pysimplesoap.client import SoapFault
from fpdf import Template
import utils
# importo funciones compartidas:
from utils import leer, escribir, leer_dbf, guardar_dbf, N, A, I, json, BaseWS, inicializar_y_capturar_excepciones, get_install_dir
WSDL = "https://fwshomo.afip.gov.ar/wslpg/LpgService?wsdl"
#WSDL = "https://serviciosjava.afip.gob.ar/wslpg/LpgService?wsdl"
#WSDL = "file:wslpg.wsdl"
DEBUG = False
XML = False
CONFIG_FILE = "wslpg.ini"
TIMEOUT = 30
HOMO = False
# definición del formato del archivo de intercambio:
ENCABEZADO = [
('tipo_reg', 1, A), # 0: encabezado liquidación
('nro_orden', 18, N),
('cuit_comprador', 11, N),
('nro_act_comprador', 5, N),
('nro_ing_bruto_comprador', 15, N),
('cod_tipo_operacion', 2, N),
('es_liquidacion_propia', 1, A), # S o N
('es_canje', 1, A), # S o N
('cod_puerto', 4, N),
('des_puerto_localidad', 240, A),
('cod_grano', 3, N),
('cuit_vendedor', 11, N),
('nro_ing_bruto_vendedor', 15, N),
('actua_corredor', 1, A), # S o N
('liquida_corredor', 1, A), # S o N
('cuit_corredor', 11, N),
('nro_ing_bruto_corredor', 15, N),
('comision_corredor', 5, I, 2), # 3.2
('fecha_precio_operacion', 10, A), # 26/02/2013
('precio_ref_tn', 8, I, 3), # 4.3
('cod_grado_ref', 2, A),
('cod_grado_ent', 2, A),
('factor_ent', 6, I, 3), # 3.3
('precio_flete_tn', 7, I, 2), # 5.2
('cont_proteico', 6, I, 3), # 3.3
('alic_iva_operacion', 5, I, 2), # 3.2
('campania_ppal', 4, N),
('cod_localidad_procedencia', 6, N),
('reservado1', 200, A), # datos_adicionales (compatibilidad hacia atras)
('coe', 12, N),
('coe_ajustado', 12, N),
('estado', 2, A),
('total_deduccion', 17, I, 2), # 17.2
('total_retencion', 17, I, 2), # 17.2
('total_retencion_afip', 17, I, 2), # 17.2
('total_otras_retenciones', 17, I, 2), # 17.2
('total_neto_a_pagar', 17, I, 2), # 17.2
('total_iva_rg_2300_07', 17, I, 2), # 17.2
('total_pago_segun_condicion', 17, I, 2), # 17.2
('fecha_liquidacion', 10, A),
('nro_op_comercial', 10, N),
('precio_operacion', 17, I, 3), # 17.3
('subtotal', 17, I, 2), # 17.2
('importe_iva', 17, I, 2), # 17.2
('operacion_con_iva', 17, I, 2), # 17.2
('total_peso_neto', 8, N), # 17.2
# Campos WSLPGv1.1:
('pto_emision', 4, N),
('cod_prov_procedencia', 2, N),
('peso_neto_sin_certificado', 8, N),
('cod_tipo_ajuste', 2, N),
('val_grado_ent', 4, I, 3), # 1.3
# Campos WSLPGv1.3:
('cod_prov_procedencia_sin_certificado', 2, N),
('cod_localidad_procedencia_sin_certificado', 6, N),
# Campos WSLPGv1.4 (ajustes):
('nro_contrato', 15, N),
('tipo_formulario', 2, N),
('nro_formulario', 12, N),
# datos devuetos:
('total_iva_10_5', 17, I, 2), # 17.2
('total_iva_21', 17, I, 2), # 17.2
('total_retenciones_ganancias', 17, I, 2), # 17.2
('total_retenciones_iva', 17, I, 2), # 17.2
('datos_adicionales', 400, A), # max 400 desde WSLPGv1.2
# Campos agregados WSLPGv1.5 (ajustes):
('iva_deducciones', 17, I, 2), # 17.2
('subtotal_deb_cred', 17, I, 2), # 17.2
('total_base_deducciones', 17, I, 2), # 17.2
# Campos agregados WSLPGv1.6 (liquidación secundaria base):
('cantidad_tn', 11, I, 3), # 8.3
('nro_act_vendedor', 5, N),
# Campos agregados WSLPGv1.9 (liquidación secundaria base):
('total_deducciones', 19, I , 2),
('total_percepciones', 19, I , 2),
]
CERTIFICADO = [
('tipo_reg', 1, A), # 1: Certificado
('reservado1', 2, N), # en WSLPGv1.7 se amplio el campo
('nro_certificado_deposito', 12, N),
('peso_neto', 8, N), # usado peso ajustado WSLPGv1.17
('cod_localidad_procedencia', 6, N),
('cod_prov_procedencia', 2, N),
('reservado', 2, N),
('campania', 4, N),
('fecha_cierre', 10, A),
('peso_neto_total_certificado', 8, N), # para ajuste unificado (WSLPGv1.4)
('coe_certificado_deposito', 12, N), # para certificacion (WSLPGv1.6)
('tipo_certificado_deposito', 3, N), # wSLPGv1.7 agrega valor 332
]
RETENCION = [
('tipo_reg', 1, A), # 2: Retencion
('codigo_concepto', 2, A),
('detalle_aclaratorio', 30, A),
('base_calculo', 10, I, 2), # 8.2
('alicuota', 6, I, 2), # 3.2
('nro_certificado_retencion', 14, N),
('fecha_certificado_retencion', 10, A),
('importe_certificado_retencion', 17, I, 2), # 17.2
('importe_retencion', 17, I, 2), # 17.2
]
DEDUCCION = [
('tipo_reg', 1, A), # 3: Deducción
('codigo_concepto', 2, A),
('detalle_aclaratorio', 30, A), # max 50 por WSLPGv1.2
('dias_almacenaje', 4, N),
('reservado1', 6, I, 3),
('comision_gastos_adm', 5, I, 2), # 3.2
('base_calculo', 10, I, 2), # 8.2
('alicuota', 6, I, 2), # 3.2
('importe_iva', 17, I, 2), # 17.2
('importe_deduccion', 17, I, 2), # 17.2
('precio_pkg_diario', 11, I, 8), # 3.8, ajustado WSLPGv1.2
]
PERCEPCION = [
('tipo_reg', 1, A), # P: Percepcion
('detalle_aclaratoria', 50, A), # max 50 por WSLPGv1.8
('base_calculo', 10, I, 2), # 8.2
('alicuota', 6, I, 2), # 3.2
('importe_final', 19, I, 2), # 17.2 (LPG WSLPGv1.16)
]
OPCIONAL = [
('tipo_reg', 1, A), # O: Opcional
('codigo', 50, A),
('descripcion', 250, A),
]
AJUSTE = [
('tipo_reg', 1, A), # 4: ajuste débito / 5: crédito (WSLPGv1.4)
('concepto_importe_iva_0', 20, A),
('importe_ajustar_iva_0', 15, I, 2), # 11.2
('concepto_importe_iva_105', 20, A),
('importe_ajustar_iva_105', 15, I, 2), # 11.2
('concepto_importe_iva_21', 20, A),
('importe_ajustar_iva_21', 15, I, 2), # 11.2
('diferencia_peso_neto', 8, N),
('diferencia_precio_operacion', 17, I, 3), # 17.3
('cod_grado', 2, A),
('val_grado', 4, I, 3), # 1.3
('factor', 6, I, 3), # 3.3
('diferencia_precio_flete_tn', 7, I, 2), # 5.2
('datos_adicionales', 400, A),
# datos devueltos:
('fecha_liquidacion', 10, A),
('nro_op_comercial', 10, N),
('precio_operacion', 17, I, 3), # 17.3
('subtotal', 17, I, 2), # 17.2
('importe_iva', 17, I, 2), # 17.2
('operacion_con_iva', 17, I, 2), # 17.2
('total_peso_neto', 8, N), # 17.2
('total_deduccion', 17, I, 2), # 17.2
('total_retencion', 17, I, 2), # 17.2
('total_retencion_afip', 17, I, 2), # 17.2
('total_otras_retenciones', 17, I, 2), # 17.2
('total_neto_a_pagar', 17, I, 2), # 17.2
('total_iva_rg_2300_07', 17, I, 2), # 17.2
('total_pago_segun_condicion', 17, I, 2), # 17.2
('iva_calculado_iva_0', 15, I, 2), # 15.2
('iva_calculado_iva_105', 15, I, 2), # 15.2
('iva_calculado_iva_21', 15, I, 2), # 15.2
]
CERTIFICACION = [
('tipo_reg', 1, A), # 7: encabezado certificación
# campos de la cabecera para todas las certificaciones (WSLPGv1.6)
('pto_emision', 4, N),
('nro_orden', 8, N),
('tipo_certificado', 1, A), # P:Primaria,R:Retiro,T:Transferencia,E:Preexistente
('nro_planta', 6, N),
('nro_ing_bruto_depositario', 15, N),
('titular_grano', 1, A), # "P" (Propio) "T" (Tercero)
('cuit_depositante', 11, N), # obligatorio si titular_grano es T
('nro_ing_bruto_depositante', 15, N),
('cuit_corredor', 11, N),
('cod_grano', 3, N),
('campania', 4, N),
('datos_adicionales', 400, A),
('reservado1', 14, A), # reservado para futuros campos (no usar)
# campos para CgAutorizarPrimariaType ex-cgAutorizarDeposito (WSLPGv1.6-1.8)
('nro_act_depositario', 5, N), # nuevo WSLPGv1.8 tambien R/T
('descripcion_tipo_grano', 20, A),
('monto_almacenaje', 10, I, 2),
('monto_acarreo', 10, I, 2),
('monto_gastos_generales', 10, I, 2),
('monto_zarandeo', 10, I, 2),
('porcentaje_secado_de', 5, I, 2),
('porcentaje_secado_a', 5, I, 2),
('monto_secado', 10, I, 2),
('monto_por_cada_punto_exceso', 10, I, 2),
('monto_otros', 10, I, 2),
('reservado_calidad', 35, A), # ver subestructura WSLPGv1.10
('peso_neto_merma_volatil', 10, I , 2),
('porcentaje_merma_secado', 5, I, 2),
('peso_neto_merma_secado', 10, I, 2),
('porcentaje_merma_zarandeo', 5, I, 2),
('peso_neto_merma_zarandeo', 10, I, 2),
('peso_neto_certificado', 10, I, 2), # WSLPGv1.9 2 decimales!
('servicios_secado', 8, I, 3),
('servicios_zarandeo', 8, I, 3),
('servicios_otros', 7, I, 3),
('servicios_forma_de_pago', 20, A),
# campos para cgAutorizarRetiroTransferencia (WSLPGv1.6):
('cuit_receptor', 11, N),
('fecha', 10, A), # no usado WSLPGv1.8
('nro_carta_porte_a_utilizar', 9, N), # obligatorio para retiro
('cee_carta_porte_a_utilizar', 14, N), # no usado WSLPGv1.8
# para cgAutorizarPreexistente (WSLPGv1.6):
('tipo_certificado_deposito_preexistente', 1, N), # "R": Retiro "T": Tra.
('nro_certificado_deposito_preexistente', 12, N),
('cac_certificado_deposito_preexistente', 14, N), # cambio WSLPGv1.8
('fecha_emision_certificado_deposito_preexistente', 10, A),
('peso_neto', 8, N),
# nro_planta definido previamente - agregado WSLPGv1.8
# datos devueltos por el webservice:
('reservado2', 183, N), # padding para futuros campos (no usar)
('coe', 12, N),
('fecha_certificacion', 10, A),
('estado', 2, A),
('reservado3', 101, A), # padding para futuros campos (no usar)
# otros campos devueltos (opcionales)
# 'pesosResumen'
('peso_bruto_certificado', 10, I , 2),
('peso_merma_secado', 10, I , 2),
('peso_merma_zarandeo', 10, I , 2),
# peso_neto_certificado definido arriba
# serviciosResumen
('importe_iva', 10, I , 2),
('servicio_gastos_generales', 10, I , 2),
('servicio_otros', 10, I , 2),
('servicio_total', 10, I , 2),
('servicio_zarandeo', 10, I , 2),
# planta
('cuit_titular_planta', 11, N),
('razon_social_titular_planta', 11, A),
# campos no documentados por AFIP (agregados luego de WSLPGv1.15 a fines Sept)
('servicios_conceptos_no_gravados', 10, I, 2),
('servicios_percepciones_iva', 10, I, 2),
('servicios_otras_percepciones', 10, I, 2),
]
CTG = [ # para cgAutorizarDeposito (WSLPGv1.6)
('tipo_reg', 1, A), # C: CTG
('nro_ctg', 8, N),
('nro_carta_porte', 9, N),
('porcentaje_secado_humedad', 5, I, 2),
('importe_secado', 10, I, 2),
('peso_neto_merma_secado', 10, I, 2),
('tarifa_secado', 10, I, 2),
('importe_zarandeo', 10, I, 2),
('peso_neto_merma_zarandeo', 10, I, 2),
('tarifa_zarandeo', 10, I, 2),
('peso_neto_confirmado_definitivo', 10, I, 2),
]
DET_MUESTRA_ANALISIS = [ # para cgAutorizarDeposito (WSLPGv1.6)
('tipo_reg', 1, A), # D: detalle muestra analisis
('descripcion_rubro', 400, A),
('tipo_rubro', 1, A), # "B" (Bonificación) y "R" (Rebaja)
('porcentaje', 5, I, 2),
('valor', 5, I, 2),
]
CALIDAD = [ # para cgAutorizar y cgInformarCalidad (WSLPGv1.10)
('tipo_reg', 1, A), # Q: caldiad
('analisis_muestra', 10, N),
('nro_boletin', 10, N),
('cod_grado', 2, A), # nuevo WSLPGv1.10: G1 G2 ....
('valor_grado', 4, I, 3), # solo para cod_grado F1 F2 ...
('valor_contenido_proteico', 5, I, 3),
('valor_factor', 6, I, 3),
]
FACTURA_PAPEL = [ # para lsgAjustar (WSLPGv1.15)
('tipo_reg', 1, A), # F: factura papel
('nro_cai', 14, N),
('nro_factura_papel', 12, N),
('fecha_factura', 10, A),
('tipo_comprobante', 3, N),
]
FUSION = [ # para liquidacionAjustarUnificado (WSLPGv1.19)
('tipo_reg', 1, A), # f: fusion
('nro_ing_brutos', 15, N),
('nro_actividad', 5, N),
]
EVENTO = [
('tipo_reg', 1, A), # E: Evento
('codigo', 4, A),
('descripcion', 250, A),
]
ERROR = [
('tipo_reg', 1, A), # R: Error
('codigo', 4, A),
('descripcion', 250, A),
]
DATO = [
('tipo_reg', 1, A), # 9: Dato adicional
('campo', 25, A),
('valor', 250, A),
]
class WSLPG(BaseWS):
"Interfaz para el WebService de Liquidación Primaria de Granos"
_public_methods_ = ['Conectar', 'Dummy', 'SetTicketAcceso', 'DebugLog',
'AutorizarLiquidacion',
'AutorizarLiquidacionSecundaria',
'AnularLiquidacionSecundaria','AnularLiquidacion',
'AutorizarAnticipo', 'CancelarAnticipo',
'CrearLiquidacion', 'CrearLiqSecundariaBase',
'AgregarCertificado', 'AgregarRetencion',
'AgregarDeduccion', 'AgregarPercepcion',
'AgregarOpcional', 'AgregarCalidad',
'AgregarFacturaPapel', 'AgregarFusion',
'ConsultarLiquidacion', 'ConsultarUltNroOrden',
'ConsultarLiquidacionSecundaria',
'ConsultarLiquidacionSecundariaUltNroOrden',
'CrearAjusteBase',
'CrearAjusteDebito', 'CrearAjusteCredito',
'AjustarLiquidacionUnificado',
'AjustarLiquidacionUnificadoPapel',
'AjustarLiquidacionContrato',
'AjustarLiquidacionSecundaria',
'AnalizarAjusteDebito', 'AnalizarAjusteCredito',
'AsociarLiquidacionAContrato', 'ConsultarAjuste',
'ConsultarLiquidacionesPorContrato',
'ConsultarLiquidacionesSecundariasPorContrato',
'AsociarLiquidacionSecundariaAContrato',
'CrearCertificacionCabecera',
'AgregarCertificacionPrimaria',
'AgregarCertificacionRetiroTransferencia',
'AgregarCertificacionPreexistente',
'AgregarDetalleMuestraAnalisis', 'AgregarCTG',
'AutorizarCertificacion',
'InformarCalidadCertificacion', 'BuscarCTG',
'AnularCertificacion',
'ConsultarCertificacion',
'ConsultarCertificacionUltNroOrden',
'BuscarCertConSaldoDisponible',
'LeerDatosLiquidacion',
'ConsultarCampanias',
'ConsultarTipoGrano',
'ConsultarGradoEntregadoXTipoGrano',
'ConsultarCodigoGradoReferencia',
'ConsultarTipoCertificadoDeposito',
'ConsultarTipoDeduccion',
'ConsultarTipoRetencion',
'ConsultarPuerto',
'ConsultarTipoActividad',
'ConsultarTipoActividadRepresentado',
'ConsultarProvincias',
'ConsultarLocalidadesPorProvincia',
'ConsultarTiposOperacion',
'BuscarLocalidades',
'AnalizarXml', 'ObtenerTagXml', 'LoadTestXML',
'SetParametros', 'SetParametro', 'GetParametro',
'CargarFormatoPDF', 'AgregarCampoPDF', 'AgregarDatoPDF',
'CrearPlantillaPDF', 'ProcesarPlantillaPDF',
'GenerarPDF', 'MostrarPDF',
]
_public_attrs_ = ['Token', 'Sign', 'Cuit',
'AppServerStatus', 'DbServerStatus', 'AuthServerStatus',
'Excepcion', 'ErrCode', 'ErrMsg', 'LanzarExcepciones', 'Errores',
'XmlRequest', 'XmlResponse', 'Version', 'Traceback', 'InstallDir',
'COE', 'COEAjustado', 'Estado', 'Resultado', 'NroOrden',
'TotalDeduccion', 'TotalRetencion', 'TotalRetencionAfip',
'TotalOtrasRetenciones', 'TotalNetoAPagar', 'TotalPagoSegunCondicion',
'TotalIvaRg2300_07', 'Subtotal', 'TotalIva105', 'TotalIva21',
'TotalRetencionesGanancias', 'TotalRetencionesIVA', 'NroContrato',
'FechaCertificacion',
]
_reg_progid_ = "WSLPG"
_reg_clsid_ = "{9D21C513-21A6-413C-8592-047357692608}"
# Variables globales para BaseWS:
HOMO = HOMO
WSDL = WSDL
LanzarExcepciones = False
Version = "%s %s" % (__version__, HOMO and 'Homologación' or '')
def inicializar(self):
BaseWS.inicializar(self)
self.AppServerStatus = self.DbServerStatus = self.AuthServerStatus = None
self.errores = []
self.COE = self.COEAjustado = ""
self.Estado = self.Resultado = self.NroOrden = self.NroContrato = ''
self.TotalDeduccion = ""
self.TotalRetencion = ""
self.TotalRetencionAfip = ""
self.TotalOtrasRetenciones = ""
self.TotalNetoAPagar = ""
self.TotalIvaRg2300_07 = ""
self.TotalPagoSegunCondicion = ""
self.Subtotal = self.TotalIva105 = self.TotalIva21 = ""
self.TotalRetencionesGanancias = self.TotalRetencionesIVA = ""
self.TotalPercepcion = ""
self.FechaCertificacion = ""
self.datos = {}
@inicializar_y_capturar_excepciones
def Conectar(self, cache=None, url="", proxy="", wrapper="", cacert=None, timeout=30):
"Establecer la conexión a los servidores de la AFIP"
# llamo al constructor heredado:
ok = BaseWS.Conectar(self, cache, url, proxy, wrapper, cacert, timeout)
if ok:
# corrijo ubicación del servidor (puerto htttp 80 en el WSDL)
location = self.client.services['LpgService']['ports']['LpgEndPoint']['location']
if location.startswith("http://"):
print "Corrigiendo WSDL ...", location,
location = location.replace("http://", "https://").replace(":80", ":443")
self.client.services['LpgService']['ports']['LpgEndPoint']['location'] = location
print location
try:
# intento abrir el diccionario persistente de localidades
import wslpg_datos
localidades_db = os.path.join(self.cache, "localidades.dat")
# verificar que puede escribir en el dir, sino abrir solo lectura
flag = os.access(self.cache, os.W_OK) and 'c' or 'r'
wslpg_datos.LOCALIDADES = shelve.open(localidades_db, flag=flag)
if DEBUG: print "Localidades en BD:", len(wslpg_datos.LOCALIDADES)
self.Traceback = "Localidades en BD: %s" % len(wslpg_datos.LOCALIDADES)
except Exception, e:
print "ADVERTENCIA: No se pudo abrir la bbdd de localidades:", e
self.Excepcion = str(e)
return ok
def __analizar_errores(self, ret):
"Comprueba y extrae errores si existen en la respuesta XML"
errores = []
if 'errores' in ret:
errores.extend(ret['errores'])
if 'erroresFormato' in ret:
errores.extend(ret['erroresFormato'])
if errores:
self.Errores = ["%(codigo)s: %(descripcion)s" % err['error']
for err in errores]
self.errores = [
{'codigo': err['error']['codigo'],
'descripcion': err['error']['descripcion'].replace("\n", "")
.replace("\r", "")}
for err in errores]
self.ErrCode = ' '.join(self.Errores)
self.ErrMsg = '\n'.join(self.Errores)
@inicializar_y_capturar_excepciones
def Dummy(self):
"Obtener el estado de los servidores de la AFIP"
results = self.client.dummy()['return']
self.AppServerStatus = str(results['appserver'])
self.DbServerStatus = str(results['dbserver'])
self.AuthServerStatus = str(results['authserver'])
return True
@inicializar_y_capturar_excepciones
def CrearLiquidacion(self, nro_orden=None, cuit_comprador=None,
nro_act_comprador=None, nro_ing_bruto_comprador=None,
cod_tipo_operacion=None,
es_liquidacion_propia=None, es_canje=None,
cod_puerto=None, des_puerto_localidad=None, cod_grano=None,
cuit_vendedor=None, nro_ing_bruto_vendedor=None,
actua_corredor=None, liquida_corredor=None, cuit_corredor=None,
comision_corredor=None, nro_ing_bruto_corredor=None,
fecha_precio_operacion=None,
precio_ref_tn=None, cod_grado_ref=None, cod_grado_ent=None,
factor_ent=None, precio_flete_tn=None, cont_proteico=None,
alic_iva_operacion=None, campania_ppal=None,
cod_localidad_procedencia=None,
datos_adicionales=None, pto_emision=1, cod_prov_procedencia=None,
peso_neto_sin_certificado=None, val_grado_ent=None,
cod_localidad_procedencia_sin_certificado=None,
cod_prov_procedencia_sin_certificado=None,
nro_contrato=None,
**kwargs
):
"Inicializa internamente los datos de una liquidación para autorizar"
# limpio los campos especiales (segun validaciones de AFIP)
if alic_iva_operacion == 0:
alic_iva_operacion = None # no informar alicuota p/ monotributo
if val_grado_ent == 0:
val_grado_ent = None
# borrando datos corredor si no corresponden
if actua_corredor == "N":
cuit_corredor = None
comision_corredor = None
nro_ing_bruto_corredor = None
# si no corresponde elimino el peso neto certificado campo opcional
if not peso_neto_sin_certificado or not int(peso_neto_sin_certificado):
peso_neto_sin_certificado = None
if cod_puerto and int(cod_puerto) != 14:
des_puerto_localidad = None # validacion 1630
# limpio los campos opcionales para no enviarlos si no corresponde:
if cod_grado_ref == "":
cod_grado_ref = None
if cod_grado_ent == "":
cod_grado_ent = None
if val_grado_ent == 0:
val_grado_ent = None
# creo el diccionario con los campos generales de la liquidación:
self.liquidacion = dict(
ptoEmision=pto_emision,
nroOrden=nro_orden,
cuitComprador=cuit_comprador,
nroActComprador=nro_act_comprador,
nroIngBrutoComprador=nro_ing_bruto_comprador,
codTipoOperacion=cod_tipo_operacion,
esLiquidacionPropia=es_liquidacion_propia,
esCanje=es_canje,
codPuerto=cod_puerto,
desPuertoLocalidad=des_puerto_localidad,
codGrano=cod_grano,
cuitVendedor=cuit_vendedor,
nroIngBrutoVendedor=nro_ing_bruto_vendedor,
actuaCorredor=actua_corredor,
liquidaCorredor=liquida_corredor,
cuitCorredor=cuit_corredor,
comisionCorredor=comision_corredor,
nroIngBrutoCorredor=nro_ing_bruto_corredor,
fechaPrecioOperacion=fecha_precio_operacion,
precioRefTn=precio_ref_tn,
codGradoRef=cod_grado_ref,
codGradoEnt=cod_grado_ent,
valGradoEnt=val_grado_ent,
factorEnt=factor_ent,
precioFleteTn=precio_flete_tn,
contProteico=cont_proteico,
alicIvaOperacion=alic_iva_operacion,
campaniaPPal=campania_ppal,
codLocalidadProcedencia=cod_localidad_procedencia,
codProvProcedencia=cod_prov_procedencia,
datosAdicionales=datos_adicionales,
pesoNetoSinCertificado=peso_neto_sin_certificado,
numeroContrato=nro_contrato or None,
certificados=[],
)
# para compatibilidad hacia atras, "copiar" los campos si no hay cert:
if peso_neto_sin_certificado:
if cod_localidad_procedencia_sin_certificado is None:
cod_localidad_procedencia_sin_certificado = cod_localidad_procedencia
if cod_prov_procedencia_sin_certificado is None:
cod_prov_procedencia_sin_certificado = cod_prov_procedencia
self.liquidacion.update(dict(
codLocalidadProcedenciaSinCertificado=cod_localidad_procedencia_sin_certificado,
codProvProcedenciaSinCertificado=cod_prov_procedencia_sin_certificado,
))
# inicializo las listas que contentran las retenciones y deducciones:
self.retenciones = []
self.deducciones = []
self.percepciones = []
self.opcionales = [] # para anticipo
# limpio las estructuras internas no utilizables en este caso
self.certificacion = None
return True
@inicializar_y_capturar_excepciones
def CrearLiqSecundariaBase(self, pto_emision=1, nro_orden=None,
nro_contrato=None,
cuit_comprador=None, nro_ing_bruto_comprador=None,
cod_puerto=None, des_puerto_localidad=None,
cod_grano=None, cantidad_tn=None,
cuit_vendedor=None, nro_act_vendedor=None, # nuevo!!
nro_ing_bruto_vendedor=None,
actua_corredor=None, liquida_corredor=None, cuit_corredor=None,
nro_ing_bruto_corredor=None,
fecha_precio_operacion=None, precio_ref_tn=None,
precio_operacion=None, alic_iva_operacion=None, campania_ppal=None,
cod_localidad_procedencia=None, cod_prov_procedencia=None,
datos_adicionales=None,
**kwargs):
"Inicializa los datos de una liquidación secundaria de granos (base)"
# creo el diccionario con los campos generales de la liquidación:
self.liquidacion = dict(
ptoEmision=pto_emision, nroOrden=nro_orden,
numeroContrato=nro_contrato or None, cuitComprador=cuit_comprador,
nroIngBrutoComprador=nro_ing_bruto_comprador,
codPuerto=cod_puerto, desPuertoLocalidad=des_puerto_localidad,
codGrano=cod_grano, cantidadTn=cantidad_tn,
cuitVendedor=cuit_vendedor, nroActVendedor=nro_act_vendedor,
nroIngBrutoVendedor=nro_ing_bruto_vendedor,
actuaCorredor=actua_corredor, liquidaCorredor=liquida_corredor,
cuitCorredor=cuit_corredor or None,
nroIngBrutoCorredor=nro_ing_bruto_corredor or None,
fechaPrecioOperacion=fecha_precio_operacion,
precioRefTn=precio_ref_tn, precioOperacion=precio_operacion,
alicIvaOperacion=alic_iva_operacion or None,
campaniaPPal=campania_ppal,
codLocalidad=cod_localidad_procedencia,
codProvincia=cod_prov_procedencia,
datosAdicionales=datos_adicionales,
)
# inicializo las listas que contentran las retenciones y deducciones:
self.deducciones = []
self.percepciones = []
self.opcionales = []
self.factura_papel = None
return True
@inicializar_y_capturar_excepciones
def AgregarCertificado(self, tipo_certificado_deposito=None,
nro_certificado_deposito=None,
peso_neto=None,
cod_localidad_procedencia=None,
cod_prov_procedencia=None,
campania=None, fecha_cierre=None,
peso_neto_total_certificado=None,
coe_certificado_deposito=None, # WSLPGv1.6
**kwargs):
"Agrego el certificado a la liquidación / certificación de granos"
# limpio campos opcionales:
if not peso_neto_total_certificado:
peso_neto_total_certificado = None # 0 no es válido
# coe_certificado_deposito no es para LPG, unificar en futuras versiones
if tipo_certificado_deposito and int(tipo_certificado_deposito) == 332:
if coe_certificado_deposito and long(coe_certificado_deposito):
nro_certificado_deposito = coe_certificado_deposito
coe_certificado_deposito = None
cert = dict(
tipoCertificadoDeposito=tipo_certificado_deposito,
nroCertificadoDeposito=nro_certificado_deposito,
pesoNeto=peso_neto,
codLocalidadProcedencia=cod_localidad_procedencia,
codProvProcedencia=cod_prov_procedencia,
campania=campania,
fechaCierre=fecha_cierre,
pesoNetoTotalCertificado=peso_neto_total_certificado,
coeCertificadoDeposito=coe_certificado_deposito,
coe=coe_certificado_deposito, # WSLPGv1.17
pesoAjustado=peso_neto, # WSLPGv1.17
)
if self.liquidacion:
self.liquidacion['certificados'].append({'certificado': cert})
else:
self.certificacion['retiroTransferencia']['certificadoDeposito'] = cert
return True
@inicializar_y_capturar_excepciones
def AgregarRetencion(self, codigo_concepto, detalle_aclaratorio,
base_calculo, alicuota,
nro_certificado_retencion=None,
fecha_certificado_retencion=None,
importe_certificado_retencion=None,
**kwargs):
"Agrega la información referente a las retenciones de la liquidación"
# limpio los campos opcionales:
if fecha_certificado_retencion is not None and not fecha_certificado_retencion.strip():
fecha_certificado_retencion = None
if importe_certificado_retencion is not None and not float(importe_certificado_retencion):
importe_certificado_retencion = None
if nro_certificado_retencion is not None and not int(nro_certificado_retencion):
nro_certificado_retencion = None
self.retenciones.append(dict(
retencion=dict(
codigoConcepto=codigo_concepto,
detalleAclaratorio=detalle_aclaratorio,
baseCalculo=base_calculo,
alicuota=alicuota,
nroCertificadoRetencion=nro_certificado_retencion,
fechaCertificadoRetencion=fecha_certificado_retencion,
importeCertificadoRetencion=importe_certificado_retencion,
))
)
return True
@inicializar_y_capturar_excepciones
def AgregarDeduccion(self, codigo_concepto=None, detalle_aclaratorio=None,
dias_almacenaje=None, precio_pkg_diario=None,
comision_gastos_adm=None, base_calculo=None,
alicuota=None, **kwargs):
"Agrega la información referente a las deducciones de la liquidación."
# limpiar campo según validación (comision_gastos_adm puede ser 0.00!)
if codigo_concepto != "CO" and comision_gastos_adm is not None \
and float(comision_gastos_adm) == 0:
comision_gastos_adm = None
# no enviar campos para prevenir errores AFIP 1705, 1707, 1708
if base_calculo is not None:
if codigo_concepto == "AL":
base_calculo = None
if codigo_concepto == "CO" and float(base_calculo) == 0:
base_calculo = None # no enviar, por retrocompatibilidad
if codigo_concepto != "AL":
dias_almacenaje = None
precio_pkg_diario = None
self.deducciones.append(dict(
deduccion=dict(
codigoConcepto=codigo_concepto,
detalleAclaratorio=detalle_aclaratorio,
diasAlmacenaje=dias_almacenaje,
precioPKGdiario=precio_pkg_diario,
comisionGastosAdm=comision_gastos_adm,
baseCalculo=base_calculo,
alicuotaIva=alicuota,
))
)
return True
@inicializar_y_capturar_excepciones
def AgregarPercepcion(self, codigo_concepto=None, detalle_aclaratoria=None,
base_calculo=None, alicuota=None, importe_final=None,
**kwargs):
"Agrega la información referente a las percepciones de la liquidación"
# liquidación secundaria (sin importe final)
self.percepciones.append(dict(
percepcion=dict(
detalleAclaratoria=detalle_aclaratoria,
baseCalculo=base_calculo,
alicuota=alicuota,
importeFinal=importe_final,
))
)
return True
@inicializar_y_capturar_excepciones
def AgregarOpcional(self, codigo=None, descripcion=None, **kwargs):
"Agrega la información referente a los opcionales de la liq. seq."
self.opcionales.append(dict(
opcional=dict(
codigo=codigo,
descripcion=descripcion,
))
)
return True
@inicializar_y_capturar_excepciones
def AgregarFacturaPapel(self, nro_cai=None, nro_factura_papel=None,
fecha_factura=None, tipo_comprobante=None,
**kwargs):
self.factura_papel = dict(
nroCAI=nro_cai,
nroFacturaPapel=nro_factura_papel,
fechaFactura=fecha_factura,
tipoComprobante=tipo_comprobante,
)
return True
@inicializar_y_capturar_excepciones
def AutorizarLiquidacion(self):
"Autorizar Liquidación Primaria Electrónica de Granos"
# limpio los elementos que no correspondan por estar vacios:
if not self.liquidacion['certificados']:
del self.liquidacion['certificados']
if not self.retenciones:
self.retenciones = None
if not self.deducciones:
self.deducciones = None
if not self.percepciones:
self.percepciones = None
else:
# ajustar los nombres de campos que varian entre LPG y LSG
for it in self.percepciones:
per = it['percepcion']
per['descripcion'] = per.pop("detalleAclaratoria")
del per['baseCalculo']
del per['alicuota']
# llamo al webservice:
ret = self.client.liquidacionAutorizar(
auth={
'token': self.Token, 'sign': self.Sign,
'cuit': self.Cuit, },
liquidacion=self.liquidacion,
retenciones=self.retenciones,
deducciones=self.deducciones,
percepciones=self.percepciones,
)
# analizo la respusta
ret = ret['liqReturn']
self.__analizar_errores(ret)
self.AnalizarLiquidacion(ret.get('autorizacion'), self.liquidacion)
return True
@inicializar_y_capturar_excepciones
def AutorizarLiquidacionSecundaria(self):
"Autorizar Liquidación Secundaria Electrónica de Granos"
# extraer y adaptar los campos para liq. sec.
if self.deducciones:
self.liquidacion['deduccion'] = []
for it in self.deducciones:
ded = it['deduccion'] # no se agrupa
self.liquidacion['deduccion'].append({
'detalleAclaratoria': ded['detalleAclaratorio'],
'baseCalculo': ded['baseCalculo'],
'alicuotaIVA': ded['alicuotaIva']})
if self.percepciones:
self.liquidacion['percepcion'] = []
for it in self.percepciones:
per = it['percepcion'] # no se agrupa
self.liquidacion['percepcion'].append(per)
if self.opcionales:
self.liquidacion['opcionales'] = self.opcionales # agrupado ok
# llamo al webservice:
ret = self.client.lsgAutorizar(
auth={
'token': self.Token, 'sign': self.Sign,
'cuit': self.Cuit, },
liqSecundariaBase=self.liquidacion,
facturaPapel=self.factura_papel,
)
# analizo la respusta
ret = ret['oReturn']
self.__analizar_errores(ret)
self.AnalizarLiquidacion(ret.get('autorizacion'), self.liquidacion)
return True
@inicializar_y_capturar_excepciones
def AutorizarAnticipo(self):
"Autorizar Anticipo de una Liquidación Primaria Electrónica de Granos"
# extraer y adaptar los campos para el anticipo
anticipo = {"liquidacion": self.liquidacion}
liq = anticipo["liquidacion"]
liq["campaniaPpal"] = self.liquidacion["campaniaPPal"]
liq["codLocProcedencia"] = self.liquidacion["codLocalidadProcedencia"]
liq["descPuertoLocalidad"] = self.liquidacion["desPuertoLocalidad"]
if self.opcionales:
liq['opcionales'] = self.opcionales
if self.retenciones:
anticipo['retenciones'] = self.retenciones
if self.deducciones:
anticipo['deducciones'] = self.deducciones
# llamo al webservice:
ret = self.client.lpgAutorizarAnticipo(
auth={
'token': self.Token, 'sign': self.Sign,
'cuit': self.Cuit, },
anticipo=anticipo,
)
# analizo la respusta
ret = ret['liqReturn']
self.__analizar_errores(ret)
self.AnalizarLiquidacion(ret.get('autorizacion'), self.liquidacion)
return True
@inicializar_y_capturar_excepciones
def CancelarAnticipo(self, pto_emision=None, nro_orden=None, coe=None,
pdf=None):
"Cancelar Anticipo de una Liquidación Primaria Electrónica de Granos"
# llamo al webservice:
ret = self.client.lpgCancelarAnticipo(
auth={
'token': self.Token, 'sign': self.Sign,
'cuit': self.Cuit, },
coe=coe,
ptoEmision=pto_emision,
nroOrden=nro_orden,
pdf="S" if pdf else "N",
)
# analizo la respusta
ret = ret['liqConsReturn']
self.__analizar_errores(ret)
if 'liquidacion' in ret:
aut = ret['autorizacion']
liq = ret['liquidacion']
self.AnalizarLiquidacion(aut, liq)
# guardo el PDF si se indico archivo y vino en la respuesta:
if pdf and 'pdf' in ret:
open(pdf, "wb").write(ret['pdf'])
return True
def AnalizarLiquidacion(self, aut, liq=None, ajuste=False):
"Método interno para analizar la respuesta de AFIP"
# proceso los datos básicos de la liquidación (devuelto por consultar):
if liq:
self.params_out = dict(
pto_emision=liq.get('ptoEmision'),
nro_orden=liq.get('nroOrden'),
cuit_comprador=liq.get('cuitComprador'),
nro_act_comprador=liq.get('nroActComprador'),
nro_ing_bruto_comprador=liq.get('nroIngBrutoComprador'),
cod_tipo_operacion=liq.get('codTipoOperacion'),
es_liquidacion_propia=liq.get('esLiquidacionPropia'),
es_canje=liq.get('esCanje'),
cod_puerto=liq.get('codPuerto'),
des_puerto_localidad=liq.get('desPuertoLocalidad'),
cod_grano=liq.get('codGrano'),
cuit_vendedor=liq.get('cuitVendedor'),
nro_ing_bruto_vendedor=liq.get('nroIngBrutoVendedor'),
actua_corredor=liq.get('actuaCorredor'),
liquida_corredor=liq.get('liquidaCorredor'),
cuit_corredor=liq.get('cuitCorredor'),
comision_corredor=liq.get('comisionCorredor'),
nro_ing_bruto_corredor=liq.get('nroIngBrutoCorredor'),
fecha_precio_operacion=liq.get('fechaPrecioOperacion'),
precio_ref_tn=liq.get('precioRefTn'),
cod_grado_ref=liq.get('codGradoRef'),
cod_grado_ent=liq.get('codGradoEnt'),
factor_ent=liq.get('factorEnt'),
precio_flete_tn=liq.get('precioFleteTn'),
cont_proteico=liq.get('contProteico'),
alic_iva_operacion=liq.get('alicIvaOperacion'),
campania_ppal=liq.get('campaniaPPal'),
cod_localidad_procedencia=liq.get('codLocalidadProcedencia'),
cod_prov_procedencia=liq.get('codProvProcedencia'),
datos_adicionales=liq.get('datosAdicionales'),
peso_neto_sin_certificado=liq.get('pesoNetoSinCertificado'),
cod_localidad_procedencia_sin_certificado=liq.get('codLocalidadProcedenciaSinCertificado'),
cod_prov_procedencia_sin_certificado=liq.get('codProvProcedenciaSinCertificado'),
certificados=[],
)
if ajuste:
self.params_out.update(
# ajustes:
diferencia_peso_neto=liq.get('diferenciaPesoNeto'),
diferencia_precio_operacion=liq.get('diferenciaPrecioOperacion'),
cod_grado=liq.get('codGrado'),
val_grado=liq.get('valGrado'),
factor=liq.get('factor'),
diferencia_precio_flete_tn=liq.get('diferenciaPrecioFleteTn'),
concepto_importe_iva_0=liq.get('conceptoImporteIva0'),
importe_ajustar_iva_0=liq.get('importeAjustarIva0'),
concepto_importe_iva_105=liq.get('conceptoImporteIva105'),
importe_ajustar_iva_105=liq.get('importeAjustarIva105'),
concepto_importe_iva_21=liq.get('conceptoImporteIva21'),
importe_ajustar_iva_21=liq.get('importeAjustarIva21'),
)
# analizar detalle de importes ajustados discriminados por alicuota
# (por compatibildiad y consistencia se usan los mismos campos)
for it in liq.get("importes", liq.get("importe")):
# en ajustes LSG no se agrupan los importes en un subtipo...
if 'importeReturn' in it:
it = it['importeReturn'][0] # TODO: revisar SOAP
tasa = "iva_%s" % str(it['alicuota']).replace(".", "").strip()
self.params_out["concepto_importe_%s" % tasa] = it['concepto']
self.params_out["importe_ajustar_%s" % tasa] = it['importe']
self.params_out["iva_calculado_%s" % tasa] = it['ivaCalculado']
if 'certificados' in liq:
for c in liq['certificados']:
cert = c['certificado']
self.params_out['certificados'].append(dict(
tipo_certificado_deposito=cert['tipoCertificadoDeposito'],
nro_certificado_deposito=cert['nroCertificadoDeposito'],
peso_neto=cert['pesoNeto'],
cod_localidad_procedencia=cert['codLocalidadProcedencia'],
cod_prov_procedencia=cert['codProvProcedencia'],
campania=cert['campania'],
fecha_cierre=cert['fechaCierre'],
))
self.params_out['errores'] = self.errores
# proceso la respuesta de autorizar, ajustar (y consultar):
if aut:
self.TotalDeduccion = aut.get('totalDeduccion')
self.TotalRetencion = aut.get('totalRetencion')
self.TotalRetencionAfip = aut.get('totalRetencionAfip')
self.TotalOtrasRetenciones = aut.get('totalOtrasRetenciones')
self.TotalNetoAPagar = aut.get('totalNetoAPagar')
self.TotalIvaRg2300_07 = aut.get('totalIvaRg2300_07')
self.TotalPagoSegunCondicion = aut.get('totalPagoSegunCondicion')
self.COE = str(aut.get('coe', ''))
self.COEAjustado = aut.get('coeAjustado')
self.Estado = aut.get('estado', '')
self.NroContrato = aut.get('numeroContrato', '')
# actualizo parámetros de salida:
self.params_out['coe'] = self.COE
self.params_out['coe_ajustado'] = self.COEAjustado
self.params_out['estado'] = self.Estado
self.params_out['total_deduccion'] = self.TotalDeduccion
self.params_out['total_retencion'] = self.TotalRetencion
self.params_out['total_retencion_afip'] = self.TotalRetencionAfip
self.params_out['total_otras_retenciones'] = self.TotalOtrasRetenciones
self.params_out['total_neto_a_pagar'] = self.TotalNetoAPagar
self.params_out['total_iva_rg_2300_07'] = self.TotalIvaRg2300_07
self.params_out['total_pago_segun_condicion'] = self.TotalPagoSegunCondicion
# datos adicionales:
self.NroOrden = self.params_out['nro_orden'] = aut.get('nroOrden')
self.params_out['cod_tipo_ajuste'] = aut.get('codTipoAjuste')
fecha = aut.get('fechaLiquidacion')
if fecha:
fecha = str(fecha)
self.params_out['fecha_liquidacion'] = fecha
self.params_out['importe_iva'] = aut.get('importeIva')
self.params_out['nro_op_comercial'] = aut.get('nroOpComercial')
self.params_out['operacion_con_iva'] = aut.get('operacionConIva')
self.params_out['precio_operacion'] = aut.get('precioOperacion')
self.params_out['total_peso_neto'] = aut.get('totalPesoNeto')
self.params_out['subtotal'] = aut.get('subTotal')
# LSG (especificos):
self.params_out['total_deducciones'] = aut.get('totalDeducciones')
if 'todalPercepciones' in aut:
# error de tipeo en el WSDL de AFIP...
self.params_out['total_percepciones'] = aut.get('todalPercepciones')
else:
self.params_out['total_percepciones'] = aut.get('totalPercepciones')
# sub estructuras:
self.params_out['retenciones'] = []
self.params_out['deducciones'] = []
self.params_out['percepciones'] = []
for retret in aut.get("retenciones", []):
retret = retret['retencionReturn']
self.params_out['retenciones'].append({
'importe_retencion': retret['importeRetencion'],
'alicuota': retret['retencion'].get('alicuota'),
'base_calculo': retret['retencion'].get('baseCalculo'),
'codigo_concepto': retret['retencion'].get('codigoConcepto'),
'detalle_aclaratorio': (retret['retencion'].get('detalleAclaratorio') or "").replace("\n", ""),
'importe_certificado_retencion': retret['retencion'].get('importeCertificadoRetencion'),
'nro_certificado_retencion': retret['retencion'].get('nroCertificadoRetencion'),
'fecha_certificado_retencion': retret['retencion'].get('fechaCertificadoRetencion'),
})
for dedret in aut.get("deducciones", []):
dedret = dedret['deduccionReturn']
self.params_out['deducciones'].append({
'importe_deduccion': dedret['importeDeduccion'],
'importe_iva': dedret.get('importeIva'),
'alicuota': dedret['deduccion'].get('alicuotaIva'),
'base_calculo': dedret['deduccion'].get('baseCalculo'),
'codigo_concepto': dedret['deduccion'].get('codigoConcepto'),
'detalle_aclaratorio': dedret['deduccion'].get('detalleAclaratorio', "").replace("\n", ""),
'dias_almacenaje': dedret['deduccion'].get('diasAlmacenaje'),
'precio_pkg_diario': dedret['deduccion'].get('precioPKGdiario'),
'comision_gastos_adm': dedret['deduccion'].get('comisionGastosAdm'),
})
for perret in aut.get("percepciones", []):
perret = perret.get('percepcionReturn', perret)
self.params_out['percepciones'].append({
'importe_final': perret['percepcion']['importeFinal'],
'alicuota': perret['percepcion'].get('alicuota'),
'base_calculo': perret['percepcion'].get('baseCalculo'),
'descripcion': perret['percepcion'].get('descripcion', "").replace("\n", ""),
})
@inicializar_y_capturar_excepciones
def CrearAjusteBase(self,
pto_emision=1, nro_orden=None, # unificado, contrato, papel
coe_ajustado=None, # unificado
nro_contrato=None, # contrato
tipo_formulario=None, # papel
nro_formulario=None, # papel
actividad=None, # contrato / papel
cod_grano=None, # contrato / papel
cuit_vendedor=None, # contrato / papel
cuit_comprador=None, # contrato / papel
cuit_corredor=None, # contrato / papel
nro_ing_bruto_vendedor=None, # papel
nro_ing_bruto_comprador=None, # papel
nro_ing_bruto_corredor=None, # papel
tipo_operacion=None, # papel
precio_ref_tn=None, # contrato
cod_grado_ent=None, # contrato
val_grado_ent=None, # contrato
precio_flete_tn=None, # contrato
cod_puerto=None, # contrato
des_puerto_localidad=None, # contrato
cod_provincia=None, # unificado, contrato, papel
cod_localidad=None, # unificado, contrato, papel
comision_corredor=None, # papel
**kwargs
):
"Inicializa internamente los datos de una liquidación para ajustar"
# ajusto nombre de campos para compatibilidad hacia atrás (encabezado):
if 'cod_localidad_procedencia' in kwargs:
cod_localidad = kwargs['cod_localidad_procedencia']
if 'cod_provincia_procedencia' in kwargs:
cod_provincia = kwargs['cod_provincia_procedencia']
if 'nro_act_comprador' in kwargs:
actividad = kwargs['nro_act_comprador']
if 'cod_tipo_operacion' in kwargs:
tipo_operacion = kwargs['cod_tipo_operacion']
# limpio los campos especiales (segun validaciones de AFIP)
if val_grado_ent == 0:
val_grado_ent = None
# borrando datos si no corresponden
if cuit_corredor and int(cuit_corredor) == 0:
cuit_corredor = None
comision_corredor = None
nro_ing_bruto_corredor = None
if cod_puerto and int(cod_puerto) != 14:
des_puerto_localidad = None # validacion 1630
# limpio los campos opcionales para no enviarlos si no corresponde:
if cod_grado_ent == "":
cod_grado_ent = None
if val_grado_ent == 0:
val_grado_ent = None
# creo el diccionario con los campos generales del ajuste base:
self.ajuste = { 'ajusteBase': {
'ptoEmision': pto_emision,
'nroOrden': nro_orden,
'coeAjustado': coe_ajustado,
'nroContrato': nro_contrato,
'tipoFormulario': tipo_formulario,
'nroFormulario': nro_formulario,
'actividad': actividad,
'codGrano': cod_grano,
'cuitVendedor': cuit_vendedor,
'cuitComprador': cuit_comprador,
'cuitCorredor': cuit_corredor,
'nroIngBrutoVendedor': nro_ing_bruto_vendedor,
'nroIngBrutoComprador': nro_ing_bruto_comprador,
'nroIngBrutoCorredor': nro_ing_bruto_corredor,
'tipoOperacion': tipo_operacion,
'codPuerto': cod_puerto,
'desPuertoLocalidad': des_puerto_localidad,
'comisionCorredor': comision_corredor,
'precioRefTn': precio_ref_tn,
'codGradoEnt': cod_grado_ent,
'valGradoEnt': val_grado_ent,
'precioFleteTn': precio_flete_tn,
'codLocalidad': cod_localidad,
'codProv': cod_provincia,
'certificados': [],
}
}
# para compatibilidad con AgregarCertificado
self.liquidacion = self.ajuste['ajusteBase']
# inicializar temporales
self.__ajuste_base = None
self.__ajuste_debito = None
self.__ajuste_credito = None
return True
@inicializar_y_capturar_excepciones
def CrearAjusteCredito(self,
datos_adicionales=None, # unificado, contrato, papel
concepto_importe_iva_0=None, # unificado, contrato, papel
importe_ajustar_iva_0=None, # unificado, contrato, papel
concepto_importe_iva_105=None, # unificado, contrato, papel
importe_ajustar_iva_105=None, # unificado, contrato, papel
concepto_importe_iva_21=None, # unificado, contrato, papel
importe_ajustar_iva_21=None, # unificado, contrato, papel
diferencia_peso_neto=None, # unificado
diferencia_precio_operacion=None, # unificado
cod_grado=None, # unificado
val_grado=None, # unificado
factor=None, # unificado
diferencia_precio_flete_tn=None, # unificado
**kwargs
):
"Inicializa internamente los datos del crédito del ajuste"
self.ajuste['ajusteCredito'] = {
'diferenciaPesoNeto': diferencia_peso_neto,
'diferenciaPrecioOperacion': diferencia_precio_operacion,
'codGrado': cod_grado,
'valGrado': val_grado,
'factor': factor,
'diferenciaPrecioFleteTn': diferencia_precio_flete_tn,
'datosAdicionales': datos_adicionales,
'opcionales': None,
'conceptoImporteIva0': concepto_importe_iva_0,
'importeAjustarIva0': importe_ajustar_iva_0,
'conceptoImporteIva105': concepto_importe_iva_105,
'importeAjustarIva105': importe_ajustar_iva_105,
'conceptoImporteIva21': concepto_importe_iva_21,
'importeAjustarIva21': importe_ajustar_iva_21,
'deducciones': [],
'retenciones': [],
'percepciones': [],
'certificados': [],
}
# vinculación con AgregarOpcional:
self.opcionales = self.ajuste['ajusteCredito']['opcionales']
# vinculación con AgregarRetencion y AgregarDeduccion
self.deducciones = self.ajuste['ajusteCredito']['deducciones']
self.retenciones = self.ajuste['ajusteCredito']['retenciones']
# para LSG:
self.percepciones = self.ajuste['ajusteCredito']['percepciones']
# para compatibilidad con AgregarCertificado (WSLPGv1.17)
self.liquidacion = self.ajuste['ajusteCredito']
return True
@inicializar_y_capturar_excepciones
def CrearAjusteDebito(self,
datos_adicionales=None, # unificado, contrato, papel
concepto_importe_iva_0=None, # unificado, contrato, papel
importe_ajustar_iva_0=None, # unificado, contrato, papel
concepto_importe_iva_105=None, # unificado, contrato, papel
importe_ajustar_iva_105=None, # unificado, contrato, papel
concepto_importe_iva_21=None, # unificado, contrato, papel
importe_ajustar_iva_21=None, # unificado, contrato, papel
diferencia_peso_neto=None, # unificado
diferencia_precio_operacion=None, # unificado
cod_grado=None, # unificado
val_grado=None, # unificado
factor=None, # unificado
diferencia_precio_flete_tn=None, # unificado
**kwargs
):
"Inicializa internamente los datos del crédito del ajuste"
self.ajuste['ajusteDebito'] = {
'diferenciaPesoNeto': diferencia_peso_neto,
'diferenciaPrecioOperacion': diferencia_precio_operacion,
'codGrado': cod_grado,
'valGrado': val_grado,
'factor': factor,
'diferenciaPrecioFleteTn': diferencia_precio_flete_tn,
'datosAdicionales': datos_adicionales,
'opcionales': None,
'conceptoImporteIva0': concepto_importe_iva_0,
'importeAjustarIva0': importe_ajustar_iva_0,
'conceptoImporteIva105': concepto_importe_iva_105,
'importeAjustarIva105': importe_ajustar_iva_105,
'conceptoImporteIva21': concepto_importe_iva_21,
'importeAjustarIva21': importe_ajustar_iva_21,
'deducciones': [],
'retenciones': [],
'percepciones': [],
'certificados': [],
}
# vinculación con AgregarOpcional:
self.opcionales = self.ajuste['ajusteDebito']['opcionales']
# vinculación con AgregarRetencion y AgregarDeduccion
self.deducciones = self.ajuste['ajusteDebito']['deducciones']
self.retenciones = self.ajuste['ajusteDebito']['retenciones']
# para LSG:
self.percepciones = self.ajuste['ajusteDebito']['percepciones']
# para compatibilidad con AgregarCertificado (WSLPGv1.17)
self.liquidacion = self.ajuste['ajusteDebito']
return True
def AgregarFusion(self, nro_ing_brutos, nro_actividad, **kwargs):
"Datos de comprador o vendedor según liquidación a ajustar (fusión.)"
self.ajuste['ajusteBase']['fusion'] = {'nroIngBrutos': nro_ing_brutos,
'nroActividad': nro_actividad,
}
return True
@inicializar_y_capturar_excepciones
def AjustarLiquidacionUnificado(self):
"Ajustar Liquidación Primaria de Granos"
# limpiar estructuras no utilizadas (si no hay deducciones / retenciones)
for k in ('ajusteDebito', 'ajusteCredito'):
if not any(self.ajuste[k].values()):
del self.ajuste[k]
else:
if not self.ajuste[k]['deducciones']:
del self.ajuste[k]['deducciones']
if not self.ajuste[k]['retenciones']:
del self.ajuste[k]['retenciones']
# llamar al webservice:
ret = self.client.liquidacionAjustarUnificado(
auth={
'token': self.Token, 'sign': self.Sign,
'cuit': self.Cuit, },
**self.ajuste
)
# analizar el resultado:
ret = ret['ajusteUnifReturn']
self.__analizar_errores(ret)
if 'ajusteUnificado' in ret:
aut = ret['ajusteUnificado']
self.AnalizarAjuste(aut)
return True
@inicializar_y_capturar_excepciones
def AjustarLiquidacionUnificadoPapel(self):
"Ajustar Liquidación realizada en un formulario F1116 B / C (papel)"
# limpiar arrays no enviados:
if not self.ajuste['ajusteBase']['certificados']:
del self.ajuste['ajusteBase']['certificados']
for k1 in ('ajusteCredito', 'ajusteDebito'):
for k2 in ('retenciones', 'deducciones'):
if not self.ajuste[k1][k2]:
del self.ajuste[k1][k2]
ret = self.client.liquidacionAjustarUnificadoPapel(
auth={
'token': self.Token, 'sign': self.Sign,
'cuit': self.Cuit, },
**self.ajuste
)
ret = ret['ajustePapelReturn']
self.__analizar_errores(ret)
if 'ajustePapel' in ret:
aut = ret['ajustePapel']
self.AnalizarAjuste(aut)
return True
@inicializar_y_capturar_excepciones
def AjustarLiquidacionContrato(self):
"Ajustar Liquidación activas relacionadas a un contrato"
# limpiar arrays no enviados:
if not self.ajuste['ajusteBase']['certificados']:
del self.ajuste['ajusteBase']['certificados']
for k1 in ('ajusteCredito', 'ajusteDebito'):
for k2 in ('retenciones', 'deducciones'):
if not self.ajuste[k1][k2]:
del self.ajuste[k1][k2]
ret = self.client.liquidacionAjustarContrato(
auth={
'token': self.Token, 'sign': self.Sign,
'cuit': self.Cuit, },
**self.ajuste
)
ret = ret['ajusteContratoReturn']
self.__analizar_errores(ret)
if 'ajusteContrato' in ret:
aut = ret['ajusteContrato']
self.AnalizarAjuste(aut)
return True
@inicializar_y_capturar_excepciones
def AjustarLiquidacionSecundaria(self):
"Ajustar Liquidación Secundaria de Granos"
# limpiar estructuras no utilizadas (si no hay deducciones / retenciones)
for k in ('ajusteDebito', 'ajusteCredito'):
if k not in self.ajuste:
# ignorar si no se agrego estructura ajuste credito / debito
continue
elif not any(self.ajuste[k].values()):
# eliminar estructura vacia credito / debito
del self.ajuste[k]
else:
# ajustar cambios de nombre entre LSG y LPG
for tasa in ("0", "105", "21"):
tasa_lsg = "10" if tasa == "105" else tasa
self.ajuste[k]['importeAjustar%s' % tasa_lsg] = self.ajuste[k]['importeAjustarIva%s' % tasa]
self.ajuste[k]['conceptoIva%s' % tasa_lsg] = self.ajuste[k]['conceptoImporteIva%s' % tasa]
# no enviar tag percepciones vacio (no agrupar en subtipo)
if self.ajuste[k]['percepciones']:
self.ajuste[k]['percepcion'] = [
per["percepcion"] for per
in self.ajuste[k]['percepciones']]
del self.ajuste[k]['percepciones']
base = self.ajuste['ajusteBase']
base['coe'] = base['coeAjustado']
base['codProvincia'] = base['codProv']
# llamar al webservice:
if base['nroContrato'] is not None and long(base['nroContrato']):
metodo = self.client.lsgAjustarXContrato
else:
metodo = self.client.lsgAjustarXCoe
ret = metodo(
auth={
'token': self.Token, 'sign': self.Sign,
'cuit': self.Cuit, },
ajusteCredito=self.ajuste.get('ajusteCredito'),
ajusteDebito=self.ajuste.get('ajusteDebito'),
**base
)
# analizar el resultado:
ret = ret['oReturn']
self.__analizar_errores(ret)
if ret:
self.AnalizarAjuste(ret)
return True
def AnalizarAjuste(self, aut, base=True):
"Método interno para analizar la respuesta de AFIP (ajustes)"
self.__ajuste_base = None
self.__ajuste_debito = None
self.__ajuste_credito = None
# para compatibilidad con la generacion de PDF (completo datos)
if hasattr(self, "liquidacion") and self.liquidacion and base:
self.AnalizarLiquidacion(aut=None, liq=self.liquidacion)
self.params_out['errores'] = self.errores
# proceso la respuesta de autorizar, ajustar (y consultar):
if aut:
# en caso de anulación o no ser ajuste, ahora no devuelve datos:
self.COE = str(aut.get('coe', ""))
self.COEAjustado = aut.get('coeAjustado')
self.NroContrato = aut.get('nroContrato')
self.Estado = aut.get('estado', "")
totunif = aut.get("totalesUnificados") or {}
self.Subtotal = totunif.get('subTotalGeneral')
self.TotalIva105 = totunif.get('iva105')
self.TotalIva21 = totunif.get('iva21')
self.TotalRetencionesGanancias = totunif.get('retencionesGanancias')
self.TotalRetencionesIVA = totunif.get('retencionesIVA')
self.TotalOtrasRetenciones = totunif.get('importeOtrasRetenciones')
self.TotalNetoAPagar = totunif.get('importeNeto')
self.TotalIvaRg2300_07 = totunif.get('ivaRG2300_2007')
self.TotalPagoSegunCondicion = totunif.get('pagoSCondicion')
# actualizo parámetros de salida:
self.params_out['coe'] = self.COE
self.params_out['coe_ajustado'] = self.COEAjustado
self.params_out['estado'] = self.Estado
self.params_out['nro_orden'] = aut.get('nroOrden')
self.params_out['cod_tipo_operacion'] = aut.get('codTipoOperacion')
self.params_out['nro_contrato'] = aut.get('nroContrato')
self.params_out['nro_op_comercial'] = aut.get('nroOpComercial', "")
# actualizo totales solo para ajuste base (liquidacion general)
if base:
self.params_out['subtotal'] = self.Subtotal
self.params_out['iva_deducciones'] = totunif.get('ivaDeducciones')
self.params_out['subtotal_deb_cred'] = totunif.get('subTotalDebCred')
self.params_out['total_base_deducciones'] = totunif.get('totalBaseDeducciones')
self.params_out['total_iva_10_5'] = self.TotalIva105
self.params_out['total_iva_21'] = self.TotalIva21
self.params_out['total_retenciones_ganancias'] = self.TotalRetencionesGanancias
self.params_out['total_retenciones_iva'] = self.TotalRetencionesIVA
self.params_out['total_otras_retenciones'] = self.TotalOtrasRetenciones
self.params_out['total_neto_a_pagar'] = self.TotalNetoAPagar
self.params_out['total_iva_rg_2300_07'] = self.TotalIvaRg2300_07
self.params_out['total_pago_segun_condicion'] = self.TotalPagoSegunCondicion
# almaceno los datos de ajustes crédito y débito para usarlos luego
self.__ajuste_base = aut
self.__ajuste_debito = aut.get('ajusteDebito') or {}
self.__ajuste_credito = aut.get('ajusteCredito') or {}
return True
@inicializar_y_capturar_excepciones
def AnalizarAjusteDebito(self):
"Método para analizar la respuesta de AFIP para Ajuste Debito"
# para compatibilidad con la generacion de PDF (completo datos)
liq = {}
if hasattr(self, "liquidacion") and self.liquidacion:
liq.update(self.liquidacion)
if hasattr(self, "ajuste") and 'ajusteDebito' in self.ajuste:
liq.update(self.ajuste['ajusteDebito'])
if self.__ajuste_debito:
liq.update(self.__ajuste_debito)
self.AnalizarLiquidacion(aut=self.__ajuste_debito, liq=liq, ajuste=True)
self.AnalizarAjuste(self.__ajuste_base, base=False) # datos generales
return True
@inicializar_y_capturar_excepciones
def AnalizarAjusteCredito(self):
"Método para analizar la respuesta de AFIP para Ajuste Credito"
liq = {}
if hasattr(self, "liquidacion") and self.liquidacion:
liq.update(self.liquidacion)
if hasattr(self, "ajuste") and 'ajusteCredito' in self.ajuste:
liq.update(self.ajuste['ajusteCredito'])
if self.__ajuste_credito:
liq.update(self.__ajuste_credito)
self.AnalizarLiquidacion(aut=self.__ajuste_credito, liq=liq, ajuste=True)
self.AnalizarAjuste(self.__ajuste_base, base=False) # datos generales
return True
@inicializar_y_capturar_excepciones
def CrearCertificacionCabecera(self, pto_emision=1, nro_orden=None,
tipo_certificado=None, nro_planta=None,
nro_ing_bruto_depositario=None, titular_grano=None,
cuit_depositante=None, nro_ing_bruto_depositante=None,
cuit_corredor=None, cod_grano=None, campania=None,
datos_adicionales=None,
**kwargs):
"Inicializa los datos de una certificación de granos (cabecera)"
self.certificacion = {}
self.certificacion['cabecera'] = dict(
ptoEmision=pto_emision,
nroOrden=nro_orden,
tipoCertificado=tipo_certificado,
nroPlanta=nro_planta or None, # opcional
nroIngBrutoDepositario=nro_ing_bruto_depositario,
titularGrano=titular_grano,
cuitDepositante=cuit_depositante or None, # opcional
nroIngBrutoDepositante=nro_ing_bruto_depositante or None, # opcional
cuitCorredor=cuit_corredor or None, # opcional
codGrano=cod_grano,
campania=campania,
datosAdicionales=datos_adicionales, # opcional
)
# limpio las estructuras internas no utilizables en este caso
self.liquidacion = None
return True
@inicializar_y_capturar_excepciones
def AgregarCertificacionPrimaria(self,
nro_act_depositario=None,
descripcion_tipo_grano=None,
monto_almacenaje=None, monto_acarreo=None,
monto_gastos_generales=None, monto_zarandeo=None,
porcentaje_secado_de=None, porcentaje_secado_a=None,
monto_secado=None, monto_por_cada_punto_exceso=None,
monto_otros=None,
porcentaje_merma_volatil=None, peso_neto_merma_volatil=None,
porcentaje_merma_secado=None, peso_neto_merma_secado=None,
porcentaje_merma_zarandeo=None, peso_neto_merma_zarandeo=None,
peso_neto_certificado=None, servicios_secado=None,
servicios_zarandeo=None, servicios_otros=None,
servicios_forma_de_pago=None,
**kwargs):
# compatibilidad hacia atras: utilizar nuevos campos mas amplio
v = None
if 'servicio_otros' in kwargs:
v = kwargs.get('servicio_otros')
if isinstance(v, basestring) and v and not v.isalpha():
v = float(v)
if v:
servicios_otros = v
if not v:
warnings.warn("Usar servicio_otros para mayor cantidad de digitos")
self.certificacion['primaria'] = dict(
nroActDepositario=nro_act_depositario,
ctg=[], # <!--0 or more repetitions:-->
descripcionTipoGrano=descripcion_tipo_grano,
montoAlmacenaje=monto_almacenaje,
montoAcarreo=monto_acarreo,
montoGastosGenerales=monto_gastos_generales,
montoZarandeo=monto_zarandeo,
porcentajeSecadoDe=porcentaje_secado_de,
porcentajeSecadoA=porcentaje_secado_a,
montoSecado=monto_secado,
montoPorCadaPuntoExceso=monto_por_cada_punto_exceso,
montoOtros=monto_otros,
porcentajeMermaVolatil=porcentaje_merma_volatil,
pesoNetoMermaVolatil=peso_neto_merma_volatil,
porcentajeMermaSecado=porcentaje_merma_secado,
pesoNetoMermaSecado=peso_neto_merma_secado,
porcentajeMermaZarandeo=porcentaje_merma_zarandeo,
pesoNetoMermaZarandeo=peso_neto_merma_zarandeo,
pesoNetoCertificado=peso_neto_certificado,
serviciosSecado=servicios_secado or None, # opcional
serviciosZarandeo=servicios_zarandeo or None,
serviciosOtros=servicios_otros or None,
serviciosFormaDePago=servicios_forma_de_pago or None,
)
# si se pasan campos no documentados por AFIP, intentar enviarlo:
for k, kk in {
'servicios_conceptos_no_gravados': 'serviciosConceptosNoGravados',
'servicios_percepciones_iva': 'serviciosPercepcionesIva',
'servicios_otras_percepciones': 'serviciosOtrasPercepciones',
}.items():
v = kwargs.get(k)
# cuidado: si AFIP retira el campo, puede fallar si se pasa en 0
if isinstance(v, basestring) and v and not v.isalpha():
v = float(v)
if v:
self.certificacion['primaria'][kk] = v
return True
@inicializar_y_capturar_excepciones
def AgregarCertificacionRetiroTransferencia(self,
nro_act_depositario=None,
cuit_receptor=None,
fecha=None,
nro_carta_porte_a_utilizar=None,
cee_carta_porte_a_utilizar=None,
**kwargs):
self.certificacion['retiroTransferencia'] = dict(
nroActDepositario=nro_act_depositario,
cuitReceptor=cuit_receptor or None, # opcional
fecha=fecha,
nroCartaPorteAUtilizar=nro_carta_porte_a_utilizar or None,
ceeCartaPorteAUtilizar=cee_carta_porte_a_utilizar or None,
certificadoDeposito=[], # <!--0 or more repetitions:-->
)
return True
@inicializar_y_capturar_excepciones
def AgregarCertificacionPreexistente(self,
tipo_certificado_deposito_preexistente=None,
nro_certificado_deposito_preexistente=None,
cac_certificado_deposito_preexistente=None,
fecha_emision_certificado_deposito_preexistente=None,
peso_neto=None, nro_planta=None,
**kwargs):
self.certificacion['preexistente'] = dict(
tipoCertificadoDepositoPreexistente=tipo_certificado_deposito_preexistente,
nroCertificadoDepositoPreexistente=nro_certificado_deposito_preexistente,
cacCertificadoDepositoPreexistente=cac_certificado_deposito_preexistente,
fechaEmisionCertificadoDepositoPreexistente=fecha_emision_certificado_deposito_preexistente,
pesoNeto=peso_neto, nroPlanta=nro_planta,
)
return True
@inicializar_y_capturar_excepciones
def AgregarCalidad(self, analisis_muestra=None, nro_boletin=None,
cod_grado=None, valor_grado=None,
valor_contenido_proteico=None, valor_factor=None,
**kwargs):
"Agrega la información sobre la calidad, al autorizar o posteriormente"
self.certificacion['primaria']['calidad'] = dict(
analisisMuestra=analisis_muestra,
nroBoletin=nro_boletin,
codGrado=cod_grado, # G1 G2 G3 F1 F2 F3
valorGrado=valor_grado or None, # opcional
valorContProteico=valor_contenido_proteico,
valorFactor=valor_factor,
detalleMuestraAnalisis=[], # <!--1 or more repetitions:-->
)
return True
@inicializar_y_capturar_excepciones
def AgregarDetalleMuestraAnalisis(self, descripcion_rubro=None,
tipo_rubro=None, porcentaje=None,
valor=None,
**kwargs):
"Agrega la información referente al detalle de la certificación"
det = dict(
descripcionRubro=descripcion_rubro,
tipoRubro=tipo_rubro,
porcentaje=porcentaje,
valor=valor,
)
self.certificacion['primaria']['calidad']['detalleMuestraAnalisis'].append(det)
return True
@inicializar_y_capturar_excepciones
def BuscarCTG(self, tipo_certificado="P", cuit_depositante=None,
nro_planta=None, cod_grano=2, campania=1314,
nro_ctg=None, tipo_ctg=None, nro_carta_porte=None,
fecha_confirmacion_ctg_des=None,
fecha_confirmacion_ctg_has=None,
):
"Devuelve los CTG/Carta de porte que se puede incluir en un certificado"
ret = self.client.cgBuscarCtg(
auth={
'token': self.Token, 'sign': self.Sign,
'cuit': self.Cuit, },
tipoCertificado=tipo_certificado,
cuitDepositante=cuit_depositante or self.Cuit,
nroPlanta=nro_planta,
codGrano=cod_grano, campania=campania,
nroCtg=nro_ctg, tipoCtg=tipo_ctg,
nroCartaPorte=nro_carta_porte,
fechaConfirmacionCtgDes=fecha_confirmacion_ctg_des,
fechaConfirmacionCtgHas=fecha_confirmacion_ctg_has,
)['oReturn']
self.__analizar_errores(ret)
array = ret.get('ctg', [])
self.Excepcion = self.Traceback = ""
self.params_out['ctgs'] = []
for ctg in array:
self.params_out['ctgs'].append({
'campania': ctg.get('campania'),
'nro_planta': ctg.get('nroPlanta'),
'nro_ctg': ctg.get('nroCtg'),
'tipo_ctg': ctg.get('tipoCtg'),
'nro_carta_porte': ctg.get('nroCartaPorte'),
'kilos_confirmados': ctg.get('kilosConfirmados'),
'fecha_confirmacion_ctg': ctg.get('fechaConfirmacionCtg'),
'cod_grano': ctg.get('codGrano'),
'cuit_remitente_comercial': ctg.get('cuitRemitenteComercial'),
'cuit_liquida': ctg.get('cuitLiquida'),
'cuit_certifica': ctg.get('cuitCertifica'),
})
return True
@inicializar_y_capturar_excepciones
def AgregarCTG(self, nro_ctg=None, nro_carta_porte=None,
porcentaje_secado_humedad=None, importe_secado=None,
peso_neto_merma_secado=None, tarifa_secado=None,
importe_zarandeo=None, peso_neto_merma_zarandeo=None,
tarifa_zarandeo=None,
peso_neto_confirmado_definitivo=None,
**kwargs):
"Agrega la información referente a una CTG de la certificación"
ctg = dict(
nroCTG=nro_ctg,
nroCartaDePorte=nro_carta_porte,
pesoNetoConfirmadoDefinitivo=peso_neto_confirmado_definitivo,
porcentajeSecadoHumedad=porcentaje_secado_humedad,
importeSecado=importe_secado,
pesoNetoMermaSecado=peso_neto_merma_secado,
tarifaSecado=tarifa_secado,
importeZarandeo=importe_zarandeo,
pesoNetoMermaZarandeo=peso_neto_merma_zarandeo,
tarifaZarandeo=tarifa_zarandeo,
)
self.certificacion['primaria']['ctg'].append(ctg)
return True
@inicializar_y_capturar_excepciones
def BuscarCertConSaldoDisponible(self, cuit_depositante=None,
cod_grano=2, campania=1314, coe=None,
fecha_emision_des=None,
fecha_emision_has=None,
):
"""Devuelve los certificados de depósito en los que un productor tiene
saldo disponible para Liquidar/Retirar/Transferir"""
ret = self.client.cgBuscarCertConSaldoDisponible(
auth={
'token': self.Token, 'sign': self.Sign,
'cuit': self.Cuit, },
cuitDepositante=cuit_depositante or self.Cuit,
codGrano=cod_grano, campania=campania,
coe=coe,
fechaEmisionDes=fecha_emision_des,
fechaEmisionHas=fecha_emision_has,
)['oReturn']
self.__analizar_errores(ret)
array = ret.get('certificado', [])
self.Excepcion = self.Traceback = ""
self.params_out['certificados'] = []
for cert in array:
self.params_out['certificados'].append(dict(
coe=cert['coe'],
tipo_certificado=cert['tipoCertificado'],
campania=cert['campania'],
cuit_depositante=cert['cuitDepositante'],
cuit_depositario=cert['cuitDepositario'],
nro_planta=cert['nroPlanta'],
kilos_disponibles=cert['kilosDisponibles'],
cod_grano=cert['codGrano'],
))
return True
@inicializar_y_capturar_excepciones
def AutorizarCertificacion(self):
"Autoriza una Certificación Primaria de Depósito de Granos (C1116A/RT)"
# limpio los elementos que no correspondan por estar vacios:
for k1 in ('primaria', 'retiroTransferencia'):
dic = self.certificacion.get(k1)
if not dic: continue
for k2 in ('ctg', 'detalleMuestraAnalisis', 'certificadoDeposito'):
if k2 in dic and not dic[k2]:
del dic[k2]
# llamo al webservice:
ret = self.client.cgAutorizar(
auth={
'token': self.Token, 'sign': self.Sign,
'cuit': self.Cuit, },
**self.certificacion
)
# analizo la respusta
ret = ret['oReturn']
self.__analizar_errores(ret)
self.AnalizarAutorizarCertificadoResp(ret)
return True
def AnalizarAutorizarCertificadoResp(self, ret):
"Metodo interno para extraer datos de la Respuesta de Certificación"
aut = ret.get('autorizacion')
if aut:
self.PtoEmision = aut['ptoEmision']
self.NroOrden = aut['nroOrden']
self.FechaCertificacion = str(aut.get('fechaCertificacion', ""))
self.COE = str(aut['coe'])
self.Estado = aut['estado']
# actualizo parámetros de salida:
self.params_out['coe'] = self.COE
self.params_out['estado'] = self.Estado
self.params_out['nro_orden'] = self.NroOrden
self.params_out['fecha_certificacion'] = self.FechaCertificacion.replace("-", "")
if "planta" in aut:
p = aut.get("planta")
self.params_out['nro_planta'] = p.get("nroPlanta")
self.params_out['cuit_titular_planta'] = p.get("cuitTitularPlanta")
self.params_out['razon_social_titular_planta'] = p.get("razonSocialTitularPlanta")
# otros campos devueltos (opcionales)
p = aut.get('pesosResumen', {})
self.params_out['peso_bruto_certificado'] = p.get("pesoBrutoCertificado")
self.params_out['peso_merma_secado'] = p.get("pesoMermaSecado")
self.params_out['peso_merma_volatil'] = p.get("pesoMermaVolatil")
self.params_out['peso_merma_zarandeo'] = p.get("pesoMermaZarandeo")
self.params_out['peso_neto_certificado'] = p.get("pesoNetoCertificado")
p = aut.get('serviciosResumen', {})
self.params_out['importe_iva'] = p.get("importeIVA")
self.params_out['servicio_gastos_generales'] = p.get("servicioGastosGenerales")
self.params_out['servicio_otros'] = p.get("servicioOtros")
self.params_out['servicio_total'] = p.get("servicioTotal")
self.params_out['servicio_zarandeo'] = p.get("servicioZarandeo")
# datos devueltos según el tipo de certificacion (consultas):
cab = ret.get('cabecera')
if cab:
self.params_out['pto_emision'] = cab.get('ptoEmision')
self.params_out['nro_orden'] = cab.get('nroOrden')
self.params_out['tipo_certificado'] = cab.get('tipoCertificado')
self.params_out['nro_planta'] = cab.get('nroPlanta')
self.params_out['nro_ing_bruto_depositario'] = cab.get('nroIngBrutoDepositario')
self.params_out['titular_grano'] = cab.get('titularGrano')
self.params_out['cuit_depositante'] = cab.get('cuitDepositante')
self.params_out['nro_ing_bruto_depositante'] = cab.get('nroIngBrutoDepositante')
self.params_out['cuit_corredor'] = cab.get('cuitCorredor')
self.params_out['cod_grano'] = cab.get('codGrano')
self.params_out['campania'] = cab.get('campania')
self.params_out['datos_adicionales'] = cab.get('datosAdicionales')
pri = ret.get('primaria')
if pri:
self.params_out['nro_act_depositario'] = pri.get('nroActDepositario')
self.params_out['descripcion_tipo_grano'] = pri.get('descripcionTipoGrano')
self.params_out['monto_almacenaje'] = pri.get('montoAlmacenaje')
self.params_out['monto_acarreo'] = pri.get('montoAcarreo')
self.params_out['monto_gastos_generales'] = pri.get('montoGastosGenerales')
self.params_out['monto_zarandeo'] = pri.get('montoZarandeo')
self.params_out['porcentaje_secado_de'] = pri.get('porcentajeSecadoDe')
self.params_out['porcentaje_secado_a'] = pri.get('porcentajeSecadoA')
self.params_out['monto_secado'] = pri.get('montoSecado')
self.params_out['monto_por_cada_punto_exceso'] = pri.get('montoPorCadaPuntoExceso')
self.params_out['monto_otros'] = pri.get('montoOtros')
self.params_out['porcentaje_merma_volatil'] = pri.get('porcentajeMermaVolatil')
self.params_out['porcentaje_merma_secado'] = pri.get('porcentajeMermaSecado')
self.params_out['peso_neto_merma_secado'] = pri.get('pesoNetoMermaSecado')
self.params_out['porcentaje_merma_zarandeo'] = pri.get('pesoNetoMermaZarandeo')
self.params_out['peso_neto_certificado'] = pri.get('pesoNetoCertificado')
self.params_out['servicios_secado'] = pri.get('serviciosSecado')
self.params_out['servicios_zarandeo'] = pri.get('serviciosZarandeo')
self.params_out['servicios_otros'] = pri.get('serviciosOtros')
self.params_out['servicios_forma_de_pago'] = pri.get('serviciosFormaDePago')
# otros campos no documentados:
self.params_out['servicios_conceptos_no_gravados'] = pri.get("serviciosConceptosNoGravados")
self.params_out['servicios_percepciones_iva'] = pri.get("serviciosPercepcionesIVA")
self.params_out['servicios_otras_percepciones'] = pri.get("serviciosOtrasPercepciones")
# sub estructuras:
self.params_out['ctgs'] = []
self.params_out['det_muestra_analisis'] = []
for ctg in pri.get("ctg", []):
self.params_out['ctgs'].append({
'nro_ctg': ctg.get('nroCTG'),
'nro_carta_porte': ctg.get('nroCartaDePorte'),
'peso_neto_confirmado_definitivo': ctg.get('pesoNetoConfirmadoDefinitivo'),
'porcentaje_secado_humedad': ctg.get('porcentajeSecadoHumedad'),
'importe_secado': ctg.get('importeSecado'),
'peso_neto_merma_secado': ctg.get('pesoNetoMermaSecado'),
'importe_zarandeo': ctg.get('importeZarandeo'),
'peso_neto_merma_zarandeo': ctg.get('pesoNetoMermaZarandeo'),
'tarifa_zarandeo': ctg.get('tarifaZarandeo'),
})
self.params_out['calidad'] = []
for cal in [pri.get("calidad", {})]:
self.params_out['calidad'].append({
'analisis_muestra': cal.get('analisisMuestra'),
'nro_boletin': cal.get('nroBoletin'),
'nro_act_depositario': cal.get('nroActDepositario'),
'cod_grado': cal.get('codGrado'),
'valor_grado': cal.get('valorGrado'),
'valor_contenido_proteico': cal.get('valorContProteico'),
'valor_factor': cal.get('valorFactor')
})
for det in cal.get("detalleMuestraAnalisis", []):
self.params_out['det_muestra_analisis'].append({
'descripcion_rubro': det.get('descripcionRubro'),
'tipo_rubro': det.get('tipoRubro'),
'porcentaje': det.get('porcentaje'),
'valor': det.get('valor'),
})
rt = ret.get('retiroTransferencia')
if rt:
self.params_out['nro_act_depositario'] = rt.get('nroActDepositario')
self.params_out['cuit_receptor'] = rt.get('cuitReceptor')
self.params_out['nro_carta_porte_a_utilizar'] = rt.get('nroCartaPorteAUtilizar')
# sub estructuras:
self.params_out['certificados'] = []
cert = rt.get("certificadoDeposito")
if cert:
self.params_out['certificados'].append({
'coe_certificado_deposito': cert.get('coeCertificadoDeposito'),
'peso_neto': cert.get('pesoNeto'),
})
pre = ret.get('preexistente')
if pre:
self.params_out['nro_planta'] = pre.get('nroPlanta')
self.params_out['tipo_certificado_deposito_preexistente'] = pre.get('tipoCertificadoDepositoPreexistente')
self.params_out['nro_certificado_deposito_preexistente'] = pre.get('nroCertificadoDepositoPreexistente')
self.params_out['cac_certificado_deposito_preexistente'] = pre.get('cacCertificadoDepositoPreexistente')
self.params_out['fecha_emision_certificado_deposito_preexistente'] = pre.get('fechaEmisionCertificadoDepositoPreexistente')
self.params_out['peso_neto'] = pre.get('pesoNeto')
self.params_out['errores'] = self.errores
@inicializar_y_capturar_excepciones
def InformarCalidadCertificacion(self, coe):
"Informar calidad de un certificado (C1116A/RT)"
# llamo al webservice:
ret = self.client.cgInformarCalidad(
auth={
'token': self.Token, 'sign': self.Sign,
'cuit': self.Cuit, },
coe=coe,
calidad=self.certificacion['primaria']['calidad'],
)
# analizo la respusta
ret = ret['oReturn']
self.__analizar_errores(ret)
self.AnalizarAutorizarCertificadoResp(ret)
return True
@inicializar_y_capturar_excepciones
def AnularCertificacion(self, coe):
"Anular liquidación activa"
ret = self.client.cgSolicitarAnulacion(
auth={
'token': self.Token, 'sign': self.Sign,
'cuit': self.Cuit, },
coe=coe,
)
ret = ret['oReturn']
self.__analizar_errores(ret)
self.Estado = ret.get('estadoCertificado', "")
return self.COE
@inicializar_y_capturar_excepciones
def AsociarLiquidacionAContrato(self, coe=None, nro_contrato=None,
cuit_comprador=None,
cuit_vendedor=None,
cuit_corredor=None,
cod_grano=None,
**kwargs):
"Asociar una Liquidación a un contrato"
ret = self.client.asociarLiquidacionAContrato(
auth={
'token': self.Token, 'sign': self.Sign,
'cuit': self.Cuit, },
coe=coe,
nroContrato=nro_contrato,
cuitComprador=cuit_comprador,
cuitVendedor=cuit_vendedor,
cuitCorredor=cuit_corredor,
codGrano=cod_grano,
)
ret = ret['liquidacion']
self.__analizar_errores(ret)
if 'liquidacion' in ret:
# analizo la respusta
liq = ret['liquidacion']
aut = ret['autorizacion']
self.AnalizarLiquidacion(aut, liq)
return True
@inicializar_y_capturar_excepciones
def ConsultarLiquidacionesPorContrato(self, nro_contrato=None,
cuit_comprador=None,
cuit_vendedor=None,
cuit_corredor=None,
cod_grano=None,
**kwargs):
"Obtener los COE de liquidaciones relacionadas a un contrato"
ret = self.client.liquidacionPorContratoConsultar(
auth={
'token': self.Token, 'sign': self.Sign,
'cuit': self.Cuit, },
nroContrato=nro_contrato,
cuitComprador=cuit_comprador,
cuitVendedor=cuit_vendedor,
cuitCorredor=cuit_corredor,
codGrano=cod_grano,
)
ret = ret['liqPorContratoCons']
self.__analizar_errores(ret)
if 'coeRelacionados' in ret:
# analizo la respuesta = [{'coe': "...."}]
self.DatosLiquidacion = sorted(ret['coeRelacionados'])
# establezco el primer COE
self.LeerDatosLiquidacion()
return True
@inicializar_y_capturar_excepciones
def ConsultarLiquidacion(self, pto_emision=None, nro_orden=None, coe=None,
pdf=None):
"Consulta una liquidación por No de orden"
if coe:
ret = self.client.liquidacionXCoeConsultar(
auth={
'token': self.Token, 'sign': self.Sign,
'cuit': self.Cuit, },
coe=coe,
pdf='S' if pdf else 'N',
)
else:
ret = self.client.liquidacionXNroOrdenConsultar(
auth={
'token': self.Token, 'sign': self.Sign,
'cuit': self.Cuit, },
ptoEmision=pto_emision,
nroOrden=nro_orden,
)
ret = ret['liqConsReturn']
self.__analizar_errores(ret)
if 'liquidacion' in ret:
aut = ret['autorizacion']
liq = ret['liquidacion']
self.AnalizarLiquidacion(aut, liq)
# guardo el PDF si se indico archivo y vino en la respuesta:
if pdf and 'pdf' in ret:
open(pdf, "wb").write(ret['pdf'])
return True
@inicializar_y_capturar_excepciones
def ConsultarLiquidacionSecundaria(self, pto_emision=None, nro_orden=None,
coe=None, pdf=None):
"Consulta una liquidación sequndaria por No de orden o coe"
if coe:
ret = self.client.lsgConsultarXCoe(
auth={
'token': self.Token, 'sign': self.Sign,
'cuit': self.Cuit, },
coe=coe,
pdf='S' if pdf else 'N',
)
else:
ret = self.client.lsgConsultarXNroOrden(
auth={
'token': self.Token, 'sign': self.Sign,
'cuit': self.Cuit, },
ptoEmision=pto_emision,
nroOrden=nro_orden,
)
ret = ret['oReturn']
self.__analizar_errores(ret)
for it in ret['liquidaciones']:
aut = it['autorizacion']
if 'liquidacion' in it:
liq = it['liquidacion']
elif 'ajuste' in it:
liq = it['ajuste']
self.AnalizarLiquidacion(aut, liq)
# guardo el PDF si se indico archivo y vino en la respuesta:
if pdf and 'pdf' in ret:
open(pdf, "wb").write(ret['pdf'])
return True
@inicializar_y_capturar_excepciones
def ConsultarLiquidacionesSecundariasPorContrato(self, nro_contrato=None,
cuit_comprador=None,
cuit_vendedor=None,
cuit_corredor=None,
cod_grano=None,
**kwargs):
"Obtener los COE de liquidaciones relacionadas a un contrato"
ret = self.client.lsgConsultarXContrato(
auth={
'token': self.Token, 'sign': self.Sign,
'cuit': self.Cuit, },
nroContrato=nro_contrato,
cuitComprador=cuit_comprador,
cuitVendedor=cuit_vendedor,
cuitCorredor=cuit_corredor,
codGrano=cod_grano,
)
ret = ret['liqPorContratoCons']
self.__analizar_errores(ret)
if 'coeRelacionados' in ret:
# analizo la respuesta = [{'coe': "...."}]
self.DatosLiquidacion = sorted(ret['coeRelacionados'])
# establezco el primer COE
self.LeerDatosLiquidacion()
return True
@inicializar_y_capturar_excepciones
def AsociarLiquidacionSecundariaAContrato(self, coe=None, nro_contrato=None,
cuit_comprador=None,
cuit_vendedor=None,
cuit_corredor=None,
cod_grano=None,
**kwargs):
"Asociar una Liquidación a un contrato"
ret = self.client.lsgAsociarAContrato(
auth={
'token': self.Token, 'sign': self.Sign,
'cuit': self.Cuit, },
coe=coe,
nroContrato=nro_contrato,
cuitComprador=cuit_comprador,
cuitVendedor=cuit_vendedor,
cuitCorredor=cuit_corredor,
codGrano=cod_grano,
)
ret = ret['oReturn']
self.__analizar_errores(ret)
if 'liquidacion' in ret:
# analizo la respusta
liq = ret['liquidacion']
aut = ret['autorizacion']
self.AnalizarLiquidacion(aut, liq)
return True
@inicializar_y_capturar_excepciones
def ConsultarCertificacion(self, pto_emision=None, nro_orden=None,
coe=None, pdf=None):
"Consulta una certificacion por No de orden o COE"
if coe:
ret = self.client.cgConsultarXCoe(
auth={
'token': self.Token, 'sign': self.Sign,
'cuit': self.Cuit, },
coe=coe,
pdf='S' if pdf else 'N',
)
else:
ret = self.client.cgConsultarXNroOrden(
auth={
'token': self.Token, 'sign': self.Sign,
'cuit': self.Cuit, },
ptoEmision=pto_emision,
nroOrden=nro_orden,
)
ret = ret['oReturn']
self.__analizar_errores(ret)
if 'autorizacion' in ret:
self.AnalizarAutorizarCertificadoResp(ret)
# guardo el PDF si se indico archivo y vino en la respuesta:
if pdf and 'pdf' in ret:
open(pdf, "wb").write(ret['pdf'])
return True
@inicializar_y_capturar_excepciones
def ConsultarAjuste(self, pto_emision=None, nro_orden=None, nro_contrato=None,
coe=None, pdf=None):
"Consulta un ajuste de liquidación por No de orden o numero de contrato"
if nro_contrato:
ret = self.client.ajustePorContratoConsultar(
auth={
'token': self.Token, 'sign': self.Sign,
'cuit': self.Cuit, },
nroContrato=nro_contrato,
)
ret = ret['ajusteContratoReturn']
elif coe is None or pdf is None:
ret = self.client.ajusteXNroOrdenConsultar(
auth={
'token': self.Token, 'sign': self.Sign,
'cuit': self.Cuit, },
ptoEmision=pto_emision,
nroOrden=nro_orden,
pdf='S' if pdf else 'N',
)
ret = ret['ajusteXNroOrdenConsReturn']
else:
ret = self.client.ajusteXCoeConsultar(
auth={
'token': self.Token, 'sign': self.Sign,
'cuit': self.Cuit, },
coe=coe,
pdf='S' if pdf else 'N',
)
ret = ret['ajusteConsReturn']
self.__analizar_errores(ret)
if 'ajusteUnificado' in ret:
aut = ret['ajusteUnificado']
self.AnalizarAjuste(aut)
# guardo el PDF si se indico archivo y vino en la respuesta:
if pdf and 'pdf' in ret:
open(pdf, "wb").write(ret['pdf'])
return True
@inicializar_y_capturar_excepciones
def ConsultarUltNroOrden(self, pto_emision=1):
"Consulta el último No de orden registrado"
ret = self.client.liquidacionUltimoNroOrdenConsultar(
auth={
'token': self.Token, 'sign': self.Sign,
'cuit': self.Cuit, },
ptoEmision=pto_emision,
)
ret = ret['liqUltNroOrdenReturn']
self.__analizar_errores(ret)
self.NroOrden = ret['nroOrden']
return True
@inicializar_y_capturar_excepciones
def ConsultarLiquidacionSecundariaUltNroOrden(self, pto_emision=1):
"Consulta el último No de orden registrado para LSG"
ret = self.client.lsgConsultarUltimoNroOrden(
auth={
'token': self.Token, 'sign': self.Sign,
'cuit': self.Cuit, },
ptoEmision=pto_emision,
)
ret = ret['liqUltNroOrdenReturn']
self.__analizar_errores(ret)
self.NroOrden = ret['nroOrden']
return True
@inicializar_y_capturar_excepciones
def ConsultarCertificacionUltNroOrden(self, pto_emision=1):
"Consulta el último No de orden registrado para CG"
ret = self.client.cgConsultarUltimoNroOrden(
auth={
'token': self.Token, 'sign': self.Sign,
'cuit': self.Cuit, },
ptoEmision=pto_emision,
)
ret = ret['liqUltNroOrdenReturn']
self.__analizar_errores(ret)
self.NroOrden = ret['nroOrden']
return True
@inicializar_y_capturar_excepciones
def LeerDatosLiquidacion(self, pop=True):
"Recorro los datos devueltos y devuelvo el primero si existe"
if self.DatosLiquidacion:
# extraigo el primer item
if pop:
datos_liq = self.DatosLiquidacion.pop(0)
else:
datos_liq = self.DatosLiquidacion[0]
self.COE = str(datos_liq['coe'])
self.Estado = unicode(datos_liq.get('estado', ""))
return self.COE
else:
return ""
@inicializar_y_capturar_excepciones
def AnularLiquidacion(self, coe):
"Anular liquidación activa"
ret = self.client.liquidacionAnular(
auth={
'token': self.Token, 'sign': self.Sign,
'cuit': self.Cuit, },
coe=coe,
)
ret = ret['anulacionReturn']
self.__analizar_errores(ret)
self.Resultado = ret['resultado']
return self.COE
@inicializar_y_capturar_excepciones
def AnularLiquidacionSecundaria(self, coe):
"Anular liquidación secundaria activa"
ret = self.client.lsgAnular(
auth={
'token': self.Token, 'sign': self.Sign,
'cuit': self.Cuit, },
coe=coe,
)
ret = ret['anulacionReturn']
self.__analizar_errores(ret)
self.Resultado = ret['resultado']
return self.COE
def ConsultarCampanias(self, sep="||"):
ret = self.client.campaniasConsultar(
auth={
'token': self.Token, 'sign': self.Sign,
'cuit': self.Cuit, },
)['campaniaReturn']
self.__analizar_errores(ret)
array = ret.get('campanias', [])
return [("%s %%s %s %%s %s" % (sep, sep, sep)) %
(it['codigoDescripcion']['codigo'],
it['codigoDescripcion']['descripcion'])
for it in array]
def ConsultarTipoGrano(self, sep="||"):
ret = self.client.tipoGranoConsultar(
auth={
'token': self.Token, 'sign': self.Sign,
'cuit': self.Cuit, },
)['tipoGranoReturn']
self.__analizar_errores(ret)
array = ret.get('granos', [])
if sep is None:
return dict([(it['codigoDescripcion']['codigo'],
it['codigoDescripcion']['descripcion'])
for it in array])
else:
return [("%s %%s %s %%s %s" % (sep, sep, sep)) %
(it['codigoDescripcion']['codigo'],
it['codigoDescripcion']['descripcion'])
for it in array]
def ConsultarCodigoGradoReferencia(self, sep="||"):
"Consulta de Grados según Grano."
ret = self.client.codigoGradoReferenciaConsultar(
auth={
'token': self.Token, 'sign': self.Sign,
'cuit': self.Cuit, },
)['gradoRefReturn']
self.__analizar_errores(ret)
array = ret.get('gradosRef', [])
if sep is None:
return dict([(it['codigoDescripcion']['codigo'],
it['codigoDescripcion']['descripcion'])
for it in array])
else:
return [("%s %%s %s %%s %s" % (sep, sep, sep)) %
(it['codigoDescripcion']['codigo'],
it['codigoDescripcion']['descripcion'])
for it in array]
def ConsultarGradoEntregadoXTipoGrano(self, cod_grano, sep="||"):
"Consulta de Grado y Valor según Grano Entregado."
ret = self.client.codigoGradoEntregadoXTipoGranoConsultar(
auth={
'token': self.Token, 'sign': self.Sign,
'cuit': self.Cuit, },
codGrano=cod_grano,
)['gradoEntReturn']
self.__analizar_errores(ret)
array = ret.get('gradoEnt', [])
if sep is None:
return dict([(it['gradoEnt']['codigoDescripcion']['codigo'],
it['gradoEnt']['valor'])
for it in array])
else:
return [("%s %%s %s %%s %s %%s %s" % (sep, sep, sep, sep)) %
(it['gradoEnt']['codigoDescripcion']['codigo'],
it['gradoEnt']['codigoDescripcion']['descripcion'],
it['gradoEnt']['valor'],
)
for it in array]
def ConsultarTipoCertificadoDeposito(self, sep="||"):
"Consulta de tipos de Certificados de Depósito"
ret = self.client.tipoCertificadoDepositoConsultar(
auth={
'token': self.Token, 'sign': self.Sign,
'cuit': self.Cuit, },
)['tipoCertDepReturn']
self.__analizar_errores(ret)
array = ret.get('tiposCertDep', [])
return [("%s %%s %s %%s %s" % (sep, sep, sep)) %
(it['codigoDescripcion']['codigo'],
it['codigoDescripcion']['descripcion'])
for it in array]
def ConsultarTipoDeduccion(self, sep="||"):
"Consulta de tipos de Deducciones"
ret = self.client.tipoDeduccionConsultar(
auth={
'token': self.Token, 'sign': self.Sign,
'cuit': self.Cuit, },
)['tipoDeduccionReturn']
self.__analizar_errores(ret)
array = ret.get('tiposDeduccion', [])
return [("%s %%s %s %%s %s" % (sep, sep, sep)) %
(it['codigoDescripcion']['codigo'],
it['codigoDescripcion']['descripcion'])
for it in array]
def ConsultarTipoRetencion(self, sep="||"):
"Consulta de tipos de Retenciones."
ret = self.client.tipoRetencionConsultar(
auth={
'token': self.Token, 'sign': self.Sign,
'cuit': self.Cuit, },
)['tipoRetencionReturn']
self.__analizar_errores(ret)
array = ret.get('tiposRetencion', [])
return [("%s %%s %s %%s %s" % (sep, sep, sep)) %
(it['codigoDescripcion']['codigo'],
it['codigoDescripcion']['descripcion'])
for it in array]
def ConsultarPuerto(self, sep="||"):
"Consulta de Puertos habilitados"
ret = self.client.puertoConsultar(
auth={
'token': self.Token, 'sign': self.Sign,
'cuit': self.Cuit, },
)['puertoReturn']
self.__analizar_errores(ret)
array = ret.get('puertos', [])
return [("%s %%s %s %%s %s" % (sep, sep, sep)) %
(it['codigoDescripcion']['codigo'],
it['codigoDescripcion']['descripcion'])
for it in array]
def ConsultarTipoActividad(self, sep="||"):
"Consulta de Tipos de Actividad."
ret = self.client.tipoActividadConsultar(
auth={
'token': self.Token, 'sign': self.Sign,
'cuit': self.Cuit, },
)['tipoActividadReturn']
self.__analizar_errores(ret)
array = ret.get('tiposActividad', [])
return [("%s %%s %s %%s %s" % (sep, sep, sep)) %
(it['codigoDescripcion']['codigo'],
it['codigoDescripcion']['descripcion'])
for it in array]
def ConsultarTipoActividadRepresentado(self, sep="||"):
"Consulta de Tipos de Actividad inscripta en el RUOCA."
try:
ret = self.client.tipoActividadRepresentadoConsultar(
auth={
'token': self.Token, 'sign': self.Sign,
'cuit': self.Cuit, },
)['tipoActividadReturn']
self.__analizar_errores(ret)
array = ret.get('tiposActividad', [])
self.Excepcion = self.Traceback = ""
return [("%s %%s %s %%s %s" % (sep, sep, sep)) %
(it['codigoDescripcion']['codigo'],
it['codigoDescripcion']['descripcion'])
for it in array]
except Exception:
ex = utils.exception_info()
self.Excepcion = ex['msg']
self.Traceback = ex['tb']
if sep:
return ["ERROR"]
def ConsultarProvincias(self, sep="||"):
"Consulta las provincias habilitadas"
ret = self.client.provinciasConsultar(
auth={
'token': self.Token, 'sign': self.Sign,
'cuit': self.Cuit, },
)['provinciasReturn']
self.__analizar_errores(ret)
array = ret.get('provincias', [])
if sep is None:
return dict([(int(it['codigoDescripcion']['codigo']),
it['codigoDescripcion']['descripcion'])
for it in array])
else:
return [("%s %%s %s %%s %s" % (sep, sep, sep)) %
(it['codigoDescripcion']['codigo'],
it['codigoDescripcion']['descripcion'])
for it in array]
def ConsultarLocalidadesPorProvincia(self, codigo_provincia, sep="||"):
ret = self.client.localidadXProvinciaConsultar(
auth={
'token': self.Token, 'sign': self.Sign,
'cuit': self.Cuit, },
codProvincia=codigo_provincia,
)['localidadesReturn']
self.__analizar_errores(ret)
array = ret.get('localidades', [])
if sep is None:
return dict([(str(it['codigoDescripcion']['codigo']),
it['codigoDescripcion']['descripcion'])
for it in array])
else:
return [("%s %%s %s %%s %s" % (sep, sep, sep)) %
(it['codigoDescripcion']['codigo'],
it['codigoDescripcion']['descripcion'])
for it in array]
def BuscarLocalidades(self, cod_prov, cod_localidad=None, consultar=True):
"Devuelve la localidad o la consulta en AFIP (uso interno)"
# si no se especifíca cod_localidad, es util para reconstruir la cache
import wslpg_datos as datos
if not str(cod_localidad) in datos.LOCALIDADES and consultar:
d = self.ConsultarLocalidadesPorProvincia(cod_prov, sep=None)
try:
# actualizar el diccionario persistente (shelve)
datos.LOCALIDADES.update(d)
except Exception, e:
print "EXCEPCION CAPTURADA", e
# capturo errores por permisos (o por concurrencia)
datos.LOCALIDADES = d
return datos.LOCALIDADES.get(str(cod_localidad), "")
def ConsultarTiposOperacion(self, sep="||"):
"Consulta tipo de Operación por Actividad."
ops = []
ret = self.client.tipoActividadConsultar(
auth={
'token': self.Token, 'sign': self.Sign,
'cuit': self.Cuit, },
)['tipoActividadReturn']
self.__analizar_errores(ret)
for it_act in ret.get('tiposActividad', []):
ret = self.client.tipoOperacionXActividadConsultar(
auth={
'token': self.Token, 'sign': self.Sign,
'cuit': self.Cuit, },
nroActLiquida=it_act['codigoDescripcion']['codigo'],
)['tipoOperacionReturn']
self.__analizar_errores(ret)
array = ret.get('tiposOperacion', [])
if sep:
ops.extend([("%s %%s %s %%s %s %%s %s" % (sep, sep, sep, sep)) %
(it_act['codigoDescripcion']['codigo'],
it['codigoDescripcion']['codigo'],
it['codigoDescripcion']['descripcion'])
for it in array])
else:
ops.extend([(it_act['codigoDescripcion']['codigo'],
it['codigoDescripcion']['codigo'],
it['codigoDescripcion']['descripcion'])
for it in array])
return ops
# Funciones para generar PDF:
def CargarFormatoPDF(self, archivo="liquidacion_form_c1116b_wslpg.csv"):
"Cargo el formato de campos a generar desde una planilla CSV"
# si no encuentro archivo, lo busco en el directorio predeterminado:
if not os.path.exists(archivo):
archivo = os.path.join(self.InstallDir, "plantillas", os.path.basename(archivo))
if DEBUG: print "abriendo archivo ", archivo
# inicializo la lista de los elementos:
self.elements = []
for lno, linea in enumerate(open(archivo.encode('latin1')).readlines()):
if DEBUG: print "procesando linea ", lno, linea
args = []
for i,v in enumerate(linea.split(";")):
if not v.startswith("'"):
v = v.replace(",",".")
else:
v = v#.decode('latin1')
if v.strip()=='':
v = None
else:
v = eval(v.strip())
args.append(v)
# corrijo path relativo para las imágenes:
if args[1] == 'I':
if not os.path.exists(args[14]):
args[14] = os.path.join(self.InstallDir, "plantillas", os.path.basename(args[14]))
if DEBUG: print "NUEVO PATH:", args[14]
self.AgregarCampoPDF(*args)
self.AgregarCampoPDF("anulado", 'T', 150, 250, 0, 0,
size=70, rotate=45, foreground=0x808080,
priority=-1)
if HOMO:
self.AgregarCampoPDF("homo", 'T', 100, 250, 0, 0,
size=70, rotate=45, foreground=0x808080,
priority=-1)
# cargo los elementos en la plantilla
self.template.load_elements(self.elements)
return True
def AgregarCampoPDF(self, nombre, tipo, x1, y1, x2, y2,
font="Arial", size=12,
bold=False, italic=False, underline=False,
foreground= 0x000000, background=0xFFFFFF,
align="L", text="", priority=0, **kwargs):
"Agrego un campo a la plantilla"
# convierto colores de string (en hexadecimal)
if isinstance(foreground, basestring): foreground = int(foreground, 16)
if isinstance(background, basestring): background = int(background, 16)
if isinstance(text, unicode): text = text.encode("latin1")
field = {
'name': nombre,
'type': tipo,
'x1': x1, 'y1': y1, 'x2': x2, 'y2': y2,
'font': font, 'size': size,
'bold': bold, 'italic': italic, 'underline': underline,
'foreground': foreground, 'background': background,
'align': align, 'text': text, 'priority': priority}
field.update(kwargs)
self.elements.append(field)
return True
def CrearPlantillaPDF(self, papel="A4", orientacion="portrait"):
"Iniciar la creación del archivo PDF"
# genero el renderizador con propiedades del PDF
t = Template(
format=papel, orientation=orientacion,
title="F 1116 B/C %s" % (self.NroOrden),
author="CUIT %s" % self.Cuit,
subject="COE %s" % self.params_out.get('coe'),
keywords="AFIP Liquidacion Electronica Primaria de Granos",
creator='wslpg.py %s (http://www.PyAfipWs.com.ar)' % __version__,)
self.template = t
return True
def AgregarDatoPDF(self, campo, valor, pagina='T'):
"Agrego un dato a la factura (internamente)"
# corrijo path relativo para las imágenes (compatibilidad hacia atrás):
if campo == 'fondo' and valor.startswith(self.InstallDir):
if not os.path.exists(valor):
valor = os.path.join(self.InstallDir, "plantillas", os.path.basename(valor))
if DEBUG: print "NUEVO PATH:", valor
self.datos[campo] = valor
return True
def ProcesarPlantillaPDF(self, num_copias=1, lineas_max=24, qty_pos='izq',
clave=''):
"Generar el PDF según la factura creada y plantilla cargada"
try:
f = self.template
liq = self.params_out
# actualizo los campos según la clave (ajuste debitos / creditos)
if clave and clave in liq:
liq = liq.copy()
liq.update(liq[clave]) # unificar con AnalizarAjusteCredito/Debito
if HOMO:
self.AgregarDatoPDF("homo", u"HOMOLOGACIÓN")
copias = {1: 'Original', 2: 'Duplicado', 3: 'Triplicado',
4: 'Cuadruplicado', 5: 'Quintuplicado'}
# convierto el formato de intercambio para representar los valores:
fmt_encabezado = dict([(v[0], v[1:]) for v in ENCABEZADO])
fmt_deduccion = dict([(v[0], v[1:]) for v in DEDUCCION])
fmt_retencion = dict([(v[0], v[1:]) for v in RETENCION])
def formatear(campo, valor, formato):
"Convertir el valor a una cadena correctamente s/ formato ($ % ...)"
if campo in formato and v is not None:
fmt = formato[campo]
if fmt[1] == N:
if 'cuit' in campo:
c = str(valor)
if len(c) == 11:
valor = "%s-%s-%s" % (c[0:2], c[2:10], c[10:])
else:
valor = ""
elif 'peso' in campo:
valor = "%s Kg" % valor
elif valor is not None and valor != "":
valor = "%d" % int(valor)
else:
valor = ""
elif fmt[1] == I:
valor = ("%%0.%df" % fmt[2]) % valor
if 'alic' in campo or 'comision' in campo:
valor = valor + " %"
elif 'factor' in campo or 'cont' in campo or 'cant' in campo:
pass
else:
valor = "$ " + valor
elif 'fecha' in campo:
d = valor
if isinstance(d, (datetime.date, datetime.datetime)):
valor = d.strftime("%d/%m/%Y")
else:
valor = "%s/%s/%s" % (d[8:10], d[5:7], d[0:4])
return valor
def buscar_localidad_provincia(cod_prov, cod_localidad):
"obtener la descripción de la provincia/localidad (usar cache)"
cod_prov = int(cod_prov)
cod_localidad = str(cod_localidad)
provincia = datos.PROVINCIAS[cod_prov]
localidad = self.BuscarLocalidades(cod_prov, cod_localidad)
return localidad, provincia
# divido los datos adicionales (debe haber renglones 1 al 9):
if liq.get('datos_adicionales') and f.has_key('datos_adicionales1'):
d = liq.get('datos_adicionales')
for i, ds in enumerate(f.split_multicell(d, 'datos_adicionales1')):
liq['datos_adicionales%s' % (i + 1)] = ds
for copia in range(1, num_copias+1):
# completo campos y hojas
f.add_page()
f.set('copia', copias.get(copia, "Adicional %s" % copia))
f.set('anulado', {'AC': '', '': 'SIN ESTADO',
'AN': "ANULADO"}.get(liq['estado'], "ERROR"))
try:
cod_tipo_ajuste = int(liq["cod_tipo_ajuste"] or '0')
except:
cod_tipo_ajuste = None
f.set('tipo_ajuste', {3: u'Liquidación de Débito',
4: u'Liquidación de Crédito',
}.get(cod_tipo_ajuste, ''))
# limpio datos del corredor si no corresponden:
if liq.get('actua_corredor', 'N') == 'N':
if liq.get('cuit_corredor', None) == 0:
del liq['cuit_corredor']
# establezco campos según tabla encabezado:
for k,v in liq.items():
v = formatear(k, v, fmt_encabezado)
if isinstance(v, (basestring, int, long, float)):
f.set(k, v)
elif isinstance(v, decimal.Decimal):
f.set(k, str(v))
elif isinstance(v, datetime.datetime):
f.set(k, str(v))
import wslpg_datos as datos
campania = int(liq.get('campania_ppal') or 0)
f.set("campania_ppal", datos.CAMPANIAS.get(campania, campania))
f.set("tipo_operacion", datos.TIPOS_OP.get(int(liq.get('cod_tipo_operacion') or 0), ""))
f.set("actividad", datos.ACTIVIDADES.get(int(liq.get('nro_act_comprador') or 0), ""))
if 'cod_grano' in liq and liq['cod_grano']:
cod_grano = int(liq['cod_grano'])
else:
cod_grano = int(self.datos.get('cod_grano') or 0)
f.set("grano", datos.GRANOS.get(cod_grano, ""))
cod_puerto = int(liq.get('cod_puerto', self.datos.get('cod_puerto')) or 0)
if cod_puerto in datos.PUERTOS:
f.set("des_puerto_localidad", datos.PUERTOS[cod_puerto])
cod_grado_ref = liq.get('cod_grado_ref', self.datos.get('cod_grado_ref')) or ""
if cod_grado_ref in datos.GRADOS_REF:
f.set("des_grado_ref", datos.GRADOS_REF[cod_grado_ref])
else:
f.set("des_grado_ref", cod_grado_ref)
cod_grado_ent = liq.get('cod_grado_ent', self.datos.get('cod_grado_ent'))
if 'val_grado_ent' in liq and int(liq.get('val_grado_ent') or 0):
val_grado_ent = liq['val_grado_ent']
elif 'val_grado_ent' in self.datos:
val_grado_ent = self.datos.get('val_grado_ent')
elif cod_grano in datos.GRADO_ENT_VALOR:
valores = datos.GRADO_ENT_VALOR[cod_grano]
if cod_grado_ent in valores:
val_grado_ent = valores[cod_grado_ent]
else:
val_grado_ent = ""
else:
val_grado_ent = ""
f.set("valor_grado_ent", "%s %s" % (cod_grado_ent or "", val_grado_ent or ""))
f.set("cont_proteico", liq.get('cont_proteico', self.datos.get('cont_proteico', "")))
if liq.get('certificados'):
# uso la procedencia del certificado de depósito
cert = liq['certificados'][0]
localidad, provincia = buscar_localidad_provincia(
cert['cod_prov_procedencia'],
cert['cod_localidad_procedencia'])
elif liq.get('cod_prov_procedencia_sin_certificado'):
localidad, provincia = buscar_localidad_provincia(
liq['cod_prov_procedencia_sin_certificado'],
liq['cod_localidad_procedencia_sin_certificado'])
else:
localidad, provincia = "", ""
f.set("procedencia", "%s - %s" % (localidad, provincia))
# si no se especifíca, uso la procedencia para el lugar
if not self.datos.get('lugar_y_fecha'):
localidad, provincia = buscar_localidad_provincia(
liq['cod_prov_procedencia'],
liq['cod_localidad_procedencia'])
lugar = "%s - %s " % (localidad, provincia)
fecha = datetime.datetime.today().strftime("%d/%m/%Y")
f.set("lugar_y_fecha", "%s, %s" % (fecha, lugar))
if 'lugar_y_fecha' in self.datos:
del self.datos['lugar_y_fecha']
if HOMO:
homo = "(pruebas)"
else:
homo = ""
if int(liq['cod_tipo_operacion'] or 0) == 1:
f.set("comprador.L", "COMPRADOR:")
f.set("vendedor.L", "VENDEDOR:")
f.set("formulario", u"Form. Electrónico 1116 B %s" % homo)
else:
f.set("comprador.L", "MANDATARIO/CONSIGNATARIO:")
f.set("vendedor.L", "MANDANTE/COMITENTE:")
f.set("formulario", u"Form. Electrónico 1116 C %s" % homo)
if int(liq.get("coe_ajustado") or 0) or int(liq.get("nro_contrato") or 0):
f.set("formulario", u"Ajuste Unificado %s" % homo)
certs = []
for cert in liq.get('certificados', []):
certs.append(u"%s%s" % (
datos.TIPO_CERT_DEP[int(cert['tipo_certificado_deposito'])],
cert['nro_certificado_deposito']))
f.set("certificados_deposito", ', '.join(certs))
for i, deduccion in enumerate(liq.get('deducciones', [])):
for k, v in deduccion.items():
v = formatear(k, v, fmt_deduccion)
f.set("deducciones_%s_%02d" % (k, i + 1), v)
for i, retencion in enumerate(liq.get('retenciones', [])):
for k, v in retencion.items():
v = formatear(k, v, fmt_retencion)
f.set("retenciones_%s_%02d" % (k, i + 1), v)
if retencion['importe_certificado_retencion']:
d = retencion['fecha_certificado_retencion']
f.set('retenciones_cert_retencion_%02d' % (i + 1),
"%s $ %0.2f %s" % (
retencion['nro_certificado_retencion'] or '',
retencion['importe_certificado_retencion'],
"%s/%s/%s" % (d[8:10], d[5:7], d[2:4]),
))
# cargo campos adicionales ([PDF] en .ini y AgregarDatoPDF)
for k,v in self.datos.items():
f.set(k, v)
# Ajustes: