from fastapi import FastAPI, BackgroundTasks, HTTPException, Form
from fastapi.responses import PlainTextResponse, HTMLResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel
import requests
from requests.exceptions import ProxyError, ConnectTimeout
from faker import Faker
import re
from concurrent.futures import ThreadPoolExecutor
import urllib.parse
import time
import gradio as gr
app = FastAPI()
fake = Faker()
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 ProxyError as e:
return PlainTextResponse(f"ProxyError: {str(e)}", status_code=500)
except ConnectTimeout as e:
return PlainTextResponse(f"ConnectTimeoutError: {str(e)}", status_code=500)
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) # Decode the 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 simulate_view(video_id: str, platform: str, proxy: str, delay: int):
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:
if platform == "instagram":
view_url = f"http://www.instagram.com/p/{video_id}?autoplay=true"
elif platform == "tiktok":
view_url = f"http://www.tiktok.com/@{video_id}?autoplay=true"
elif platform == "youtube":
view_url = f"http://www.youtube.com/watch?v={video_id}&autoplay=1"
elif platform == "facebook":
view_url = f"http://www.facebook.com/watch/?v={video_id}&autoplay=1"
elif platform == "twitch":
view_url = f"http://www.twitch.tv/videos/{video_id}?autoplay=true"
elif platform == "spotify":
view_url = f"http://open.spotify.com/track/{video_id}?autoplay=true"
else:
raise ValueError("Invalid platform")
response = requests.get(view_url, headers=headers, proxies=proxies, timeout=10)
time.sleep(delay)
except Exception as e:
print(f"Error in simulate_view: {e}")
def perform_views(url: str, platform: str, count: int, delay: int, parallel_processes: int):
video_id = extract_video_id(url, platform)
def task():
try:
proxy_response = requests.get("http://0.0.0.0:8000/rotate_ip")
if proxy_response.status_code == 200:
proxy = proxy_response.text.strip()
if proxy:
simulate_view(video_id, platform, proxy, delay)
except Exception as e:
print(f"Error in perform_views: {e}")
with ThreadPoolExecutor(max_workers=parallel_processes) as executor:
for _ in range(count):
executor.submit(task)
@app.post("/increase_views", response_class=PlainTextResponse)
def increase_views(
url: str = Form(...),
platform: str = Form(...),
count: int = Form(...),
delay: int = Form(...),
parallel_processes: int = Form(...),
background_tasks: BackgroundTasks = BackgroundTasks()
):
if platform not in ["instagram", "tiktok", "youtube", "facebook", "twitch", "spotify"]:
raise HTTPException(status_code=400, detail="Invalid platform. Choose 'instagram', 'tiktok', 'youtube', 'facebook', 'twitch', or 'spotify'.")
background_tasks.add_task(perform_views, url, platform, count, delay, parallel_processes)
return PlainTextResponse(f"Started {count} view tasks for {platform} at {url} with {delay} seconds delay and {parallel_processes} parallel processes.")
def increase_views_gradio(url, platform, count, delay, parallel_processes):
return increase_views(url, platform, count, delay, parallel_processes, background_tasks=BackgroundTasks())
iface = gr.Interface(
fn=increase_views_gradio,
inputs=[
gr.Textbox(label="URL"),
gr.Dropdown(["instagram", "tiktok", "youtube", "facebook", "twitch", "spotify"], label="Platform"),
gr.Number(label="View Count", minimum=1),
gr.Number(label="Delay (seconds)", minimum=1),
gr.Number(label="Parallel Processes", minimum=1),
],
outputs="text",
title="Increase Views",
description="Increase views on your favorite social media platforms.",
)
@app.get("/form", response_class=HTMLResponse)
def form():
return iface.render()
@app.get("/", response_class=HTMLResponse)
def read_root():
html_content = """
Main Page
"""
return HTMLResponse(content=html_content)
if __name__ == "__main__":
import uvicorn
iface.launch(share=True)