Skip to content

Instantly share code, notes, and snippets.

@cgrinham
Last active August 31, 2022 12:51
Show Gist options
  • Save cgrinham/d8faec9b8d0a93ae4ba82100ac633aca to your computer and use it in GitHub Desktop.
Save cgrinham/d8faec9b8d0a93ae4ba82100ac633aca to your computer and use it in GitHub Desktop.
Vicidial Python Wrapper
import os
import requests
SERVER_URL = os.getenv("VICIDIAL_SERVER_URL", "")
USER = os.getenv("VICIDIAL_USER", "")
PASSWORD = os.getenv("VICIDIAL_PASSWORD", "")
HTTP_PROXY = os.getenv("HTTP_PROXY", "")
HTTPS_PROXY = os.getenv("HTTPS_PROXY", "")
class VicidialClient:
def __init__(self, server_url, username, password, source="script"):
self.server_url = server_url
self.username = username
self.password = password
self.source = source
@staticmethod
def process_response(response_content, headers):
headers = None
for row in response_content.splitlines():
row_data = row.split("|")
if not headers:
headers = row_data
continue
yield dict(zip(headers, row_data))
def make_request(self, function, **params):
request_args = {
"url": f"{self.server_url}/vicidial/non_agent_api.php",
"params": {
"function": function,
"user": self.username,
"pass": self.password,
"source": self.source,
"headers": True,
},
"proxies": {
"http": HTTP_PROXY,
"https": HTTPS_PROXY,
}
}
request_args["params"].update(params)
response = requests.get(**request_args)
print(f"Made request to {response.request.url}, response {response.content}")
if response.text.startswith("ERROR"):
setattr(response, "success", False)
elif response.text.startswith("SUCCESS"):
setattr(response, "success", True)
return response
def get_version(self):
""" http://server/vicidial/non_agent_api.php?function=version """
return self.make_request("version")
def list_campaigns(self):
response = self.make_request("campaigns_list", source="script")
if not response.success:
raise ValueError(response.text)
return self.process_response(response.text)
def add_lead(
self,
phone_number, # must be all numbers, 6-16 digits
phone_code, # must be all numbers, 1-4 digits, defaults to 1 if not set
list_id, # must be all numbers, 3-12 digits, defaults to 999 if not set
campaign_id=None, # 2-8 Character campaign ID, required if using campaign_dnc_check or callbacks
vendor_lead_code=None, # 1-20 characters
first_name=None, # 1-30 characters
last_name=None, # 1-30 characters
alt_phone=None, # 1-12 characters
email=None, # 1-70 characters
comments=None, # 1-255 characters
):
params = {}
if campaign_id:
params["campaign_id"] = campaign_id
if vendor_lead_code:
params["vendor_lead_code"] = vendor_lead_code
if first_name:
params["first_name"] = first_name
if last_name:
params["last_name"] = last_name
if alt_phone:
params["alt_phone"] = alt_phone
if email:
params["email"] = email
if comments:
params["comments"] = comments
response = self.make_request(
"add_lead",
phone_number=phone_number,
phone_code=phone_code,
list_id=list_id,
**params,
)
if response.success:
return self.process_response(response.content)
raise ValueError(response.text)
def update_lead(
self,
lead_id,
**kwargs,
):
params = {k: v for k, v in kwargs.items() if k in ALLOWED_FIELDS}
response = self.make_request(
"update_lead",
lead_id=lead_id,
list_id_field=list_id_field,
)
if response.success:
return self.process_response(response.content)
raise ValueError(response.text)
def batch_update_lead(
self,
lead_ids,
**kwargs,
):
ALLOWED_FIELDS = [
"list_exists_check", # or N, default is N. If the list_id_field is not a defined list in the system, it will ERROR and not update the leads.
"reset_lead", # Y or N, Setting this to Y will reset the called-since-last-reset flag of the lead, default is N.
"records", # maximum number of records to update (defaults to '100'[most recently created leads])
"user_field", # 1-20 characters, this updates the 'user' field in the vicidial_list table
"list_id_field", # 3-12 digits, this updates the 'list_id' field in the vicidial_list table
"status", # 1-6 characters, not punctuation or spaces
"vendor_lead_code", # 1-20 characters
"source_id", # 1-50 characters
"gmt_offset_now", # overridden by auto-lookup of phone_code and area_code portion of phone number if applicable
"title", # 1-4 characters
"first_name", # 1-30 characters
"middle_initial", # 1 character
"last_name", # 1-30 characters
"address1", # 1-100 characters
"address2", # 1-100 characters
"address3", # 1-100 characters
"city", # 1-50 characters
"state", # 2 characters
"province", # 1-50 characters
"postal_code", # 1-10 characters
"country_code", # 3 characters
"gender", # U, M, F (Undefined, Male, Female) - defaults to 'U'
"date_of_birth", # YYYY-MM-DD
"alt_phone", # 1-12 characters
"email", # 1-70 characters
"security_phrase", # 1-100 characters
"comments", # 1-255 characters
"rank", # 1-5 digits
"owner", # 1-20 characters (user ID, Territory or user group)
"called_count", # digits only, the number of attempts dialing the lead
"phone_code", # digits only, can be 1-4 digits
]
params = {k: v for k, v in kwargs.items() if k in ALLOWED_FIELDS}
response = self.make_request(
"update_lead",
lead_ids=lead_ids,
**params
)
if response.success:
return self.process_response(response.content)
raise ValueError(response.text)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment