Update app.py

app.py

@@ -6,11 +6,45 @@ from selenium.webdriver.common.by import By
 from selenium.webdriver.support import expected_conditions as EC
 from selenium.webdriver.common.action_chains import ActionChains
 import gradio as gr
-import random
-from selenium.common.exceptions import TimeoutException, NoSuchElementException, WebDriverException
+import re
+import requests
+from bs4 import BeautifulSoup
+from faker import Faker
+from webdriver_manager.chrome import ChromeDriverManager
+import multiprocessing
+import queue
+
+fake = Faker()
+
+PROXY_IP = '47.236.149.233'
+PROXY_PORT = 9090
+TEMP_MAIL_WEBSITE_URL = "https://temp-mail.org/es/"
+FREE_SMS_WEBSITE_URL = "https://smsreceivefree.net/"
+FREE_SMS_GET_NUMBER_URL = f"{FREE_SMS_WEBSITE_URL}"
+FREE_SMS_INBOX_URL_BASE = f"{FREE_SMS_WEBSITE_URL}/lc/"
+
+view_queue = multiprocessing.Queue()
 
 def get_driver():
     options = webdriver.ChromeOptions()
+    options.add_argument(f'user-agent={fake.user_agent()}')
+    if PROXY_IP and PROXY_PORT:
+        options.add_argument(f'--proxy-server=http://{PROXY_IP}:{PROXY_PORT}')
+        print(f"Using proxy: {PROXY_IP}:{PROXY_PORT}")
+    else:
+        print("Using direct connection (no proxy).")
+    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')
     options.add_argument('--headless=new')
     driver = webdriver.Chrome(options=options)
     return driver
@@ -18,19 +52,235 @@ def get_driver():
 def random_sleep(min_seconds, max_seconds):
     time.sleep(random.uniform(min_seconds, max_seconds))
 
+def generate_random_username(base_username="user"):
+    return f"{base_username}_{random.randint(100000, 999999)}"
+
+def generate_random_password(length=15):
+    characters = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789!@#$%^&*()"
+    return ''.join(random.choice(characters) for _ in range(length))
+
+def get_temp_email_from_website(driver):
+    try:
+        driver.get(TEMP_MAIL_WEBSITE_URL)
+        email_element = WebDriverWait(driver, 20).until(EC.presence_of_element_located((By.ID, "mail")))
+        temp_email = email_element.get_attribute('value')
+        return temp_email
+    except:
+        return None
+
+def get_verification_code_from_website(driver):
+    try:
+        start_time = time.time()
+        while time.time() - start_time < 120:
+            try:
+                driver.refresh()
+                random_sleep(5, 8)
+                email_list_element = WebDriverWait(driver, 20).until(EC.presence_of_element_located((By.ID, "inbox-id-0")))
+                email_list_element.click()
+                email_frame = WebDriverWait(driver, 20).until(EC.frame_to_be_available_and_switch_to_it((By.ID, "iframeMail")))
+                email_body = WebDriverWait(driver, 20).until(EC.presence_of_element_located((By.TAG_NAME, "body"))).text
+                verification_code_match = re.search(r'Verification code: (\d{6})', email_body)
+                if verification_code_match:
+                    return verification_code_match.group(1)
+                driver.switch_to.default_content()
+            except:
+                driver.switch_to.default_content()
+            time.sleep(random.uniform(5, 10))
+        return None
+    except:
+        return None
+
+def get_phone_number_from_free_sms_service():
+    try:
+        response = requests.get(FREE_SMS_GET_NUMBER_URL, timeout=10)
+        response.raise_for_status()
+        soup = BeautifulSoup(response.content, 'html.parser')
+        phone_number_element = soup.select_one('.number-line')
+        if phone_number_element:
+            phone_number = phone_number_element.text.strip()
+            return phone_number
+    except:
+        return None
+
+def get_sms_code_from_free_sms_service(phone_number):
+    if not phone_number:
+        return None
+    try:
+        start_time = time.time()
+        while time.time() - start_time < 120:
+            try:
+                inbox_url = f"{FREE_SMS_INBOX_URL_BASE}{phone_number.replace('+', '')}"
+                response = requests.get(inbox_url, timeout=10)
+                response.raise_for_status()
+                soup = BeautifulSoup(response.content, 'html.parser')
+                message_element = soup.select_one('.messages > div:nth-child(2) > div:nth-child(2)')
+                if message_element:
+                    message_text = message_element.text.strip()
+                    verification_code_match = re.search(r'Verification code: (\d{6})', message_text)
+                    if verification_code_match:
+                        return verification_code_match.group(1)
+            except:
+                pass
+            time.sleep(random.uniform(5, 10))
+        return None
+    except:
+        return None
+
+def create_instagram_account():
+    driver = get_driver()
+    try:
+        temp_email = get_temp_email_from_website(driver)
+        driver.get("https://www.instagram.com/accounts/emailsignup/")
+        random_sleep(5, 8)
+        username = generate_random_username("insta_user")
+        password = generate_random_password()
+        full_name = "Test User " + str(random.randint(100, 999))
+        email_field = WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.NAME, "emailOrPhone")))
+        email_field.send_keys(temp_email)
+        fullname_field = WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.NAME, "fullName")))
+        fullname_field.send_keys(full_name)
+        username_field = WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.NAME, "username")))
+        username_field.send_keys(username)
+        password_field = WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.NAME, "password")))
+        password_field.send_keys(password)
+        signup_button = WebDriverWait(driver, 10).until(EC.element_to_be_clickable((By.XPATH, "//button[contains(text(), 'Sign up')]")))
+        signup_button.click()
+        random_sleep(8, 12)
+        month_select = WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.XPATH, "//select[@title='Month']")))
+        month_select.send_keys("January")
+        day_select = WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.XPATH, "//select[@title='Day']")))
+        day_select.send_keys("1")
+        year_select = WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.XPATH, "//select[@title='Year']")))
+        year_select.send_keys("2000")
+        next_button = WebDriverWait(driver, 10).until(EC.element_to_be_clickable((By.XPATH, "//button[contains(text(), 'Next')]")))
+        next_button.click()
+        random_sleep(8, 12)
+        return driver, f"Instagram Account Creation Attempted: Username: {username}, Email: {temp_email}."
+    except:
+        if driver:
+            driver.quit()
+        return None, "Instagram Account Creation Attempted: Error."
+
+def create_google_account():
+    driver = get_driver()
+    try:
+        temp_email = get_temp_email_from_website(driver)
+        driver.get("https://accounts.google.com/signup")
+        random_sleep(5, 8)
+        first_name = "Test"
+        last_name = "User" + str(random.randint(100, 999))
+        username = generate_random_username("google_user")
+        password = generate_random_password()
+        first_name_field = WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.ID, "firstName")))
+        first_name_field.send_keys(first_name)
+        last_name_field = WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.ID, "lastName")))
+        last_name_field.send_keys(last_name)
+        username_field = WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.ID, "username")))
+        username_field.send_keys(username)
+        password_field = WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.NAME, "Passwd")))
+        password_field.send_keys(password)
+        confirm_password_field = WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.NAME, "ConfirmPasswd")))
+        confirm_password_field.send_keys(password)
+        next_button = WebDriverWait(driver, 10).until(EC.element_to_be_clickable((By.XPATH, "//span[text()='Next']")))
+        next_button.click()
+        random_sleep(8, 12)
+        return driver, f"Google Account Creation Attempted (without phone/SMS): Username: {username}, Email: {temp_email}."
+    except:
+        if driver:
+            driver.quit()
+        return None, "Google Account Creation Attempted: Error."
+
+def create_spotify_account():
+    driver = get_driver()
+    try:
+        temp_email = get_temp_email_from_website(driver)
+        driver.get("https://www.spotify.com/signup/")
+        random_sleep(5, 8)
+        password = generate_random_password()
+        profile_name = "Spotify User " + str(random.randint(100, 999))
+        birth_month = "January"
+        birth_day = "1"
+        birth_year = "2000"
+        email_field = WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.ID, "email")))
+        email_field.send_keys(temp_email)
+        password_field = WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.ID, "password")))
+        password_field.send_keys(password)
+        confirm_password_field = WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.ID, "confirm")))
+        confirm_password_field.send_keys(password)
+        name_field = WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.ID, "name")))
+        name_field.send_keys(profile_name)
+        month_select = WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.ID, "month")))
+        month_select.send_keys(birth_month)
+        day_select = WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.ID, "day")))
+        day_select.send_keys(birth_day)
+        year_select = WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.ID, "year")))
+        year_select.send_keys("2000")
+        signup_button = WebDriverWait(driver, 10).until(EC.element_to_be_clickable((By.XPATH, "//button[@data-encore-id='buttonPrimary']")))
+        signup_button.click()
+        random_sleep(10, 15)
+        return driver, f"Spotify Account Creation Attempted: Email: {temp_email}."
+    except:
+        if driver:
+            driver.quit()
+        return None, "Spotify Account Creation Attempted: Error."
+
+def detect_captcha(driver):
+    captcha_detected = False
+    captcha_message = None
+    captcha_element = None
+    try:
+        captcha_element = driver.find_element(By.ID, "recaptcha-anchor")
+        captcha_detected = True
+        captcha_message = "Google reCAPTCHA v2 Checkbox detected. Attempting auto-click (not solving)."
+    except:
+        pass
+    try:
+        captcha_element = driver.find_element(By.CLASS_NAME, "grecaptcha-badge")
+        captcha_detected = True
+        captcha_message = "Google reCAPTCHA v3 Badge detected. Attempting auto-click (not solving)."
+    except:
+        pass
+    try:
+        captcha_element = driver.find_element(By.CSS_SELECTOR, 'iframe[src*="turnstile"]')
+        captcha_detected = True
+        captcha_message = "Cloudflare Turnstile CAPTCHA detected. Attempting auto-click (not solving)."
+    except:
+        pass
+    return captcha_detected, captcha_message, captcha_element
+
 def view_instagram_video(video_url, max_playback_time):
     status_list = []
     driver = get_driver()
+    account_created = False
     try:
+        try:
+            driver, creation_status = create_instagram_account()
+            status_list.append(creation_status)
+            if driver:
+                account_created = True
+        except:
+            status_list.append("Instagram Account Creation Attempted: Error during creation.")
+
+        captcha_detected, captcha_message, captcha_element = detect_captcha(driver)
+        if captcha_detected:
+            status_list.append(captcha_message)
+            if captcha_element:
+                try:
+                    actions = ActionChains(driver)
+                    actions.move_to_element(captcha_element).click().perform()
+                    status_list.append("Auto-clicked CAPTCHA element.")
+                    random_sleep(5,8)
+                except:
+                    pass
+
         try:
             driver.get(video_url)
             random_sleep(5, 8)
         except:
             status_list.append(f"Error loading video URL: {video_url}")
-            return status_list
 
         try:
-            not_now_popup = WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.CLASS_NAME, '_a9_1')))
+            not_now_popup = WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.CLASS_NAME, 'mt3GC')))
             not_now_buttons = not_now_popup.find_elements(By.TAG_NAME, "button")
             if len(not_now_buttons) > 1:
                 not_now_button = not_now_buttons[1]
@@ -43,40 +293,236 @@ def view_instagram_video(video_url, max_playback_time):
             pass
 
         try:
-            not_now_element = WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.CLASS_NAME, 'mt3GC')))
-            buttons = not_now_element.find_elements(By.TAG_NAME, "button")
-            if len(buttons) > 1:  # Check if there are at least 2 buttons
-                button_to_click = buttons[1]
-                button_to_click.click()
-            else:
-                print("Warning: Popup 'mt3GC' found, but not enough buttons to click.") # More specific warning
-        except (TimeoutException, NoSuchElementException, WebDriverException) as e:
-            print(f"Warning: Could not handle initial popup: {type(e).__name__} - {e}") # Include exception info
+            video_element = WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.TAG_NAME, 'video')))
+            driver.execute_script("arguments[0].play();", video_element)
+        except:
             pass
+        random_sleep(5, 7)
+        time.sleep(max_playback_time)
+        status_list.append(f"Viewed Instagram video (attempted): {video_url}")
+    except:
+        status_list.append(f"Error processing Instagram video view: {video_url}")
+    finally:
+        if driver:
+            driver.quit()
+    return status_list
 
-        status_list.append(f"Attempted to handle Instagram popup for: {video_url}")
-    except Exception as e:
-        status_list.append(f"Error processing Instagram video popup handling: {str(e)}")
+def view_youtube_video(video_url, max_playback_time):
+    status_list = []
+    driver = get_driver()
+    account_created = False
+    try:
+        try:
+            driver, creation_status = create_google_account()
+            status_list.append(creation_status)
+            if driver:
+                account_created = True
+        except:
+            status_list.append("Google Account Creation Attempted: Error during creation.")
+
+        captcha_detected, captcha_message, captcha_element = detect_captcha(driver)
+        if captcha_detected:
+            status_list.append(captcha_message)
+            if captcha_element:
+                try:
+                    actions = ActionChains(driver)
+                    actions.move_to_element(captcha_element).click().perform()
+                    status_list.append("Auto-clicked CAPTCHA element.")
+                    random_sleep(5,8)
+                except:
+                    pass
+
+        try:
+            driver.get(video_url)
+            random_sleep(5, 8)
+        except:
+            status_list.append(f"Error loading video URL: {video_url}")
+
+        try:
+            cookie_button = WebDriverWait(driver, 10).until(EC.element_to_be_clickable((By.XPATH, "//button[contains(., 'Aceptar')]")))
+            cookie_button.click()
+        except:
+            pass
+        try:
+            not_now_button = WebDriverWait(driver, 10).until(EC.element_to_be_clickable((By.XPATH, "//div[@id='notification-dialog']//button[contains(., 'No gracias')]")))
+            not_now_button.click()
+        except:
+            pass
+        try:
+            play_button_container = WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.ID, "movie_player")))
+            actions = ActionChains(driver)
+            actions.click(play_button_container).perform()
+        except:
+            pass
+        random_sleep(15, 22)
+        time.sleep(max_playback_time)
+        status_list.append(f"Viewed YouTube video (attempted): {video_url}")
+    except:
+        status_list.append(f"Error processing YouTube video view: {video_url}")
     finally:
         if driver:
             driver.quit()
     return status_list
 
-def run_button_click(video_link, max_playback_time, _):
-    return view_instagram_video(video_link, max_playback_time)
+def view_spotify_track(track_url, max_playback_time):
+    status_list = []
+    driver = get_driver()
+    account_created = False
+    try:
+        try:
+            driver, creation_status = create_spotify_account()
+            status_list.append(creation_status)
+            if driver:
+                account_created = True
+        except:
+            status_list.append("Spotify Account Creation Attempted: Error during creation.")
+
+        captcha_detected, captcha_message, captcha_element = detect_captcha(driver)
+        if captcha_detected:
+            status_list.append(captcha_message)
+            if captcha_element:
+                try:
+                    actions = ActionChains(driver)
+                    actions.move_to_element(captcha_element).click().perform()
+                    status_list.append("Auto-clicked CAPTCHA element.")
+                    random_sleep(5,8)
+                except:
+                    pass
+        try:
+            driver.get(track_url)
+            random_sleep(5, 8)
+        except:
+            status_list.append(f"Error loading track URL: {track_url}")
+
+        try:
+            cookie_consent_button = WebDriverWait(driver, 10).until(EC.element_to_be_clickable((By.XPATH, "//button[@data-testid='cookie-banner-accept']")))
+            cookie_consent_button.click()
+        except:
+            pass
+        try:
+            play_button = WebDriverWait(driver, 10).until(EC.element_to_be_clickable((By.XPATH, "//button[@data-testid='control-button-playpause']")))
+            play_button.click()
+        except:
+            pass
+        random_sleep(20, 30)
+        time.sleep(max_playback_time)
+        status_list.append(f"Viewed Spotify track (attempted): {track_url}")
+    except:
+        status_list.append(f"Error processing Spotify track view: {track_url}")
+    finally:
+        if driver:
+            driver.quit()
+    return status_list
+
+def process_single_view_queued(task_id):
+    while True:
+        try:
+            params = view_queue.get(timeout=5)
+        except queue.Empty:
+            break
+
+        video_link = params['video_link']
+        max_playback_time = params['max_playback_time']
+        platform_type = params['platform_type']
+        status_list = []
+
+        try:
+            if "instagram.com" in video_link:
+                if not ("/reel/" in video_link or "/p/" in video_link):
+                    status_list.append("Invalid Instagram video link.")
+                else:
+                    status_list.extend(view_instagram_video(video_link, max_playback_time))
+                    platform_type = "instagram"
+            elif "youtube.com" in video_link or "youtu.be" in video_link:
+                status_list.extend(view_youtube_video(video_link, max_playback_time))
+                platform_type = "youtube"
+            elif "open.spotify.com" in video_link and "/track/" in video_link:
+                status_list.extend(view_spotify_track(track_url=video_link, max_playback_time=max_playback_time))
+                platform_type = "spotify"
+            else:
+                status_list.append("Invalid video link. Please use Instagram, YouTube, or Spotify track links.")
+        except:
+            status_list.append(f"Critical error during view processing.")
+        finally:
+            if status_list:
+                print(f"View completed. Status: {status_list[-1] if status_list else 'No Status'}")
+            return status_list
+
+def process_request_wrapper(video_link, num_views, max_playback_time, num_processes):
+    return list(process_request(video_link, num_views, max_playback_time, num_processes))
+
+def process_request(video_link, num_views, max_playback_time, num_processes):
+    status_all_views = []
+    if not video_link:
+        status_all_views.append("Please enter a video link.")
+        return [[status] for status in status_all_views]
+    try:
+        num_views = int(num_views)
+        if num_views <= 0 or num_views > 10000000:
+            status_all_views.append("Number of views must be between 1 and 10,000,000.")
+            return [[status] for status in status_all_views]
+    except ValueError:
+        status_all_views.append("Invalid number of views.")
+        return [[status] for status in status_all_views]
+
+    platform_type = None
+    video_link_check = video_link if not isinstance(video_link, list) else video_link[0]
+    if "instagram.com" in video_link_check:
+        platform_type = "instagram"
+    elif "youtube.com" in video_link_check or "youtu.be" in video_link_check:
+        platform_type = "youtube"
+    elif "open.spotify.com" in video_link_check and "/track/" in video_link_check:
+        platform_type = "spotify"
+
+    for _ in range(num_views):
+        view_queue.put({
+            'video_link': video_link,
+            'max_playback_time': max_playback_time,
+            'platform_type': platform_type
+        })
+
+    num_refreshes = 100
+    refresh_interval = 5
+    refreshes_done = 0
+    last_refresh_time = time.time()
+
+    with multiprocessing.Pool(processes=num_processes) as pool:
+        results_iterator = pool.imap_unordered(process_single_view_queued, range(num_views))
+
+        for statuses in results_iterator:
+            if statuses:
+                status_all_views.extend(statuses)
+                if any("CAPTCHA" in status for status in statuses):
+                    status_all_views.append("CAPTCHA detected and auto-clicked. CAPTCHA is unlikely to be solved automatically. Manual solving or a paid CAPTCHA service is required for full automation. Account creation and views may be stopped or degraded in quality.")
+
+            current_time = time.time()
+            if current_time - last_refresh_time >= refresh_interval and refreshes_done < num_refreshes:
+                yield [[status] for status in status_all_views]
+                last_refresh_time = current_time
+                refreshes_done += 1
+
+    yield [[status] for status in status_all_views]
+
+
+def run_button_click(video_link, num_views, max_playback_time, num_processes, _):
+    return process_request_wrapper(video_link, num_views, max_playback_time, num_processes)
 
 def launch_gradio_interface():
     iface = gr.Interface(
         fn=run_button_click,
         inputs=[
-            gr.Textbox(label="Instagram Video Link", placeholder="Enter Instagram Reel or Post Link"),
+            gr.Textbox(label="Video Link", placeholder="Enter Instagram, YouTube, or Spotify Track Link"),
+            gr.Number(label="Number of Views (Max 10,000,000)", value=1, minimum=1, maximum=10000000, step=1),
             gr.Slider(label="Max Playback Time (seconds)", minimum=5, maximum=300, value=30, step=5),
-            gr.Button("Run Popup Handler")
+            gr.Slider(label="Number of Processes (Simultaneous Views - Max 100, Be Resourceful!)", minimum=1, maximum=100, value=1, step=1, visible=True),
+            gr.Button("Run View Bot")
         ],
-        outputs = gr.Dataframe(headers=["Status"], label="Status"),
-        live=False
+        outputs = gr.Dataframe(headers=["Status"], label="View Status"),
+        live=False # Changed to False
     )
+    iface.queue(max_size=1000000)
     iface.launch(share=False)
 
+
 if __name__ == "__main__":
     launch_gradio_interface()