Spaces:
Paused
Paused
| 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() |