· 5 years ago · Jan 12, 2021, 04:42 AM
1import requests
2import os.path as path
3import sys
4from lxml.etree import Element, SubElement, tostring
5import re
6import time
7import hashlib
8import base64
9import logging
10import calendar
11
12
13MONTHLY_RETURN_URL = 'https://www.tpvs.hmrc.gov.uk/new-cis/monthly_return'
14CIS_CONSTRUCT_URL = 'https://secure.gateway.gov.uk/submission'
15POLL_URL = 'https://secure.gateway.gov.uk/poll'
16DEFAULT_POLLINTERVAL = '10'
17BASEPATH = path.dirname(__file__)
18ACTIONS = ['cis_verify', 'cis_submit']
19CONSOLE_COLOURS = {
20 'Header': '\033[94m',
21 'Message': '\033[92m',
22}
23
24
25def _get_last_month_day():
26 """Calculates last day for current month
27 and return's string with date in format YYYY-MM-DD
28 """
29 last_month_day = calendar.monthrange(int(time.strftime('%Y')), int(time.strftime('%m')))[1] # Last day of current month
30 return time.strftime('%Y-%m-') + str(last_month_day)
31
32
33LAST_MONTH_DAY = _get_last_month_day()
34
35
36def get_xml_value(tag, xml):
37 """Extracts value from given xml string between given tag"""
38 pattern = re.compile('<%s>(.*?)</%s>' % (tag, tag), re.DOTALL)
39# print(pattern, xml)
40 value = '\n'.join(pattern.findall(xml.decode('utf-8')))
41 return value
42
43
44class GovTalk(object):
45 """Main class for HMRC services"""
46
47 CONSTRUCT_URL = False # URL xml request is send to, could be either MONTHLY_RETURN_URL ot CIS_CONSTRUCT_URL
48 testing = False
49 partner_params = {}
50 return_data = {}
51 _logger = logger = logging.getLogger('Govtalk API')
52
53 def __init__(self, action=None):
54 super(GovTalk, self).__init__()
55 assert (action in ['cis_verify', 'cis_return']), "Not implemented yet! You can use CIS Verification/monthly return only"
56 self.CONSTRUCT_URL = 'https://secure.gateway.gov.uk/submission'
57 if action == 'cis_verify':
58 self.test_file = path.join(BASEPATH, 'xml_samples', 'cis_verification_sample.xml')
59 self.poll_file = path.join(BASEPATH, 'xml_samples', 'verification_poll.xml')
60 self.action = action
61 elif action == 'cis_return':
62 self.action = action
63
64 def build_verification_message(self, correlation_value=False, mesage_type=False):
65 """Build CIS Verification XML, If correlation value is passed it will
66 build verification poll message to get status of previously sent document.
67 IRmark is added as empty element at this point and value is added further.
68 @Returns: xml to be passed to DSP.
69 """
70 SenderDetails_vals = self.partner_params['SenderDetails']
71 Channel_vals = self.partner_params['Channel']
72 IRheader_vals = self.partner_params['IRheader']
73 Contractor_vals = self.partner_params['Contractor']
74 Subcontractor_vals = self.partner_params['Subcontractor']
75
76 GovTalkMessage = Element('GovTalkMessage', xmlns="http://www.govtalk.gov.uk/CM/envelope")
77 EnvelopeVersion = SubElement(GovTalkMessage, 'EnvelopeVersion')
78 EnvelopeVersion.text = '2.0'
79
80 # Building Header
81 Header = SubElement(GovTalkMessage, 'Header')
82 MessageDetails = SubElement(Header, 'MessageDetails')
83 message_class = SubElement(MessageDetails, 'Class')
84 message_class.text = 'IR-CIS-VERIFY'
85 qualifier = SubElement(MessageDetails, 'Qualifier')
86 qualifier.text = 'request' if not correlation_value else 'poll'
87 function = SubElement(MessageDetails, 'Function')
88 function.text = 'submit'
89 if correlation_value:
90 correlation = SubElement(MessageDetails, 'CorrelationID')
91 correlation.text = correlation_value
92 transformation = SubElement(MessageDetails, 'Transformation')
93 transformation.text = 'XML'
94 if not correlation_value:
95 gateway_test = SubElement(MessageDetails, 'GatewayTest')
96 gateway_test.text = '1' if self.testing else '0'
97 SenderDetails = SubElement(Header, 'SenderDetails')
98 IDAuthentication = SubElement(SenderDetails, 'IDAuthentication')
99 SenderID = SubElement(IDAuthentication, 'SenderID')
100 SenderID.text = SenderDetails_vals['SenderID']
101 Authentication = SubElement(IDAuthentication, 'Authentication')
102 Method = SubElement(Authentication, 'Method')
103 Method.text = SenderDetails_vals['Method']
104 Value = SubElement(Authentication, 'Value')
105 Value.text = SenderDetails_vals['Value']
106
107 GovTalkDetails = SubElement(GovTalkMessage, 'GovTalkDetails')
108 Keys = SubElement(GovTalkDetails, 'Keys')
109 if not correlation_value:
110 Key1 = SubElement(Keys, 'Key', Type="TaxOfficeNumber")
111 Key1.text = IRheader_vals['TaxOfficeNumber']
112 Key2 = SubElement(Keys, 'Key', Type="TaxOfficeReference")
113 Key2.text = IRheader_vals['TaxOfficeReference']
114 TargetDetails = SubElement(GovTalkDetails, 'TargetDetails')
115 Organisation = SubElement(TargetDetails, 'Organisation')
116 Organisation.text = 'IR'
117 ChannelRouting = SubElement(GovTalkDetails, 'ChannelRouting')
118 Channel = SubElement(ChannelRouting, 'Channel')
119 URI = SubElement(Channel, 'URI')
120
121 URI.text = Channel_vals['URI']
122 Product = SubElement(Channel, 'Product')
123 Product.text = Channel_vals['Product']
124 Version = SubElement(Channel, 'Version')
125 Version.text = Channel_vals['Version']
126
127 # Building Body
128 Body = SubElement(GovTalkMessage, 'Body')
129 IRenvelope = SubElement(Body, 'IRenvelope', xmlns="http://www.govtalk.gov.uk/taxation/CISrequest")
130 # IR header
131 IRheader = SubElement(IRenvelope, 'IRheader')
132 IRheaderKeys = SubElement(IRheader, 'Keys')
133 IRheaderKey1 = SubElement(IRheaderKeys, 'Key', Type="TaxOfficeNumber")
134 IRheaderKey1.text = IRheader_vals['TaxOfficeNumber']
135 IRheaderKey2 = SubElement(IRheaderKeys, 'Key', Type="TaxOfficeReference")
136 IRheaderKey2.text = IRheader_vals['TaxOfficeReference']
137 PeriodEnd = SubElement(IRheader, 'PeriodEnd')
138 PeriodEnd.text = IRheader_vals['PeriodEnd']
139 DefaultCurrency = SubElement(IRheader, 'DefaultCurrency')
140 DefaultCurrency.text = 'GBP'
141 SubElement(IRheader, 'IRmark', Type="generic")
142 Sender = SubElement(IRheader, 'Sender')
143 Sender.text = 'Employer'
144
145 # CIS request
146 CISrequest = SubElement(IRenvelope, 'CISrequest')
147 Contractor = SubElement(CISrequest, 'Contractor')
148 UTR = SubElement(Contractor, 'UTR')
149 UTR.text = Contractor_vals['UTR']
150 AOref = SubElement(Contractor, 'AOref')
151 AOref.text = Contractor_vals['AOref']
152 Subcontractor = SubElement(CISrequest, 'Subcontractor')
153 Action = SubElement(Subcontractor, 'Action')
154 Action.text = 'verify'
155 Type = SubElement(Subcontractor, 'Type')
156 Type.text = Subcontractor_vals['Type']
157 TradingName = SubElement(Subcontractor, 'TradingName')
158 TradingName.text = Subcontractor_vals['TradingName']
159 WorksRef = SubElement(Subcontractor, 'WorksRef')
160 WorksRef.text = Subcontractor_vals['WorksRef']
161 SubcontractorUTR = SubElement(Subcontractor, 'UTR')
162 SubcontractorUTR.text = Subcontractor_vals['UTR']
163 if Subcontractor_vals['Type'] == 'company':
164 CRN = SubElement(Subcontractor, 'CRN')
165 CRN.text = Subcontractor_vals['CRN']
166 else:
167 NINO = SubElement(Subcontractor, 'NINO')
168 NINO.text = Subcontractor_vals['NINO']
169
170 if self.partner_params['Subcontractor']['Type'] == 'partnership':
171 Partnership = SubElement(Subcontractor, 'Partnership')
172 Name = SubElement(Partnership, 'Name')
173 Name.text = Subcontractor_vals['PartnershipName']
174 PartnershipUTR = SubElement(Partnership, 'UTR')
175 PartnershipUTR.text = Subcontractor_vals['PartnershipUTR']
176 Declaration = SubElement(CISrequest, 'Declaration')
177 Declaration.text = 'yes'
178
179 xml = tostring(GovTalkMessage, xml_declaration=True, encoding='UTF-8')
180 if not correlation_value:
181 xml = self.add_irmark(xml)
182 return xml
183
184 def build_delete_request(self, message_class, correlation_id, transaction_id):
185 """Build CIS delete XML request."""
186 GovTalkMessage = Element('GovTalkMessage', xmlns="http://www.govtalk.gov.uk/CM/envelope")
187 EnvelopeVersion = SubElement(GovTalkMessage, 'EnvelopeVersion')
188 EnvelopeVersion.text = '2.0'
189 # Building Header
190 Header = SubElement(GovTalkMessage, 'Header')
191 MessageDetails = SubElement(Header, 'MessageDetails')
192 MessageClass = SubElement(MessageDetails, 'Class')
193 MessageClass.text = message_class
194 qualifier = SubElement(MessageDetails, 'Qualifier')
195 qualifier.text = 'request'
196 function = SubElement(MessageDetails, 'Function')
197 function.text = 'delete'
198 CorrelationID = SubElement(MessageDetails, 'CorrelationID')
199 if correlation_id:
200 CorrelationID.text = correlation_id
201 transformation = SubElement(MessageDetails, 'Transformation')
202 transformation.text = 'XML'
203 GovTalkDetails = SubElement(GovTalkMessage, 'GovTalkDetails')
204 SubElement(GovTalkDetails, 'Keys')
205 SubElement(GovTalkMessage, 'Body')
206 xml = tostring(GovTalkMessage, xml_declaration=True, encoding='UTF-8')
207 return xml
208
209 def delete_request(self, response):
210 """Delete request must be sent to DSP after success verification response,
211 if not sent it will be deleted from the DSP within 60 days.
212 """
213 correlation = get_xml_value('CorrelationID', response)
214 document_class = get_xml_value('Class', response)
215 transaction_id = get_xml_value('TransactionID', response)
216 xml = self.build_delete_request(document_class, correlation, transaction_id)
217 delete_response = self.send_message(xml)
218 return delete_response
219
220 def generate_irmark(self, xml):
221 """Generates IRmark for given xml document."""
222 body = '<Body xmlns="http://www.govtalk.gov.uk/CM/envelope">%s</Body>' % get_xml_value('Body', xml)
223 print(xml)
224 body = re.sub('<IRmark Type="generic"/>', '', body) # IRmark must be not included to encryption
225 print(body)
226 IRmark = hashlib.sha1(body.encode('utf-8')).digest()
227 IRmark = base64.b64encode(IRmark)
228 return IRmark
229
230 def add_irmark(self, xml):
231 ir_mark = self.generate_irmark(xml)
232 ir_mark_string = '<IRmark Type="generic">%s</IRmark>' % ir_mark
233 xml = re.sub('<IRmark Type="generic"/>', ir_mark_string, xml)
234 return xml
235
236 def get_correlation(self, xml):
237 """Extracts correlation value from given xml."""
238 return get_xml_value('CorrelationID', xml)
239
240 def get_pollinterval(self, xml_response):
241 """Extracts PollInterval from xml provided by Gateway response"""
242 pattern = re.compile('PollInterval="(.*?)"', re.DOTALL)
243 PollInterval = pattern.search(xml_response).group(1)
244 if not PollInterval:
245 PollInterval = DEFAULT_POLLINTERVAL
246 PollInterval = int(PollInterval)
247 return PollInterval
248
249 def build_request_message(self, correlation=False):
250 # if self.testing:
251 # if self.action == 'cis_verify':
252 # if not correlation:
253 # with open(self.test_file, 'r') as test_file:
254 # xml = test_file.read()
255 # request_message = self.add_irmark(xml)
256 # test_file.close()
257 # else:
258 # with open(self.poll_file, 'r') as test_file:
259 # request_message = test_file.read()
260 # test_file.close()
261 if self.action == 'cis_verify':
262 request_message = self.build_verification_message(correlation_value=correlation)
263 elif self.action == 'cis_return':
264 request_message = self.build_return_request_message(correlation_value=correlation)
265 return request_message
266
267 def send_message(self, xml, poll=False):
268 assert xml, 'No XML data passed!'
269 url = self.CONSTRUCT_URL if not poll else POLL_URL
270 if self.testing and poll:
271 url = 'https://secure.dev.gateway.gov.uk/poll'
272 result = requests.post(url, data=xml)
273 response = result.text
274 result.close()
275 return response
276
277 def check_verification_params(self, partner_params):
278 required_fields = ['SenderDetails', 'IRheader', 'Contractor', 'Subcontractor']
279 assert set(required_fields) <= set(partner_params), 'Reuired Params: %s' % required_fields
280
281 SenderDetails = partner_params['SenderDetails']
282 SenderDetails_fields = ['SenderID', 'Value']
283 assert set(SenderDetails_fields) <= set(SenderDetails.keys()), 'Reqired Sender Details Params: %s' % SenderDetails_fields
284
285 IRheader = partner_params['IRheader']
286 IRheader_fields = ['TaxOfficeNumber', 'TaxOfficeReference', 'PeriodEnd', 'Sender']
287 assert set(IRheader_fields) <= set(IRheader.keys()), 'Reqired IRheader Params: %s' % IRheader_fields
288
289 Contractor = partner_params['Contractor']
290 Contractor_fields = ['UTR', 'AOref']
291 assert set(Contractor_fields) <= set(Contractor.keys()), 'Reqired IRheader Params: %s' % Contractor_fields
292
293 Subcontractor = partner_params['Subcontractor']
294 if Subcontractor['Type'] == 'partnership' and not Subcontractor['PartnershipName']:
295 raise Exception('No Partner name defined for partnership! Please define one!')
296 Subcontractor_fields = ['Type', 'TradingName', 'WorksRef', 'CRN', 'NINO'] # UTR
297 assert set(Subcontractor_fields) <= set(Subcontractor.keys()), 'Reqired IRheader Params: %s' % Subcontractor_fields
298
299 Channel = partner_params['Channel']
300 Channel_fields = ['URI', 'Product', 'Version']
301 assert set(Channel_fields) <= set(Channel.keys()), 'Reqired IRheader Params: %s' % Channel_fields
302
303 def submit_cis(self, testing=False, **args):
304 self.partner_params = args.get('partner_data', {})
305 self.return_data = args.get('return_data', {})
306 if testing:
307 self.testing = True
308 self.CONSTRUCT_URL = 'https://secure.dev.gateway.gov.uk/submission'
309 else:
310 self.testing = False
311 self.CONSTRUCT_URL = CIS_CONSTRUCT_URL
312 if self.action == 'cis_verify':
313 self.check_verification_params(self.partner_params)
314 elif self.action == 'cis_return':
315 self.check_return_data(self.return_data)
316 xml = self.build_request_message()
317 response = self.send_message(xml)
318 PollInterval = self.get_pollinterval(response)
319 correlation_value = self.get_correlation(response)
320 xml = self.build_request_message(correlation=correlation_value)
321 counter = 0
322 while 'acknowledgement' in response:
323 counter += 1
324 self._logger.info("SlEEPING for %s seconds." % PollInterval)
325 time.sleep(PollInterval)
326 response = self.send_message(xml, poll=True)
327 PollInterval = self.get_pollinterval(response)
328 if counter == 6:
329 if self.action == 'cis_verify':
330 Exception('HMRC is not responding, please try again later.')
331 else:
332 return {'CorrelationID': correlation_value,
333 'Message': 'It takes too long to check subcontractors return. Please try again later.'}
334 self._logger.info('GovTalk response %s' % response)
335 if '</Error>' in response:
336 self.raise_error(response)
337 time.sleep(PollInterval)
338 self.delete_request(response)
339 return response
340
341 def check_return_data(self, params):
342 # nino = params
343 pass
344
345 def build_return_request_message(self, correlation_value=False, mesage_type=False):
346 """Builds XML to pass in request for CIS return"""
347
348 GovTalkMessage = Element('GovTalkMessage', xmlns="http://www.govtalk.gov.uk/CM/envelope")
349 EnvelopeVersion = SubElement(GovTalkMessage, 'EnvelopeVersion')
350 EnvelopeVersion.text = '2.0'
351
352 # Building Header
353 Header = SubElement(GovTalkMessage, 'Header')
354 MessageDetails = SubElement(Header, 'MessageDetails')
355 message_class = SubElement(MessageDetails, 'Class')
356 message_class.text = 'IR-CIS-CIS300MR'
357 qualifier = SubElement(MessageDetails, 'Qualifier')
358 qualifier.text = 'request' if not correlation_value else 'poll'
359 function = SubElement(MessageDetails, 'Function')
360 function.text = 'submit'
361 if correlation_value:
362 correlation = SubElement(MessageDetails, 'CorrelationID')
363 correlation.text = correlation_value
364 transformation = SubElement(MessageDetails, 'Transformation')
365 transformation.text = 'XML'
366 if not correlation_value:
367 SenderDetails_vals = self.return_data['SenderDetails']
368 gateway_test = SubElement(MessageDetails, 'GatewayTest')
369 gateway_test.text = '1' if self.testing else '0'
370 SenderDetails = SubElement(Header, 'SenderDetails')
371 IDAuthentication = SubElement(SenderDetails, 'IDAuthentication')
372 SenderID = SubElement(IDAuthentication, 'SenderID')
373 SenderID.text = SenderDetails_vals['SenderID']
374 Authentication = SubElement(IDAuthentication, 'Authentication')
375 Method = SubElement(Authentication, 'Method')
376 Method.text = 'clear'
377 Value = SubElement(Authentication, 'Value')
378 Value.text = SenderDetails_vals['Value']
379
380 GovTalkDetails = SubElement(GovTalkMessage, 'GovTalkDetails')
381 Keys = SubElement(GovTalkDetails, 'Keys')
382 if not correlation_value:
383 Channel_vals = self.return_data['Channel']
384 IRheader_vals = self.return_data['IRheader']
385 Contractor_vals = self.return_data['Contractor']
386 Key1 = SubElement(Keys, 'Key', Type="TaxOfficeNumber")
387 Key1.text = IRheader_vals['TaxOfficeNumber']
388 Key2 = SubElement(Keys, 'Key', Type="TaxOfficeReference")
389 Key2.text = IRheader_vals['TaxOfficeReference']
390 TargetDetails = SubElement(GovTalkDetails, 'TargetDetails')
391 Organisation = SubElement(TargetDetails, 'Organisation')
392 Organisation.text = 'IR'
393 ChannelRouting = SubElement(GovTalkDetails, 'ChannelRouting')
394 Channel = SubElement(ChannelRouting, 'Channel')
395 URI = SubElement(Channel, 'URI')
396
397 URI.text = Channel_vals['URI']
398 Product = SubElement(Channel, 'Product')
399 Product.text = 'Enapps CIS'
400 Version = SubElement(Channel, 'Version')
401 Version.text = '1.0'
402
403 # Building Body
404 Body = SubElement(GovTalkMessage, 'Body')
405 IRenvelope = SubElement(Body, 'IRenvelope', xmlns="http://www.govtalk.gov.uk/taxation/CISreturn")
406 # IR header
407 IRheader = SubElement(IRenvelope, 'IRheader')
408 IRheaderKeys = SubElement(IRheader, 'Keys')
409 IRheaderKey1 = SubElement(IRheaderKeys, 'Key', Type="TaxOfficeNumber")
410 IRheaderKey1.text = IRheader_vals['TaxOfficeNumber']
411 IRheaderKey2 = SubElement(IRheaderKeys, 'Key', Type="TaxOfficeReference")
412 IRheaderKey2.text = IRheader_vals['TaxOfficeReference']
413 PeriodEnd = SubElement(IRheader, 'PeriodEnd')
414 PeriodEnd.text = IRheader_vals['PeriodEnd']
415 DefaultCurrency = SubElement(IRheader, 'DefaultCurrency')
416 DefaultCurrency.text = 'GBP'
417 SubElement(IRheader, 'IRmark', Type="generic")
418 Sender = SubElement(IRheader, 'Sender')
419 Sender.text = IRheader_vals['Sender']
420
421 # CIS request
422 CISreturn = SubElement(IRenvelope, 'CISreturn')
423 Contractor = SubElement(CISreturn, 'Contractor')
424 UTR = SubElement(Contractor, 'UTR')
425 UTR.text = Contractor_vals['UTR']
426 AOref = SubElement(Contractor, 'AOref')
427 AOref.text = Contractor_vals['AOref']
428 for Subcontractor_vals in self.return_data['Subcontractors']:
429 Subcontractor = SubElement(CISreturn, 'Subcontractor')
430 TradingName = SubElement(Subcontractor, 'TradingName')
431 TradingName.text = Subcontractor_vals['TradingName']
432 UnmatchedRate = SubElement(Subcontractor, 'UnmatchedRate')
433 UnmatchedRate.text = 'yes'
434 SubUTR = SubElement(Subcontractor, 'UTR')
435 SubUTR.text = Subcontractor_vals['UTR']
436 if Subcontractor_vals['NINO']:
437 Nino = SubElement(Subcontractor, 'NINO')
438 Nino.text = Subcontractor_vals['NINO']
439 elif Subcontractor_vals['CRN']:
440 Crn = SubElement(Subcontractor, 'CRN')
441 Crn.text = Subcontractor_vals['CRN']
442 if Subcontractor_vals['VerificationNumber']:
443 VerificationNumber = SubElement(Subcontractor, 'VerificationNumber')
444 VerificationNumber.text = Subcontractor_vals['VerificationNumber']
445 if eval(Subcontractor_vals['TotalPayments']):
446 TotalPayments = SubElement(Subcontractor, 'TotalPayments')
447 TotalPayments.text = Subcontractor_vals['TotalPayments']
448 if eval(Subcontractor_vals['CostOfMaterials']):
449 CostOfMaterials = SubElement(Subcontractor, 'CostOfMaterials')
450 CostOfMaterials.text = Subcontractor_vals['CostOfMaterials']
451 if eval(Subcontractor_vals['TotalDeducted']):
452 TotalDeducted = SubElement(Subcontractor, 'TotalDeducted')
453 TotalDeducted.text = Subcontractor_vals['TotalDeducted']
454
455 Declarations = SubElement(CISreturn, 'Declarations')
456 EmploymentStatus = SubElement(Declarations, 'EmploymentStatus')
457 EmploymentStatus.text = 'yes'
458 Verification = SubElement(Declarations, 'Verification')
459 Verification.text = 'yes'
460 InformationCorrect = SubElement(Declarations, 'InformationCorrect')
461 InformationCorrect.text = 'yes'
462 # Inactivity = SubElement(Declarations, 'Inactivity') #Please indicate if you do not anticipate paying subcontractors in the next six months
463 # Inactivity.text = 'yes'
464
465 xml = tostring(GovTalkMessage, xml_declaration=True, encoding='UTF-8')
466 if not correlation_value:
467 xml = self.add_irmark(xml)
468 return xml
469
470 def return_cis(self, testing=False, **args):
471 self.return_data = args.get('return_data', {})
472 if testing:
473 self.testing = True
474 self.CONSTRUCT_URL = 'https://secure.dev.gateway.gov.uk/submission'
475 else:
476 self.testing = False
477 self.CONSTRUCT_URL = CIS_CONSTRUCT_URL
478 self.check_return_data(self.return_data)
479 xml = self.build_request_message()
480 response = self.send_message(xml)
481 PollInterval = self.get_pollinterval(response)
482 correlation_value = self.get_correlation(response)
483 xml = self.build_request_message(correlation=correlation_value)
484 counter = 0
485 while 'acknowledgement' in response:
486 counter += 1
487 self._logger.info("SlEEPING for %s seconds." % PollInterval)
488 time.sleep(PollInterval)
489 response = self.send_message(xml, poll=True)
490 PollInterval = self.get_pollinterval(response)
491 if counter == 6:
492 return {'CorrelationID': correlation_value,
493 'Message': 'It takes too long to check subcontractors return. Please try again later.'}
494 if '</Error>' in response:
495 self.raise_error(response)
496 time.sleep(PollInterval)
497 self.delete_request(response)
498 return response
499
500 def raise_error(self, xml_response):
501 error_instance = GovtalkError('cis_verify', xml_response)
502 error_message = error_instance.construct_message()
503 raise Exception(error_message)
504
505
506class GovtalkError(object):
507 """Main class for constructing erorr message from GovTalk xml response"""
508
509 action = False
510 xml_response = ''
511
512 def __init__(self, action, xml_response):
513 assert (action in ACTIONS), "Invalid action! You can use cis verification and submissin only!"
514 assert xml_response, 'No xml passed!'
515 super(GovtalkError, self).__init__()
516 self.action = action
517 self.xml_response = xml_response
518
519 def construct_message(self):
520 error_message = get_xml_value('Text', self.xml_response)
521 error_code = get_xml_value('Number', self.xml_response)
522 return 'Message: %s\n Error codes: %s' % (error_message, error_code)
523
524
525def get_default_return_data():
526 return {
527 'SenderDetails': {'SenderID': 'CISRUSER1028',
528 'Value': 'testing1',
529 'Method': 'clear',
530 },
531 'Channel': {'URI': '0920',
532 },
533 'IRheader': {'TaxOfficeNumber': '123',
534 'TaxOfficeReference': 'R229',
535 'PeriodEnd': '2009-05-05',
536 'Sender': 'Partnership',
537 },
538 'Contractor': {'UTR': '2325648152', # Company UTR
539 'AOref': '123PP87654321', # Company Aoref
540 },
541 'Subcontractors': [{
542 'TradingName': 'Foundations',
543 'VerificationNumber': 'V1234567890A',
544 'PartnershipName': '234234',
545 'UTR': '1234567890',
546 'NINO': 'JK901234D',
547 'TotalPayments': '2000.00',
548 'CostOfMaterials': '500.00',
549 'TotalDeducted': '450.33',
550 }]
551 }
552
553
554def get_default_verification_data():
555 return {
556 'SenderDetails': {'SenderID': 'CISRUSER1107',
557 'Value': 'testing1',
558 'Method': 'clear',
559 },
560 'Channel': {'URI': '0920', # Enapps Vendor ID
561 'Product': 'sdstesting',
562 'Version': '1',
563 },
564 'IRheader': {'TaxOfficeNumber': '123',
565 'TaxOfficeReference': 'R308',
566 'PeriodEnd': LAST_MONTH_DAY,
567 'Sender': 'Employer',
568 },
569 'Contractor': {'UTR': '2325648152',
570 'AOref': '123PP87654321',
571 },
572 'Subcontractor': {'Type': 'partnership',
573 'TradingName': 'Foundations',
574 'WorksRef': 'WR3000',
575 'PartnershipName': '234234',
576 'PartnershipUTR': 'sfes',
577 'UTR': '1234567890',
578 'CRN': 'AB123456',
579 'NINO': 'PR456789B'
580 },
581 }
582
583if __name__ == '__main__':
584 if len(sys.argv) == 1:
585 partner_data = get_default_return_data()
586 print("Usage: python govtalk.py verify|return")
587 sys.exit(1)
588 elif len(sys.argv) > 3:
589 raise TypeError('Too many arguments passed!')
590 # elif not isinstance(eval(sys.argv[-1]), dict): # TODO rewrite me to not use eval!
591 # raise AttributeError('Argument passed must be a dictionaty!')
592 # else:
593 # partner_data = eval(sys.argv[-1])
594 if sys.argv[-1] == 'verify':
595 govtalk_instance = GovTalk(action='cis_verify')
596 result = govtalk_instance.submit_cis(testing=True, partner_data=get_default_verification_data())
597 elif sys.argv[-1] == 'return':
598 govtalk_instance = GovTalk(action='cis_return')
599 result = govtalk_instance.submit_cis(testing=True, return_data=get_default_return_data())
600 else:
601 print("Usage: python govtalk.py verify|return")
602 sys.exit(1)
603 print("%s%s\nGOVTALK Response-->\n%s%s" % ('-'*40, CONSOLE_COLOURS['Header'], CONSOLE_COLOURS['Message'], result))
604