soti.py

# alphattspro-main/soti.py

import os
import asyncio
import ssl
from fastapi import APIRouter, WebSocket, WebSocketDisconnect, Request
from fastapi.responses import HTMLResponse, JSONResponse, StreamingResponse, FileResponse
import httpx
import websockets

router = APIRouter()

# تعریف آدرس‌های مربوط به هر دو اسپیس مستقل هوش مصنوعی صوتی
SOTI_PRO_HTTP = "https://ezmarynoori-sotipro.hf.space"
SOTI_PRO_WS = "wss://ezmarynoori-sotipro.hf.space/ws"

SOTI_FREE_HTTP = "https://ezmarynoori-sotifree.hf.space"
SOTI_FREE_WS = "wss://ezmarynoori-sotifree.hf.space/ws"

# هدرهای امنیتی جهت اعطای رسمی دسترسی دوربین و میکروفون به مرورگر و وب‌ویو موبایل
PERMISSION_HEADERS = {
    "Permissions-Policy": "camera=*, microphone=*, geolocation=*, autoplay=*",
    "Feature-Policy": "camera *; microphone *; autoplay *",
    "Access-Control-Allow-Origin": "*",
    "Access-Control-Allow-Methods": "GET, POST, OPTIONS",
    "Access-Control-Allow-Headers": "*"
}

# غیرفعال کردن بررسی سخت‌گیرانه SSL برای جلوگیری از خطای تایید گواهینامه در داکر رانفلر
ssl_context = ssl.create_default_context()
ssl_context.check_hostname = False
ssl_context.verify_mode = ssl.CERT_NONE

def get_target_urls(request_or_ws) -> tuple[str, str]:
    """
    مسیریابی هوشمند و پویا بر اساس هدر Referer مرورگر کاربر.
    اگر کاربر در صفحه sotifree باشد به سرور رایگان و در غیر این صورت به سرور پرو متصل می‌شود.
    """
    referer = request_or_ws.headers.get("referer", "").lower()
    if "sotifree" in referer:
        return SOTI_FREE_HTTP, SOTI_FREE_WS
    else:
        return SOTI_PRO_HTTP, SOTI_PRO_WS

@router.websocket("/ws")
async def websocket_proxy(client_ws: WebSocket):
    """
    پروکسی هوشمند وب‌سوکت با مسیریابی پویا بین پرو و رایگان بر اساس ارجاع دهنده (Referer)
    """
    await client_ws.accept()
    
    # تشخیص خودکار مقصد بر اساس آدرس صفحه‌ای که کاربر در آن قرار دارد
    target_http, target_ws = get_target_urls(client_ws)
    
    # هدرهای شبیه‌سازی شده برای عبور از سد امنیتی هاگینگ‌فیس
    headers = {
        "Host": target_http.replace("https://", ""),
        "Origin": target_http,
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
    }
    
    # تشخیص پویا و هوشمند نسخه کتابخانه websockets رانفلر جهت ارسال پارامتر صحیح دست‌تکانی
    ws_kwargs = {}
    try:
        ws_version = getattr(websockets, "__version__", "10.0")
        try:
            major_version = int(ws_version.split('.')[0])
        except ValueError:
            major_version = 10
            
        if major_version >= 14:
            ws_kwargs["additional_headers"] = headers
        else:
            ws_kwargs["extra_headers"] = headers
    except Exception:
        ws_kwargs["extra_headers"] = headers

    try:
        # اتصال امن به وب‌سوکت هدف با غیرفعال کردن چک کردن SSL جهت تضمین پایداری ارتباط
        async with websockets.connect(
            target_ws, 
            ssl=ssl_context,
            **ws_kwargs
        ) as server_ws:
            
            async def client_to_server():
                try:
                    while True:
                        data = await client_ws.receive()
                        if "text" in data and data["text"] is not None:
                            await server_ws.send(data["text"])
                        elif "bytes" in data and data["bytes"] is not None:
                            await server_ws.send(data["bytes"])
                except Exception:
                    pass

            async def server_to_client():
                try:
                    async for message in server_ws:
                        if isinstance(message, str):
                            await client_ws.send_text(message)
                        else:
                            await client_ws.send_bytes(message)
                except Exception:
                    pass

            await asyncio.gather(client_to_server(), server_to_client())
            
    except WebSocketDisconnect:
        pass
    except Exception as e:
        print(f"Soti WS Proxy Error: {e}")
    finally:
        try:
            await client_ws.close()
        except:
            pass

@router.get("/api/instructions")
async def proxy_instructions(request: Request):
    """
    دریافت هوشمند و پروکسی دستورالعمل‌های شخصیت‌ها بر اساس نسخه کاربر (پرو یا رایگان)
    """
    target_http, _ = get_target_urls(request)
    async with httpx.AsyncClient(verify=False) as client:
        try:
            resp = await client.get(f"{target_http}/api/instructions", timeout=10.0)
            return JSONResponse(content=resp.json(), status_code=resp.status_code, headers=PERMISSION_HEADERS)
        except Exception as e:
            return JSONResponse(content={"error": str(e)}, status_code=500, headers=PERMISSION_HEADERS)

# --- صفحات اصلی هردو نسخه ---
@router.get("/soti")
@router.get("/soti/")
async def serve_soti_pro_page():
    async with httpx.AsyncClient(verify=False) as client:
        try:
            resp = await client.get(f"{SOTI_PRO_HTTP}/", timeout=15.0)
            return HTMLResponse(content=resp.text, status_code=resp.status_code, headers=PERMISSION_HEADERS)
        except Exception as e:
            return HTMLResponse(f"<h1>خطا در بارگذاری نرم‌افزار صوتی پرو: {e}</h1>", status_code=500, headers=PERMISSION_HEADERS)

@router.get("/sotifree")
@router.get("/sotifree/")
async def serve_soti_free_page():
    async with httpx.AsyncClient(verify=False) as client:
        try:
            resp = await client.get(f"{SOTI_FREE_HTTP}/", timeout=15.0)
            return HTMLResponse(content=resp.text, status_code=resp.status_code, headers=PERMISSION_HEADERS)
        except Exception as e:
            return HTMLResponse(f"<h1>خطا در بارگذاری نرم‌افزار صوتی رایگان: {e}</h1>", status_code=500, headers=PERMISSION_HEADERS)


@router.get("/static/{file_path:path}")
async def proxy_static_files(request: Request, file_path: str):
    """
    انتقال زنده استاتیک ری‌اکت با تفکیک هوشمند مسیرها بین پرو، رایگان و دیسک رانفلر
    """
    if file_path.startswith("images/"):
        # 🟢 بهینه‌سازی بسیار مهم: لود مستقیم تصویر و ویدیوهای گالری‌ها از هارد دیسک داخلی رانفلر بدون تاخیر شبکه
        filename = file_path.replace("images/", "")
        local_path = os.path.join("static/images", filename)
        if os.path.exists(local_path):
            media_type = "application/octet-stream"
            if filename.endswith(".mp4"): media_type = "video/mp4"
            elif filename.endswith(".png"): media_type = "image/png"
            elif filename.endswith(".webp"): media_type = "image/webp"
            elif filename.endswith(".jpg") or filename.endswith(".jpeg"): media_type = "image/jpeg"
            elif filename.endswith(".txt"): media_type = "text/plain"
            return FileResponse(local_path, media_type=media_type, headers=PERMISSION_HEADERS)
        return HTMLResponse("File not found", status_code=404)
    else:
        # تشخیص هوشمند مبدا فایل استاتیک ری‌اکت صوتی (پرو یا رایگان) بر اساس Referer
        target_http, _ = get_target_urls(request)
        url = f"{target_http}/static/{file_path}"

    async def stream_file():
        async with httpx.AsyncClient(verify=False) as client:
            try:
                async with client.stream("GET", url, timeout=20.0) as resp:
                    async for chunk in resp.aiter_bytes():
                        yield chunk
            except Exception:
                yield b""
    
    media_type = "application/octet-stream"
    if file_path.endswith(".js"):
        media_type = "application/javascript"
    elif file_path.endswith(".css"):
        media_type = "text/css"
    elif file_path.endswith(".png"):
        media_type = "image/png"
    elif file_path.endswith(".jpg") or file_path.endswith(".jpeg"):
        media_type = "image/jpeg"
    elif file_path.endswith(".svg"):
        media_type = "image/svg+xml"
    elif file_path.endswith(".mp4"):
        media_type = "video/mp4"
    elif file_path.endswith(".webp"):
        media_type = "image/webp"
        
    return StreamingResponse(stream_file(), media_type=media_type, headers=PERMISSION_HEADERS)

@router.get("/manifest.json")
async def proxy_manifest(request: Request):
    target_http, _ = get_target_urls(request)
    async with httpx.AsyncClient(verify=False) as client:
        try:
            resp = await client.get(f"{target_http}/manifest.json", timeout=10.0)
            return JSONResponse(content=resp.json(), status_code=resp.status_code, headers=PERMISSION_HEADERS)
        except Exception:
            return JSONResponse(content={}, status_code=404, headers=PERMISSION_HEADERS)

@router.get("/favicon.ico")
async def proxy_favicon(request: Request):
    target_http, _ = get_target_urls(request)
    async def stream_fav():
        async with httpx.AsyncClient(verify=False) as client:
            try:
                async with client.stream("GET", f"{target_http}/favicon.ico", timeout=10.0) as r:
                    async for chunk in r.aiter_bytes():
                        yield chunk
            except Exception:
                yield b""
    return StreamingResponse(stream_fav(), media_type="image/x-icon", headers=PERMISSION_HEADERS)

@router.get("/logo192.png")
async def proxy_logo192(request: Request):
    target_http, _ = get_target_urls(request)
    async def stream_logo():
        async with httpx.AsyncClient(verify=False) as client:
            try:
                async with client.stream("GET", f"{target_http}/logo192.png", timeout=10.0) as r:
                    async for chunk in r.aiter_bytes():
                        yield chunk
            except Exception:
                yield b""
    return StreamingResponse(stream_logo(), media_type="image/png", headers=PERMISSION_HEADERS)

@router.get("/logo512.png")
async def proxy_logo512(request: Request):
    target_http, _ = get_target_urls(request)
    async def stream_logo():
        async with httpx.AsyncClient(verify=False) as client:
            try:
                async with client.stream("GET", f"{target_http}/logo512.png", timeout=10.0) as r:
                    async for chunk in r.aiter_bytes():
                        yield chunk
            except Exception:
                yield b""
    return StreamingResponse(stream_logo(), media_type="image/png", headers=PERMISSION_HEADERS)