diff --git a/README.md b/README.md index c0f99532c..4fca00f70 100644 --- a/README.md +++ b/README.md @@ -97,7 +97,7 @@ You'll need a digital certificate (.crt) and private key (.key) to authenticate Provisionally, you can use author's testing certificate/key: ``` -wget https://www.sistemasagiles.com.ar/soft/pyafipws/reingart.zip +wget https://www.sistemasagiles.com.ar/soft/pyafipws/reingart2021.zip unzip reingart.zip ``` diff --git a/setup.cfg b/setup.cfg index 496266bde..0a88f53b1 100644 --- a/setup.cfg +++ b/setup.cfg @@ -31,6 +31,7 @@ console_scripts = wsbfev1 = pyafipws.wsbfev1:main wscdc = pyafipws.wscdc:main wscoc = pyafipws.wscoc:main + wscpe = pyafipws.wscpe:main wsct = pyafipws.wsct:main wsctg = pyafipws.wsctg:main wsfecred = pyafipws.wsfecred:main diff --git a/wscpe.py b/wscpe.py index 171cf9fe3..c33d3bc01 100644 --- a/wscpe.py +++ b/wscpe.py @@ -44,7 +44,7 @@ http://www.sistemasagiles.com.ar/trac/wiki/PyAfipWs """ -AYUDA=""" +AYUDA = """ Opciones: --ayuda: este mensaje @@ -224,12 +224,18 @@ def __analizar_errores(self, ret): if errores: errores = errores[0] self.ErrCode = " ".join(["%(codigo)s" % err for err in errores]) - self.ErrMsg = "\n".join(["%(codigo)s - %(descripcion)s" % err for err in errores]) + self.ErrMsg = "\n".join( + ["%(codigo)s - %(descripcion)s" % err for err in errores] + ) def __analizar_observaciones(self, ret): "Comprueba y extrae observaciones si existen en la respuesta XML" - self.Observaciones = [obs["codigoDescripcion"] for obs in ret.get("arrayObservaciones", [])] - self.Obs = "\n".join(["%(codigo)s: %(descripcion)s" % obs for obs in self.Observaciones]) + self.Observaciones = [ + obs["codigoDescripcion"] for obs in ret.get("arrayObservaciones", []) + ] + self.Obs = "\n".join( + ["%(codigo)s: %(descripcion)s" % obs for obs in self.Observaciones] + ) def __analizar_evento(self, ret): "Comprueba y extrae el wvento informativo si existen en la respuesta XML" @@ -362,7 +368,9 @@ def AgregarIntervinientes( return True @inicializar_y_capturar_excepciones - def AgregarDatosCarga(self, cod_grano=None, cosecha=None, peso_bruto=None, peso_tara=None, **kwargs): + def AgregarDatosCarga( + self, cod_grano=None, cosecha=None, peso_bruto=None, peso_tara=None, **kwargs + ): """Inicializa internamente los datos de carga para una cpe.""" datos_carga = { "codGrano": cod_grano, @@ -485,9 +493,9 @@ def AgregarCerrarContingencia( ): """Inicializa datos para el cierre, la reactivacion, extension de una contingencias en cpe. - A: Reactivación para descarga en destino. - B: Extensión cierre contingencia. - C: Desactivar definitivamente la CP. + A: Reactivación para descarga en destino. + B: Extensión cierre contingencia. + C: Desactivar definitivamente la CP. """ if concepto == "A": motivo = { @@ -495,12 +503,15 @@ def AgregarCerrarContingencia( "reactivacionDestino": { "cuitTransportista": cuit_transportista, "nroOperativo": nro_operativo, - } + }, } else: motivo = { "concepto": concepto, # A, B, C - distintos para la desactivacion - "motivoDesactivacionCP": {"concepto": concepto_desactivacion, "descripcion": descripcion} + "motivoDesactivacionCP": { + "concepto": concepto_desactivacion, + "descripcion": descripcion, + }, } self.cpe.update(motivo) @@ -513,7 +524,7 @@ def AutorizarCPEFerroviaria(self, archivo="cpe_ferroviaria.pdf"): "sign": self.Sign, "cuitRepresentada": self.Cuit, }, - solicitud=self.cpe + solicitud=self.cpe, ) ret = response.get("respuesta") if ret: @@ -661,17 +672,19 @@ def EditarCPEFerroviaria( **kwargs ): """Modificar datos de una CP Ferroviaria en estado Activo.""" - self.cpe.update({ - "nroCTG": nro_ctg, - "cuitCorredorVentaPrimaria": cuit_corredor_venta_primaria, - "cuitCorredorVentaSecundaria": cuit_corredor_venta_secundaria, - "cuitRemitenteComercialVentaPrimaria": cuit_remitente_comercial_venta_primaria, - "cuitRemitenteComercialVentaSecundaria": cuit_remitente_comercial_venta_secundaria, - "cuitRemitenteComercialVentaSecundaria2": cuit_remitente_comercial_venta_secundaria2, - "cuitTransportista": cuit_transportista, - "pesoBruto": peso_bruto, - "codGrano": cod_grano, - }) + self.cpe.update( + { + "nroCTG": nro_ctg, + "cuitCorredorVentaPrimaria": cuit_corredor_venta_primaria, + "cuitCorredorVentaSecundaria": cuit_corredor_venta_secundaria, + "cuitRemitenteComercialVentaPrimaria": cuit_remitente_comercial_venta_primaria, + "cuitRemitenteComercialVentaSecundaria": cuit_remitente_comercial_venta_secundaria, + "cuitRemitenteComercialVentaSecundaria2": cuit_remitente_comercial_venta_secundaria2, + "cuitTransportista": cuit_transportista, + "pesoBruto": peso_bruto, + "codGrano": cod_grano, + } + ) response = self.client.editarCPEFerroviaria( auth={ "token": self.Token, @@ -689,7 +702,13 @@ def EditarCPEFerroviaria( @inicializar_y_capturar_excepciones def ConsultarCPEFerroviaria( - self, tipo_cpe=None, cuit_solicitante=None, sucursal=None, nro_orden=None, nro_ctg=None, archivo="cpe.pdf" + self, + tipo_cpe=None, + cuit_solicitante=None, + sucursal=None, + nro_orden=None, + nro_ctg=None, + archivo="cpe.pdf", ): """Busca una CPE existente según parámetros de búsqueda y retorna información de la misma.""" if not nro_ctg: @@ -735,12 +754,15 @@ def ConsultaCPEFerroviariaPorNroOperativo(self, nro_operativo=1111111111): # respuesta distinta devuelve un array de resultados 'resumenCartaPorte' if ret: self.__analizar_errores(ret) - if 'resumenCartaPorte' in ret: + if "resumenCartaPorte" in ret: cps = [ - '\n'.join('{}: {}'.format(campo, valor) for campo, valor in carta_porte.items()) - for carta_porte in ret['resumenCartaPorte'] + "\n".join( + "{}: {}".format(campo, valor) + for campo, valor in carta_porte.items() + ) + for carta_porte in ret["resumenCartaPorte"] ] - return '\n==========\n'.join(cps) + return "\n==========\n".join(cps) @inicializar_y_capturar_excepciones def ConfirmacionDefinitivaCPEFerroviaria(self, archivo="cpe.pdf"): @@ -769,7 +791,7 @@ def CerrarContingenciaCPE(self, archivo="cpe.pdf"): "sign": self.Sign, "cuitRepresentada": self.Cuit, }, - solicitud=self.cpe + solicitud=self.cpe, ) ret = response.get("respuesta") if ret: @@ -787,7 +809,7 @@ def NuevoDestinoDestinatarioCPEFerroviaria(self, archivo="cpe.pdf"): "sign": self.Sign, "cuitRepresentada": self.Cuit, }, - solicitud=self.cpe + solicitud=self.cpe, ) ret = response.get("respuesta") if ret: @@ -805,7 +827,7 @@ def RegresoOrigenCPEFerroviaria(self, archivo="cpe.pdf"): "sign": self.Sign, "cuitRepresentada": self.Cuit, }, - solicitud=self.cpe + solicitud=self.cpe, ) ret = response.get("respuesta") if ret: @@ -850,19 +872,21 @@ def EditarCPEAutomotor( **kwargs ): """Modificar datos de una CP Automotor en estado Activo.""" - self.cpe.update({ - "nroCTG": nro_ctg, - "cuitCorredorVentaPrimaria": cuit_corredor_venta_primaria, - "cuitCorredorVentaSecundaria": cuit_corredor_venta_secundaria, - "cuitRemitenteComercialVentaPrimaria": cuit_remitente_comercial_venta_primaria, - "cuitRemitenteComercialVentaSecundaria": cuit_remitente_comercial_venta_secundaria, - "cuitRemitenteComercialVentaSecundaria2": cuit_remitente_comercial_venta_secundaria2, - "cuitChofer": cuit_chofer, - "cuitTransportista": cuit_transportista, - "pesoBruto": peso_bruto, - "codGrano": cod_grano, - "dominio": dominio, - }) + self.cpe.update( + { + "nroCTG": nro_ctg, + "cuitCorredorVentaPrimaria": cuit_corredor_venta_primaria, + "cuitCorredorVentaSecundaria": cuit_corredor_venta_secundaria, + "cuitRemitenteComercialVentaPrimaria": cuit_remitente_comercial_venta_primaria, + "cuitRemitenteComercialVentaSecundaria": cuit_remitente_comercial_venta_secundaria, + "cuitRemitenteComercialVentaSecundaria2": cuit_remitente_comercial_venta_secundaria2, + "cuitChofer": cuit_chofer, + "cuitTransportista": cuit_transportista, + "pesoBruto": peso_bruto, + "codGrano": cod_grano, + "dominio": dominio, + } + ) response = self.client.editarCPEAutomotor( auth={ "token": self.Token, @@ -880,7 +904,13 @@ def EditarCPEAutomotor( @inicializar_y_capturar_excepciones def ConsultarCPEAutomotor( - self, tipo_cpe=None, cuit_solicitante=None, sucursal=None, nro_orden=None, nro_ctg=None, archivo="cpe.pdf" + self, + tipo_cpe=None, + cuit_solicitante=None, + sucursal=None, + nro_orden=None, + nro_ctg=None, + archivo="cpe.pdf", ): """Busca una CPE existente según parámetros de búsqueda y retorna información de la misma.""" if not nro_ctg: @@ -987,7 +1017,7 @@ def ConsultarCPEPorDestino( fecha_partida_desde=None, fecha_partida_hasta=None, tipo_cpe=None, - sep="||" + sep="||", ): """Consulta de CPE en calidad de destino para una planta y rango de fechas específicos.""" solicitud = { @@ -1002,7 +1032,7 @@ def ConsultarCPEPorDestino( "sign": self.Sign, "cuitRepresentada": self.Cuit, }, - solicitud=solicitud + solicitud=solicitud, ) ret = response.get("respuesta") self.__analizar_errores(ret) @@ -1018,8 +1048,13 @@ def ConsultarCPEPorDestino( ] array.extend(ret.get("cartaPorte", [])) return [ - ("%s {nroCTG} %s {fechaPartida} %s {estado} %s {fechaUltimaModificacion} %s" % (sep, sep, sep, sep, sep)).format(**it) - if sep else it for it in array + ( + "%s {nroCTG} %s {fechaPartida} %s {estado} %s {fechaUltimaModificacion} %s" + % (sep, sep, sep, sep, sep) + ).format(**it) + if sep + else it + for it in array ] @inicializar_y_capturar_excepciones @@ -1035,7 +1070,7 @@ def ConsultarCPEPendientesDeResolucion(self, perfil=None, planta=None, sep="||") "sign": self.Sign, "cuitRepresentada": self.Cuit, }, - solicitud=solicitud + solicitud=solicitud, ) ret = response.get("respuesta") self.__analizar_errores(ret) @@ -1051,8 +1086,13 @@ def ConsultarCPEPendientesDeResolucion(self, perfil=None, planta=None, sep="||") ] array.extend(ret.get("cartaPorte", [])) return [ - ("%s {nroCTG} %s {fechaPartida} %s {estado} %s {fechaUltimaModificacion} %s" % (sep, sep, sep, sep, sep)).format(**it) - if sep else it for it in array + ( + "%s {nroCTG} %s {fechaPartida} %s {estado} %s {fechaUltimaModificacion} %s" + % (sep, sep, sep, sep, sep) + ).format(**it) + if sep + else it + for it in array ] @inicializar_y_capturar_excepciones @@ -1085,7 +1125,12 @@ def ConsultarProvincias(self, sep="||"): ret = response.get("respuesta") self.__analizar_errores(ret) array = ret.get("provincia", []) - return [("%s {codigo} %s {descripcion} %s" % (sep, sep, sep)).format(**it) if sep else it for it in array] + return [ + ("%s {codigo} %s {descripcion} %s" % (sep, sep, sep)).format(**it) + if sep + else it + for it in array + ] @inicializar_y_capturar_excepciones def ConsultarLocalidadesPorProvincia(self, cod_provincia=1, sep="||"): @@ -1101,7 +1146,12 @@ def ConsultarLocalidadesPorProvincia(self, cod_provincia=1, sep="||"): ret = response.get("respuesta") self.__analizar_errores(ret) array = ret.get("localidad", []) - return [("%s {codigo} %s {descripcion} %s" % (sep, sep, sep)).format(**it) if sep else it for it in array] + return [ + ("%s {codigo} %s {descripcion} %s" % (sep, sep, sep)).format(**it) + if sep + else it + for it in array + ] @inicializar_y_capturar_excepciones def ConsultarTiposGrano(self, sep="||"): @@ -1116,7 +1166,12 @@ def ConsultarTiposGrano(self, sep="||"): ret = response.get("respuesta") self.__analizar_errores(ret) array = ret.get("grano", []) - return [("%s {codigo} %s {descripcion} %s" % (sep, sep, sep)).format(**it) if sep else it for it in array] + return [ + ("%s {codigo} %s {descripcion} %s" % (sep, sep, sep)).format(**it) + if sep + else it + for it in array + ] @inicializar_y_capturar_excepciones def ConsultarLocalidadesProductor(self, cuit_productor=None, sep="||"): @@ -1132,7 +1187,12 @@ def ConsultarLocalidadesProductor(self, cuit_productor=None, sep="||"): ret = response.get("respuesta") self.__analizar_errores(ret) array = ret.get("localidad", []) - return [("%s {codigo} %s {descripcion} %s" % (sep, sep, sep)).format(**it) if sep else it for it in array] + return [ + ("%s {codigo} %s {descripcion} %s" % (sep, sep, sep)).format(**it) + if sep + else it + for it in array + ] @inicializar_y_capturar_excepciones def ConsultarPlantas(self, cuit, sep="||"): @@ -1153,13 +1213,18 @@ def ConsultarPlantas(self, cuit, sep="||"): { "nroPlanta": "Nro Planta", "codProvincia": "Cod Provincia", - "codLocalidad": "Cod Localidad" + "codLocalidad": "Cod Localidad", } ] array.extend(ret.get("planta", [])) return [ - ("%s {nroPlanta} %s {codProvincia} %s {codLocalidad} %s" % (sep, sep, sep, sep)).format(**it) - if sep else it for it in array + ( + "%s {nroPlanta} %s {codProvincia} %s {codLocalidad} %s" + % (sep, sep, sep, sep) + ).format(**it) + if sep + else it + for it in array ] @inicializar_y_capturar_excepciones @@ -1173,7 +1238,8 @@ def Dummy(self): INSTALL_DIR = WSCPE.InstallDir = get_install_dir() -if __name__ == "__main__": + +def main(): # obteniendo el TA from .wsaa import WSAA @@ -1212,14 +1278,14 @@ def Dummy(self): cuit_solicitante=CUIT, sucursal=221, nro_orden=nro_orden, - observaciones="Notas del transporte" + observaciones="Notas del transporte", ) ok = wscpe.AgregarOrigen( # planta=1, # cod_provincia_operador=12, # cod_localidad_operador=7717, cod_provincia_productor=1, - cod_localidad_productor=14310 + cod_localidad_productor=14310, ) ok = wscpe.AgregarDestino( planta=1938, @@ -1281,10 +1347,7 @@ def Dummy(self): if "--autorizar_cpe_ferroviaria" in sys.argv: ok = wscpe.CrearCPE() ok = wscpe.AgregarCabecera( - sucursal=1, - nro_orden=1, - planta=1, - observaciones="Notas del transporte" + sucursal=1, nro_orden=1, planta=1, observaciones="Notas del transporte" ) ok = wscpe.AgregarDestino( cuit_destinatario=30000000006, @@ -1352,15 +1415,21 @@ def Dummy(self): wscpe.AnularCPE() if "--rechazo_cpe" in sys.argv: - wscpe.AgregarCabecera(cuit_solicitante=CUIT, tipo_cpe=74, sucursal=1, nro_orden=1) + wscpe.AgregarCabecera( + cuit_solicitante=CUIT, tipo_cpe=74, sucursal=1, nro_orden=1 + ) wscpe.RechazoCPE() if "--confirmar_arribo_cpe" in sys.argv: - wscpe.AgregarCabecera(cuit_solicitante=CUIT, tipo_cpe=74, sucursal=1, nro_orden=1) + wscpe.AgregarCabecera( + cuit_solicitante=CUIT, tipo_cpe=74, sucursal=1, nro_orden=1 + ) wscpe.ConfirmarArriboCPE() if "--informar_contingencia" in sys.argv: - wscpe.AgregarCabecera(cuit_solicitante=CUIT, tipo_cpe=74, sucursal=1, nro_orden=1) + wscpe.AgregarCabecera( + cuit_solicitante=CUIT, tipo_cpe=74, sucursal=1, nro_orden=1 + ) wscpe.AgregarContingencia(concepto="F", descripcion="XXXXX") wscpe.InformarContingencia() @@ -1395,7 +1464,9 @@ def Dummy(self): ) if "--consultar_cpe_ferroviaria" in sys.argv: - wscpe.ConsultarCPEFerroviaria(tipo_cpe=75, sucursal=1, nro_orden=1, cuit_solicitante=CUIT) + wscpe.ConsultarCPEFerroviaria( + tipo_cpe=75, sucursal=1, nro_orden=1, cuit_solicitante=CUIT + ) if "--consulta_cpe_ferroviaria_por_nro_operativo" in sys.argv: wscpe.ConsultaCPEFerroviariaPorNroOperativo(nro_operativo=1111111111) @@ -1453,7 +1524,7 @@ def Dummy(self): wscpe.AgregarTransporte( codigo_ramal=99, descripcion_ramal="Ok", - fecha_hora_partida=datetime.datetime.now()+datetime.timedelta(days=1), + fecha_hora_partida=datetime.datetime.now() + datetime.timedelta(days=1), km_recorrer=333, ) wscpe.NuevoDestinoDestinatarioCPEFerroviaria() @@ -1519,10 +1590,14 @@ def Dummy(self): ) if "--consultar_cpe_automotor" in sys.argv: - wscpe.ConsultarCPEAutomotor(tipo_cpe=74, sucursal=1, nro_orden=1, cuit_solicitante=CUIT) + wscpe.ConsultarCPEAutomotor( + tipo_cpe=74, sucursal=1, nro_orden=1, cuit_solicitante=CUIT + ) if "--confirmacion_definitiva_cpe_automotor" in sys.argv: - wscpe.AgregarCabecera(cuit_solicitante=CUIT, tipo_cpe=74, sucursal=1, nro_orden=1) + wscpe.AgregarCabecera( + cuit_solicitante=CUIT, tipo_cpe=74, sucursal=1, nro_orden=1 + ) # wscpe.AgregarIntervinientes(cuit_representante_recibidor=20222222223) wscpe.AgregarDatosCarga(peso_bruto=1000, peso_tara=10000) wscpe.ConfirmacionDefinitivaCPEAutomotor() @@ -1530,31 +1605,55 @@ def Dummy(self): if "--nuevo_destino_destinatario_cpe_automotor" in sys.argv: wscpe.AgregarCabecera(tipo_cpe=74, sucursal=1, nro_orden=1) wscpe.AgregarDestino( - cuit_destino=20111111112, cod_provincia=1, cod_localidad=10216, planta=1, es_destino_campo=True, cuit_destinatario=30000000006 + cuit_destino=20111111112, + cod_provincia=1, + cod_localidad=10216, + planta=1, + es_destino_campo=True, + cuit_destinatario=30000000006, + ) + wscpe.AgregarTransporte( + fecha_hora_partida=datetime.datetime.now(), + km_recorrer=333, + codigo_turno="00", ) - wscpe.AgregarTransporte(fecha_hora_partida=datetime.datetime.now(), km_recorrer=333, codigo_turno="00") wscpe.NuevoDestinoDestinatarioCPEAutomotor() if "--regreso_origen_cpe_automotor" in sys.argv: wscpe.AgregarCabecera(tipo_cpe=74, sucursal=1, nro_orden=1) - wscpe.AgregarTransporte(fecha_hora_partida=datetime.datetime.now(), km_recorrer=333, codigo_turno="00") + wscpe.AgregarTransporte( + fecha_hora_partida=datetime.datetime.now(), + km_recorrer=333, + codigo_turno="00", + ) wscpe.RegresoOrigenCPEAutomotor() if "--desvio_cpe_automotor" in sys.argv: - wscpe.AgregarCabecera(cuit_solicitante=CUIT, tipo_cpe=74, sucursal=1, nro_orden=1) + wscpe.AgregarCabecera( + cuit_solicitante=CUIT, tipo_cpe=74, sucursal=1, nro_orden=1 + ) wscpe.AgregarDestino( - cuit_destino=20111111112, cod_provincia=1, cod_localidad=10216, planta=1, es_destino_campo=True # newton + cuit_destino=20111111112, + cod_provincia=1, + cod_localidad=10216, + planta=1, + es_destino_campo=True, # newton + ) + wscpe.AgregarTransporte( + fecha_hora_partida=datetime.datetime.now(), + km_recorrer=333, + codigo_turno="00", ) - wscpe.AgregarTransporte(fecha_hora_partida=datetime.datetime.now(), km_recorrer=333, codigo_turno="00") wscpe.DesvioCPEAutomotor() if "--consultar_cpe_por_destino" in sys.argv: today = datetime.datetime.now().date() ret = wscpe.ConsultarCPEPorDestino( planta=1938, - fecha_partida_desde=today - datetime.timedelta(days=3), # solo hasta 3 dias antes + fecha_partida_desde=today + - datetime.timedelta(days=3), # solo hasta 3 dias antes fecha_partida_hasta=today, - tipo_cpe=74 # opcional + tipo_cpe=74, # opcional ) if ret: print("\n".join(ret)) @@ -1596,3 +1695,7 @@ def Dummy(self): if wscpe.Errores: print("Error:", wscpe.ErrMsg) + + +if __name__ == "__main__": + main()