import os import time import base64 import json import httpx import uuid import hmac import hashlib from fastapi import APIRouter, Form, File, UploadFile from fastapi.responses import HTMLResponse, JSONResponse router = APIRouter() # توکن ثابت ربات بله شما جهت بازسازی خودکار آدرس‌ها در حافظه موقت (بدون دیتابیس) BALE_BOT_TOKEN = os.environ.get("BALE_BOT_TOKEN", "2030030646:atvybsEWn6HNfx7ciG_sM6MY-majP0hEQgE").strip() # کلید امنیتی مشترک برای جلوگیری از هک و دستکاری وضعیت اشتراک در لینک‌ها SHARED_SECRET = "AI_ALPHA_SECURE_TOKEN_2026" CONTEXT_TRANSLATIONS = { "edit_image": "ویرایش عکس 🪄", "voice_change": "تغییر صدا 🎙️", "voice_clone": "کلون کردن صدا 👤", "animate_image": "متحرک‌سازی تصویر 🎞️", "video_image": "ساخت ویدیو با عکس 🖼️", "image_analysis": "تحلیل تصویر 🔍", "stt": "تبدیل صدا به متن 📝", "chat_file": "چت با فایل/عکس 🤖" } # پوسته زیبای خطای انقضای لینک به صورت کاملاً فارسی EXPIRED_HTML = """ لینک منقضی شده است | هوش مصنوعی آلفا

⚠️ لینک منقضی شده است!

متأسفانه این لینک منقضی شده است.

لطفاً از طریق ربات مجدداً روی دکمه مورد نظر کلیک کنید تا لینک جدید و فعال دریافت کنید.

برگشت به روبیکا 🤖
""" BRIDGE_HTML = """ آپلود فایل | هوش مصنوعی آلفا

آپلود سنتر آلفا 🚀

بخش: {CONTEXT_NAME}

فایل خود را اینجا انتخاب کنید تا مستقیماً و پرسرعت به ربات متصل شود.

⏳ این لینک یک‌بار مصرف است و تا ۲ دقیقه اعتبار دارد.
✅ فایل با موفقیت به ربات ارسال شد!

لطفاً این صفحه را ببندید و به روبیکا برگردید. ربات در حال پردازش است.
""" VIEWER_HTML = """ {FILE_NAME} | دانلود سنتر آلفا

{FILE_NAME}

{MEDIA_CONTENT}
دانلود مستقیم فایل 📥
قدرت گرفته از هسته هوش مصنوعی آلفا 🚀
""" def verify_signature(user_code: str, unique_token: str, ctx: str, t: int, p: str, sig: str) -> bool: """بررسی و راستی‌آزمایی امضای دیجیتال برای جلوگیری از دستکاری دستی پارامترها""" message = f"{user_code}:{unique_token}:{ctx}:{t}:{p}" expected_sig = hmac.new(SHARED_SECRET.encode(), message.encode(), hashlib.sha256).hexdigest() return hmac.compare_digest(expected_sig, sig) @router.get("/view/{encoded}") async def view_file_page(encoded: str): """رمزگشایی ریاضیاتی آدرس بله در رم و نمایش صفحه لودر بدون مصرف هارد رانفلر""" try: missing_padding = len(encoded) % 4 if missing_padding: encoded += '=' * (4 - missing_padding) data_bytes = base64.urlsafe_b64decode(encoded) data = json.loads(data_bytes.decode('utf-8')) bale_url = data.get("u") file_name = data.get("n", "file") file_type = data.get("t", "File") media_content = "" ft_lower = file_type.lower() if ft_lower in ["photo", "image"]: media_content = f'{file_name}' elif ft_lower in ["video"]: media_content = f'' elif ft_lower in ["voice", "music", "audio"]: media_content = f'' else: media_content = '
📄نمایش پیش‌نمایش برای این نوع سند در دسترس نیست.
' final_html = VIEWER_HTML.replace("{FILE_NAME}", file_name).replace("{MEDIA_CONTENT}", media_content).replace("{BALE_URL}", bale_url) return HTMLResponse(content=final_html, status_code=200) except: return HTMLResponse("

❌ فایل معتبر نیست یا لینک مخدوش شده است.

", status_code=400) @router.get("/v") @router.get("/v/") async def view_file_short(t: str = "File", id: str = ""): """🟢 لودر اختصاصی بسیار کوتاه بدون مصرف هارد و دیسک رانفلر""" try: bale_url = f"https://tapi.bale.ai/file/bot{BALE_BOT_TOKEN}/{id}" titles = { "photo": "تصویر هوش مصنوعی 🎨", "image": "تصویر هوش مصنوعی 🎨", "video": "ویدیوی ساخته شده 🎬", "voice": "صدای تولید شده 🎙️", "music": "موسیقی ساخته شده 🎵", "audio": "فایل صوتی 🎧", "file": "سند متنی آلفا 📄" } display_name = titles.get(t.lower(), "فایل خروجی آلفا 📁") media_content = "" ft_lower = t.lower() if ft_lower in ["photo", "image"]: media_content = f'{display_name}' elif ft_lower in ["video"]: media_content = f'' elif ft_lower in ["voice", "music", "audio"]: media_content = f'' else: media_content = '
📄فرمت فایل عمومی
' final_html = VIEWER_HTML.replace("{FILE_NAME}", display_name).replace("{MEDIA_CONTENT}", media_content).replace("{BALE_URL}", bale_url) return HTMLResponse(content=final_html, status_code=200) except: return HTMLResponse("

❌ ارتباط با سرور برقرار نشد.

", status_code=400) @router.get("/upload/{user_code}/{unique_token}") async def upload_web_page(user_code: str, unique_token: str, ctx: str = "unknown", b: str = "", t: int = 0, p: str = "0", sig: str = ""): """صفحه‌ای که کاربر باز میکند (همراه با بررسی اعتبار زمانی لینک و لود پویای ادیتور روبیکا)""" current_time = int(time.time()) if t == 0 or (current_time - t) > 120: # منقضی شدن لینک وب پس از ۱۲۰ ثانیه (۲ دقیقه) return HTMLResponse(content=EXPIRED_HTML, status_code=200) is_premium_verified = False if sig and verify_signature(user_code, unique_token, ctx, t, p, sig): if p == "1": is_premium_verified = True premium_status_val = "paid" if is_premium_verified else "free" file_id = f"{user_code}_{ctx}_{unique_token}" # 🟢 اگر کانتکست مربوط به ویرایش عکس بود، فایل شیک ادیتور را لود کن if ctx == "edit_image" and os.path.exists("boteditor.html"): with open("boteditor.html", "r", encoding="utf-8") as f: boteditor_content = f.read() final_html = boteditor_content.replace("{FILE_ID}", file_id).replace("{BOT_B64}", b).replace("{TIMESTAMP}", str(t)).replace("{PREMIUM_STATUS}", premium_status_val) return HTMLResponse(content=final_html, status_code=200) # 🟢 جدید: اگر کانتکست مربوط به کلون کردن صدا بود، فایل شیک بات کلون را لود کن if ctx == "voice_clone" and os.path.exists("botclone.html"): with open("botclone.html", "r", encoding="utf-8") as f: botclone_content = f.read() final_html = botclone_content.replace("{FILE_ID}", file_id).replace("{BOT_B64}", b).replace("{TIMESTAMP}", str(t)).replace("{PREMIUM_STATUS}", premium_status_val) return HTMLResponse(content=final_html, status_code=200) context_name = CONTEXT_TRANSLATIONS.get(ctx, "فایل عمومی 📁") html_content = BRIDGE_HTML.replace("{FILE_ID}", file_id).replace("{CONTEXT_NAME}", context_name).replace("{BOT_B64}", b).replace("{TIMESTAMP}", str(t)) return HTMLResponse(content=html_content, status_code=200) @router.post("/api/bridge/upload") async def bridge_upload( file_id: str = Form(...), bot_b64: str = Form(...), t: int = Form(...), prompt: str = Form(""), # دریافت خودکار پرامپت (دستور متنی) ویرایش عکس file: UploadFile = File(...) ): """دریافت فایل و پرامپت از کاربر و ارسال مستقیم به سرور هاگینگ فیس""" current_time = int(time.time()) if (current_time - t) > 120: # لغو فرآیند آپلود پس از ۲ دقیقه return JSONResponse({"status": "error", "message": "این لینک منقضی شده است. مجدداً از ربات لینک بگیرید."}) try: bot_url = base64.b64decode(bot_b64).decode('utf-8').strip() file_content = await file.read() target_url = f"{bot_url}/api/bridge/webhook" async with httpx.AsyncClient(verify=False, timeout=90.0) as client: resp = await client.post( target_url, files={"file": (file.filename, file_content, file.content_type)}, data={ "file_id": file_id, "prompt": prompt # ارسال پرامپت به وب‌هوک هاگینگ فیس } ) if resp.status_code == 200: return JSONResponse({"status": "success"}) else: return JSONResponse({"status": "error", "message": f"خطای ارتباط با هاگینگ فیس: {resp.text[:120]}"}) except Exception as e: return JSONResponse({"status": "error", "message": f"خطای داخلی رانفلر: {str(e)}"}) @router.post("/api/bridge/upload_clone") async def bridge_upload_clone( file_id: str = Form(...), bot_b64: str = Form(...), t: int = Form(...), file1: UploadFile = File(...), # صدای ورودی کاربر file2: UploadFile = File(...) # صدای مدل الگو ): """🟢 مسیر جدید: دریافتِ همزمان دو فایل صوتی برای کلون صدا و ارسال مستقیم به هاگینگ فیس""" current_time = int(time.time()) if (current_time - t) > 120: # لغو فرآیند آپلود پس از ۲ دقیقه return JSONResponse({"status": "error", "message": "این لینک منقضی شده است. مجدداً از ربات لینک بگیرید."}) try: bot_url = base64.b64decode(bot_b64).decode('utf-8').strip() # خواندن موازی بایت‌های دو فایل file1_content = await file1.read() file2_content = await file2.read() target_url = f"{bot_url}/api/bridge/webhook_clone" async with httpx.AsyncClient(verify=False, timeout=120.0) as client: resp = await client.post( target_url, files={ "file1": (file1.filename, file1_content, file1.content_type), "file2": (file2.filename, file2_content, file2.content_type) }, data={"file_id": file_id} ) if resp.status_code == 200: return JSONResponse({"status": "success"}) else: return JSONResponse({"status": "error", "message": f"خطای ارتباط با هاگینگ فیس: {resp.text[:120]}"}) except Exception as e: return JSONResponse({"status": "error", "message": f"خطای داخلی رانفلر: {str(e)}"})