Skip to content

Commit 942883e

Browse files
ivanalemuniozivanalemunioz
committed
Create web servicesand electromic billing classes
1 parent 4c48469 commit 942883e

File tree

4 files changed

+296
-4
lines changed

4 files changed

+296
-4
lines changed

‎afip/__init__.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1 +1 @@
1-
from .afip import *
1+
from .afip import Afip

‎afip/afip.py‎

Lines changed: 15 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,20 +1,25 @@
11
import http.client
22
import json
33

4+
from .webservice import WebService
5+
from .electronicbilling import ElectronicBilling
6+
47
class Afip:
58
sdk_version_number = '1.0.0'
6-
9+
710
def __init__(self, options: dict):
811
if not(options.get("CUIT")):
912
raise Exception("CUIT field is required in options")
1013

1114
self.CUIT: int = options.get("CUIT")
12-
self.production: bool = options.get("production") if options.get("production") else False
15+
self.production: bool = options.get("production") if options.get("production") == True else False
1316
self.environment: str = "prod" if self.production == True else "dev"
1417
self.cert: str = options.get("cert")
1518
self.key: str = options.get("key")
1619
self.access_token: str = options.get("access_token")
1720

21+
self.ElectronicBilling = ElectronicBilling(self)
22+
1823

1924
# Gets token authorization for an AFIP Web Service
2025
#
@@ -75,4 +80,11 @@ def getLastRequestXML(self) -> dict:
7580
raise Exception(data.decode("utf-8"))
7681

7782
return json.loads(data.decode("utf-8"))
78-
83+
84+
85+
# Create generic Web Service
86+
def webService(self, service, options: dict = {}) -> WebService:
87+
options['service'] = service
88+
options['generic'] = True
89+
90+
return WebService(self, options)

‎afip/electronicbilling.py‎

Lines changed: 225 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,225 @@
1+
import http.client
2+
import json
3+
import re
4+
5+
from .webservice import WebService
6+
7+
class ElectronicBilling(WebService):
8+
soapv12 = True
9+
WSDL = "https://servicios1.afip.gov.ar/wsfev1/service.asmx?WSDL"
10+
URL = "https://servicios1.afip.gov.ar/wsfev1/service.asmx"
11+
WSDL_TEST = "https://wswhomo.afip.gov.ar/wsfev1/service.asmx?WSDL"
12+
URL_TEST = "https://wswhomo.afip.gov.ar/wsfev1/service.asmx"
13+
14+
def __init__(self, afip):
15+
super(ElectronicBilling, self).__init__(afip, { "service": "wsfe" })
16+
17+
# Create PDF
18+
def createPDF(self, data: dict):
19+
conn = http.client.HTTPSConnection("app.afipsdk.com")
20+
21+
headers = {
22+
"Content-Type": "application/json",
23+
"sdk-version-number": self.afip.sdk_version_number,
24+
"sdk-library": "python",
25+
"sdk-environment": self.afip.environment
26+
}
27+
28+
if self.afip.access_token: headers["Authorization"] = "Bearer %s" % self.afip.access_token
29+
30+
conn.request("POST", "/api/v1/pdfs", json.dumps(data), headers)
31+
32+
res = conn.getresponse()
33+
34+
data = res.read()
35+
36+
if res.getcode() >= 400:
37+
raise Exception(data.decode("utf-8"))
38+
39+
response_data = json.loads(data.decode("utf-8"))
40+
41+
return {
42+
"file": response_data["file"],
43+
"file_name": response_data["file_name"]
44+
}
45+
46+
# Gets last voucher number
47+
def getLastVoucher(self, sales_point: int, type: int):
48+
req = {
49+
"PtoVta": sales_point,
50+
"CbteTipo": type
51+
}
52+
53+
self.executeRequest("FECompUltimoAutorizado", req)["CbteNro"]
54+
55+
# Create a voucher from AFIP
56+
def createVoucher(self, data: dict, return_response: bool = False):
57+
# Reassign data to avoid modify te original object
58+
data = data.copy()
59+
60+
req = {
61+
"FeCAEReq": {
62+
"FeCabReq": {
63+
"CantReg": data["CbteHasta"] - data["CbteDesde"] + 1,
64+
"PtoVta": data["PtoVta"],
65+
"CbteTipo": data["CbteTipo"]
66+
},
67+
"FeDetReq": {
68+
"FECAEDetRequest": data
69+
}
70+
}
71+
}
72+
73+
data.pop("CantReg")
74+
data.pop("PtoVta")
75+
data.pop("CbteTipo")
76+
77+
if data.get("Tributos"): data["Tributos"] = { "Tributo": data["Tributos"] }
78+
if data.get("Iva"): data["Iva"] = { "AlicIva": data["Iva"] }
79+
if data.get("CbtesAsoc"): data["CbtesAsoc"] = { "CbteAsoc": data["CbtesAsoc"] }
80+
if data.get("Compradores"): data["Compradores"] = { "Comprador": data["Compradores"] }
81+
if data.get("Opcionales"): data["Opcionales"] = { "Opcional": data["Opcionales"] }
82+
83+
results = self.executeRequest("FECAESolicitar", req)
84+
85+
if return_response == True:
86+
return results
87+
88+
if type(results["FeDetResp"]["FECAEDetResponse"]) in (tuple, list):
89+
results["FeDetResp"]["FECAEDetResponse"] = results["FeDetResp"]["FECAEDetResponse"][0]
90+
91+
return {
92+
"CAE": results["FeDetResp"]["FECAEDetResponse"]["CAE"],
93+
"CAEFchVto": self.formatDate(results["FeDetResp"]["FECAEDetResponse"]["CAEFchVto"])
94+
}
95+
96+
# Create next voucher from AFIP
97+
def createNextVoucher(self, data: dict):
98+
# Reassign data to avoid modify te original object
99+
data = data.copy()
100+
101+
lastVoucher = self.getLastVoucher(data["PtoVta"], data["CbteTipo"])
102+
103+
voucherNumber = lastVoucher + 1
104+
105+
data["CbteDesde"] = voucherNumber
106+
data["CbteHasta"] = voucherNumber
107+
108+
res = self.createVoucher(data)
109+
110+
res["voucherNumber"] = voucherNumber
111+
112+
return res
113+
114+
# Get complete voucher information
115+
def getVoucherInfo(self, number: int, sales_point: int, type: int):
116+
req = {
117+
"FeCompConsReq": {
118+
"CbteNro": number,
119+
"PtoVta": sales_point,
120+
"CbteTipo": type
121+
}
122+
}
123+
124+
return self.executeRequest("FECompConsultar", req)
125+
126+
# Create CAEA
127+
def createCAEA(self, period: int, fortnight: int):
128+
req = {
129+
"Periodo": period,
130+
"Orden": fortnight
131+
}
132+
133+
return self.executeRequest("FECAEASolicitar", req)["ResultGet"]
134+
135+
# Get CAEA
136+
def getCAEA(self, period: int, fortnight: int):
137+
req = {
138+
"Periodo": period,
139+
"Orden": fortnight
140+
}
141+
142+
return self.executeRequest("FECAEAConsultar", req)["ResultGet"]
143+
144+
# Asks to AFIP Servers for available sales points
145+
def getSalesPoints(self):
146+
return self.executeRequest("FEParamGetPtosVenta")["ResultGet"]["PtoVenta"]
147+
148+
# Asks to AFIP Servers for available voucher types
149+
def getVoucherTypes(self):
150+
return self.executeRequest("FEParamGetTiposCbte")["ResultGet"]["CbteTipo"]
151+
152+
# Asks to AFIP Servers for voucher concepts availables
153+
def getConceptTypes(self):
154+
return self.executeRequest("FEParamGetTiposConcepto")["ResultGet"]["ConceptoTipo"]
155+
156+
# Asks to AFIP Servers for document types availables
157+
def getDocumentTypes(self):
158+
return self.executeRequest("FEParamGetTiposDoc")["ResultGet"]["DocTipo"]
159+
160+
# Asks to AFIP Servers for available aliquotes
161+
def getAliquotTypes(self):
162+
return self.executeRequest("FEParamGetTiposIva")["ResultGet"]["IvaTipo"]
163+
164+
# Asks to AFIP Servers for available currencies
165+
def getCurrenciesTypes(self):
166+
return self.executeRequest("FEParamGetTiposMonedas")["ResultGet"]["Moneda"]
167+
168+
# Asks to AFIP Servers for available voucher optional data
169+
def getOptionsTypes(self):
170+
return self.executeRequest("FEParamGetTiposOpcional")["ResultGet"]["OpcionalTipo"]
171+
172+
# Asks to AFIP Servers for available tax types
173+
def getTaxTypes(self):
174+
return self.executeRequest("FEParamGetTiposTributos")["ResultGet"]["TributoTipo"]
175+
176+
# Asks to web service for servers status
177+
def getServerStatus(self):
178+
return self.executeRequest("FEDummy")
179+
180+
# Change date from AFIP used format (yyyymmdd) to yyyy-mm-dd
181+
def formatDate(self, date: int) -> str:
182+
m = re.search(r"(\d{4})(\d{2})(\d{2})", str(date))
183+
184+
return "%s-%s-%s" %(m.group(1), m.group(2), m.group(3))
185+
186+
# Sends request to AFIP servers
187+
def executeRequest(self, operation: str, params: dict = {}):
188+
params.update(self.getWSInitialRequest(operation))
189+
190+
results = super(ElectronicBilling, self).executeRequest(operation, params)
191+
192+
self.__checkErrors(operation, results)
193+
194+
return results["%sResult" % operation]
195+
196+
# Prepare default request parameters for most operations
197+
def getWSInitialRequest(self, operation: str):
198+
if operation == "FEDummy":
199+
return {}
200+
201+
ta = self.getTokenAuthorization()
202+
203+
return {
204+
"Auth": {
205+
"Token": ta["token"],
206+
"Sign": ta["sign"],
207+
"Cuit": self.afip.CUIT
208+
}
209+
}
210+
211+
# Check if occurs an error on Web Service request
212+
def __checkErrors(self, operation: str, results: dict):
213+
res = results["%sResult" %operation]
214+
215+
if operation == "FECAESolicitar" and res["FeDetResp"]:
216+
if type(res["FeDetResp"]["FECAEDetResponse"]) in (tuple, list):
217+
res["FeDetResp"]["FECAEDetResponse"] = res["FeDetResp"]["FECAEDetResponse"][0]
218+
219+
if res["FeDetResp"]["FECAEDetResponse"].get("Observaciones") and res["FeDetResp"]["FECAEDetResponse"]["Resultado"] != "A":
220+
res["Errors"] = { "Err": res["FeDetResp"]["FECAEDetResponse"]["Observaciones"]["Obs"] }
221+
222+
if res.get("Errors"):
223+
err = res["Errors"]["Err"][0] if type(res["Errors"]["Err"]) in (tuple, list) else res["Errors"]["Err"]
224+
225+
raise Exception("(%s) %s") % (err["Code"], err["Msg"])

‎afip/webservice.py‎

Lines changed: 55 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,55 @@
1+
import http.client
2+
import json
3+
4+
class WebService:
5+
def __init__(self, afip, options: dict = {}):
6+
self.afip = afip
7+
self.options = options
8+
self.WSDL = options.get('WSDL')
9+
self.URL = options.get('URL')
10+
self.WSDL_TEST = options.get('WSDL_TEST')
11+
self.URL_TEST = options.get('URL_TEST')
12+
self.soapv12 = options.get('soapV1_2')
13+
14+
if not(options.get('service')):
15+
raise Exception("service field is required in options")
16+
17+
# Gets token authorization for an AFIP Web Service
18+
#
19+
# If force is true it forces to create a new TA
20+
def getTokenAuthorization(self, force = False) -> dict:
21+
return self.afip.getServiceTA(self.options.get('service'), force)
22+
23+
# Sends request to AFIP servers
24+
def executeRequest(self, method, params = {}) -> dict:
25+
conn = http.client.HTTPSConnection("app.afipsdk.com")
26+
27+
payload = {
28+
"method": method,
29+
"params": params,
30+
"environment": self.afip.environment,
31+
"wsid": self.options.get('service'),
32+
"url": self.URL if self.afip.production else self.URL_TEST,
33+
"wsdl": self.WSDL if self.afip.production else self.WSDL_TEST,
34+
"soap_v_1_2": self.soapv12
35+
}
36+
37+
headers = {
38+
"Content-Type": "application/json",
39+
"sdk-version-number": self.afip.sdk_version_number,
40+
"sdk-library": "python",
41+
"sdk-environment": self.afip.environment
42+
}
43+
44+
if self.afip.access_token: headers["Authorization"] = "Bearer %s" % self.afip.access_token
45+
46+
conn.request("POST", "/api/v1/afip/requests", json.dumps(payload), headers)
47+
48+
res = conn.getresponse()
49+
50+
data = res.read()
51+
52+
if res.getcode() >= 400:
53+
raise Exception(data.decode("utf-8"))
54+
55+
return json.loads(data.decode("utf-8"))

0 commit comments

Comments
 (0)