Create bot.py

bot.py

@@ -0,0 +1,262 @@
+class DIDx_Chatbot:
+    def __init__(self):
+        import openai
+
+
+
+        def get_balance():
+            import requests
+            response = requests.get(
+                f'https://newapi.didx.net/DidxApis/api/WebGetAccountBalance.php?UserID={user_id}&Password={password}')
+            return response.text
+
+        def get_available_dids_country():
+            import requests
+            response = requests.get(
+                f'https://newapi.didx.net/DidxApis/api/getDIDCountry.php?UserID={user_id}&Password={password}')
+            return response.text
+
+        def get_available_dids(CountryCode, AreaCode):
+            import requests
+            response = requests.get(
+                f'https://newapi.didx.net/DidxApis/api/getAvailableDIDS.php?UserID={user_id}&Password={password}&CountryCode={CountryCode}&AreaCode={AreaCode}')
+            return response.text
+
+        def get_purchased_dids():
+            import requests
+            response = requests.get(
+                f'https://newapi.didx.net/DidxApis/api/getMyPurchasedNumbers.php?UserID={user_id}&Password={password}')
+            return response.text
+
+        self.get_balance = get_balance
+        self.get_available_dids_country = get_available_dids_country
+        self.get_available_dids = get_available_dids
+        self.get_purchased_dids = get_purchased_dids
+
+        function_balance = {
+            "type": "function",
+            "function": {
+                "name": "get_balance",
+                "description": "Retrieve the available balance present in the account",
+                "parameters": {
+                    "type": "object",
+                    "properties": {
+                        "Balance": {
+                            "type": "string",
+                            "description": "A balance in the amount of dollars"
+                        },
+                        "Status": {
+                            "type": "string",
+                            "description": "A balance status due or something else"
+
+                        }
+                    },
+                    "required": ["account balance", "Status"]
+                }
+            }
+        }
+        function_dids_country = {
+            "type": "function",
+            "function": {
+                "name": "get_available_dids_country",
+                "description": "Retrieve the available DIDs country information",
+                "parameters": {
+                    "type": "object",
+                    "properties": {
+                        "country_code": {
+                            "type": "string",
+                            "description": "Country code for available DIDs"
+                        },
+                        "description": {
+                            "type": "string",
+                            "description": "Country/City/County code for available DIDs"
+
+                        },
+                        "country_id": {
+                            "type": "string",
+                            "description": "Country/City/County code for available DIDs"
+                        }
+
+                    },
+                    "required": ["country_code", "description", "country_id"]
+                }
+            }
+        }
+        function_dids = {
+            "type": "function",
+            "function": {
+                "name": "get_available_dids",
+                "description": "Retrieve all available DIDs list",
+                "parameters": {
+                    "type": "object",
+                    "properties": {
+                        "CountryCode": {
+                            "type": "string",
+                            "description": "Country code for available DIDs. Default CountryCode is 1 for US"
+                        },
+                        "description": {
+                            "type": "string",
+                            "description": "A list of available DIDs"
+
+                        },
+                        "AreaCode": {
+                            "type": "string",
+                            "description": "Area Code for the available DIDs. Default AreaCode is -1 that will show all AreaCode"
+                        }
+
+                    },
+                    "required": ["CountryCode", "description", "AreaCode"]
+                }
+            }
+        }
+
+        function_purchased_dids = {
+            "type": "function",
+            "function": {
+                "name": "get_purchased_dids",
+                "description": "Retrieve all purchased DIDs",
+                "parameters": {
+                    "type": "object",
+                    "properties": {
+                        "DIDNumber": {
+                            "type": "string",
+                            "description": "Purchased DIDNumber"
+                        },
+                        "OurSetupCost": {
+                            "type": "string",
+                            "description": "Our Setup Cost"
+
+                        },
+                        "OurMonthlyCharges": {
+                            "type": "string",
+                            "description": "Our Monthly Charges"
+                        },
+                        "OurPerMinuteCharges": {
+                            "type": "string",
+                            "description": "Our Per-Minute Charges"
+                        }
+
+                    },
+                    "required": ["CountryCode", "description", "AreaCode"]
+                }
+            }
+        }
+
+
+        import os
+        from dotenv import load_dotenv
+
+        # Load environment variables from .env file
+        load_dotenv()
+
+        self.client = openai.OpenAI(api_key=os.environ['openai_api_key'])
+        print(self.client)
+        # Step 1: Create an Assistant
+        self.assistant = self.client.beta.assistants.create(
+            name="DIDx Customer Support Chatbot",
+            instructions="You are a personal DIDx customer support chatbot.",
+            tools=[function_balance, function_dids_country, function_dids, function_purchased_dids],
+            model="gpt-4-1106-preview",
+        )
+
+    def user_auth(self, user, passw):
+        global user_id
+        global password
+
+        user_id = user
+        password = passw
+
+    def user_chat(self, query):
+        import time
+        # Step 2: Create a Thread
+        thread = self.client.beta.threads.create()
+
+        # Step 3: Add a Message to a Thread
+        message = self.client.beta.threads.messages.create(
+            thread_id=thread.id,
+            role="user",
+            content=query
+        )
+
+        # Step 4: Run the Assistant
+        run = self.client.beta.threads.runs.create(
+            thread_id=thread.id,
+            assistant_id=self.assistant.id,
+            instructions=""
+        )
+        answer = None
+        while True:
+            # Retrieve the run status
+            run_status = self.client.beta.threads.runs.retrieve(
+                thread_id=thread.id,
+                run_id=run.id
+            )
+            # print(run_status.model_dump_json(indent=4))
+            run_status.model_dump_json(indent=4)
+
+            # If run is completed, get messages
+            if run_status.status == 'completed':
+                messages = self.client.beta.threads.messages.list(
+                    thread_id=thread.id
+                )
+                # Loop through messages and print content based on role
+                for msg in messages.data:
+                    role = msg.role
+                    content = msg.content[0].text.value
+                    print(f"{role.capitalize()}: {content}")
+                    answer = f"{role.capitalize()}: {content}"
+                    break
+                break
+            elif run_status.status == 'requires_action':
+                # print("Function Calling")
+                required_actions = run_status.required_action.submit_tool_outputs.model_dump()
+                # print('required action test: ',required_actions)
+                tool_outputs = []
+                import json
+                for action in required_actions["tool_calls"]:
+                    func_name = action['function']['name']
+                    arguments = json.loads(action['function']['arguments'])
+
+                    if func_name == "get_balance":
+                        output = self.get_balance()
+                        tool_outputs.append({
+                            "tool_call_id": action['id'],
+                            "output": output
+                        })
+
+                    elif func_name == 'get_available_dids_country':
+                        output = self.get_available_dids_country()
+                        tool_outputs.append({
+                            "tool_call_id": action['id'],
+                            "output": output
+                        })
+                    elif func_name == 'get_available_dids':
+                        output = self.get_available_dids(CountryCode=arguments['CountryCode'],
+                                                         AreaCode=arguments['AreaCode'])
+                        tool_outputs.append({
+                            "tool_call_id": action['id'],
+                            "output": output
+                        })
+                    elif func_name == 'get_purchased_dids':
+                        output = self.get_purchased_dids()
+                        tool_outputs.append({
+                            "tool_call_id": action['id'],
+                            "output": output
+                        })
+
+                    else:
+                        raise ValueError(f"Unknown function: {func_name}")
+
+                print("Submitting outputs back to the Assistant...")
+                self.client.beta.threads.runs.submit_tool_outputs(
+                    thread_id=thread.id,
+                    run_id=run.id,
+                    tool_outputs=tool_outputs
+                )
+            else:
+                print("Waiting for the Assistant to process...")
+                time.sleep(5)
+
+        if answer is not None:
+            print(f'this is my answer : ', answer)
+            return answer