Update app.py

app.py

@@ -1,8 +1,6 @@
 import requests
 from bs4 import BeautifulSoup
-from fastapi import FastAPI, HTTPException, Form
-from fastapi.responses import PlainTextResponse, HTMLResponse
-from fastapi.staticfiles import StaticFiles
+from httpx import AsyncClient, HTTPError
 from pydantic import BaseModel
 import re
 import urllib.parse
@@ -18,14 +16,12 @@ from selenium.webdriver.support import expected_conditions as EC
 from selenium.webdriver.common.action_chains import ActionChains
 import time
 from webdriver_manager.chrome import ChromeDriverManager
-from fastapi import BackgroundTasks
 from tqdm import tqdm
 import asyncio
+import gradio as gr
 
 load_dotenv()
 
-app = FastAPI()
-
 class Proxy(BaseModel):
     ip: str
     port: str
@@ -38,8 +34,6 @@ class VisitRequest(BaseModel):
     delay: int
     parallel_processes: int
 
-app.mount("/static", StaticFiles(directory="static"), name="static")
-
 def get_random_proxy():
     try:
         options = webdriver.ChromeOptions()
@@ -66,7 +60,6 @@ def get_random_proxy():
     except Exception as e:
         return None
 
-@app.get("/get_proxies", response_class=PlainTextResponse)
 def get_proxies():
     try:
         proxies = []
@@ -76,9 +69,8 @@ def get_proxies():
                 proxies.append(f"{proxy.ip}:{proxy.port} ({proxy.type})")
         return "\n".join(proxies)
     except Exception as e:
-        return PlainTextResponse(str(e), status_code=500)
+        return str(e)
 
-@app.get("/rotate_ip", response_class=PlainTextResponse)
 def rotate_ip():
     try:
         fake = Faker()
@@ -90,10 +82,10 @@ def rotate_ip():
         }
         proxy = get_random_proxy()
         if not proxy:
-            return PlainTextResponse("No proxy available.", status_code=500)
-        return PlainTextResponse(f"{proxy.ip}:{proxy.port} ({proxy.type})")
+            return "No proxy available."
+        return f"{proxy.ip}:{proxy.port} ({proxy.type})"
     except Exception as e:
-        return PlainTextResponse(str(e), status_code=500)
+        return str(e)
 
 def extract_video_id(url: str, platform: str) -> str:
     url = urllib.parse.unquote(url)
@@ -116,93 +108,93 @@ def extract_video_id(url: str, platform: str) -> str:
     else:
         raise ValueError("Invalid URL format")
 
-def instagram_login(username: str, password: str):
+async def instagram_login(username: str, password: str):
     login_url = "https://www.instagram.com/accounts/login/ajax/"
-    session = requests.Session()
-    response = session.get("https://www.instagram.com/", headers={"User-Agent": "Mozilla/5.0"})
-    soup = BeautifulSoup(response.text, "html.parser")
-    csrf_token = soup.find("meta", {"name": "csrf-token"})["content"]
-    login_data = {
-        "username": username,
-        "enc_password": f"#PWD_INSTAGRAM_BROWSER:0:&:{password}"
-    }
-    headers = {
-        "User-Agent": "Mozilla/5.0",
-        "X-CSRFToken": csrf_token,
-        "X-Requested-With": "XMLHttpRequest"
-    }
-    response = session.post(login_url, data=login_data, headers=headers)
-    if response.status_code == 200 and response.json().get("authenticated"):
-        return session
-    else:
-        raise HTTPException(status_code=401, detail="Authentication failed")
+    async with AsyncClient() as session:
+        response = await session.get("https://www.instagram.com/", headers={"User-Agent": "Mozilla/5.0"})
+        soup = BeautifulSoup(response.text, "html.parser")
+        csrf_token = soup.find("meta", {"name": "csrf-token"})["content"]
+        login_data = {
+            "username": username,
+            "enc_password": f"#PWD_INSTAGRAM_BROWSER:0:&:{password}"
+        }
+        headers = {
+            "User-Agent": "Mozilla/5.0",
+            "X-CSRFToken": csrf_token,
+            "X-Requested-With": "XMLHttpRequest"
+        }
+        response = await session.post(login_url, data=login_data, headers=headers)
+        if response.status_code == 200 and response.json().get("authenticated"):
+            return session
+        else:
+            raise HTTPError("Authentication failed")
 
-def tiktok_login(username: str, password: str):
+async def tiktok_login(username: str, password: str):
     login_url = "https://www.tiktok.com/login/"
-    session = requests.Session()
-    login_data = {
-        "username": username,
-        "password": password
-    }
-    response = session.post(login_url, data=login_data)
-    if response.status_code == 200:
-        return session
-    else:
-        raise HTTPException(status_code=401, detail="Authentication failed")
+    async with AsyncClient() as session:
+        login_data = {
+            "username": username,
+            "password": password
+        }
+        response = await session.post(login_url, data=login_data)
+        if response.status_code == 200:
+            return session
+        else:
+            raise HTTPError("Authentication failed")
 
-def youtube_login(username: str, password: str):
+async def youtube_login(username: str, password: str):
     login_url = "https://accounts.google.com/ServiceLogin"
-    session = requests.Session()
-    login_data = {
-        "username": username,
-        "password": password
-    }
-    response = session.post(login_url, data=login_data)
-    if response.status_code == 200:
-        return session
-    else:
-        raise HTTPException(status_code=401, detail="Authentication failed")
+    async with AsyncClient() as session:
+        login_data = {
+            "username": username,
+            "password": password
+        }
+        response = await session.post(login_url, data=login_data)
+        if response.status_code == 200:
+            return session
+        else:
+            raise HTTPError("Authentication failed")
 
-def facebook_login(username: str, password: str):
+async def facebook_login(username: str, password: str):
     login_url = "https://www.facebook.com/login"
-    session = requests.Session()
-    login_data = {
-        "email": username,
-        "pass": password
-    }
-    response = session.post(login_url, data=login_data)
-    if response.status_code == 200:
-        return session
-    else:
-        raise HTTPException(status_code=401, detail="Authentication failed")
+    async with AsyncClient() as session:
+        login_data = {
+            "email": username,
+            "pass": password
+        }
+        response = await session.post(login_url, data=login_data)
+        if response.status_code == 200:
+            return session
+        else:
+            raise HTTPError("Authentication failed")
 
-def twitch_login(username: str, password: str):
+async def twitch_login(username: str, password: str):
     login_url = "https://www.twitch.tv/login"
-    session = requests.Session()
-    login_data = {
-        "login": username,
-        "password": password
-    }
-    response = session.post(login_url, data=login_data)
-    if response.status_code == 200:
-        return session
-    else:
-        raise HTTPException(status_code=401, detail="Authentication failed")
+    async with AsyncClient() as session:
+        login_data = {
+            "login": username,
+            "password": password
+        }
+        response = await session.post(login_url, data=login_data)
+        if response.status_code == 200:
+            return session
+        else:
+            raise HTTPError("Authentication failed")
 
-def spotify_login(username: str, password: str):
+async def spotify_login(username: str, password: str):
     login_url = "https://accounts.spotify.com/api/token"
-    session = requests.Session()
-    login_data = {
-        "username": username,
-        "password": password
-    }
-    response = session.post(login_url, data=login_data)
-    if response.status_code == 200:
-        return session
-    else:
-        raise HTTPException(status_code=401, detail="Authentication failed")
+    async with AsyncClient() as session:
+        login_data = {
+            "username": username,
+            "password": password
+        }
+        response = await session.post(login_url, data=login_data)
+        if response.status_code == 200:
+            return session
+        else:
+            raise HTTPError("Authentication failed")
 
-def simulate_instagram_view(video_id: str, proxy: Proxy, session: requests.Session):
+def simulate_instagram_view(video_id: str, proxy: Proxy, session: AsyncClient):
     webdriver.DesiredCapabilities.CHROME['proxy'] = {
         "httpProxy": f"{proxy.ip}:{proxy.port}",
         "ftpProxy": f"{proxy.ip}:{proxy.port}",
@@ -238,7 +230,7 @@ def simulate_instagram_view(video_id: str, proxy: Proxy, session: requests.Sessi
         view_url = f"https://www.instagram.com/reel/{video_id}"
 
         if proxy.type == "http" or proxy.type == "https":
-            response = session.get(view_url, headers=headers, proxies={"http": f"http://{proxy.ip}:{proxy.port}", "https": f"https://{proxy.ip}:{proxy.port}"})
+            asyncio.run(session.get(view_url, headers=headers, proxies={"http": f"http://{proxy.ip}:{proxy.port}", "https": f"https://{proxy.ip}:{proxy.port}"})) 
 
         driver.get(view_url)
 
@@ -257,7 +249,7 @@ def simulate_instagram_view(video_id: str, proxy: Proxy, session: requests.Sessi
     finally:
         driver.quit()
 
-def simulate_tiktok_view(video_id: str, proxy: Proxy, session: requests.Session):
+def simulate_tiktok_view(video_id: str, proxy: Proxy, session: AsyncClient):
     webdriver.DesiredCapabilities.CHROME['proxy'] = {
         "httpProxy": f"{proxy.ip}:{proxy.port}",
         "ftpProxy": f"{proxy.ip}:{proxy.port}",
@@ -293,7 +285,7 @@ def simulate_tiktok_view(video_id: str, proxy: Proxy, session: requests.Session)
         view_url = f"https://www.tiktok.com/@user/video/{video_id}"
 
         if proxy.type == "http" or proxy.type == "https":
-            response = session.get(view_url, headers=headers, proxies={"http": f"http://{proxy.ip}:{proxy.port}", "https": f"https://{proxy.ip}:{proxy.port}"})
+            asyncio.run(session.get(view_url, headers=headers, proxies={"http": f"http://{proxy.ip}:{proxy.port}", "https": f"https://{proxy.ip}:{proxy.port}"})) 
 
         driver.get(view_url)
 
@@ -312,7 +304,7 @@ def simulate_tiktok_view(video_id: str, proxy: Proxy, session: requests.Session)
     finally:
         driver.quit()
 
-def simulate_youtube_view(video_id: str, proxy: Proxy, session: requests.Session):
+def simulate_youtube_view(video_id: str, proxy: Proxy, session: AsyncClient):
     webdriver.DesiredCapabilities.CHROME['proxy'] = {
         "httpProxy": f"{proxy.ip}:{proxy.port}",
         "ftpProxy": f"{proxy.ip}:{proxy.port}",
@@ -348,7 +340,7 @@ def simulate_youtube_view(video_id: str, proxy: Proxy, session: requests.Session
         view_url = f"https://www.youtube.com/watch?v={video_id}"
 
         if proxy.type == "http" or proxy.type == "https":
-            response = session.get(view_url, headers=headers, proxies={"http": f"http://{proxy.ip}:{proxy.port}", "https": f"https://{proxy.ip}:{proxy.port}"})
+            asyncio.run(session.get(view_url, headers=headers, proxies={"http": f"http://{proxy.ip}:{proxy.port}", "https": f"https://{proxy.ip}:{proxy.port}"}))
 
         driver.get(view_url)
 
@@ -367,7 +359,7 @@ def simulate_youtube_view(video_id: str, proxy: Proxy, session: requests.Session
     finally:
         driver.quit()
 
-def simulate_facebook_view(video_id: str, proxy: Proxy, session: requests.Session):
+def simulate_facebook_view(video_id: str, proxy: Proxy, session: AsyncClient):
     webdriver.DesiredCapabilities.CHROME['proxy'] = {
         "httpProxy": f"{proxy.ip}:{proxy.port}",
         "ftpProxy": f"{proxy.ip}:{proxy.port}",
@@ -403,7 +395,7 @@ def simulate_facebook_view(video_id: str, proxy: Proxy, session: requests.Sessio
         view_url = f"https://www.facebook.com/{video_id}"
 
         if proxy.type == "http" or proxy.type == "https":
-            response = session.get(view_url, headers=headers, proxies={"http": f"http://{proxy.ip}:{proxy.port}", "https": f"https://{proxy.ip}:{proxy.port}"})
+            asyncio.run(session.get(view_url, headers=headers, proxies={"http": f"http://{proxy.ip}:{proxy.port}", "https": f"https://{proxy.ip}:{proxy.port}"}))
 
         driver.get(view_url)
 
@@ -422,7 +414,7 @@ def simulate_facebook_view(video_id: str, proxy: Proxy, session: requests.Sessio
     finally:
         driver.quit()
 
-def simulate_twitch_view(video_id: str, proxy: Proxy, session: requests.Session):
+def simulate_twitch_view(video_id: str, proxy: Proxy, session: AsyncClient):
     webdriver.DesiredCapabilities.CHROME['proxy'] = {
         "httpProxy": f"{proxy.ip}:{proxy.port}",
         "ftpProxy": f"{proxy.ip}:{proxy.port}",
@@ -458,7 +450,7 @@ def simulate_twitch_view(video_id: str, proxy: Proxy, session: requests.Session)
         view_url = f"https://www.twitch.tv/videos/{video_id}"
 
         if proxy.type == "http" or proxy.type == "https":
-            response = session.get(view_url, headers=headers, proxies={"http": f"http://{proxy.ip}:{proxy.port}", "https": f"https://{proxy.ip}:{proxy.port}"})
+            asyncio.run(session.get(view_url, headers=headers, proxies={"http": f"http://{proxy.ip}:{proxy.port}", "https": f"https://{proxy.ip}:{proxy.port}"}))
 
         driver.get(view_url)
 
@@ -477,7 +469,7 @@ def simulate_twitch_view(video_id: str, proxy: Proxy, session: requests.Session)
     finally:
         driver.quit()
 
-def simulate_spotify_view(video_id: str, proxy: Proxy, session: requests.Session):
+def simulate_spotify_view(video_id: str, proxy: Proxy, session: AsyncClient):
     webdriver.DesiredCapabilities.CHROME['proxy'] = {
         "httpProxy": f"{proxy.ip}:{proxy.port}",
         "ftpProxy": f"{proxy.ip}:{proxy.port}",
@@ -513,7 +505,7 @@ def simulate_spotify_view(video_id: str, proxy: Proxy, session: requests.Session
         view_url = f"https://open.spotify.com/track/{video_id}"
 
         if proxy.type == "http" or proxy.type == "https":
-            response = session.get(view_url, headers=headers, proxies={"http": f"http://{proxy.ip}:{proxy.port}", "https": f"https://{proxy.ip}:{proxy.port}"})
+            asyncio.run(session.get(view_url, headers=headers, proxies={"http": f"http://{proxy.ip}:{proxy.port}", "https": f"https://{proxy.ip}:{proxy.port}"}))
 
         driver.get(view_url)
 
@@ -532,12 +524,11 @@ def simulate_spotify_view(video_id: str, proxy: Proxy, session: requests.Session
     finally:
         driver.quit()
 
-
-def simulate_views_background(url: str, platform: str, count: int, delay: int, parallel_processes: int, session: requests.Session = None):
+def simulate_views_background(url: str, platform: str, count: int, delay: int, parallel_processes: int, session: AsyncClient = None):
     for _ in range(parallel_processes):
         asyncio.run(simulate_one_view_process(url, platform, count, delay, session))
 
-async def simulate_one_view_process(url: str, platform: str, count: int, delay: int, session: requests.Session = None):
+async def simulate_one_view_process(url: str, platform: str, count: int, delay: int, session: AsyncClient = None):
     video_id = extract_video_id(url, platform)
     proxy = get_random_proxy()
     if not proxy:
@@ -561,176 +552,95 @@ async def simulate_one_view_process(url: str, platform: str, count: int, delay:
         except Exception as e:
             pass
 
-@app.post("/simulate_views")
-async def simulate_views_endpoint(request: VisitRequest, background_tasks: BackgroundTasks):
+async def simulate_views_endpoint(request: VisitRequest):
     try:
         session = None
         if os.getenv(f'{request.platform.upper()}_USER') and os.getenv(f'{request.platform.upper()}_PASSWORD'):
-            session = authenticate(os.getenv(f'{request.platform.upper()}_USER'), os.getenv(f'{request.platform.upper()}_PASSWORD'), request.platform)
-
-        background_tasks.add_task(
-            simulate_views_background,
-            url=request.url,
-            platform=request.platform,
-            count=request.count,
-            delay=request.delay,
-            parallel_processes=request.parallel_processes,
-            session=session
+            session = await authenticate(os.getenv(f'{request.platform.upper()}_USER'), os.getenv(f'{request.platform.upper()}_PASSWORD'), request.platform)
+
+        asyncio.create_task(
+            simulate_views_background(
+                url=request.url,
+                platform=request.platform,
+                count=request.count,
+                delay=request.delay,
+                parallel_processes=request.parallel_processes,
+                session=session
+            )
         )
 
-        return PlainTextResponse("Views simulation started in the background.")
+        return "Views simulation started in the background."
 
     except Exception as e:
-        return PlainTextResponse(str(e), status_code=500)
+        return str(e)
 
-def authenticate(username: str, password: str, platform: str):
+async def authenticate(username: str, password: str, platform: str):
     if platform == "instagram":
-        return instagram_login(username, password)
+        return await instagram_login(username, password)
     elif platform == "tiktok":
-        return tiktok_login(username, password)
+        return await tiktok_login(username, password)
     elif platform == "youtube":
-        return youtube_login(username, password)
+        return await youtube_login(username, password)
     elif platform == "facebook":
-        return facebook_login(username, password)
+        return await facebook_login(username, password)
     elif platform == "twitch":
-        return twitch_login(username, password)
+        return await twitch_login(username, password)
     elif platform == "spotify":
-        return spotify_login(username, password)
+        return await spotify_login(username, password)
     else:
-        raise HTTPException(status_code=400, detail="Invalid platform")
+        raise ValueError("Invalid platform")
 
-@app.post("/login")
-async def login(username: str = Form(...), password: str = Form(...), platform: str = Form(...)):
+async def login(username: str, password: str, platform: str):
     try:
-        session = authenticate(username, password, platform)
-        return PlainTextResponse(f"Authenticated on {platform}")
-    except HTTPException as e:
-        return e
+        session = await authenticate(username, password, platform)
+        return f"Authenticated on {platform}"
+    except Exception as e:
+        return str(e)
 
-@app.post("/simulate")
-async def simulate(background_tasks: BackgroundTasks, urls: str = Form(...), platform: str = Form(...), count: int = Form(...), delay: int = Form(...), parallel_processes: int = Form(...)):
+async def simulate(urls: str, platform: str, count: int, delay: int, parallel_processes: int):
     try:
         session = None
         if os.getenv(f'{platform.upper()}_USER') and os.getenv(f'{platform.upper()}_PASSWORD'):
-            session = authenticate(os.getenv(f'{platform.upper()}_USER'), os.getenv(f'{platform.upper()}_PASSWORD'), platform)
+            session = await authenticate(os.getenv(f'{platform.upper()}_USER'), os.getenv(f'{platform.upper()}_PASSWORD'), platform)
 
         for index, url in enumerate(urls.split("\n")):
-            background_tasks.add_task(
-                simulate_views_background,
-                url=url,
-                platform=platform,
-                count=count,
-                delay=delay,
-                parallel_processes=parallel_processes,
-                session=session
+            asyncio.create_task(
+                simulate_views_background(
+                    url=url,
+                    platform=platform,
+                    count=count,
+                    delay=delay,
+                    parallel_processes=parallel_processes,
+                    session=session
+                )
             )
-        return PlainTextResponse("Simulations started in the background.")
+        return "Simulations started in the background."
 
     except Exception as e:
-        return PlainTextResponse(str(e), status_code=500)
-
-@app.get("/")
-async def main_page():
-    return HTMLResponse(
-        """
-        <!DOCTYPE html>
-        <html>
-        <head>
-            <title>Video Views Simulator</title>
-            <style>
-                body {
-                    background-color: #f0f0f0;
-                    font-family: sans-serif;
-                    display: flex;
-                    justify-content: center;
-                    align-items: center;
-                    min-height: 100vh;
-                }
-                .container {
-                    background-color: #fff;
-                    padding: 30px;
-                    border-radius: 10px;
-                    box-shadow: 0 2px 5px rgba(0, 0, 0, 0.1);
-                    text-align: center;
-                    animation: fadeIn 1s ease-in-out;
-                }
-                h1 {
-                    color: #333;
-                    margin-bottom: 20px;
-                    text-shadow: 0 1px 2px rgba(0, 0, 0, 0.1);
-                }
-                form {
-                    display: flex;
-                    flex-direction: column;
-                    width: 350px;
-                }
-                label {
-                    margin-bottom: 5px;
-                    font-weight: bold;
-                }
-                input[type="text"],
-                textarea,
-                select,
-                input[type="number"] {
-                    padding: 10px;
-                    margin-bottom: 15px;
-                    border: 1px solid #ddd;
-                    border-radius: 5px;
-                }
-                input[type="submit"] {
-                    background-color: #007bff;
-                    color: #fff;
-                    border: none;
-                    padding: 12px 20px;
-                    border-radius: 5px;
-                    cursor: pointer;
-                    transition: background-color 0.3s;
-                }
-                input[type="submit"]:hover {
-                    background-color: #0056b3;
-                }
-                @keyframes fadeIn {
-                    from {
-                        opacity: 0;
-                    }
-                    to {
-                        opacity: 1;
-                    }
-                }
-            </style>
-        </head>
-        <body>
-            <div class="container">
-                <h1>Video Views Simulator</h1>
-                <form action="/simulate" method="post">
-                    <label for="urls">Video URLs (one per line):</label><br>
-                    <textarea id="urls" name="urls"></textarea><br><br>
-                    <label for="platform">Platform:</label><br>
-                    <select id="platform" name="platform">
-                        <option value="instagram">Instagram</option>
-                        <option value="tiktok">TikTok</option>
-                        <option value="youtube">YouTube</option>
-                        <option value="facebook">Facebook</option>
-                        <option value="twitch">Twitch</option>
-                        <option value="spotify">Spotify</option>
-                    </select><br><br>
-                    <label for="count">Number of Views:</label><br>
-                    <input type="number" id="count" name="count" value="1"><br><br>
-                    <label for="delay">Delay (seconds):</label><br>
-                    <input type="number" id="delay" name="delay" value="1"><br><br>
-                    <label for="parallel_processes">Parallel Processes:</label><br>
-                    <input type="number" id="parallel_processes" name="parallel_processes" value="1"><br><br>
-                    <input type="submit" value="Simulate">
-                </form>
-            </div>
-        </body>
-        </html>
-        """
-    )
+        return str(e)
 
 def rand(min, max):
     return random.randint(min, max)
 
-if __name__ == "__main__":
-    import uvicorn
-    uvicorn.run(app, host="0.0.0.0", port=7860)
+with gr.Blocks() as interface:
+    gr.Markdown("<h1>Video Views Simulator</h1>")
+    with gr.Row():
+        urls = gr.Textbox(label="Video URLs (one per line)")
+        platform = gr.Dropdown(
+            ["instagram", "tiktok", "youtube", "facebook", "twitch", "spotify"],
+            label="Platform",
+        )
+    with gr.Row():
+        count = gr.Number(label="Number of Views", value=1)
+        delay = gr.Number(label="Delay (seconds)", value=1)
+        parallel_processes = gr.Number(label="Parallel Processes", value=1)
+    simulate_btn = gr.Button("Simulate")
+    output = gr.Textbox(label="Output")
+
+    simulate_btn.click(
+        fn=simulate,
+        inputs=[urls, platform, count, delay, parallel_processes],
+        outputs=output,
+    )
+
+interface.launch(server_name="0.0.0.0", server_port=7860)