Create app.py

app.py

@@ -0,0 +1,468 @@
+import requests
+from bs4 import BeautifulSoup
+from fastapi import FastAPI, BackgroundTasks, HTTPException, Form
+from fastapi.responses import PlainTextResponse, HTMLResponse
+from fastapi.staticfiles import StaticFiles
+from pydantic import BaseModel
+import re
+from concurrent.futures import ThreadPoolExecutor
+import urllib.parse
+import gradio as gr
+import fake_useragent as fake
+import random
+from dotenv import load_dotenv
+import os
+
+load_dotenv()
+
+app = FastAPI()
+
+class Proxy(BaseModel):
+    ip: str
+    port: str
+
+class VisitRequest(BaseModel):
+    url: str
+    platform: str
+    count: int
+    delay: int
+    parallel_processes: int
+
+app.mount("/static", StaticFiles(directory="static"), name="static")
+
+def get_random_proxy():
+    pubproxy_url = "http://pubproxy.com/api/proxy?limit=1"
+    response = requests.get(pubproxy_url, verify=False)
+    data = response.json()
+    if data['data']:
+        proxy_data = data['data'][0]
+        return f"{proxy_data['ip']}:{proxy_data['port']}"
+    return None
+
+@app.get("/get_proxies", response_class=PlainTextResponse)
+def get_proxies():
+    try:
+        proxies = []
+        pubproxy_url = "http://pubproxy.com/api/proxy?limit=5"
+        response = requests.get(pubproxy_url, verify=False)
+        data = response.json()
+
+        for proxy_data in data['data']:
+            ip, port = proxy_data['ipPort'].split(":")
+            proxy = f"{ip}:{port}"
+            proxies.append(proxy)
+        
+        return "\n".join(proxies)
+    except Exception as e:
+        return PlainTextResponse(str(e), status_code=500)
+
+@app.get("/rotate_ip", response_class=PlainTextResponse)
+def rotate_ip():
+    try:
+        random_ip = fake.ipv4()
+
+        headers = {
+            "X-Forwarded-For": random_ip,
+            "Client-IP": random_ip,
+            "X-Real-IP": random_ip
+        }
+
+        test_url = "http://pubproxy.com/api/proxy?limit=1&type=http&https=true&speed=60"
+        response = requests.get(test_url, headers=headers, verify=False)
+        data = response.json()
+
+        proxies = []
+        for proxy_data in data['data']:
+            ip, port = proxy_data['ipPort'].split(":")
+            proxy = f"{ip}:{port}"
+            proxies.append(proxy)
+
+        return "\n".join(proxies)
+    except Exception as e:
+        return PlainTextResponse(str(e), status_code=500)
+
+def extract_video_id(url: str, platform: str) -> str:
+    url = urllib.parse.unquote(url)
+    if platform == "instagram":
+        match = re.search(r"instagram\.com/reel/([^/?]+)", url)
+    elif platform == "tiktok":
+        match = re.search(r"tiktok\.com/@[^/]+/video/(\d+)", url)
+    elif platform == "youtube":
+        match = re.search(r"youtube\.com/watch\?v=([^&]+)", url)
+    elif platform == "facebook":
+        match = re.search(r"facebook\.com/.*/videos/(\d+)", url)
+    elif platform == "twitch":
+        match = re.search(r"twitch\.tv/videos/(\d+)", url)
+    elif platform == "spotify":
+        match = re.search(r"spotify\.com/track/([^/?]+)", url)
+    else:
+        match = None
+
+    if match:
+        return match.group(1)
+    else:
+        raise ValueError("Invalid URL format")
+
+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")
+
+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")
+
+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")
+
+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")
+
+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")
+
+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")
+
+def simulate_instagram_view(video_id: str, proxy: str, session: requests.Session):
+    proxies = {
+        "http": f"http://{proxy}",
+        "https": f"http://{proxy}"
+    }
+    fake_ipv4 = fake.ipv4()
+    headers = {
+        "User-Agent": fake.user_agent(),
+        "X-Forwarded-For": fake_ipv4,
+        "Client-IP": fake_ipv4,
+        "X-Real-IP": fake_ipv4
+    }
+    try:
+        view_url = f"https://www.instagram.com/reel/{video_id}"
+        response = session.get(view_url, headers=headers, proxies=proxies)
+        embed_link = f"https://www.instagram.com/p/{video_id}/embed/"
+        embed_link_with_start = f"{embed_link}?start={rand(4, 8)}"
+        print(embed_link_with_start)
+    except Exception as e:
+        print(f"Error in simulate_instagram_view: {e}")
+
+def simulate_tiktok_view(video_id: str, proxy: str, session: requests.Session):
+    proxies = {
+        "http": f"http://{proxy}",
+        "https": f"http://{proxy}"
+    }
+    fake_ipv4 = fake.ipv4()
+    headers = {
+        "User-Agent": fake.user_agent(),
+        "X-Forwarded-For": fake_ipv4,
+        "Client-IP": fake_ipv4,
+        "X-Real-IP": fake_ipv4
+    }
+    try:
+        view_url = f"https://www.tiktok.com/@user/video/{video_id}"
+        response = session.get(view_url, headers=headers, proxies=proxies)
+        embed_link = f"https://www.tiktok.com/embed/v/{video_id}"
+        embed_link_with_start = f"{embed_link}?start={rand(4, 8)}"
+        print(embed_link_with_start)
+    except Exception as e:
+        print(f"Error in simulate_tiktok_view: {e}")
+
+def simulate_youtube_view(video_id: str, proxy: str, session: requests.Session):
+    proxies = {
+        "http": f"http://{proxy}",
+        "https": f"http://{proxy}"
+    }
+    fake_ipv4 = fake.ipv4()
+    headers = {
+        "User-Agent": fake.user_agent(),
+        "X-Forwarded-For": fake_ipv4,
+        "Client-IP": fake_ipv4,
+        "X-Real-IP": fake_ipv4
+    }
+    try:
+        view_url = f"https://www.youtube.com/watch?v={video_id}"
+        response = session.get(view_url, headers=headers, proxies=proxies)
+        embed_link = f"https://www.youtube.com/embed/{video_id}"
+        embed_link_with_start = f"{embed_link}?controls=0&start={rand(4, 8)}"
+        print(embed_link_with_start)
+    except Exception as e:
+        print(f"Error in simulate_youtube_view: {e}")
+
+def simulate_facebook_view(video_id: str, proxy: str, session: requests.Session):
+    proxies = {
+        "http": f"http://{proxy}",
+        "https": f"http://{proxy}"
+    }
+    fake_ipv4 = fake.ipv4()
+    headers = {
+        "User-Agent": fake.user_agent(),
+        "X-Forwarded-For": fake_ipv4,
+        "Client-IP": fake_ipv4,
+        "X-Real-IP": fake_ipv4
+    }
+    try:
+        view_url = f"https://www.facebook.com/{video_id}"
+        response = session.get(view_url, headers=headers, proxies=proxies)
+        embed_link = f"https://www.facebook.com/plugins/video.php?href=https://www.facebook.com/{video_id}/&show_text=0&width=560&height=315"
+        embed_link_with_start = f"{embed_link}&start={rand(4, 8)}"
+        print(embed_link_with_start)
+    except Exception as e:
+        print(f"Error in simulate_facebook_view: {e}")
+
+def simulate_twitch_view(video_id: str, proxy: str, session: requests.Session):
+    proxies = {
+        "http": f"http://{proxy}",
+        "https": f"http://{proxy}"
+    }
+    fake_ipv4 = fake.ipv4()
+    headers = {
+        "User-Agent": fake.user_agent(),
+        "X-Forwarded-For": fake_ipv4,
+        "Client-IP": fake_ipv4,
+        "X-Real-IP": fake_ipv4
+    }
+    try:
+        view_url = f"https://www.twitch.tv/videos/{video_id}"
+        response = session.get(view_url, headers=headers, proxies=proxies)
+        embed_link = f"https://player.twitch.tv/?video=v{video_id}&parent=www.example.com&autoplay=false&muted=false"
+        embed_link_with_start = f"{embed_link}&start={rand(4, 8)}"
+        print(embed_link_with_start)
+    except Exception as e:
+        print(f"Error in simulate_twitch_view: {e}")
+
+def simulate_spotify_view(video_id: str, proxy: str, session: requests.Session):
+    proxies = {
+        "http": f"http://{proxy}",
+        "https": f"http://{proxy}"
+    }
+    fake_ipv4 = fake.ipv4()
+    headers = {
+        "User-Agent": fake.user_agent(),
+        "X-Forwarded-For": fake_ipv4,
+        "Client-IP": fake_ipv4,
+        "X-Real-IP": fake_ipv4
+    }
+    try:
+        view_url = f"https://open.spotify.com/track/{video_id}"
+        response = session.get(view_url, headers=headers, proxies=proxies)
+        embed_link = f"https://open.spotify.com/embed/track/{video_id}"
+        embed_link_with_start = f"{embed_link}?start={rand(4, 8)}"
+        print(embed_link_with_start)
+    except Exception as e:
+        print(f"Error in simulate_spotify_view: {e}")
+
+def simulate_views(url: str, platform: str, count: int, delay: int, parallel_processes: int, session: requests.Session):
+    video_id = extract_video_id(url, platform)
+    proxy = get_random_proxy()
+    
+    if not proxy:
+        print("No proxy available.")
+        return
+
+    with ThreadPoolExecutor(max_workers=parallel_processes) as executor:
+        for _ in range(count):
+            if platform == "instagram":
+                executor.submit(simulate_instagram_view, video_id, proxy, session)
+            elif platform == "tiktok":
+                executor.submit(simulate_tiktok_view, video_id, proxy, session)
+            elif platform == "youtube":
+                executor.submit(simulate_youtube_view, video_id, proxy, session)
+            elif platform == "facebook":
+                executor.submit(simulate_facebook_view, video_id, proxy, session)
+            elif platform == "twitch":
+                executor.submit(simulate_twitch_view, video_id, proxy, session)
+            elif platform == "spotify":
+                executor.submit(simulate_spotify_view, video_id, proxy, session)
+
+@app.post("/simulate_views")
+async def simulate_views_endpoint(request: VisitRequest):
+    try:
+        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)
+            simulate_views(
+                url=request.url,
+                platform=request.platform,
+                count=request.count,
+                delay=request.delay,
+                parallel_processes=request.parallel_processes,
+                session=session
+            )
+            return PlainTextResponse("Views simulation started")
+        else:
+            return PlainTextResponse("Credentials not found for this platform.")
+    except Exception as e:
+        return PlainTextResponse(str(e), status_code=500)
+
+def authenticate(username: str, password: str, platform: str):
+    if platform == "instagram":
+        return instagram_login(username, password)
+    elif platform == "tiktok":
+        return tiktok_login(username, password)
+    elif platform == "youtube":
+        return youtube_login(username, password)
+    elif platform == "facebook":
+        return facebook_login(username, password)
+    elif platform == "twitch":
+        return twitch_login(username, password)
+    elif platform == "spotify":
+        return spotify_login(username, password)
+    else:
+        raise HTTPException(status_code=400, detail="Invalid platform")
+
+@app.post("/login")
+async def login(username: str = Form(...), password: str = Form(...), platform: str = Form(...)):
+    try:
+        session = authenticate(username, password, platform)
+        return PlainTextResponse(f"Authenticated on {platform}")
+    except HTTPException as e:
+        return e
+
+def gradio_interface(file, platform, count, delay, parallel_processes):
+    try:
+        # Suponiendo que el archivo es un CSV con columnas: URL, Count, Delay, Parallel Processes
+        import pandas as pd
+        df = pd.read_csv(file)
+
+        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)
+            for index, row in df.iterrows():
+                try:
+                    url = row['URL']
+                    simulate_views(url, platform, count, delay, parallel_processes, session)
+                    print(f"Simulation started for row {index}")
+                except Exception as e:
+                    print(f"Error processing row {index}: {e}")
+            
+            return "Simulation completed."
+        else:
+            return "Credentials not found for this platform."
+    except Exception as e:
+        return f"Error: {e}"
+
+
+def rand(min, max):
+    return random.randint(min, max)
+
+def run_simulation(file, platform, count, delay, parallel_processes):
+    try:
+        # Suponiendo que el archivo es un CSV con columnas: URL, Count, Delay, Parallel Processes
+        import pandas as pd
+        df = pd.read_csv(file)
+
+        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)
+
+            # Create a thread pool executor with maximum workers based on parallel_processes
+            with ThreadPoolExecutor(max_workers=parallel_processes) as executor:
+                for index, row in df.iterrows():
+                    try:
+                        url = row['URL']
+
+                        # Execute the simulate_views function in a separate thread
+                        executor.submit(simulate_views, url, platform, count, delay, parallel_processes, session)
+
+                        print(f"Simulation started for row {index}")
+                    except Exception as e:
+                        print(f"Error processing row {index}: {e}")
+            
+            return "Simulation completed."
+        else:
+            return "Credentials not found for this platform."
+    except Exception as e:
+        return f"Error: {e}"
+
+gr.Interface(
+    fn=run_simulation, 
+    inputs=[
+        gr.File(label="Upload CSV File"),
+        gr.Dropdown(choices=["instagram", "tiktok", "youtube", "facebook", "twitch", "spotify"], label="Platform"),
+        gr.Slider(minimum=1, maximum=1000, label="Count"),
+        gr.Slider(minimum=1, maximum=60, label="Delay (seconds)"),
+        gr.Slider(minimum=1, maximum=10, label="Parallel Processes")
+    ], 
+    outputs=gr.Textbox(label="Output")
+).launch(share=True)