4.1.2.3.2 Internet Gateway Faxing - SRFax API
For Ubuntu 18.04
Install Python2.7 and python-suds for soap
sudo apt-get install python2.7
python-suds
The original suds has a non standard import
scheme that actually allow you to use suds.client (and other names like
suds.cache, suds.metrics etc) after doing an "import suds". suds-jurko cleaned up the
imports, so now you need to import those names specifically. You need
to do either "import suds.client" and refer to it as suds.client or
"from suds import client".
Now Uncoil Your Python Script
#!/usr/bin/env python #from srfax import srfax # Copyright 2012 Vingd, Inc. under MIT liscence # https://github.com/vingd/srfax-api-python/blob/master/LICENSE.txt # extended by PHC import sys, getopt # -*- coding: utf-8 -*- '''SRFax (www.srfax.com) python library''' import re import json import os.path import base64 import logging import time import suds # uncomment the following for more modern python-suds #import suds.client URL = 'https://www.srfax.com/SRF_UserFaxWebSrv.php?wsdl' LOGGER = logging.getLogger(__name__) # E.164 phone numbers are formatted +17045550100 RE_E164 = re.compile(r'^\+\d{7,15}$') RE_NANP = re.compile(r'^\+1') class SRFaxError(Exception): '''SRFax Exception''' def __init__(self, error_code, message, cause=None, retry=False): self.error_code = error_code self.message = message self.cause = cause self.retry = retry super(SRFaxError, self).__init__(error_code, message, cause, retry) LOGGER.exception("%s" % (self)) def get_error_code(self): '''Get exception error code''' return self.error_code def get_cause(self): '''Get exception cause''' return self.cause def get_retry(self): '''Get retry option (should we retry the request?)''' return self.retry class SRFax(object): '''SRFax class''' def __init__(self, access_id, access_pwd, caller_id=None, sender_email=None, account_code=None, url=None): self.access_id = access_id self.access_pwd = access_pwd self.caller_id = caller_id self.sender_email = sender_email self.account_code = account_code self.url = url or URL self.client = suds.client.Client(self.url) def queue_fax(self, to_fax_number, filepath, cover_subject, caller_id=None, sender_email=None, account_code=None): '''Queue fax for sending''' to_fax_number = SRFax.verify_fax_numbers(to_fax_number) fax_type = 'BROADCAST' if len(to_fax_number) > 1 else 'SINGLE' to_fax_number = '|'.join(to_fax_number) if isinstance(filepath, basestring): filepath = [filepath] if not isinstance(filepath, list): raise TypeError('filepath not properly defined') if len(filepath) > 5: raise Exception('More than 5 files defined in filepath') # Display input and output file name passed as the args print ("2fax number : %s and input file: %s with subject: %s" % (fax_no,fax_file,cover_subject) ) params = { 'access_id': self.access_id, 'access_pwd': self.access_pwd, 'sCallerID': caller_id or self.caller_id, 'sSenderEmail': sender_email or self.sender_email, 'sFaxType': fax_type, 'sToFaxNumber': to_fax_number, 'sAccountCode': account_code or self.account_code or '', 'sRetries': '2', 'sCPSubject': cover_subject, 'sFaxFromHeader': 'Haileybury Family Health Team', # 'sRetries': retries or self.retries or '', # 'sCoverPage': cover_page or self.cover_page or '', # 'sFaxFromHeader': fax_from_header or self.fax_from_header or '', # 'sCPFromName': cover_from or self.cover_from or '', # 'sCPToName': cover_to or self.cover_to or '', # 'sCPSubject': cover_subject or self.cover_subject or '', # 'sCPComments': cover_comments or self.cover_comments or '', # 'sQueueFaxDate': fax_date or self.fax_date or '', # 'sQueueFaxTime': fax_time or self.fax_time or '', } SRFax.verify_parameters(params) for i in range(len(filepath)): path = filepath[i] basename = os.path.basename(path) if not isinstance(basename, unicode): basename = basename.decode('utf-8') params['sFileName_%d' % (i + 1)] = basename params['sFileContent_%d' % (i + 1)] = SRFax.get_file_content(path) return self.process_request('Queue_Fax', params) def get_fax_status(self, fax_id): '''Get fax status''' params = { 'access_id': self.access_id, 'access_pwd': self.access_pwd, 'sFaxDetailID': fax_id, } SRFax.verify_parameters(params) response = self.process_request('Get_FaxStatus', params) if len(response) == 1: response = response[0] return response def get_fax_inbox(self, period='ALL'): '''Get fax inbox''' params = { 'access_id': self.access_id, 'access_pwd': self.access_pwd, 'sPeriod': period, } SRFax.verify_parameters(params) return self.process_request('Get_Fax_Inbox', params) def get_fax_outbox(self, period='ALL'): '''Get fax outbox''' params = { 'access_id': self.access_id, 'access_pwd': self.access_pwd, 'sPeriod': period, } SRFax.verify_parameters(params) return self.process_request('Get_Fax_Outbox', params) def retrieve_fax(self, fax_filename, folder): '''Retrieve fax content in Base64 format''' params = { 'access_id': self.access_id, 'access_pwd': self.access_pwd, 'sFaxFileName': fax_filename, 'sDirection': folder, } SRFax.verify_parameters(params) response = self.process_request('Retrieve_Fax', params) if len(response) == 1: response = response[0] return response def delete_fax(self, fax_filename, folder): '''Delete fax files from server''' if isinstance(fax_filename, str): fax_filename = [fax_filename] if not isinstance(fax_filename, list): raise TypeError('fax_filename not properly defined') if len(fax_filename) > 5: raise Exception('More than 5 files defined in fax_filename') params = { 'access_id': self.access_id, 'access_pwd': self.access_pwd, 'sDirection': folder, } SRFax.verify_parameters(params) for i in range(len(fax_filename)): params['sFileName_%d' % (i + 1)] = fax_filename[i] return self.process_request('Delete_Fax', params) def process_request(self, method, params): '''Process SRFax SOAP request''' method = getattr(self.client.service, method) try: response = method(**params) # pylint: disable-msg=W0142 except Exception as exc: raise SRFaxError('REQUESTFAILED', 'SOAP request failed', cause=exc, retry=True) return SRFax.process_response(response) @staticmethod def process_response(response): '''Process SRFax SOAP response''' if not response: raise SRFaxError('INVALIDRESPONSE', 'Empty response', retry=True) if 'Status' not in response or 'Result' not in response: raise SRFaxError('INVALIDRESPONSE', 'Status and/or Result not in response: %s' % (response), retry=True) result = response['Result'] try: if isinstance(result, list): for i in range(len(result)): if not result[i]: continue if isinstance(result[i], suds.sax.text.Text): result[i] = str(result[i]) else: result[i] = json.loads(json.dumps(dict(result[i]))) elif isinstance(result, suds.sax.text.Text): result = str(result) except Exception as exc: raise SRFaxError('INVALIDRESPONSE', 'Error converting SOAP response', cause=exc, retry=True) LOGGER.debug('Result: %s' % (result)) if response['Status'] != 'Success': errmsg = result if (isinstance(errmsg, list) and len(errmsg) == 1 and 'ErrorCode' in errmsg[0]): errmsg = errmsg[0]['ErrorCode'] raise SRFaxError('REQUESTFAILED', errmsg) if result is None: result = True return result @staticmethod def verify_parameters(params): '''Verify that dict values are set''' for key in params.keys(): print ("key : %s and value : %s " % (key, params[key]) ) if params[key] is None: raise TypeError('%s not set' % (key)) @staticmethod def is_e164_number(number): '''Simple check if number is in E.164 format''' if isinstance(number, str) and RE_E164.match(number): return True return False @staticmethod def is_nanp_number(number): '''Simple check if number is inside North American Numbering Plan''' if isinstance(number, str) and RE_NANP.match(number): return True return False @staticmethod def verify_fax_numbers(to_fax_number): '''Verify and prepare fax numbers for use at SRFax''' if isinstance(to_fax_number, basestring): to_fax_number = [to_fax_number] if not isinstance(to_fax_number, list): raise TypeError('to_fax_number not properly defined') for i in range(len(to_fax_number)): number = str(to_fax_number[i]) if not SRFax.is_e164_number(number): raise TypeError('Number not in E.164 format: %s' % (number)) if SRFax.is_nanp_number(number): to_fax_number[i] = number[1:] else: to_fax_number[i] = '011' + number[1:] return to_fax_number @staticmethod def get_file_content(filepath): '''Read and return file content Base64 encoded''' if not os.path.exists(filepath): raise Exception('File does not exists: %s' % (filepath)) if not os.path.isfile(filepath): raise Exception('Not a file: %s' % (filepath)) content = None try: fdp = open(filepath, 'rb') except IOError: raise else: content = fdp.read() fdp.close() if not content: raise Exception('Error reading file or file empty: %s' % (filepath)) return base64.b64encode(content) # Store input and output file names fax_no='' fax_file='' cover_subject='' # Read command line args myopts, args = getopt.getopt(sys.argv[1:],"n:f:s:") ############################### # o == option # a == argument passed to the o ############################### for o, a in myopts: if o == '-n': fax_no=a elif o == '-f': fax_file=a elif o == '-s': cover_subject=a else: print("Usage: %s -n faxnumber -f file -s subject" % sys.argv[0]) sys.exit # Display input and output file name passed as the args print ("fax number : %s and input file: %s with subject: %s" % (fax_no,fax_file,cover_subject) ) srfax_client = SRFax(122222, "password", caller_id=8662441234, sender_email="ddd@hhhh.org") fax_id = srfax_client.queue_fax(fax_no, fax_file, cover_subject) time.sleep(30) status = srfax_client.get_fax_status(fax_id) print ("fax status : %s" % (status) ) #outbox=srfax_client.get_fax_outbox() #print ("foutbox : %s" % (outbox) )
Now a cron job
Document Actions