Vicidial Python Wrapper

vicidial_python.py

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)