| 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): |
| 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) |
|
|
| @app.post("/simulate_views") |
| 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") |
|
|
| @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(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") |
|
|
| @app.get("/random_login") |
| 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 |
|
|
