Spaces:
Sleeping
Sleeping
| 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 | |
| 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) | |
| 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): | |
| 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 = requests.get(view_url, headers=headers, proxies=proxies) | |
| except Exception as e: | |
| print(f"Error in simulate_instagram_view: {e}") | |
| def simulate_tiktok_view(video_id: str, proxy: str): | |
| 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 = requests.get(view_url, headers=headers, proxies=proxies) | |
| except Exception as e: | |
| print(f"Error in simulate_tiktok_view: {e}") | |
| def simulate_youtube_view(video_id: str, proxy: str): | |
| 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 = requests.get(view_url, headers=headers, proxies=proxies) | |
| except Exception as e: | |
| print(f"Error in simulate_youtube_view: {e}") | |
| def simulate_facebook_view(video_id: str, proxy: str): | |
| 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 = requests.get(view_url, headers=headers, proxies=proxies) | |
| except Exception as e: | |
| print(f"Error in simulate_facebook_view: {e}") | |
| def simulate_twitch_view(video_id: str, proxy: str): | |
| 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 = requests.get(view_url, headers=headers, proxies=proxies) | |
| except Exception as e: | |
| print(f"Error in simulate_twitch_view: {e}") | |
| def simulate_spotify_view(video_id: str, proxy: str): | |
| 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 = requests.get(view_url, headers=headers, proxies=proxies) | |
| 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): | |
| 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) | |
| elif platform == "tiktok": | |
| executor.submit(simulate_tiktok_view, video_id, proxy) | |
| elif platform == "youtube": | |
| executor.submit(simulate_youtube_view, video_id, proxy) | |
| elif platform == "facebook": | |
| executor.submit(simulate_facebook_view, video_id, proxy) | |
| elif platform == "twitch": | |
| executor.submit(simulate_twitch_view, video_id, proxy) | |
| elif platform == "spotify": | |
| executor.submit(simulate_spotify_view, video_id, proxy) | |
| async def simulate_views_endpoint(request: VisitRequest): | |
| try: | |
| simulate_views( | |
| url=request.url, | |
| platform=request.platform, | |
| count=request.count, | |
| delay=request.delay, | |
| parallel_processes=request.parallel_processes | |
| ) | |
| return PlainTextResponse("Views simulation started") | |
| 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") | |
| 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(url: str, platform: str, count: int, delay: int, parallel_processes: int): | |
| try: | |
| simulate_views(url, platform, count, delay, parallel_processes) | |
| return "Simulation started successfully." | |
| except Exception as e: | |
| return f"Error: {e}" | |
| gr.Interface( | |
| fn=gradio_interface, | |
| inputs=[ | |
| gr.Textbox(label="URL"), | |
| 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() | |
| def get_credentials(): | |
| credentials = [] | |
| with open(".env", "r") as file: | |
| lines = file.readlines() | |
| for line in lines: | |
| line = line.strip() | |
| if "=" in line: | |
| platform, details = line.split("=", 1) | |
| if "|" in details: | |
| user, password = details.split("|", 1) | |
| credentials.append((platform.strip(), user.strip(), password.strip())) | |
| return credentials | |
| def get_random_credentials(): | |
| credentials = get_credentials() | |
| if credentials: | |
| return random.choice(credentials) | |
| else: | |
| raise HTTPException(status_code=404, detail="No credentials found") | |
| async def random_login(platform: str): | |
| try: | |
| platform_cred = next((plat for plat in get_credentials() if plat[0] == platform), None) | |
| if platform_cred: | |
| _, username, password = platform_cred | |
| session = authenticate(username, password, platform) | |
| return PlainTextResponse(f"Authenticated on {platform} with user {username}") | |
| else: | |
| raise HTTPException(status_code=404, detail="Platform not found") | |
| except HTTPException as e: | |
| return e | |