Blame for app.py

import requests
import json
from uuid import UUID
from typing import Dict, Any, Optional
import gradio as gr
import os
import threading
import time
import pandas as pd

ZGM_API_TOKEN = os.getenv("ZGM_API_TOKEN")
if not ZGM_API_TOKEN:
    raise ValueError("ZGM_API_TOKEN not found in environment variables!")

VPS_IP = os.getenv("VPS_IP")
if not VPS_IP:
    raise ValueError("VPS_IP not found in environment variables!")

KEY_API_URL = f"http://{VPS_IP}:3000"

class ZGMApiClient:
    def __init__(self, base_url: str, api_token: str):
        self.base_url = base_url.rstrip('/')
        self.api_token = api_token
        self.headers = {
            'ZGM-API-TOKEN': api_token,
            'Content-Type': 'application/json',
            'Accept': 'text/plain'
        }

    def _make_request(self, method: str, endpoint: str, params: Dict = None, data: Dict = None) -> Dict:
        url = f"{self.base_url}{endpoint}"
        try:
            response = requests.request(
                method=method,
                url=url,
                headers=self.headers,
                params=params,
                json=data,
                timeout=15
            )
            response.raise_for_status()
            return response.json()
        except requests.exceptions.ConnectTimeout:
            return {"error": f"Cannot connect to {self.base_url}. Please check URL or network.", "status_code": 0}
        except requests.exceptions.HTTPError as e:
            if response.status_code == 401:
                return {"error": "Unauthorized. Please check API Token.", "status_code": 401}
            elif response.status_code == 500:
                return {"error": "Server error from ZGM API.", "status_code": 500}
            return {"error": f"HTTP Error: {str(e)}", "status_code": response.status_code}
        except requests.exceptions.RequestException as e:
            return {"error": f"Connection error: {str(e)}", "status_code": 0}

    def rent_otp(self, service: str) -> Dict:
        data = {"service": service}
        return self._make_request('POST', '/api/mail/otp-rent', data=data)

    def get_otp(self, order_id: str) -> Dict:
        try:
            UUID(order_id)
            return self._make_request('GET', '/api/mail/otp-get', params={"orderId": order_id})
        except ValueError:
            return {"error": "order_id must be a valid UUID", "status_code": 400}

client = ZGMApiClient("https://gm.haivl.wtf", ZGM_API_TOKEN)

def validate_key_from_api(access_key: str) -> tuple[bool, str]:
    try:
        response = requests.get(f"{KEY_API_URL}/api/keys/validate", params={"key": access_key}, timeout=5)
        response.raise_for_status()
        data = response.json()
        if data["valid"]:
            return True, f"Remaining {data['remaining']} keys."
        else:
            return False, "Invalid key!"
    except requests.exceptions.RequestException as e:
        return False, f"Error checking key from API: {str(e)}"

def remove_key_from_api(access_key: str) -> tuple[bool, str]:
    try:
        response = requests.delete(f"{KEY_API_URL}/api/keys/remove", params={"key": access_key}, timeout=5)
        response.raise_for_status()
        data = response.json()
        if data["success"]:
            return True, f"Remaining {data['remaining']} keys."
        else:
            return False, "Cannot remove key."
    except requests.exceptions.RequestException as e:
        return False, f"Error removing key from API: {str(e)}"

def validate_and_remove_key(access_key: str, result: Dict) -> tuple[bool, str]:
    is_valid, message = validate_key_from_api(access_key)
    if not is_valid:
        return False, message
    if result.get("code") == 200 and result.get("data") and "orderId" in result["data"] and "email" in result["data"]:
        success, remove_message = remove_key_from_api(access_key)
        if not success:
            return True, f"OTP rented successfully but {remove_message}"
        remaining_keys = int(remove_message.split("Remaining ")[1].split(" key")[0])
        remaining_message = f"Remaining {remaining_keys} keys."
        if remaining_keys < 100:
            remaining_message += " Low key count! Please create more keys."
        return True, remaining_message
    else:
        return True, f"Error: {result.get('message', 'Unable to rent Gmail.')}\nKey not deleted."

def execute_with_loading(func, *args, **kwargs):
    result = [None]
    is_loading = [True]

    def target():
        result[0] = func(*args, **kwargs)
        is_loading[0] = False

    api_thread = threading.Thread(target=target)
    api_thread.start()

    while is_loading[0]:
        time.sleep(0.1)
    api_thread.join()
    return result[0]

def format_rent_otp_result(data: Dict) -> pd.DataFrame:
    if not data or "data" not in data:
        return pd.DataFrame([["No data to display."]], columns=["Message"])
    order_info = data["data"]
    df = pd.DataFrame([{
        "Order ID": order_info.get("orderId", "N/A"),
        "Email": order_info.get("email", "N/A")
    }])
    return df

def format_get_otp_result(data: Dict) -> pd.DataFrame:
    if not data or "data" not in data:
        return pd.DataFrame([["No data to display."]], columns=["Message"])
    otp_data = data["data"]
    if isinstance(otp_data, dict):
        otp_data = [otp_data]
    if not isinstance(otp_data, list):
        return pd.DataFrame([["Invalid data."]], columns=["Message"])
    df = pd.DataFrame([
        {
            "Email": item.get("email", "N/A"),
            "OTP Code": item.get("otpCode", "N/A"),
            "Content": item.get("otpContent", "N/A")
        }
        for item in otp_data
    ])
    return df

def clear_all_logs():
    return pd.DataFrame([["No data yet"]], columns=["Message"]), pd.DataFrame([["No data yet"]], columns=["Message"]), ""

def gradio_interface(access_key, action, service, order_id, rent_otp_log, get_otp_log):
    message = ""
    new_order_id = order_id
    if rent_otp_log is None:
        rent_otp_log = pd.DataFrame([["No data yet"]], columns=["Message"])
    if get_otp_log is None:
        get_otp_log = pd.DataFrame([["No data yet"]], columns=["Message"])
    if action == "Rent OTP":
        if not access_key:
            rent_otp_log = pd.DataFrame([["Please enter a key to rent OTP!"]], columns=["Message"])
            return rent_otp_log, get_otp_log, message, order_id
        if not service:
            rent_otp_log = pd.DataFrame([["Please select a service!"]], columns=["Message"])
            return rent_otp_log, get_otp_log, message, order_id
        result = execute_with_loading(client.rent_otp, service)
        is_valid, message = validate_and_remove_key(access_key, result)
        if not is_valid:
            rent_otp_log = pd.DataFrame([[message]], columns=["Message"])
            return rent_otp_log, get_otp_log, message, order_id
        if result.get("code") == 200:
            rent_otp_log = format_rent_otp_result(result)
            new_order_id = result["data"].get("orderId", order_id)
        else:
            rent_otp_log = pd.DataFrame([[json.dumps(result, indent=2, ensure_ascii=False)]], columns=["Message"])
        return rent_otp_log, get_otp_log, message, new_order_id
    elif action == "Get OTP":
        if not order_id:
            get_otp_log = pd.DataFrame([["Please enter Order ID!"]], columns=["Message"])
            return rent_otp_log, get_otp_log, "", order_id
        result = execute_with_loading(client.get_otp, order_id)
        if result.get("code") == 200 and "data" in result:
            get_otp_log = format_get_otp_result(result)
        else:
            get_otp_log = pd.DataFrame([[json.dumps(result, indent=2, ensure_ascii=False)]], columns=["Message"])
        return rent_otp_log, get_otp_log, "", order_id
    else:
        return rent_otp_log, get_otp_log, "Invalid action!", order_id

SERVICES = [
    "ICLOUD", "ZSK", "FACEBOOK", "TELEGRAM",
    "GGPLAY", "INSTAGRAM", "FMCPAY", "AMAZON"
]

NOTIFICATION_FILE = "notification.txt"

def read_notification() -> str:
    if os.path.exists(NOTIFICATION_FILE):
        with open(NOTIFICATION_FILE, "r", encoding="utf-8") as f:
            return f.read()
    return ""

def update_notification(msg_input, password):
    if password == "2312":
        with open(NOTIFICATION_FILE, "w", encoding="utf-8") as f:
            f.write(msg_input)
        return msg_input
    return read_notification()

with gr.Blocks() as demo:
    gr.Markdown("## Gmail Trust 247 - Enter Key to Rent OTP")

    with gr.Row():
        access_key = gr.Textbox(label="Access Key (Required for Rent OTP)", placeholder="Enter 32-char key...")
        action = gr.Dropdown(label="Select Action", choices=["Rent OTP", "Get OTP"], value="Rent OTP")

    with gr.Row():
        service = gr.Dropdown(label="Service (For Rent OTP)", choices=SERVICES)
        order_id = gr.Textbox(label="Order ID (For Get OTP)", placeholder="Enter Order ID")

    rent_otp_log = gr.State(value=pd.DataFrame([["No data yet"]], columns=["Message"]))
    get_otp_log = gr.State(value=pd.DataFrame([["No data yet"]], columns=["Message"]))

    rent_otp_output = gr.Dataframe(
        label="Rent OTP Log",
        headers=["Order ID", "Email"],
        datatype=["str", "str"],
        interactive=False
    )

    get_otp_output = gr.Dataframe(
        label="Get OTP Log",
        headers=["Email", "OTP Code", "Content"],
        datatype=["str", "str", "str"],
        interactive=False
    )

    message_output = gr.Textbox(label="Message", interactive=False)

    with gr.Row():
        submit_btn = gr.Button("Submit", variant="primary")
        clear_btn = gr.Button("Clear Log", variant="secondary")

    submit_btn.click(
        fn=gradio_interface,
        inputs=[access_key, action, service, order_id, rent_otp_log, get_otp_log],
        outputs=[rent_otp_output, get_otp_output, message_output, order_id]
    )

    clear_btn.click(
        fn=clear_all_logs,
        inputs=None,
        outputs=[rent_otp_output, get_otp_output, message_output]
    )

    gr.Markdown("## System Notification")
    with gr.Row():
        notification_input = gr.Textbox(label="View the number of gmail in stock", lines=3)
        password_input = gr.Textbox(label="Password", type="password")
    with gr.Row():
        update_btn = gr.Button("View the number of gmail in stock")
        current_notification = gr.Textbox(label="gmail in stock", value=read_notification(), interactive=False)

    update_btn.click(update_notification, inputs=[notification_input, password_input], outputs=current_notification)

    gr.Markdown("## Get OTP Log")

demo.launch()