forked from reingart/pyafipws
-
Notifications
You must be signed in to change notification settings - Fork 0
/
soap.py
176 lines (159 loc) · 7.31 KB
/
soap.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
#!/usr/bin/python
# -*- coding: latin-1 -*-
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU Lesser 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.
"Implementación pythónica de cliente SOAP"
__author__ = "Mariano Reingart (mariano@nsis.com.ar)"
__copyright__ = "Copyright (C) 2008 Mariano Reingart"
__license__ = "LGPL 3.0"
__version__ = "1.0"
import httplib2
from simplexml import SimpleXMLElement
class SoapFault(RuntimeError):
def __init__(self,faultcode,faultstring):
self.faultcode = faultcode
self.faultstring = faultstring
class SoapClient(object):
"Manejo de Cliente SOAP Simple (símil PHP)"
def __init__(self, location = None, action = None, namespace = None,
cert = None, trace = False, exceptions = False, proxy = None ):
self.certssl = cert
self.keyssl = None
self.location = location # server location (url)
self.action = action # SOAP base action
self.namespace = namespace # message
self.trace = trace # show debug messages
self.exceptions = exceptions # lanzar execpiones? (Soap Faults)
if not proxy:
self.http = httplib2.Http('.cache')
else:
import socks
##httplib2.debuglevel=4
self.http = httplib2.Http(proxy_info = httplib2.ProxyInfo(
proxy_type=socks.PROXY_TYPE_HTTP, **proxy))
#if self.certssl: # esto funciona para validar al server?
# self.http.add_certificate(self.keyssl, self.keyssl, self.certssl)
def __getattr__(self, attr):
"Devuelve un pseudo-método que puede ser llamado"
return lambda self=self,xml="", *args, **kwargs: self.call(attr,xml,*args,**kwargs)
def call(self, method, *args, **kwargs):
"Prepara el xml y realiza la llamada SOAP, devuelve un SimpleXMLElement"
# Mensaje de Solicitud SOAP básico:
xml = """<?xml version="1.0" encoding="utf-8"?>
<soap:Envelope xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:xsd="http://www.w3.org/2001/XMLSchema"
xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/">
<soap:Body>
<%(method)s xmlns="%(namespace)s">
</%(method)s>
</soap:Body>
</soap:Envelope>""" % dict(method=method, namespace= self.namespace)
request = SimpleXMLElement(xml)
# parsear argumentos
for k,v in kwargs.items(): # dict: tag=valor
self.parse(getattr(request,method),k,v)
self.xml_request = request.asXML()
self.xml_response = self.send(method, self.xml_request)
response = SimpleXMLElement(self.xml_response)
if self.exceptions and ("soapenv:Fault" in response or "soap:Fault" in response):
raise SoapFault(unicode(response.faultcode), unicode(response.faultstring))
return response
def parse(self, node, tag, value, add_child=True):
"Analiza un objeto y devuelve su representación XML"
if isinstance(value, dict): # serializar diccionario (<key>value</key>)
child = add_child and node.addChild(tag) or node
for k,v in value.items():
self.parse(child, k, v)
elif isinstance(value, tuple): # serializar diccionario (<key>value</key>)
child = add_child and node.addChild(tag) or node
for k,v in value:
self.parse(getattr(node,tag), k, v)
elif isinstance(value, list): # serializar listas
child=node.addChild(tag)
for t in value:
self.parse(child,tag,t, False)
elif isinstance(value, basestring): # no volver a convertir los strings y unicodes
node.addChild(tag,value)
else: # el resto de los objetos se convierten a string
node.addChild(tag,str(value)) # habria que agregar un método asXML?
def send(self, method, xml):
"Envía el pedido SOAP por HTTP (llama al método con el xml como cuerpo)"
if self.location == 'test': return
location = "%s" % self.location #?op=%s" % (self.location, method)
headers={
'Content-type': 'text/xml; charset="UTF-8"',
'Content-length': str(len(xml)),
"SOAPAction": "\"%s%s\"" % (self.action,method)
}
if self.trace:
print "-"*80
print "POST %s" % location
print '\n'.join(["%s: %s" % (k,v) for k,v in headers.items()])
print "\n%s" % xml
response, content = self.http.request(
location,"POST", body=xml, headers=headers )
self.response = response
self.content = content
if self.trace:
print
print '\n'.join(["%s: %s" % (k,v) for k,v in response.items()])
print content
print "="*80
return content
def parse_proxy(proxy_str):
"Parses proxy address user:pass@host:port into a dict suitable for httplib2"
proxy_dict = {}
if proxy_str is None:
return
if "@" in proxy_str:
user_pass, host_port = proxy_str.split("@")
else:
user_pass, host_port = "", proxy_str
if ":" in host_port:
host, port = host_port.split(":")
proxy_dict['proxy_host'], proxy_dict['proxy_port'] = host, int(port)
if ":" in user_pass:
proxy_dict['proxy_user'], proxy_dict['proxy_pass'] = user_pass.split(":")
return proxy_dict
if __name__=="__main__":
# Demo & Test: Feriados (Ministerio del Interior):
from datetime import datetime, timedelta
client = SoapClient(
location = "http://webservices.mininterior.gov.ar/Feriados/Service.svc",
action = 'http://tempuri.org/IMyService/', # SOAPAction
namespace = "http://tempuri.org/FeriadoDS.xsd",
trace = True)
dt1 = datetime.today() - timedelta(days=60)
dt2 = datetime.today() + timedelta(days=60)
feriadosXML = client.FeriadosEntreFechasAsXml(dt1=dt1.isoformat(), dt2=dt2.isoformat());
print feriadosXML
##print parse_proxy(None)
##print parse_proxy("host:1234")
##print parse_proxy("user:pass@host:1234")
##sys.exit(0)
# Demo & Test:
token = "PD94bWwgdmVyc2lvbj0iMS4wIiBlbmNvZGluZz0iVVRGLTgiIHN0YW5kYW"
sign = "gXVvzVwRrfkUAZKoy8ZqA3AL8IZgVxUvOHQH6g1/XzZJns1/k0lUdJslkzW"
cuit = long(30199999)
id = 1234
cbte =199
client = SoapClient(
location = "https://wswhomo.afip.gov.ar/wsfe/service.asmx",
action = 'http://ar.gov.afip.dif.facturaelectronica/', # SOAPAction
namespace = "http://ar.gov.afip.dif.facturaelectronica/",
trace = True)
results = client.FERecuperaQTYRequest(
argAuth= {"Token": token, "Sign": sign, "cuit":long(cuit)}
)
if int(results.FERecuperaQTYRequestResult.RError.percode) != 0:
print "Percode: %s" % results.FERecuperaQTYRequestResult.RError.percode
print "MSGerror: %s" % results.FERecuperaQTYRequestResult.RError.perrmsg
else:
print int(results.FERecuperaQTYRequestResult.qty.value)