Update for Sign Up Forms

.gitignore

@@ -1 +1,11 @@
 .venv/
+__pycache__/
+*.pyc
+*.pyo
+*.pyd
+.env
+*.ipynb
+.ipynb_checkpoints/
+*.log
+*.log.*
+*.log-*

app.py

@@ -1,27 +1,38 @@
 """gradio app for Chat Interface for DataStax Langflow calls"""
 
-import json
+import re
+from datetime import datetime
 from typing import Optional, Sequence, Tuple
 from uuid import uuid4
+import os
 import requests
 import gradio as gr
-import os
+from dotenv import load_dotenv
+
+load_dotenv()
+
 
 BASE_API_URL = "https://api.langflow.astra.datastax.com"
 LANGFLOW_ID = "e99b525a-84dd-4cf2-bac6-8e5ebc6a7d17"
 FLOW_ID = "bfbf7237-50dd-40dd-baba-cc680288d788"
-APPLICATION_TOKEN = os.getenv('DATASTAX_TOKEN')
+ASTRA_DB_CLIENT_ID = os.getenv('ASTRA_DB_CLIENT_ID')
+ASTRA_DB_ID = os.getenv('ASTRA_DB_ID')
+ASTRA_DB_KEYSPACE = os.getenv('ASTRA_DB_KEYSPACE')
+ASTRA_DB_REGION = os.getenv('ASTRA_DB_REGION')
+ASTRA_DB_SECRET = os.getenv('ASTRA_DB_SECRET')
+ASTRA_DB_TOKEN = os.getenv('ASTRA_DB_TOKEN')
+DATASTAX_TOKEN = os.getenv('DATASTAX_TOKEN')
 
 ENDPOINT = 'ai_traveller'
 
 TWEAKS = {
     "Memory-fCuCs": {
-    "n_messages": 100,
-    "order": "Ascending",
-    "sender": "Machine and User",
-    "sender_name": "",
-    "template": "{sender_name}: {text}"
-  },
+        "n_messages": 100,
+        "order": "Ascending",
+        "sender": "Machine and User",
+        "sender_name": "",
+        "template": "{sender_name}: {text}"
+    },
     "ChatInput-PN5E4": {},
     "Prompt-vGvVG": {},
     "GoogleGenerativeAIModel-6n0Ft": {},
@@ -29,6 +40,52 @@ TWEAKS = {
 }
 
 
+def email_validator(_email: str) -> bool:
+    """RegEx based Validate the email address."""
+    return bool(re.match(r"[^@]+@[^@]+\.[^@]+", _email))
+
+
+def contact_validator(_contact: str) -> bool:
+    """RegEx based Validate the contact number."""
+    return bool(re.match(r'^\d{10}$', _contact))
+
+
+def required_validator(_name: str, _email: str, _contact: str) -> bool:
+    """Validate the required fields."""
+    return bool(_name.strip() and _email.strip() and _contact.strip())
+
+
+def submit_signup(_email: str, _name: str, _contact: str):
+    """Add signups to the database."""
+
+    if not email_validator(_email):
+        return 'Invalid Email'
+
+    if not contact_validator(_contact):
+        return 'Invalid Contact Number'
+
+    if not required_validator(_name, _email, _contact):
+        return 'All Fields Required for Sign Up'
+
+    response = requests.post(
+        f'https://{ASTRA_DB_ID}-{ASTRA_DB_REGION}.apps.astra.datastax.com/api/rest/v2/keyspaces/{ASTRA_DB_KEYSPACE}/signups',
+        headers={
+            'Content-Type': 'application/json',
+            'X-Cassandra-Token': ASTRA_DB_TOKEN
+        },
+        json={
+            'email': _email,
+            'name': _name,
+            'contact_no': _contact,
+            'created_at': datetime.now().strftime('%Y-%m-%dT%H:%M:%SZ')
+        },
+        timeout=360
+    )
+    if response.status_code != 201:
+        return 'Error'
+    return 'Success'
+
+
 def call_travel_ai(
     message: str,
     history: Sequence[Tuple[str, str]],
@@ -37,7 +94,7 @@ def call_travel_ai(
     output_type: str = "chat",
     input_type: str = "chat",
     tweaks: Optional[dict] = None,
-    application_token: Optional[str] = APPLICATION_TOKEN
+    application_token: Optional[str] = DATASTAX_TOKEN
 ) -> dict:
     """
     Run a flow with a given message and optional tweaks.
@@ -47,6 +104,7 @@ def call_travel_ai(
     :param tweaks: Optional tweaks to customize the flow
     :return: The JSON response from the flow
     """
+    del history
     api_url = f"{BASE_API_URL}/lf/{LANGFLOW_ID}/api/v1/run/{endpoint}"
 
     payload = {
@@ -78,27 +136,62 @@ def call_travel_ai(
 
     return resp_message
 
+
 def generate_session_id():
+    """Generate a unique session ID."""
     return str(uuid4())
 
-def render_value(value):
-    return value
 
 with gr.Blocks() as demo:
-    session_id = gr.Textbox(
+    random_session_id = gr.Textbox(
         value=generate_session_id,
         visible=False,
         interactive=False,
         label="Session ID",
     )
 
-    chat_interface = gr.ChatInterface(
-        call_travel_ai,
-        type='messages',
-        title=f"AI Travel Partner",
-        additional_inputs=[session_id],
-        autofocus=True,
-        fill_height=True,
+    with gr.Accordion(
+        'Sign Up',
+    ) as signup:
+
+        with gr.Column():
+            name = gr.Textbox(label="Name")
+            email = gr.Textbox(label="Email")
+            phone = gr.Textbox(label="Phone")
+            submit_button = gr.Button("Submit")
+
+    with gr.Accordion(
+        'Chat With Agent',
+        visible=False,
+    ) as chat:
+        chat_interface = gr.ChatInterface(
+            call_travel_ai,
+            type='messages',
+            title="AI Travel Partner",
+            additional_inputs=[random_session_id],
+            autofocus=True,
+            fill_height=True,
+        )
+
+    def update_visibility(
+        _name: str,
+        _email: str,
+        _phone: str,
+
+    ):
+        """Update the visibility of the sign up and chat interfaces."""
+        _status = submit_signup(_email, _name, _phone)
+        gr.Info(_status, duration=5, title='Sign Up Status')
+
+        return (
+            gr.update(visible=_status != 'Success'),
+            gr.update(visible=_status == 'Success')
+        )
+
+    submit_button.click(  # pylint: disable=no-member
+        update_visibility,
+        [name, email, phone],
+        outputs=[signup, chat],
     )
 
 demo.launch(share=False, debug=True,)

requirements.txt

@@ -2,3 +2,5 @@ gradio
 pandas
 requests
 numpy
+cassandra-driver
+python-dotenv