Glowquest

#!/usr/bin/env python3
import asyncio
import random
from asyncio import Semaphore, Task
from typing import Any, Dict, List, Optional, Union

from aiohttp import ClientSession, ClientError, ClientTimeout, TCPConnector

class Colors:
    """
    @brief ANSI color codes for terminal output
    """
    BLACK: str = '\033[30m'
    RED: str = '\033[31m'
    GREEN: str = '\033[32m'
    YELLOW: str = '\033[33m'
    BLUE: str = '\033[34m'
    MAGENTA: str = '\033[35m'
    CYAN: str = '\033[36m'
    WHITE: str = '\033[37m'

    BRIGHT_BLACK: str = '\033[90m'
    BRIGHT_RED: str = '\033[91m'
    BRIGHT_GREEN: str = '\033[92m'
    BRIGHT_YELLOW: str = '\033[93m'
    BRIGHT_BLUE: str = '\033[94m'
    BRIGHT_MAGENTA: str = '\033[95m'
    BRIGHT_CYAN: str = '\033[96m'
    BRIGHT_WHITE: str = '\033[97m'

    BOLD: str = '\033[1m'
    DIM: str = '\033[2m'
    UNDERLINE: str = '\033[4m'
    RESET: str = '\033[0m'

    @staticmethod
    def task(text: str) -> str:
        """
        @brief Color for Task keyword
        @param text Text to color
        @return Colored text
        """
        return f"{Colors.BRIGHT_CYAN}{Colors.BOLD}Task{Colors.RESET}{text}"
    
    @staticmethod
    def proxy(text: str) -> str:
        """
        @brief Color for Proxy keyword
        @param text Text to color
        @return Colored text
        """
        return f"{Colors.BRIGHT_MAGENTA}{Colors.BOLD}Proxy{Colors.RESET}{text}"
    
    @staticmethod
    def tokens(text: str) -> str:
        """
        @brief Color for Tokens keyword
        @param text Text to color
        @return Colored text
        """
        return f"{Colors.BRIGHT_GREEN}{Colors.BOLD}Tokens{Colors.RESET}{text}"
    
    @staticmethod
    def error(text: str) -> str:
        """
        @brief Color for Error keyword
        @param text Text to color
        @return Colored text
        """
        return f"{Colors.BRIGHT_RED}{Colors.BOLD}Error{Colors.RESET}{text}"
    
    @staticmethod
    def response(text: str) -> str:
        """
        @brief Color for Response keyword
        @param text Text to color
        @return Colored text
        """
        return f"{Colors.BRIGHT_YELLOW}{Colors.BOLD}Response{Colors.RESET}{text}"
    
    @staticmethod
    def http_status(status: int) -> str:
        """
        @brief Color HTTP status codes
        @param status HTTP status code
        @return Colored status code
        """
        if status == 200:
            return f"{Colors.BRIGHT_GREEN}{status}{Colors.RESET}"
        elif 400 <= status < 500:
            return f"{Colors.BRIGHT_YELLOW}{status}{Colors.RESET}"
        elif status >= 500:
            return f"{Colors.BRIGHT_RED}{status}{Colors.RESET}"
        else:
            return f"{Colors.CYAN}{status}{Colors.RESET}"
    
    @staticmethod
    def number(num: int) -> str:
        """
        @brief Color for numbers
        @param num Number to color
        @return Colored number
        """
        return f"{Colors.BRIGHT_WHITE}{num}{Colors.RESET}"

PROMPT: str = """
I need a comprehensive and specific explanation of general relativity that covers the following points in detail:

Fundamental Concepts: Explain the key ideas of spacetime, gravity, and the equivalence principle in the context of general relativity. How do these concepts differ from Newtonian physics?

Mathematical Framework: Describe the role of the Einstein Field Equations (EFE) in general relativity. How do they describe the relationship between matter, energy, and the curvature of spacetime? Provide a basic overview of how these equations work without going too deep into advanced tensor calculus.

Geometrical Interpretation: Discuss the concept of spacetime curvature and how massive objects like stars and planets "bend" spacetime. What is the significance of the geodesic equation in this context?

Experimental Evidence: Provide a summary of key experimental and observational evidence that supports general relativity, such as the bending of light by gravity (gravitational lensing), the perihelion precession of Mercury, and the detection of gravitational waves. How do these observations confirm Einstein's predictions?

Black Holes and Singularities: Explain the formation and nature of black holes in general relativity. What is meant by the "event horizon" and "singularity," and how do these concepts emerge from the Einstein Field Equations?

Relativity and Cosmology: How does general relativity influence our understanding of the universe on cosmological scales? Discuss its role in the expansion of the universe, dark energy, and the Big Bang theory.

Be sure to include the necessary context to make these concepts understandable to someone with a basic understanding of physics but who is not necessarily an expert.
"""

async def load_proxies(file_path: str) -> List[str]:
    """
    @brief Load proxy list from file and return with indices
    @param file_path Path to the proxy file
    @return List of proxy strings
    """
    try:
        with open(file_path, 'r') as file:
            proxies: List[str] = [line.strip() for line in file.readlines() if line.strip()]
        return proxies
    except FileNotFoundError:
        print(f"{Colors.BRIGHT_YELLOW}Warning:{Colors.RESET} {file_path} not found, running without proxies")
        return []

async def run_async_tasks(num_tasks: int = 200, max_concurrent: int = 200) -> None:
    """
    @brief Run multiple async tasks with proxy rotation
    @param num_tasks Number of async tasks to spawn
    @param max_concurrent Maximum concurrent requests
    """
    proxies_list: List[str] = await load_proxies("proxies.txt")
    
    print(f"{Colors.BRIGHT_GREEN}Loaded{Colors.RESET} {Colors.number(len(proxies_list))} proxies")
    print(f"{Colors.BRIGHT_BLUE}Starting{Colors.RESET} {Colors.number(num_tasks)} async tasks (max {Colors.number(max_concurrent)} concurrent)...")
    print(f"{Colors.BRIGHT_CYAN}{'-' * 50}{Colors.RESET}")
    
    semaphore: Semaphore = Semaphore(max_concurrent)
    
    connector: TCPConnector = TCPConnector(
        limit=100,
        limit_per_host=30,
        ttl_dns_cache=300,
        use_dns_cache=True,
    )
    
    timeout: ClientTimeout = ClientTimeout(total=15, connect=5)
    
    async with ClientSession(
        connector=connector,
        timeout=timeout,
        headers={
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36'
        }
    ) as session:
        tasks: List[Task] = [
            asyncio.create_task(funny(session, proxies_list, task_id, semaphore))
            for task_id in range(num_tasks)
        ]

        results: List[Union[int, None]] = await asyncio.gather(*tasks, return_exceptions=True)

        successful_tasks: int = sum(1 for r in results if isinstance(r, int) and r > 0)
        total_tokens: int = sum(r for r in results if isinstance(r, int))

        print(f"{Colors.BRIGHT_CYAN}{'-' * 50}{Colors.RESET}")
        print(f"{Colors.BRIGHT_GREEN}{Colors.BOLD}COMPLETED:{Colors.RESET} {Colors.number(successful_tasks)}/{Colors.number(num_tasks)} tasks successful")
        print(f"{Colors.BRIGHT_YELLOW}{Colors.BOLD}TOTAL {Colors.tokens(': ')} BURNED:{Colors.RESET} {Colors.number(total_tokens)}")
        print(f"{Colors.BRIGHT_CYAN}{'-' * 50}{Colors.RESET}")

async def funny(session: ClientSession, proxies_list: List[str], task_id: int, semaphore: Semaphore) -> int:
    """
    @brief Make async API requests with proxy rotation and truncated output
    @param session aiohttp ClientSession
    @param proxies_list List of available proxies
    @param task_id Unique identifier for this async task
    @param semaphore Semaphore to limit concurrent requests
    @return Number of tokens burned
    """
    tokens_burned: int = 0
    request_count: int = 0
    max_requests: int = 100

    url: str = "https://base44.app/api/apps/68b4c950a14b75d1d461b68c/integration-endpoints/Core/InvokeLLM"

    headers: Dict[str, str] = {
        'accept': 'application/json',
        'accept-language': 'en-US,en;q=0.9',
        # 'authorization': 'Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJyZXRhcmRyZXRhcmRyZUBnbWFpbC5jb20iLCJleHAiOjE3NTk4Njc2MDgsImlhdCI6MTc1OTI2MjgwOH0.-AN4zdF_UtXXjPov_dFJtleuLWFJpyYS04EaL1v5bGE',
        'cache-control': 'no-cache',
        'origin': 'https://glowquest.app',
        'pragma': 'no-cache',
        'priority': 'u=1, i',
        'referer': 'https://glowquest.app/',
        'sec-ch-ua': '"Chromium";v="140", "Not=A?Brand";v="24", "Google Chrome";v="140"',
        'sec-ch-ua-mobile': '?0',
        'sec-ch-ua-platform': '"Windows"',
        'sec-fetch-dest': 'empty',
        'sec-fetch-mode': 'cors',
        'sec-fetch-site': 'cross-site',
        'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36',
        'x-app-id': '68b4c950a14b75d1d461b68c',
        'x-origin-url': 'https://glowquest.app/Profile',
    }

    payload: Dict[str, Any] = {
        # "prompt": "Generate better site than https://glowquest.app/ in nodejs",
        "prompt": PROMPT,
        "add_context_from_internet": True
    }

    while request_count < max_requests:
        request_count += 1
        
        proxy_index: int = random.randint(0, len(proxies_list) - 1) if proxies_list else 0
        proxy: str = proxies_list[proxy_index] if proxies_list else "No proxy"
        proxy_url: Optional[str] = f'http://{proxy}' if proxies_list else None

        try:
            async with semaphore:
                async with session.post(
                    url, 
                    proxy=proxy_url,
                    headers=headers, 
                    json=payload,
                    timeout=ClientTimeout(total=10)
                ) as response:
                    response_text: str = await response.text()
                    
                    if response.status == 200:
                        tokens_burned += 1
                        truncated_response: str = response_text[:100]
                        if len(response_text) > 100:
                            truncated_response += "..."
                        
                        print(f"{Colors.task(f' {task_id:3d}')} | {Colors.proxy(f' #{proxy_index:3d}')} | {Colors.tokens(f': {tokens_burned:3d}')} | {Colors.response(f': {truncated_response}')}")
                    else:
                        print(f"{Colors.task(f' {task_id:3d}')} | {Colors.proxy(f' #{proxy_index:3d}')} | HTTP {Colors.http_status(response.status)}")
                
        except ClientError as e:
            if request_count % 10 == 1:
                print(f"{Colors.task(f' {task_id:3d}')} | {Colors.proxy(f' #{proxy_index:3d}')} | {Colors.error(f': {str(e)[:50]}...')}")
            continue
            
        except asyncio.TimeoutError:
            if request_count % 10 == 1:
                print(f"{Colors.task(f' {task_id:3d}')} | {Colors.proxy(f' #{proxy_index:3d}')} | {Colors.error(': Timeout...')}")
            continue
            
        except Exception as e:
            if request_count % 10 == 1:
                print(f"{Colors.task(f' {task_id:3d}')} | {Colors.proxy(f' #{proxy_index:3d}')} | {Colors.error(f': Unexpected: {str(e)[:50]}...')}")
            continue

    print(f"{Colors.task(f' {task_id:3d}')} | {Colors.BRIGHT_GREEN}{Colors.BOLD}Completed{Colors.RESET} | Total {Colors.tokens(f' burned: {tokens_burned}')}")
    return tokens_burned

async def main() -> None:
    """
    @brief Main async function
    """
    
    try:
        await run_async_tasks(
            num_tasks=200,
            max_concurrent=200
        )
        
    except KeyboardInterrupt:
        print(f"\n{Colors.BRIGHT_YELLOW}Interrupted by user (Ctrl+C){Colors.RESET}")
    except Exception as e:
        print(f"Unexpected {Colors.error(f': {e}')}")

if __name__ == "__main__":
    """
    @brief Main entry point
    """
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print(f"\n{Colors.BRIGHT_YELLOW}Interrupted by user (Ctrl+C){Colors.RESET}")
    except Exception as e:
        print(f"Unexpected {Colors.error(f': {e}')}")