Update app.py

app.py

@@ -29,30 +29,22 @@ class Proxy(BaseModel):
 
 class VisitRequest(BaseModel):
     url: str
-    platform: str
     count: int
     delay: int
     parallel_processes: int
 
 def get_random_proxy():
     try:
-        options = webdriver.ChromeOptions()
-        options.add_argument('--headless')
-        options.add_argument('--disable-gpu')
-        options.add_argument('--no-sandbox')
-        driver = webdriver.Chrome(ChromeDriverManager().install(), options=options)
-        driver.get("https://uhhy-fsfsfs.hf.space/valid")
-        WebDriverWait(driver, 10).until(
-            EC.presence_of_element_located((By.CSS_SELECTOR, "table tbody tr"))
-        )
-        proxy_elements = driver.find_elements(By.CSS_SELECTOR, "table tbody tr")
+        response = requests.get("https://uhhy-fsfsfs.hf.space/valid")
+        soup = BeautifulSoup(response.text, 'html.parser')
+        proxy_rows = soup.find_all('tr')[1:]
         proxies = []
-        for element in proxy_elements:
-            ip = element.find_element(By.CSS_SELECTOR, "td:nth-child(1)").text
-            port = element.find_element(By.CSS_SELECTOR, "td:nth-child(2)").text
-            proxy_type = element.find_element(By.CSS_SELECTOR, "td:nth-child(3)").text.lower()
+        for row in proxy_rows:
+            columns = row.find_all('td')
+            ip = columns[0].text
+            port = columns[1].text
+            proxy_type = columns[2].text.lower()
             proxies.append(Proxy(ip=ip, port=port, type=proxy_type))
-        driver.quit()
         if proxies:
             return random.choice(proxies)
         else:
@@ -107,7 +99,7 @@ def extract_video_id(url: str, platform: str) -> str:
     if match:
         return match.group(1)
     else:
-        raise ValueError("Invalid URL format")
+        return None
 
 def instagram_login(username: str, password: str):
     login_url = "https://www.instagram.com/accounts/login/ajax/"
@@ -195,88 +187,30 @@ def spotify_login(username: str, password: str):
     else:
         raise HTTPError("Authentication failed")
 
-def simulate_instagram_view(video_id: str, proxy: Proxy, session: Client):
-    try:
-        fake = Faker()
-        fake_ipv4 = fake.ipv4()
-        headers = {
-            "User-Agent": fake.user_agent(),
-            "X-Forwarded-For": fake_ipv4,
-            "Client-IP": fake_ipv4,
-            "X-Real-IP": fake_ipv4
-        }
-        view_url = f"https://www.instagram.com/reel/{video_id}"
-
-        if proxy.type == "http" or proxy.type == "https":
-            session.get(view_url, headers=headers, proxies={"http": f"http://{proxy.ip}:{proxy.port}", "https": f"https://{proxy.ip}:{proxy.port}"})
-        time.sleep(rand(10, 20))
-        return True 
-    except Exception as e:
-        print(f"Error in simulate_instagram_view: {e}")
-        return False
-
-def simulate_tiktok_view(video_id: str, proxy: Proxy, session: Client):
-    try:
-        fake = Faker()
-        fake_ipv4 = fake.ipv4()
-        headers = {
-            "User-Agent": fake.user_agent(),
-            "X-Forwarded-For": fake_ipv4,
-            "Client-IP": fake_ipv4,
-            "X-Real-IP": fake_ipv4
-        }
-        view_url = f"https://www.tiktok.com/@user/video/{video_id}"
-
-        if proxy.type == "http" or proxy.type == "https":
-            session.get(view_url, headers=headers, proxies={"http": f"http://{proxy.ip}:{proxy.port}", "https": f"https://{proxy.ip}:{proxy.port}"})
-
-        time.sleep(rand(10, 20))
-        return True 
-    except Exception as e:
-        print(f"Error in simulate_tiktok_view: {e}")
-        return False 
-
-def simulate_youtube_view(video_id: str, proxy: Proxy, session: Client):
-    try:
-        fake = Faker()
-        fake_ipv4 = fake.ipv4()
-        headers = {
-            "User-Agent": fake.user_agent(),
-            "X-Forwarded-For": fake_ipv4,
-            "Client-IP": fake_ipv4,
-            "X-Real-IP": fake_ipv4
-        }
-        view_url = f"https://www.youtube.com/watch?v={video_id}"
-
-        if proxy.type == "http" or proxy.type == "https":
-            session.get(view_url, headers=headers, proxies={"http": f"http://{proxy.ip}:{proxy.port}", "https": f"https://{proxy.ip}:{proxy.port}"})
-        time.sleep(rand(10, 20))
-        return True
-    except Exception as e:
-        print(f"Error in simulate_youtube_view: {e}")
-        return False 
-
-def simulate_facebook_view(video_id: str, proxy: Proxy, session: Client):
-    try:
-        fake = Faker()
-        fake_ipv4 = fake.ipv4()
-        headers = {
-            "User-Agent": fake.user_agent(),
-            "X-Forwarded-For": fake_ipv4,
-            "Client-IP": fake_ipv4,
-            "X-Real-IP": fake_ipv4
-        }
-        view_url = f"https://www.facebook.com/{video_id}"
-
-        if proxy.type == "http" or proxy.type == "https":
-            session.get(view_url, headers=headers, proxies={"http": f"http://{proxy.ip}:{proxy.port}", "https": f"https://{proxy.ip}:{proxy.port}"})
-        time.sleep(rand(10, 20))
-        return True
-    except Exception as e:
-        print(f"Error in simulate_facebook_view: {e}")
-        return False
+def simulate_view(url: str, proxy: Proxy, session: Client, delay: int):
+    webdriver.DesiredCapabilities.CHROME['proxy'] = {
+        "httpProxy": f"{proxy.ip}:{proxy.port}",
+        "ftpProxy": f"{proxy.ip}:{proxy.port}",
+        "sslProxy": f"{proxy.ip}:{proxy.port}",
+        "proxyType": "MANUAL",
+    }
+    options = webdriver.ChromeOptions()
+    options.add_argument("--headless")
+    options.add_argument("--disable-blink-features=AutomationControlled")
+    options.add_argument("--disable-popup-blocking")
+    options.add_argument("--disable-infobars")
+    options.add_argument("--disable-web-security")
+    options.add_argument("--ignore-certificate-errors")
+    options.add_argument("--disable-notifications")
+    options.add_argument("--disable-extensions")
+    options.add_argument("--disable-gpu")
+    options.add_argument("--no-sandbox")
+    options.add_argument("--disable-dev-shm-usage")
+    options.add_argument("--mute-audio")
+    options.add_argument('--ignore-ssl-errors=yes')
+    options.add_argument('--ignore-certificate-errors')
+    driver = webdriver.Chrome(ChromeDriverManager().install(), options=options)
 
-def simulate_twitch_view(video_id: str, proxy: Proxy, session: Client):
     try:
         fake = Faker()
         fake_ipv4 = fake.ipv4()
@@ -286,39 +220,30 @@ def simulate_twitch_view(video_id: str, proxy: Proxy, session: Client):
             "Client-IP": fake_ipv4,
             "X-Real-IP": fake_ipv4
         }
-        view_url = f"https://www.twitch.tv/videos/{video_id}"
 
         if proxy.type == "http" or proxy.type == "https":
-            session.get(view_url, headers=headers, proxies={"http": f"http://{proxy.ip}:{proxy.port}", "https": f"https://{proxy.ip}:{proxy.port}"})
-        time.sleep(rand(10, 20))
-        return True
-    except Exception as e:
-        print(f"Error in simulate_twitch_view: {e}")
-        return False
+            session.get(url, headers=headers, proxies={"http": f"http://{proxy.ip}:{proxy.port}", "https": f"https://{proxy.ip}:{proxy.port}"})
 
-def simulate_spotify_view(video_id: str, proxy: Proxy, session: Client):
-    try:
-        fake = Faker()
-        fake_ipv4 = fake.ipv4()
-        headers = {
-            "User-Agent": fake.user_agent(),
-            "X-Forwarded-For": fake_ipv4,
-            "Client-IP": fake_ipv4,
-            "X-Real-IP": fake_ipv4
-        }
-        view_url = f"https://open.spotify.com/track/{video_id}"
+        driver.get(url)
 
-        if proxy.type == "http" or proxy.type == "https":
-            session.get(view_url, headers=headers, proxies={"http": f"http://{proxy.ip}:{proxy.port}", "https": f"https://{proxy.ip}:{proxy.port}"})
+        try:
+            not_now = WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.CLASS_NAME, 'mt3GC')))
+            a = not_now.find_elements(By.TAG_NAME, "button")[1]
+            actions = ActionChains(driver)
+            actions.click(a)
+            actions.perform()
+        except:
+            pass
 
-        time.sleep(rand(10, 20))
+        time.sleep(delay)
         return True
     except Exception as e:
-        print(f"Error in simulate_spotify_view: {e}")
+        print(f"Error simulating view: {e}")
         return False
+    finally:
+        driver.quit()
 
-def simulate_views_background(url: str, platform: str, count: int, delay: int, session: Client = None):
-    video_id = extract_video_id(url, platform)
+def simulate_views_background(url: str, count: int, delay: int, session: Client = None):
     proxy = get_random_proxy()
     if not proxy:
         print("No proxy available.")
@@ -329,18 +254,7 @@ def simulate_views_background(url: str, platform: str, count: int, delay: int, s
     with tqdm(total=count, desc=f"Simulating views for {url}") as pbar:
         for i in range(count):
             try:
-                if platform == "instagram":
-                    success = simulate_instagram_view(video_id, proxy, session)
-                elif platform == "tiktok":
-                    success = simulate_tiktok_view(video_id, proxy, session)
-                elif platform == "youtube":
-                    success = simulate_youtube_view(video_id, proxy, session)
-                elif platform == "facebook":
-                    success = simulate_facebook_view(video_id, proxy, session)
-                elif platform == "twitch":
-                    success = simulate_twitch_view(video_id, proxy, session)
-                elif platform == "spotify":
-                    success = simulate_spotify_view(video_id, proxy, session)
+                success = simulate_view(url, proxy, session, delay)
 
                 if success:
                     successful_views += 1
@@ -350,7 +264,6 @@ def simulate_views_background(url: str, platform: str, count: int, delay: int, s
                 pbar.update(1)
                 remaining_views = count - i - 1
                 pbar.set_postfix({"Successful": successful_views, "Failed": failed_views, "Remaining": remaining_views, "Time Remaining": pbar.format_interval(pbar.remaining_time)})
-                time.sleep(delay)
             except Exception as e:
                 failed_views += 1
                 pbar.update(1)
@@ -360,14 +273,10 @@ def simulate_views_background(url: str, platform: str, count: int, delay: int, s
 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)
-
         threading.Thread(
             target=simulate_views_background,
             args=(
                 request.url,
-                request.platform,
                 request.count,
                 request.delay,
                 session,
@@ -379,41 +288,15 @@ def simulate_views_endpoint(request: VisitRequest):
     except Exception as e:
         return str(e)
 
-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 ValueError("Invalid platform")
-
-def login(username: str, password: str, platform: str):
-    try:
-        session = authenticate(username, password, platform)
-        return f"Authenticated on {platform}"
-    except Exception as e:
-        return str(e)
-
-def simulate(urls: str, platform: str, count: int, delay: int, parallel_processes: int):
+def simulate(urls: 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)
         for url in urls.split("\n"):
             for _ in range(parallel_processes):
                 threading.Thread(
                     target=simulate_views_background,
                     args=(
                         url,
-                        platform,
                         count,
                         delay,
                         session
@@ -484,21 +367,13 @@ with gr.Blocks(css="""
             }
         }
     """) as interface:
-    gr.Markdown("""
+    with gr.Row():
+        gr.HTML("""
         <div class="container">
-            <h1>Video Views Simulator</h1>
+            <h1>Website Traffic Simulator</h1>
             <form action="/simulate" method="post">
-                <label for="urls">Video URLs (one per line):</label><br>
+                <label for="urls">Website URLs or IP addresses (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>
@@ -510,11 +385,7 @@ with gr.Blocks(css="""
         </div>
     """)
     with gr.Row():
-        urls = gr.Textbox(label="Video URLs (one per line)")
-        platform = gr.Dropdown(
-            ["instagram", "tiktok", "youtube", "facebook", "twitch", "spotify"],
-            label="Platform",
-        )
+        urls = gr.Textbox(label="Website URLs or IP addresses (one per line)")
     with gr.Row():
         count = gr.Number(label="Number of Views", value=1)
         delay = gr.Number(label="Delay (seconds)", value=1)
@@ -524,7 +395,7 @@ with gr.Blocks(css="""
 
     simulate_btn.click(
         fn=simulate,
-        inputs=[urls, platform, count, delay, parallel_processes],
+        inputs=[urls, count, delay, parallel_processes],
         outputs=output,
     )