Spaces:
Sleeping
Sleeping
| import os | |
| import json | |
| import random | |
| import base64 | |
| import asyncio | |
| import urllib.request | |
| from fastapi import FastAPI, Header, HTTPException, BackgroundTasks | |
| from pydantic import BaseModel | |
| from typing import Optional | |
| from playwright.async_api import async_playwright | |
| from playwright_stealth import stealth_async | |
| app = FastAPI(title="Omni-Digital Cloud Engine (Enterprise)", version="4.0") | |
| # আপনার কাস্টম কি | |
| MASTER_API_KEY = os.getenv("API_KEY", "rh5152..") | |
| # --- 🚀 ৫টি নতুন ফিচার সহ অ্যাডভান্সড পে-লোড মডেল --- | |
| class CommandPayload(BaseModel): | |
| url: str | |
| js_code: Optional[str] = None | |
| extract_html: bool = False | |
| wait_time_ms: int = 0 | |
| # ৩. পারসিস্টেন্ট সেশন (Persistent Session) | |
| session_id: Optional[str] = None | |
| # ৪. ক্যাপচা সলভিং গেটওয়ে (Anti-Captcha) | |
| solve_captcha: bool = False | |
| # ৫. ওয়েবহুক (Webhook URL for Background tasks) | |
| webhook_url: Optional[str] = None | |
| # ৬. ভিজ্যুয়াল ডিবাগিং (Visual Debug Screenshot) | |
| visual_debug: bool = False | |
| # ৭. স্মার্ট ফিঙ্গারপ্রিন্টিং (Dynamic Fingerprint) | |
| smart_fingerprint: bool = True | |
| async def health_check(): | |
| return {"status": "System Online", "message": "Enterprise Stealth Mode v4.0 Active."} | |
| # --- ৭. স্মার্ট ফিঙ্গারপ্রিন্টিং ল্যাব (Dynamic Fingerprint Generator) --- | |
| def get_dynamic_fingerprint(): | |
| resolutions = [{"width": 1920, "height": 1080}, {"width": 1366, "height": 768}, {"width": 1440, "height": 900}, {"width": 1536, "height": 864}] | |
| memory_options = [4, 8, 16, 32] | |
| cpu_options = [4, 8, 12, 16] | |
| # Chrome-এর লেটেস্ট ভার্সনগুলো রেন্ডমাইজ করা | |
| chrome_version = random.choice(["120.0.0.0", "121.0.0.0", "122.0.0.0", "123.0.0.0"]) | |
| os_version = random.choice(["Windows NT 10.0; Win64; x64", "Macintosh; Intel Mac OS X 10_15_7", "X11; Linux x86_64"]) | |
| user_agent = f"Mozilla/5.0 ({os_version}) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/{chrome_version} Safari/537.36" | |
| return { | |
| "user_agent": user_agent, | |
| "viewport": random.choice(resolutions), | |
| "memory": random.choice(memory_options), | |
| "cpus": random.choice(cpu_options) | |
| } | |
| # --- মূল ব্রাউজার প্রসেসিং ইঞ্জিন --- | |
| async def process_engine_task(payload: CommandPayload): | |
| async with async_playwright() as p: | |
| fingerprint = get_dynamic_fingerprint() if payload.smart_fingerprint else { | |
| "user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/122.0.0.0 Safari/537.36", | |
| "viewport": {"width": 1920, "height": 1080}, | |
| "memory": 8, "cpus": 4 | |
| } | |
| browser_args = [ | |
| "--no-sandbox", | |
| "--disable-setuid-sandbox", | |
| "--disable-blink-features=AutomationControlled", | |
| "--disable-dev-shm-usage" | |
| ] | |
| # ৩. পারসিস্টেন্ট সেশন হ্যান্ডলিং | |
| if payload.session_id: | |
| user_data_dir = f"/app/sessions/{payload.session_id}" | |
| os.makedirs(user_data_dir, exist_ok=True) | |
| context = await p.chromium.launch_persistent_context( | |
| user_data_dir=user_data_dir, | |
| headless=True, | |
| user_agent=fingerprint['user_agent'], | |
| viewport=fingerprint['viewport'], | |
| args=browser_args | |
| ) | |
| page = context.pages[0] if context.pages else await context.new_page() | |
| else: | |
| browser = await p.chromium.launch(headless=True, args=browser_args) | |
| context = await browser.new_context( | |
| user_agent=fingerprint['user_agent'], | |
| viewport=fingerprint['viewport'] | |
| ) | |
| page = await context.new_page() | |
| # ৭. স্মার্ট স্টিলথ ইনজেকশন (CPU/Memory Spoofing) | |
| stealth_script = f""" | |
| Object.defineProperty(navigator, 'webdriver', {{get: () => false}}); | |
| window.chrome = {{ runtime: {{}} }}; | |
| Object.defineProperty(navigator, 'languages', {{get: () => ['en-US', 'en']}}); | |
| Object.defineProperty(navigator, 'deviceMemory', {{get: () => {fingerprint['memory']}}}); | |
| Object.defineProperty(navigator, 'hardwareConcurrency', {{get: () => {fingerprint['cpus']}}}); | |
| """ | |
| await context.add_init_script(stealth_script) | |
| await stealth_async(page) | |
| try: | |
| # পেজ নেভিগেশন | |
| await page.goto(payload.url, wait_until="domcontentloaded", timeout=60000) | |
| # ৪. ক্যাপচা সলভিং মডিউল (Simulation / Wait protocol) | |
| if payload.solve_captcha: | |
| # Cloudflare বা রেডি ক্যাপচার জন্য এক্সট্রা অপেক্ষা করা | |
| await page.wait_for_timeout(5000) | |
| # (ভবিষ্যতে এখানে 2Captcha বা AI ভিশন API যুক্ত করা যাবে) | |
| if payload.wait_time_ms > 0: | |
| await page.wait_for_timeout(payload.wait_time_ms) | |
| script_result = await page.evaluate(payload.js_code) if payload.js_code else None | |
| html_content = await page.content() if payload.extract_html else None | |
| # ৬. ভিজ্যুয়াল ডিবাগিং (Screenshot to Base64) | |
| screenshot_b64 = None | |
| if payload.visual_debug: | |
| screenshot_bytes = await page.screenshot(type='jpeg', quality=70) | |
| screenshot_b64 = base64.b64encode(screenshot_bytes).decode('utf-8') | |
| result = { | |
| "status": "success", | |
| "current_url": page.url, | |
| "js_output": script_result, | |
| "html_extracted": bool(html_content), | |
| "fingerprint_used": fingerprint['user_agent'], | |
| "screenshot_base64": screenshot_b64 | |
| } | |
| except Exception as e: | |
| result = {"status": "error", "message": str(e)} | |
| finally: | |
| if payload.session_id: | |
| await context.close() | |
| else: | |
| await browser.close() | |
| return result | |
| # --- ৫. ওয়েবহুক ট্রিগার ফাংশন --- | |
| def send_webhook(url: str, data: dict): | |
| try: | |
| req = urllib.request.Request(url, data=json.dumps(data).encode('utf-8'), headers={'Content-Type': 'application/json'}) | |
| urllib.request.urlopen(req, timeout=10) | |
| except Exception as e: | |
| print(f"Webhook delivery failed: {str(e)}") | |
| async def autonomous_background_task(payload: CommandPayload): | |
| result = await process_engine_task(payload) | |
| if payload.webhook_url: | |
| await asyncio.to_thread(send_webhook, payload.webhook_url, result) | |
| # --- রিয়েল-টাইম এপিআই (আগের মতো) --- | |
| async def execute_browser_task(payload: CommandPayload, x_api_key: str = Header(None)): | |
| if x_api_key != MASTER_API_KEY: | |
| raise HTTPException(status_code=403, detail="Access Denied") | |
| return await process_engine_task(payload) | |
| # --- ৫. নতুন এপিআই: অটোনোমাস শিডিউলার --- | |
| async def schedule_task(payload: CommandPayload, background_tasks: BackgroundTasks, x_api_key: str = Header(None)): | |
| if x_api_key != MASTER_API_KEY: | |
| raise HTTPException(status_code=403, detail="Access Denied") | |
| if not payload.webhook_url: | |
| raise HTTPException(status_code=400, detail="webhook_url is required for scheduled tasks") | |
| background_tasks.add_task(autonomous_background_task, payload) | |
| return {"status": "Task scheduled successfully", "webhook": payload.webhook_url} |