""" THE Z AI — Computer Mode Server v20 "Headless-Native" ===================================================================================== إعادة كتابة كاملة من الصفر — بنية مختلفة تماماً عن v11-v13. ── لماذا إعادة كتابة كاملة؟ ── البنية القديمة (v11-v13) اعتمدت على: Xvfb (X server وهمي منفصل) + Firefox/Chromium عادي يرسم عليه + التقاط الشاشة بأداة خارجية (import/ImageMagick، أو python-xlib، أو xwd) + xdotool لكل حركة فأرة/كيبورد. هذا يعني *عمليتين منفصلتين* لكل مستخدم (Xvfb + المتصفح) + عملية subprocess ثالثة لكل لقطة شاشة واحدة، وكلها تتواصل عبر بروتوكول X11 الثقيل. على سيرفر موارد محدودة (HF Free: 2 vCPU/16GB مشتركة)، هذا يعني استهلاك ذاكرة/CPU أعلى بكثير من اللازم، وأي بطء عابر في X11 يُفسَّر كـ"فشل التقاط" فيُفعّل ريستارت تلقائي للجلسة كاملة — وهذا بالضبط ما كان يسبب الريستارت المتكرر والأخطاء. ── البنية الجديدة v20 ── Chromium بوضع "--headless=new" (الوضع الرسمي الحديث من Google) *فقط*. بدون Xvfb. بدون X server من أي نوع. Chromium يرندر الصفحة بالكامل داخلياً في الذاكرة (نفس محرك الرندر الحقيقي، بدون نافذة مرئية)، ونتحدث معه حصرياً عبر Chrome DevTools Protocol (CDP) عبر WebSocket واحد لكل مستخدم: - Page.captureScreenshot → لقطة الشاشة (JPEG) مباشرة من المتصفح نفسه، بدون subprocess خارجي، بدون فحص "هل الصورة سوداء" (headless لا يُنتج شاشات سوداء أصلاً لأنه لا يعتمد على X server قد يتعطل). - Input.dispatchMouseEvent → حركة/نقر/سحب الفأرة (بدل xdotool mousemove/click) - Input.dispatchKeyEvent → كتابة/اختصارات لوحة المفاتيح (بدل xdotool key/type) - Page.navigate → فتح رابط/تبويب جديد - Runtime.evaluate → لصق النص العربي (clipboard بديل)، تنفيذ سكربت مساعد نتيجة هذا التبسيط: - عملية واحدة فقط لكل مستخدم (Chromium) بدل عمليتين (Xvfb+متصفح). - استهلاك ذاكرة أقل بشكل ملحوظ (لا X server منفصل يُبقي framebuffer في الذاكرة). - لا وجود لمفهوم "فشل التقاط الشاشة" الذي كان يُفعّل ريستارت — لو فشل CDP فالمشكلة الوحيدة الممكنة هي أن Chromium نفسه توقف، فنُعيد تشغيله هو فقط (لا حاجة لأي "reset" أوسع يمسح بروفايلات أو Xvfb لأنه غير موجود أصلاً). - أخف تعقيداً: أداة تحكم واحدة (CDP) بدل خليط (xdotool + import + xlib + xwd). ── ما الذي أُزيل من v13 ولماذا ── أدوات "browser_get_elements / browser_click_element / browser_type_into_element / browser_find_element / browser_get_page_text / browser_execute_js / browser_navigate (CDP) / browser_scroll_to_element / browser_get_state" — هذه لم تكن تُستخدم إطلاقاً من الواجهة الأمامية (index.html) بعد فحص الكود الفعلي؛ الواجهة تُرسل فقط: screenshot, terminal, mouse_move, mouse_click, mouse_drag, keyboard_type, keyboard_hotkey, keyboard_press, scroll, clipboard_write, clipboard_read, paste, open_app, open_tab, close_tab, browser_back, browser_forward, browser_search, screen_info, reset_computer. إزالة الكود الميت قلّلت التعقيد والـ WebSocket overhead لأدوات لا أحد يستدعيها. ── التوافق مع الواجهة الأمامية ── البروتوكول (أسماء actions، أسماء حقول رسائل screenshot/terminal_result/ computer_reset/error/connected/screen_info) مطابق 100% لما تتوقعه index_plus_prompt_enhanced_v2-3.html — لا حاجة لأي تعديل في الفرونت إند. ── v20.1 — إضافات: تعدد المتصفحات + بحث رسمي بدون كابتشا ── 1) تعدد المتصفحات (Multi-Browser Fallback): عند إقلاع السيرفر، يُفحص كل مرشحي المتصفح المعروفين (Chromium بأسمائه المختلفة حسب التوزيعة، Chrome، Edge، Brave...) بتشغيل حقيقي فعلي (مو مجرد "هل موجود بالمسار") واختيار أول واحد يستجيب فعلاً على CDP. لو المتصفح المعتمد فشل لاحقاً أثناء تشغيل جلسة مستخدم حقيقية (مثلاً تحديث كسر الثنائي)، السيرفر يجرب بقية المرشحين تلقائياً ويعتمد أول بديل ناجح لبقية الجلسات — بدون أي تدخل يدوي أو إعادة نشر. 2) Brave Search API — قناة بحث رسمية بدون كابتشا إطلاقاً: المشكلة الأصلية: فتح duckduckgo.com/google.com كموقع فعلي عبر المتصفح لأجل "البحث عن معلومة" يواجه كابتشا حتمياً، لأن هذي المحركات مصممة لمنع الطلبات الآلية تحديداً. الحل: بدل زيارة الموقع، الذكاء الاصطناعي يرسل أمر terminal بصيغة search: نص الاستعلام (بدل curl معقد) فيُستدعى Brave Search API مباشرة (بدون فتح أي متصفح، بدون كابتشا، لأنه مرخّص لهذا الغرض تحديداً من مزوّده). فتح متصفح Computer Mode الفعلي يبقى محفوظاً فقط لما يحتاج المستخدم "تصفح موقع بعينه بصرياً" أو التفاعل مع صفحة معينة — بالضبط نفس فلسفة عمل وكلاء مثل Manus: بحث نظيف عبر API لسطر الأوامر أولاً، ومتصفح حقيقي فقط للتصفح الفعلي. التفعيل: عيّن متغيّر بيئة BRAVE_API_KEY (من https://api-dashboard.search.brave.com — فيه باقة مجانية 2000 استعلام/شهر) على منصة النشر (Render/HF Spaces → Settings → Environment Variables). بدون المفتاح، يستمر السيرفر بالعمل عبر مصادر API المجانية البديلة (DuckDuckGo Instant Answer, Wikipedia, HackerNews, Reddit) تلقائياً. ── v20.2 — Camoufox: محرك متصفح ثانٍ مضاد للكشف ── المشكلة: حتى مع Brave Search API، أحياناً المستخدم يطلب فعلياً "تصفّح موقع معيّن بصرياً" (مو بحث)، وبعض المواقع (خصوصاً المحمية بـ Cloudflare/ DataDome) تكتشف Chromium العادي فوراً حتى لو headless، لأنه يسرّب علامات أتمتة واضحة (navigator.webdriver=true، بصمة toString مكشوفة، إلخ). الحل: محرك بديل اختياري (Camoufox — Firefox معدَّل من مستوى C++، مفتوح المصدر ومجاني بالكامل) يُخفي هذي العلامات وينتج بصمة متصفح واقعية إحصائياً. يُفعَّل عبر أمر جديد يرسله الفرونت إند: {"action": "switch_engine", "data": {"engine": "camoufox"}} وللرجوع للمحرك الافتراضي السريع: {"action": "switch_engine", "data": {"engine": "chromium"}} بقية كل الأوامر (screenshot, mouse_click, keyboard_type...) تبقى بنفس الصيغة والبروتوكول تماماً — فقط تُنفَّذ عبر المحرك النشط للمستخدم، بدون أي تعديل مطلوب على الفرونت إند لتوجيه الأوامر بنفسها. تنبيه واقعي مهم: Camoufox يقلل ظهور الكابتشا بشكل ملحوظ، لكنه **ليس حلاً سحرياً 100%** — مواثيق Camoufox نفسها تذكر أن مواقع متقدمة جداً قد تكتشفه أحياناً عبر تحليل بصمات متقدم. كما أن سمعة IP السيرفر (Render/HF) تبقى عاملاً منفصلاً؛ Camoufox يعالج بصمة *المتصفح*، لا بصمة *الشبكة/IP*. المتطلبات: pip install camoufox[geoip] playwright، ثم playwright install firefox (أو camoufox fetch لتنزيل الثنائي المعدَّل تلقائياً عند أول استخدام، ~150-300MB لمرة واحدة). بدون هذي الحزم، أوامر switch_engine إلى camoufox تفشل بأمان وتُبقي الجلسة على Chromium. """ import asyncio import base64 import hashlib import json import os import re import subprocess import threading import time import urllib.parse from typing import Optional from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Query from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse, HTMLResponse import uvicorn try: import websockets as ws_lib _WS_LIB_AVAILABLE = True except ImportError: _WS_LIB_AVAILABLE = False print("⚠️ websockets غير مثبّتة — pip install websockets (مطلوبة لهذا الإصدار)") # ════════════════════════════════════════════════════════════════ # ── الإعدادات العامة ─────────────────────────────────────────── # ════════════════════════════════════════════════════════════════ # أقصى عدد جلسات متزامنة. رقم واقعي لسيرفر موارد محدودة (HF Free CPU # Basic: 2 vCPU/16GB مشتركة) — كل جلسة = عملية Chromium واحدة فقط الآن # (بدل Xvfb+متصفح سابقاً)، فالحمل الفعلي لكل مستخدم أخف من النصف تقريباً # مقارنة بالبنية القديمة، لكن نبقي رقماً متحفظاً بدل الافتراض المتفائل. MAX_SESSIONS = 40 CDP_PORT_BASE = 9300 SCREEN_W, SCREEN_H = 1280, 800 _sessions_lock = threading.Lock() _cdp_port_pool_lock = threading.Lock() _used_ports: set = set() def _detect_chromium() -> str: """ يكتشف أي متصفح متوفر على هذا السيرفر — يجرب كل الأسماء الشائعة (Chromium, Chrome, Chromium الرسمي بأشكاله المختلفة حسب توزيعة Linux)، مو اسم واحد بس. """ return _pick_working_browser() # كل الأسماء الشائعة لثنائيات المتصفح على توزيعات Linux المختلفة (Debian/Ubuntu، # Alpine، RPM-based) — الذكاء الاصطناعي (عبر هذه الدالة) يجرب كل واحد بالترتيب # ويستخدم أول واحد يعمل فعلياً، بدل الاعتماد على اسم واحد قد لا يوجد على صورة # الحاوية المستخدمة. _BROWSER_CANDIDATES = [ "chromium-browser", # Debian/Ubuntu "chromium", # Alpine/Arch/Fedora "google-chrome-stable", # Google Chrome الرسمي "google-chrome", "chromium-freeworld", # بعض توزيعات RPM "ungoogled-chromium", # نسخة خصوصية من Chromium "microsoft-edge-stable", # Edge (مبني على Chromium، يدعم نفس CDP) "brave-browser", # Brave (مبني على Chromium، يدعم نفس CDP) ] def _binary_exists(name: str) -> bool: r = subprocess.run(["which", name], capture_output=True, text=True) return r.returncode == 0 and bool(r.stdout.strip()) def _test_browser_launch(binary: str, timeout: float = 6.0) -> bool: """ اختبار حقيقي: يشغّل المتصفح فعلياً بوضع headless على منفذ CDP مؤقت ويتأكد أنه يستجيب — مو مجرد فحص "موجود بالمسار"، لأن أحياناً الثنائي موجود لكن مكسور (مكتبات ناقصة، إلخ). """ test_port = CDP_PORT_BASE + 9999 # منفذ مخصص للاختبار فقط، لا يتعارض مع الجلسات try: subprocess.run(["pkill", "-9", "-f", f"[r]emote-debugging-port={test_port}"], capture_output=True, timeout=3) except Exception: pass try: proc = subprocess.Popen([ binary, "--headless=new", f"--remote-debugging-port={test_port}", "--remote-debugging-address=127.0.0.1", "--disable-gpu", "--no-sandbox", "--disable-dev-shm-usage", "--no-first-run", "about:blank", ], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) except Exception: return False ok = False deadline = time.time() + timeout try: import urllib.request while time.time() < deadline: try: with urllib.request.urlopen(f"http://127.0.0.1:{test_port}/json/version", timeout=1.0) as r: if r.status == 200: ok = True break except Exception: time.sleep(0.3) finally: try: proc.kill() proc.wait(timeout=3) except Exception: pass try: subprocess.run(["pkill", "-9", "-f", f"[r]emote-debugging-port={test_port}"], capture_output=True, timeout=3) except Exception: pass return ok def _pick_working_browser() -> str: """ يجرب كل مرشحي المتصفح بالترتيب: أولاً يتأكد أن الثنائي موجود على القرص (سريع)، وإذا وجد، يختبره فعلياً بتشغيله والتأكد أنه يستجيب على CDP (اختبار حقيقي، لا افتراض). أول متصفح "حي" فعلاً يُعتمد لبقية عمر السيرفر. """ found_but_untested = [] for name in _BROWSER_CANDIDATES: if _binary_exists(name): found_but_untested.append(name) if not found_but_untested: print("❌ لا يوجد أي متصفح مثبّت على هذا السيرفر من بين المرشحين المعروفين:") print(f" {', '.join(_BROWSER_CANDIDATES)}") print(" ثبّت واحداً عبر: apt-get install chromium") return "chromium-browser" # افتراضي، حتى لو غير موجود، ليكون اسم واضح بالسجلات print(f"🔍 متصفحات موجودة على القرص: {', '.join(found_but_untested)} — يجري اختبار كل واحد فعلياً...") for name in found_but_untested: print(f" 🧪 اختبار {name} ...") if _test_browser_launch(name): print(f" ✅ {name} يعمل فعلياً — سيُعتمد لكل الجلسات") return name else: print(f" ❌ {name} موجود لكن فشل بالتشغيل الفعلي (مكتبات ناقصة أو ثنائي معطوب)") # كل شيء وُجد لكن فشل الاختبار الحي — استخدم أول موجود كخيار أخير مع تحذير واضح print(f"⚠️ كل المتصفحات الموجودة فشلت باختبار التشغيل الحي. سيُستخدم '{found_but_untested[0]}' " f"كخيار افتراضي أخير، لكن يُنصح بفحص تثبيت المتصفح على هذا السيرفر.") return found_but_untested[0] BROWSER_BIN = _detect_chromium() def _alloc_cdp_port() -> int: """يخصص منفذ CDP حر لمستخدم جديد.""" with _cdp_port_pool_lock: p = CDP_PORT_BASE while p in _used_ports: p += 1 _used_ports.add(p) return p def _release_cdp_port(port: int): with _cdp_port_pool_lock: _used_ports.discard(port) # ════════════════════════════════════════════════════════════════ # ── Camoufox — محرك متصفح بديل مضاد للكشف (اختياري، بجانب Chromium) ── # ════════════════════════════════════════════════════════════════ # لماذا محرك ثاني منفصل؟ Camoufox هو Firefox معدَّل من مستوى C++ (مو حقن # JavaScript قابل للكشف) يُخفي علامات الأتمتة (navigator.webdriver وغيرها) # ويولّد بصمة متصفح متّسقة وواقعية إحصائياً (نظام تشغيل، دقة شاشة، خطوط، # WebGL...) عبر مكتبة BrowserForge. هذا يقلل ظهور كابتشا المواقع المحمية # بشدة (Cloudflare, DataDome) بشكل ملحوظ — لكنه **ليس ضماناً مطلقاً** (حتى # موثّق Camoufox نفسه يذكر أن مواقع متقدمة قد تكتشفه أحياناً). # # الفرق الجوهري عن Chromium/CDP: واجهة Camoufox الرسمية هي Python+Playwright # (مو CDP خام)، فتحكمنا فيه يمر عبر Playwright API (page.mouse, page.keyboard, # page.screenshot) بدل رسائل CDP JSON المباشرة المستخدمة مع Chromium. لذلك # بنيناه كـ"محرك بديل" مستقل بجانب CDPSession، بنفس بروتوكول الأكشنات الخارجي # (screenshot/mouse_click/keyboard_type...) حتى لا يتأثر الفرونت إند إطلاقاً. # # التفعيل: تلقائي بالكامل — لا حاجة لأي إعداد. Camoufox حزمة بايثون عادية # (pip install camoufox[geoip] + playwright)، أول استخدام يُنزّل ثنائي # Firefox المعدَّل تلقائياً (~150-300MB مرة واحدة فقط، يُخزَّن ويُعاد استخدامه). try: from camoufox.async_api import AsyncCamoufox _CAMOUFOX_LIB_AVAILABLE = True except ImportError: _CAMOUFOX_LIB_AVAILABLE = False print("ℹ️ مكتبة camoufox غير مثبّتة — pip install camoufox playwright لتفعيل المحرك المضاد للكشف") CAMOUFOX_ENABLED = os.environ.get("CAMOUFOX_ENABLED", "1").strip() not in ("0", "false", "no") class CamoufoxSession: """ جلسة Camoufox واحدة لكل مستخدم — بديل لـ CDPSession لكن مبني فوق Playwright بدل CDP خام. يوفر نفس نوع العمليات (screenshot, mouse, keyboard, navigate) بواجهة موحّدة (dict نتائج) لتُستهلك من نفس معالج الأكشنات (handle_action) دون تغيير بروتوكول الفرونت إند. """ def __init__(self, user_id: str): self.user_id = user_id self._cm = None # AsyncCamoufox context manager instance self._browser = None # Playwright Browser self._context = None # Playwright BrowserContext self.page = None # Playwright Page الحالية self._mouse_x = SCREEN_W // 2 self._mouse_y = SCREEN_H // 2 @property def alive(self) -> bool: return self.page is not None and not self.page.is_closed() async def start(self) -> bool: if not _CAMOUFOX_LIB_AVAILABLE: print("[camoufox] ❌ المكتبة غير مثبَّتة على هذا السيرفر") return False try: profile = _profile_dir("camoufox_" + self.user_id) self._cm = AsyncCamoufox( headless=True, # Camoufox يدعم headless حقيقي بأمان (يُخفي علامات الكشف حتى في هذا الوضع) humanize=True, # حركة فأرة شبيهة بالبشر (منحنيات Bezier بدل خطوط مستقيمة) geoip=False, # لا نحتاج ضبط جغرافي تلقائي عبر بروكسي هنا persistent_context=True, user_data_dir=profile, window=(SCREEN_W, SCREEN_H), ) self._context = await self._cm.__aenter__() # في persistent_context=True، الكائن المُعاد هو BrowserContext مباشرة pages = self._context.pages self.page = pages[0] if pages else await self._context.new_page() await self.page.set_viewport_size({"width": SCREEN_W, "height": SCREEN_H}) print(f"[camoufox] ✅ جلسة جديدة لـ '{self.user_id}' جاهزة") return True except Exception as e: print(f"[camoufox] ❌ فشل بدء الجلسة: {e}") await self.close() return False async def close(self): try: if self._cm: await self._cm.__aexit__(None, None, None) except Exception: pass self._cm = None self._context = None self.page = None async def screenshot_b64(self) -> "str | None": if not self.alive: return None try: raw = await self.page.screenshot(type="jpeg", quality=75) return base64.b64encode(raw).decode() except Exception as e: print(f"[camoufox] ❌ فشل التقاط الشاشة: {e}") return None async def mouse_move(self, x: int, y: int): if self.alive: await self.page.mouse.move(x, y) self._mouse_x, self._mouse_y = x, y async def mouse_click(self, x: int, y: int, button: str = "left", double: bool = False): if not self.alive: return await self.page.mouse.move(x, y) if double: await self.page.mouse.dblclick(x, y, button=button) else: await self.page.mouse.click(x, y, button=button) self._mouse_x, self._mouse_y = x, y async def mouse_drag(self, x1: int, y1: int, x2: int, y2: int): if not self.alive: return await self.page.mouse.move(x1, y1) await self.page.mouse.down() await self.page.mouse.move(x2, y2, steps=10) await self.page.mouse.up() self._mouse_x, self._mouse_y = x2, y2 async def scroll(self, x: int, y: int, delta_y: int): if not self.alive: return await self.page.mouse.move(x, y) await self.page.mouse.wheel(0, delta_y) async def type_text(self, text: str): if self.alive: await self.page.keyboard.type(text, delay=25) # تأخير خفيف بين الحروف — أكثر واقعية بشرياً async def key_press(self, key_name: str): if self.alive: await self.page.keyboard.press(key_name) async def hotkey(self, keys: list): if not self.alive: return combo = "+".join(keys) await self.page.keyboard.press(combo) async def navigate(self, url: str) -> bool: if not self.alive: return False try: await self.page.goto(url, timeout=20000, wait_until="domcontentloaded") return True except Exception as e: print(f"[camoufox] ⚠️ فشل التنقّل إلى {url}: {e}") return False async def go_back(self): if self.alive: try: await self.page.go_back(timeout=8000) except Exception: pass async def go_forward(self): if self.alive: try: await self.page.go_forward(timeout=8000) except Exception: pass # جلسات Camoufox منفصلة عن جلسات Chromium/CDP — مفتاحها user_id أيضاً، لكن # قاموس مستقل لأن محرك المستخدم قد يتبدّل بين الاثنين أثناء الجلسة نفسها. _camoufox_sessions: dict = {} _camoufox_lock = threading.Lock() async def get_camoufox_session(user_id: str) -> "CamoufoxSession | None": """يُرجع جلسة Camoufox موجودة أو ينشئ واحدة جديدة عند الحاجة.""" if not CAMOUFOX_ENABLED or not _CAMOUFOX_LIB_AVAILABLE: return None with _camoufox_lock: sess = _camoufox_sessions.get(user_id) if sess and sess.alive: return sess sess = CamoufoxSession(user_id) ok = await sess.start() if not ok: return None with _camoufox_lock: _camoufox_sessions[user_id] = sess return sess async def destroy_camoufox_session(user_id: str): with _camoufox_lock: sess = _camoufox_sessions.pop(user_id, None) if sess: await sess.close() # ════════════════════════════════════════════════════════════════ # ── CDP Session — الاتصال المباشر بـ Chromium عبر DevTools Protocol ── # ════════════════════════════════════════════════════════════════ class CDPSession: """ اتصال WebSocket واحد بجلسة Chromium headless واحدة. كل الأوامر (screenshot, mouse, keyboard, navigate) تمر من هنا. اتصال واحد يُعاد استخدامه طوال حياة الجلسة بدل فتح اتصال جديد لكل أمر. """ def __init__(self, port: int): self.port = port self._ws = None self._msg_id = 0 self._pending: dict = {} self._recv_task: Optional[asyncio.Task] = None self._send_lock = asyncio.Lock() self._target_id = None async def connect(self, timeout: float = 6.0) -> bool: if not _WS_LIB_AVAILABLE: return False # جرّب عدة مرات لأن Chromium قد يحتاج لحظة بعد الإقلاع قبل فتح منفذ CDP deadline = time.time() + timeout tabs = None while time.time() < deadline: try: import urllib.request with urllib.request.urlopen( f"http://127.0.0.1:{self.port}/json/list", timeout=1.5 ) as r: tabs = json.loads(r.read()) break except Exception: await asyncio.sleep(0.25) if tabs is None: print(f"[CDP:{self.port}] ❌ Chromium لا يستجيب على منفذ CDP") return False ws_url = None for t in tabs: if t.get("type") == "page": ws_url = t.get("webSocketDebuggerUrl") self._target_id = t.get("id") break if not ws_url: print(f"[CDP:{self.port}] ❌ لا يوجد تبويب صفحة متاح") return False try: self._ws = await asyncio.wait_for( ws_lib.connect(ws_url, ping_interval=15, ping_timeout=10, max_size=64 * 1024 * 1024), timeout=timeout, ) except Exception as e: print(f"[CDP:{self.port}] ❌ فشل اتصال WebSocket: {e}") return False self._recv_task = asyncio.create_task(self._recv_loop()) # فعّل النطاقات المطلوبة await self.send("Page.enable", timeout=5.0) await self.send("Runtime.enable", timeout=5.0) await self.send("DOM.enable", timeout=5.0) # اضبط حجم الشاشة الافتراضي عبر Emulation (headless بدون نافذة فعلية، # هذا يحدد أبعاد الرندر الداخلي) await self.send("Emulation.setDeviceMetricsOverride", { "width": SCREEN_W, "height": SCREEN_H, "deviceScaleFactor": 1, "mobile": False, }, timeout=5.0) print(f"[CDP:{self.port}] ✅ متصل بجلسة Chromium") return True async def _recv_loop(self): try: async for raw in self._ws: try: msg = json.loads(raw) except Exception: continue mid = msg.get("id") if mid is not None and mid in self._pending: fut = self._pending.pop(mid) if not fut.done(): fut.set_result(msg) except Exception: pass finally: for fut in self._pending.values(): if not fut.done(): fut.set_exception(ConnectionError("CDP disconnected")) self._pending.clear() self._ws = None async def send(self, method: str, params: dict = None, timeout: float = 10.0): if not self._ws: return None async with self._send_lock: self._msg_id += 1 mid = self._msg_id fut = asyncio.get_event_loop().create_future() self._pending[mid] = fut try: await self._ws.send(json.dumps({"id": mid, "method": method, "params": params or {}})) return await asyncio.wait_for(fut, timeout=timeout) except Exception: self._pending.pop(mid, None) return None @property def alive(self) -> bool: return self._ws is not None async def close(self): if self._recv_task: self._recv_task.cancel() if self._ws: try: await self._ws.close() except Exception: pass self._ws = None # ════════════════════════════════════════════════════════════════ # ── تشغيل/إيقاف عملية Chromium headless ──────────────────────── # ════════════════════════════════════════════════════════════════ def _profile_dir(user_id: str) -> str: safe = re.sub(r"[^a-zA-Z0-9_.-]", "_", user_id) or "anon" d = os.path.expanduser(f"~/.zai_chromium_profiles/{safe}") os.makedirs(d, exist_ok=True) return d def _launch_browser_process(binary: str, port: int, profile: str) -> "subprocess.Popen | None": cmd = [ binary, "--headless=new", # الوضع الرسمي الحديث — لا حاجة لأي display f"--remote-debugging-port={port}", "--remote-debugging-address=127.0.0.1", "--disable-gpu", "--no-sandbox", "--disable-dev-shm-usage", "--disable-extensions", "--disable-notifications", "--disable-popup-blocking", "--disable-translate", "--disable-sync", "--disable-background-timer-throttling", "--disable-backgrounding-occluded-windows", "--disable-renderer-backgrounding", "--disable-features=Translate,BackForwardCache", "--no-first-run", "--no-default-browser-check", "--mute-audio", "--hide-scrollbars", f"--window-size={SCREEN_W},{SCREEN_H}", f"--user-data-dir={profile}", "--js-flags=--max-old-space-size=256", # سقف ذاكرة JS — يمنع تبويب ثقيل من استهلاك كل الرام "about:blank", ] try: proc = subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) return proc except FileNotFoundError: return None except Exception as e: print(f"[chromium] ❌ فشل تشغيل {binary}: {e}") return None def _start_chromium(port: int, user_id: str) -> "subprocess.Popen | None": """ يُشغّل متصفحاً بوضع headless=new على منفذ CDP مخصص. لا Xvfb، لا DISPLAY، لا نافذة — الرندر بالكامل داخلي (in-process) لمحرك المتصفح نفسه. ذكاء إضافي: لو المتصفح الحالي (BROWSER_BIN) فشل بالإقلاع فعلياً وقت التشغيل الحقيقي (مو وقت الفحص الأولي)، يجرّب بقية المتصفحات المرشحة تلقائياً بدل الفشل الكامل، ويحدّث BROWSER_BIN للمتصفح اللي نجح ليُستخدم مباشرة بالجلسات القادمة (بدون إعادة تجربة كل المتصفحات الفاشلة من جديد كل مرة). """ global BROWSER_BIN profile = _profile_dir(user_id) try: subprocess.run(["pkill", "-9", "-f", f"[r]emote-debugging-port={port}"], capture_output=True, timeout=3) time.sleep(0.2) except Exception: pass proc = _launch_browser_process(BROWSER_BIN, port, profile) if proc is not None: print(f"[chromium] 🚀 بدأ '{BROWSER_BIN}' (PID {proc.pid}) على منفذ CDP {port} — headless=new") return proc print(f"[chromium] ⚠️ فشل '{BROWSER_BIN}' فعلياً — أجرب بقية المتصفحات المعروفة تلقائياً...") for alt in _BROWSER_CANDIDATES: if alt == BROWSER_BIN or not _binary_exists(alt): continue proc = _launch_browser_process(alt, port, profile) if proc is not None: print(f"[chromium] ✅ نجح البديل '{alt}' — سيُعتمد لبقية الجلسات من الآن") BROWSER_BIN = alt return proc print(f"[chromium] ❌ فشلت كل المتصفحات المتاحة على هذا السيرفر") return None def _kill_chromium(proc: "subprocess.Popen | None", port: int): if proc: try: proc.kill() proc.wait(timeout=3) except Exception: pass try: subprocess.run(["pkill", "-9", "-f", f"[r]emote-debugging-port={port}"], capture_output=True, timeout=3) except Exception: pass # ════════════════════════════════════════════════════════════════ # ── إدارة الجلسات (مستخدم → CDPSession + Chromium process) ───── # ════════════════════════════════════════════════════════════════ # user_id → { # "port": int, "proc": Popen, "cdp": CDPSession|None, # "active_ws": WebSocket|None, "created": float, # "consecutive_failures": int, "last_frame_hash": str, # "last_mouse_x": int, "last_mouse_y": int, # } _sessions: dict = {} async def get_or_create_session(user_id: str, ws: WebSocket) -> dict: with _sessions_lock: sess = _sessions.get(user_id) if sess: sess["active_ws"] = ws print(f"[session] 🔁 إعادة اتصال '{user_id}' على منفذ {sess['port']}") return sess if len(_sessions) >= MAX_SESSIONS: # أعد استخدام أقدم جلسة غير متصلة بدل رفض المستخدم الجديد idle = [(uid, s) for uid, s in _sessions.items() if not s.get("active_ws")] if idle: oldest_uid = min(idle, key=lambda t: t[1]["created"])[0] old = _sessions.pop(oldest_uid) _kill_chromium(old.get("proc"), old["port"]) _release_cdp_port(old["port"]) print(f"[session] ♻️ أُزيلت جلسة خاملة قديمة '{oldest_uid}' لإفساح المجال") port = _alloc_cdp_port() sess = { "user_id": user_id, "port": port, "proc": None, "cdp": None, "active_ws": ws, "created": time.time(), "consecutive_failures": 0, "last_frame_hash": "", "last_mouse_x": SCREEN_W // 2, "last_mouse_y": SCREEN_H // 2, "shot_lock": asyncio.Lock(), "engine": "chromium", # "chromium" (افتراضي، عبر CDP) أو "camoufox" (مضاد للكشف، عبر Playwright) } _sessions[user_id] = sess print(f"[session] ✅ مستخدم جديد '{user_id}' → منفذ CDP {port}") return sess async def ensure_browser_running(sess: dict) -> bool: """يتأكد أن Chromium + اتصال CDP يعملان لهذه الجلسة. يُشغّلهما إذا لزم.""" if sess.get("cdp") and sess["cdp"].alive: return True if not sess.get("proc") or sess["proc"].poll() is not None: proc = await asyncio.to_thread(_start_chromium, sess["port"], sess["user_id"]) sess["proc"] = proc if proc is None: return False cdp = CDPSession(sess["port"]) ok = await cdp.connect(timeout=8.0) if ok: sess["cdp"] = cdp return True return False async def reset_session(user_id: str) -> bool: """يُعيد تشغيل Chromium بالكامل لهذا المستخدم — الإصلاح الشامل الوحيد المطلوب الآن، بما أنه لا يوجد Xvfb أو ملفات جلسة X منفصلة يجب تنظيفها.""" with _sessions_lock: sess = _sessions.get(user_id) if not sess: return False proc, port = sess.get("proc"), sess["port"] old_cdp = sess.get("cdp") if old_cdp: await old_cdp.close() await asyncio.to_thread(_kill_chromium, proc, port) new_proc = await asyncio.to_thread(_start_chromium, port, user_id) new_cdp = CDPSession(port) ok = await new_cdp.connect(timeout=8.0) if new_proc else False with _sessions_lock: if user_id in _sessions: _sessions[user_id]["proc"] = new_proc _sessions[user_id]["cdp"] = new_cdp if ok else None _sessions[user_id]["consecutive_failures"] = 0 _sessions[user_id]["last_frame_hash"] = "" print(f"[reset] {'✅' if ok else '❌'} إعادة ضبط '{user_id}' على منفذ {port}") return ok async def destroy_session_ws(user_id: str, ws: WebSocket): """عند انقطاع WebSocket — نُبقي Chromium شغالاً (الجلسة محفوظة) وفقط نزيل الاتصال.""" with _sessions_lock: sess = _sessions.get(user_id) if sess and sess.get("active_ws") is ws: sess["active_ws"] = None print(f"[session] 📴 '{user_id}' قطع الاتصال (الجلسة محفوظة)") async def destroy_session_full(user_id: str): with _sessions_lock: sess = _sessions.pop(user_id, None) if sess: if sess.get("cdp"): await sess["cdp"].close() await asyncio.to_thread(_kill_chromium, sess.get("proc"), sess["port"]) _release_cdp_port(sess["port"]) await destroy_camoufox_session(user_id) # ════════════════════════════════════════════════════════════════ # ── لقطة الشاشة — عبر CDP مباشرة (بدون subprocess خارجي) ──────── # ════════════════════════════════════════════════════════════════ async def cdp_screenshot(sess: dict, quality: int = 75) -> "tuple[str, int, int] | None": """ يلتقط لقطة شاشة عبر Page.captureScreenshot — أمر CDP واحد يرجع JPEG base64 جاهز مباشرة من محرك الرندر الداخلي لـ Chromium. لا يوجد X server يمكن أن "يتجمد"، لا حاجة لفحص "هل الصورة سوداء"، ولا عملية subprocess خارجية (import/xwd) قد تفشل بسبب أدوات مفقودة على السيرفر. يُعيد (base64_jpeg, width, height) أو None عند الفشل. """ cdp = sess.get("cdp") if not cdp or not cdp.alive: return None result = await cdp.send("Page.captureScreenshot", { "format": "jpeg", "quality": quality, "fromSurface": True, "captureBeyondViewport": False, }, timeout=8.0) if not result: return None data = result.get("result", {}).get("data") if not data: return None return data, SCREEN_W, SCREEN_H def _draw_mouse_marker_on_jpeg(b64_jpeg: str, mx: int, my: int) -> str: """يرسم علامة صغيرة لموقع الفأرة فوق الصورة (مطلوبة بصرياً للذكاء الاصطناعي ليرى أين المؤشر). العملية خفيفة (رسم فقط، بدون resize) فتُبقي زمن كل لقطة منخفضاً.""" try: from PIL import Image, ImageDraw import io as _io raw = base64.b64decode(b64_jpeg) img = Image.open(_io.BytesIO(raw)).convert("RGB") draw = ImageDraw.Draw(img) r = 9 draw.ellipse([(mx - r, my - r), (mx + r, my + r)], outline=(255, 60, 60), width=2) draw.line([(mx - 14, my), (mx + 14, my)], fill=(255, 60, 60), width=1) draw.line([(mx, my - 14), (mx, my + 14)], fill=(255, 60, 60), width=1) buf = _io.BytesIO() img.save(buf, format="JPEG", quality=80, optimize=True) return base64.b64encode(buf.getvalue()).decode() except Exception: return b64_jpeg # لو فشل الرسم لأي سبب، أعد الصورة الأصلية بدل الفشل الكامل def _grid_variant(b64_jpeg: str, grid_step: int, color: tuple, label_every: int) -> str: """ ينتج نسخة من الصورة مع شبكة إحداثيات مرسومة فوقها — تساعد الذكاء الاصطناعي على تحديد إحداثيات دقيقة للنقر. نفس فكرة v13 لكن مبنية فوق JPEG الجاهز من CDP مباشرة بدل إعادة الرسم من raw pixels. """ try: from PIL import Image, ImageDraw import io as _io raw = base64.b64decode(b64_jpeg) img = Image.open(_io.BytesIO(raw)).convert("RGB") draw = ImageDraw.Draw(img, "RGBA") w, h = img.size major = (*color, 130) minor = (*color, 40) x = grid_step while x < w: is_major = x % label_every == 0 draw.line([(x, 0), (x, h)], fill=major if is_major else minor, width=1) if is_major: draw.rectangle([(x + 1, 2), (x + 34, 15)], fill=(0, 0, 0, 175)) draw.text((x + 2, 3), str(x), fill=color) x += grid_step y = grid_step while y < h: is_major = y % label_every == 0 draw.line([(0, y), (w, y)], fill=major if is_major else minor, width=1) if is_major: draw.rectangle([(2, y + 1), (38, y + 14)], fill=(0, 0, 0, 175)) draw.text((3, y + 2), str(y), fill=color) y += grid_step final = img.convert("RGB") buf = _io.BytesIO() final.save(buf, format="JPEG", quality=78, optimize=True) return base64.b64encode(buf.getvalue()).decode() except Exception: return b64_jpeg async def capture_full(sess: dict, quality: int = 75) -> dict: """ يلتقط لقطة واحدة وينتج منها: نسخة نظيفة + نسخة بعلامة الفأرة + شبكتان إحداثيات (عادية ودقيقة) — بنفس تنسيق حقول الرد الذي تتوقعه الواجهة الأمامية (data, data_grid, data_grid2, data_grid3, width, height, mouse_x, mouse_y). يعمل مع كلا المحركين حسب sess["engine"]: "chromium" (افتراضي) → عبر CDP مباشرة (cdp_screenshot) "camoufox" → عبر Playwright screenshot (أبطأ قليلاً لكن مضاد للكشف) """ if sess.get("engine") == "camoufox": cam = await get_camoufox_session(sess["user_id"]) if not cam: return {"data": "", "data_grid": "", "data_grid2": "", "data_grid3": "", "width": SCREEN_W, "height": SCREEN_H, "mouse_x": sess.get("last_mouse_x", 0), "mouse_y": sess.get("last_mouse_y", 0), "is_placeholder": True} raw_b64 = await cam.screenshot_b64() if not raw_b64: return {"data": "", "data_grid": "", "data_grid2": "", "data_grid3": "", "width": SCREEN_W, "height": SCREEN_H, "mouse_x": sess.get("last_mouse_x", 0), "mouse_y": sess.get("last_mouse_y", 0), "is_placeholder": True} w, h = SCREEN_W, SCREEN_H mx, my = sess.get("last_mouse_x", w // 2), sess.get("last_mouse_y", h // 2) def _build_cam(): clean = _draw_mouse_marker_on_jpeg(raw_b64, mx, my) grid1 = _grid_variant(clean, 50, (0, 255, 180), 100) grid2 = _grid_variant(clean, 20, (255, 60, 60), 60) return clean, grid1, grid2 clean, grid1, grid2 = await asyncio.to_thread(_build_cam) return {"data": clean, "data_grid": grid1, "data_grid2": grid2, "data_grid3": grid2, "width": w, "height": h, "mouse_x": mx, "mouse_y": my, "is_placeholder": False} shot = await cdp_screenshot(sess, quality) if not shot: return {"data": "", "data_grid": "", "data_grid2": "", "data_grid3": "", "width": SCREEN_W, "height": SCREEN_H, "mouse_x": sess.get("last_mouse_x", 0), "mouse_y": sess.get("last_mouse_y", 0), "is_placeholder": True} raw_b64, w, h = shot mx, my = sess.get("last_mouse_x", w // 2), sess.get("last_mouse_y", h // 2) def _build(): clean = _draw_mouse_marker_on_jpeg(raw_b64, mx, my) grid1 = _grid_variant(clean, 50, (0, 255, 180), 100) grid2 = _grid_variant(clean, 20, (255, 60, 60), 60) return clean, grid1, grid2 clean, grid1, grid2 = await asyncio.to_thread(_build) return { "data": clean, "data_grid": grid1, "data_grid2": grid2, "data_grid3": grid2, "width": w, "height": h, "mouse_x": mx, "mouse_y": my, "is_placeholder": False, } def _frame_hash(data: str) -> str: return hashlib.md5(data.encode()).hexdigest() # ════════════════════════════════════════════════════════════════ # ── فأرة/كيبورد — عبر CDP Input domain (بدون xdotool/X server) ── # ════════════════════════════════════════════════════════════════ async def cdp_mouse_move(sess: dict, x: int, y: int): cdp = sess.get("cdp") if not cdp or not cdp.alive: return await cdp.send("Input.dispatchMouseEvent", { "type": "mouseMoved", "x": x, "y": y, }, timeout=4.0) sess["last_mouse_x"], sess["last_mouse_y"] = x, y async def cdp_mouse_click(sess: dict, x: int, y: int, button: str = "left", double: bool = False): cdp = sess.get("cdp") if not cdp or not cdp.alive: return await cdp_mouse_move(sess, x, y) click_count = 2 if double else 1 await cdp.send("Input.dispatchMouseEvent", { "type": "mousePressed", "x": x, "y": y, "button": button, "clickCount": click_count, }, timeout=4.0) await asyncio.sleep(0.04) await cdp.send("Input.dispatchMouseEvent", { "type": "mouseReleased", "x": x, "y": y, "button": button, "clickCount": click_count, }, timeout=4.0) async def cdp_mouse_drag(sess: dict, x1: int, y1: int, x2: int, y2: int, steps: int = 8): cdp = sess.get("cdp") if not cdp or not cdp.alive: return await cdp_mouse_move(sess, x1, y1) await cdp.send("Input.dispatchMouseEvent", { "type": "mousePressed", "x": x1, "y": y1, "button": "left", "clickCount": 1, }, timeout=4.0) for i in range(1, steps + 1): ix = x1 + (x2 - x1) * i // steps iy = y1 + (y2 - y1) * i // steps await cdp.send("Input.dispatchMouseEvent", { "type": "mouseMoved", "x": ix, "y": iy, "button": "left", }, timeout=3.0) await asyncio.sleep(0.02) await cdp.send("Input.dispatchMouseEvent", { "type": "mouseReleased", "x": x2, "y": y2, "button": "left", "clickCount": 1, }, timeout=4.0) sess["last_mouse_x"], sess["last_mouse_y"] = x2, y2 async def cdp_scroll(sess: dict, x: int, y: int, delta_y: int): cdp = sess.get("cdp") if not cdp or not cdp.alive: return await cdp.send("Input.dispatchMouseEvent", { "type": "mouseWheel", "x": x, "y": y, "deltaX": 0, "deltaY": delta_y, }, timeout=4.0) # خريطة أسماء مفاتيح شائعة → أكواد CDP (windowsVirtualKeyCode + key) _KEY_MAP = { "Return": (13, "Enter"), "Enter": (13, "Enter"), "Tab": (9, "Tab"), "Escape": (27, "Escape"), "BackSpace": (8, "Backspace"), "Delete": (46, "Delete"), "Left": (37, "ArrowLeft"), "Right": (39, "ArrowRight"), "Up": (38, "ArrowUp"), "Down": (40, "ArrowDown"), "Home": (36, "Home"), "End": (35, "End"), "space": (32, " "), "F5": (116, "F5"), } async def cdp_key_press(sess: dict, key_name: str): cdp = sess.get("cdp") if not cdp or not cdp.alive: return code, key = _KEY_MAP.get(key_name, (0, key_name)) params = {"type": "keyDown", "key": key} if code: params["windowsVirtualKeyCode"] = code params["nativeVirtualKeyCode"] = code await cdp.send("Input.dispatchKeyEvent", params, timeout=4.0) params["type"] = "keyUp" await cdp.send("Input.dispatchKeyEvent", params, timeout=4.0) async def cdp_hotkey(sess: dict, keys: list): """اختصار متعدد المفاتيح مثل ['ctrl','l'] أو ['alt','Left'].""" cdp = sess.get("cdp") if not cdp or not cdp.alive: return modifiers = 0 mod_map = {"alt": 1, "ctrl": 2, "control": 2, "meta": 4, "shift": 8} plain_keys = [] for k in keys: kl = k.lower() if kl in mod_map: modifiers |= mod_map[kl] else: plain_keys.append(k) for k in plain_keys: code, key = _KEY_MAP.get(k, (ord(k[0].upper()) if len(k) == 1 else 0, k)) await cdp.send("Input.dispatchKeyEvent", { "type": "rawKeyDown", "key": key, "modifiers": modifiers, "windowsVirtualKeyCode": code, "nativeVirtualKeyCode": code, }, timeout=4.0) await cdp.send("Input.dispatchKeyEvent", { "type": "keyUp", "key": key, "modifiers": modifiers, "windowsVirtualKeyCode": code, "nativeVirtualKeyCode": code, }, timeout=4.0) async def cdp_type_text(sess: dict, text: str): """ يكتب نصاً عبر Input.insertText — يدعم كل اللغات (عربي/إنجليزي/فرنسي) بشكل أصلي عبر CDP نفسه، بدون حاجة لـ xclip أو حيلة clipboard+paste التي كانت ضرورية سابقاً بسبب قيود xdotool مع النصوص غير اللاتينية. """ cdp = sess.get("cdp") if not cdp or not cdp.alive: return await cdp.send("Input.insertText", {"text": text}, timeout=6.0) async def cdp_navigate(sess: dict, url: str) -> bool: cdp = sess.get("cdp") if not cdp or not cdp.alive: return False result = await cdp.send("Page.navigate", {"url": url}, timeout=15.0) return bool(result and "error" not in result.get("result", {})) async def cdp_clipboard_write(sess: dict, text: str): """ينسخ نصاً لحافظة afterward-paste داخلية عبر متغيّر JS بسيط (بديل خفيف لـ xclip، لأنه لا يوجد X server هنا أصلاً تُبنى عليه حافظة نظام حقيقية).""" cdp = sess.get("cdp") if not cdp or not cdp.alive: return js = f"window.__zclip = {json.dumps(text)};" await cdp.send("Runtime.evaluate", {"expression": js}, timeout=4.0) async def cdp_clipboard_read(sess: dict) -> str: cdp = sess.get("cdp") if not cdp or not cdp.alive: return "" result = await cdp.send("Runtime.evaluate", { "expression": "window.__zclip || ''", "returnByValue": True, }, timeout=4.0) if not result: return "" return result.get("result", {}).get("result", {}).get("value", "") or "" async def cdp_get_screen_info(sess: dict) -> dict: return {"width": SCREEN_W, "height": SCREEN_H, "mouse_x": sess.get("last_mouse_x", 0), "mouse_y": sess.get("last_mouse_y", 0)} # ════════════════════════════════════════════════════════════════ # ── SafeSearch — نفس منطق v13 محفوظاً كما هو (يعمل بشكل جيد) ──── # ════════════════════════════════════════════════════════════════ _BING_RE = re.compile(r"https?://(?:www\.)?bing\.com[^\s'\"]*") _DDG_RE = re.compile(r"https?://(?:www\.)?duckduckgo\.com[^\s'\"]*") _GOOG_RE = re.compile(r"https?://(?:www\.)?google\.[a-z.]+[^\s'\"]*") def _safe_search(text: str) -> str: def _bing(m): u = m.group(0) return re.sub(r"adlt=\w+", "adlt=strict", u) if "adlt=" in u else u + ("&" if "?" in u else "?") + "adlt=strict" def _ddg(m): u = m.group(0) return re.sub(r"kp=\d", "kp=1", u) if "kp=" in u else u + ("&" if "?" in u else "?") + "kp=1" def _goog(m): u = m.group(0) return re.sub(r"safe=\w+", "safe=strict", u) if "safe=" in u else u + ("&" if "?" in u else "?") + "safe=strict" return _GOOG_RE.sub(_goog, _DDG_RE.sub(_ddg, _BING_RE.sub(_bing, text))) # ════════════════════════════════════════════════════════════════ # ── Terminal — تنفيذ أوامر bash (منفصل تماماً عن Chromium) ────── # ════════════════════════════════════════════════════════════════ _terminal_sem = asyncio.Semaphore(6) _PKILL_RE = re.compile(r"\b(pkill|killall)\s+(-9\s+)?-f\s+(['\"]?)([a-zA-Z0-9_./-]+)\3") def _sanitize_pkill(cmd: str) -> str: def _fix(m): tool, d9, _q, p = m.group(1), m.group(2) or "", m.group(3), m.group(4) sp = f"[{p[0]}]{p[1:]}" if len(p) > 1 else p return f"{tool} {d9}-f '{sp}'" return _PKILL_RE.sub(_fix, cmd) async def run_cmd(cmd: str, timeout: int = 60) -> dict: cmd = _sanitize_pkill(_safe_search(cmd)) async with _terminal_sem: env = {**os.environ, "PYTHONIOENCODING": "utf-8", "LANG": "en_US.UTF-8"} def _exec(): try: r = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=timeout, env=env, executable="/bin/bash") return {"stdout": r.stdout[-15000:], "stderr": r.stderr[-3000:], "returncode": r.returncode} except subprocess.TimeoutExpired: return {"stdout": "", "stderr": f"⏱️ Timeout {timeout}s", "returncode": -1} except Exception as e: return {"stdout": "", "stderr": str(e), "returncode": -1} return await asyncio.to_thread(_exec) def _search_sources(query: str) -> list: q = urllib.parse.quote_plus(query) return [ {"name": "DuckDuckGo Instant", "cmd": f"curl -s --max-time 15 'https://api.duckduckgo.com/?q={q}&format=json&no_html=1&skip_disambig=1' | python3 -c \"import sys,json;d=json.load(sys.stdin);a=d.get('AbstractText','');r=d.get('RelatedTopics',[]);print('ANS:',a or 'none');[print('-',x.get('Text','')[:200]) for x in r[:6] if isinstance(x,dict)]\""}, {"name": "Wikipedia EN", "cmd": f"curl -s --max-time 12 'https://en.wikipedia.org/api/rest_v1/page/summary/{q}' | python3 -c \"import sys,json;d=json.load(sys.stdin);print(d.get('title','')+'\\n'+d.get('extract','')[:1200])\""}, {"name": "DuckDuckGo HTML", "cmd": f"curl -sL --max-time 15 -H 'User-Agent: Mozilla/5.0' 'https://html.duckduckgo.com/html/?q={q}' | python3 -c \"import sys,re;h=sys.stdin.read();s=re.findall(r'class=.result__snippet[^>]*>(.*?)',h,re.DOTALL);clean=lambda x:re.sub('<[^>]+>','',x).strip();[print(str(i+1)+'. '+clean(x)[:200]) for i,x in enumerate(s[:7])]\""}, {"name": "HackerNews", "cmd": f"curl -s --max-time 12 'https://hn.algolia.com/api/v1/search?query={q}&hitsPerPage=6&tags=story' | python3 -c \"import sys,json;d=json.load(sys.stdin);[print(str(i+1)+'. '+h.get('title','')+' Pts:'+str(h.get('points',0))) for i,h in enumerate(d.get('hits',[])[:5])]\""}, {"name": "Reddit", "cmd": f"curl -sL --max-time 12 -H 'Accept: application/json' 'https://www.reddit.com/search.json?q={q}&sort=new&limit=6' | python3 -c \"import sys,json;d=json.load(sys.stdin);[print(str(i+1)+'. '+p['data'].get('title','')[:160]) for i,p in enumerate(d.get('data',{{}}).get('children',[])[:5])]\""}, ] def _result_empty(s: str) -> bool: if not s or len(s.strip()) < 15: return True return "ans: none" in s.lower() # ════════════════════════════════════════════════════════════════ # ── Brave Search API — قناة البحث الرسمية المصرَّح بها ────────── # ════════════════════════════════════════════════════════════════ # على عكس زيارة duckduckgo.com/google.com كموقع فعلي عبر المتصفح (اللي يواجه # كابتشا لأن هذي المحركات تكره الطلبات الآلية)، Brave Search API مصمم أصلاً # ليُستخدم برمجياً — بدون كابتشا إطلاقاً، لأنه مرخّص لهذا الغرض تحديداً. # هذا بالضبط أسلوب Manus: بحث نظيف عبر API/سطر أوامر يرجع نتائج نصية مباشرة، # فتح متصفح فعلي يُحفظ فقط لما يحتاج المستخدم "تصفح موقع بعينه بصرياً". # # التفعيل: عيّن متغيّر البيئة BRAVE_API_KEY بمفتاحك من # https://api-dashboard.search.brave.com (فيه tier مجاني — 2000 استعلام/شهر). BRAVE_API_KEY = os.environ.get("BRAVE_API_KEY", "").strip() BRAVE_ENDPOINT = "https://api.search.brave.com/res/v1/web/search" async def brave_search(query: str, count: int = 8) -> dict: """ يستدعي Brave Search API مباشرة (بدون متصفح، بدون كابتشا). يُعيد نتائج منسّقة نصياً بنفس شكل مخرجات run_cmd العادية، عشان تندمج بسلاسة مع بقية منطق الترمنال والفرونت إند بدون أي تعديل إضافي مطلوب هناك. """ if not BRAVE_API_KEY: return {"stdout": "", "stderr": "BRAVE_API_KEY غير مُعرَّف — راجع الإعدادات", "returncode": -1} def _fetch(): import urllib.request import urllib.error url = f"{BRAVE_ENDPOINT}?q={urllib.parse.quote_plus(query)}&count={min(count, 20)}&safesearch=moderate" req = urllib.request.Request(url, headers={ "Accept": "application/json", "Accept-Encoding": "gzip", "X-Subscription-Token": BRAVE_API_KEY, }) try: with urllib.request.urlopen(req, timeout=15) as r: raw = r.read() if r.info().get("Content-Encoding") == "gzip": import gzip raw = gzip.decompress(raw) return json.loads(raw), None except urllib.error.HTTPError as e: body = e.read().decode(errors="ignore")[:300] return None, f"HTTP {e.code}: {body}" except Exception as e: return None, str(e) data, err = await asyncio.to_thread(_fetch) if err: return {"stdout": "", "stderr": f"Brave Search error: {err}", "returncode": -1} lines = [] # الإجابة السريعة (infobox) إن وُجدت infobox = data.get("infobox", {}) or {} if infobox.get("long_desc") or infobox.get("description"): lines.append("ANS: " + (infobox.get("long_desc") or infobox.get("description", ""))[:400]) web_results = (data.get("web", {}) or {}).get("results", []) or [] for i, r in enumerate(web_results[:count], 1): title = r.get("title", "") desc = re.sub(r"<[^>]+>", "", r.get("description", "")) url = r.get("url", "") lines.append(f"{i}. {title}\n {desc[:220]}\n {url}") if not lines: return {"stdout": "", "stderr": "لا توجد نتائج من Brave Search لهذا الاستعلام", "returncode": 0} return {"stdout": "\n".join(lines)[:15000], "stderr": "", "returncode": 0} # نمط يلتقط أمر بحث "زائف" مخصص يُرسله الذكاء الاصطناعي عبر أداة terminal # بصيغة بسيطة: search: نص الاستعلام (أو search "نص الاستعلام") _SEARCH_CMD_RE = re.compile(r"^\s*search\s*:?\s*[\"']?(.+?)[\"']?\s*$", re.IGNORECASE) async def run_cmd_smart(cmd: str, timeout: int = 60) -> dict: # ── أولوية قصوى: أمر بحث صريح "search: ..." → Brave Search API مباشرة ── m_search = _SEARCH_CMD_RE.match(cmd) if m_search and "curl" not in cmd and "http" not in cmd: query = m_search.group(1).strip() if BRAVE_API_KEY: res = await brave_search(query) if res["returncode"] == 0 and not _result_empty(res["stdout"]): return res # فشل Brave (مفتاح خاطئ/حصة منتهية) → استمر بالخطة البديلة أدناه # لا يوجد مفتاح Brave أو فشل الطلب → استخدم مصادر API المجانية الأخرى return await _fallback_search_sources(query) res = await run_cmd(cmd, timeout) if res["returncode"] == 0 and not _result_empty(res["stdout"]): return res is_search = "curl" in cmd and any(x in cmd for x in ["search", "duckduck", "bing", "google", "wikipedia", "reddit"]) if not is_search: return res m = re.search(r"[?&]q=([^&'\"\s]+)", cmd) query = urllib.parse.unquote(m.group(1).replace("+", " ")).strip() if m else "" if len(query) < 3: return res return await _fallback_search_sources(query, prior_result=res) async def _fallback_search_sources(query: str, prior_result: dict = None) -> dict: """يجرب Brave أولاً (لو متاح)، ثم مصادر API المجانية الأخرى (DuckDuckGo Instant Answer, Wikipedia, HackerNews, Reddit) — كل هذي APIs رسمية ومصرَّح باستخدامها آلياً، لا تحتاج فتح متصفح ولا تواجه كابتشا إطلاقاً.""" if BRAVE_API_KEY: res = await brave_search(query) if res["returncode"] == 0 and not _result_empty(res["stdout"]): return res results = [] for src in _search_sources(query): r2 = await run_cmd(src["cmd"], 20) if not _result_empty(r2["stdout"]): results.append(f"[{src['name']}]\n{r2['stdout']}") if len(results) >= 2: break if results: return {"stdout": "\n\n".join(results)[:15000], "stderr": "", "returncode": 0} return prior_result or {"stdout": "", "stderr": "لم يُعثر على نتائج من أي مصدر", "returncode": -1} # ════════════════════════════════════════════════════════════════ # ── FastAPI App ─────────────────────────────────────────────── # ════════════════════════════════════════════════════════════════ app = FastAPI(title="Z-Computer-Mode v20 — Headless-Native") app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"]) @app.get("/", response_class=HTMLResponse) async def root(): with _sessions_lock: n = len(_sessions) rows = "".join( f"{uid[:30]}{s['port']}" f"{'🟢 متصل' if s.get('active_ws') else '⚫ غير متصل'}" for uid, s in _sessions.items() ) return f""" Z Computer Mode v20

🖥️ THE Z AI — Computer Mode Server v20 (Headless-Native)

✅ يعمل — بدون Xvfb، Chromium headless=new فقط

🌐 المتصفح المعتمد: {BROWSER_BIN}

🔍 Brave Search API: {'✅ مفعّل' if BRAVE_API_KEY else '❌ غير مُعرَّف (عيّن BRAVE_API_KEY)'}

🦊 Camoufox (مضاد للكشف): {'✅ متاح — إرسال action: switch_engine للتفعيل' if _CAMOUFOX_LIB_AVAILABLE and CAMOUFOX_ENABLED else '❌ غير متاح'}

👥 الجلسات النشطة: {n} / {MAX_SESSIONS}

👤 الجلسات

{rows or ''}
User IDCDP Portالحالة
لا توجد جلسات

Endpoints: /health · /ws?user_id=email (WebSocket)

""" # ════════════════════════════════════════════════════════════════ # ── معالج الأوامر — بروتوكول مطابق 100% للواجهة الأمامية ──────── # ════════════════════════════════════════════════════════════════ # الأوامر المدعومة (كل ما ترسله index_plus_prompt_enhanced_v2-3.html فعلياً): # screenshot, terminal, mouse_move, mouse_click, mouse_drag, keyboard_type, # keyboard_hotkey, keyboard_press, scroll, clipboard_write, clipboard_read, # paste, open_app, open_tab, close_tab, browser_back, browser_forward, # browser_search, screen_info, reset_computer # عتبة أبسط بكثير من v13: بما أن لا يوجد X server منفصل يمكن أن "يتجمد" دون # أن يتوقف Chromium نفسه، أي فشل التقاط الآن يعني عملياً أن Chromium توقف أو # اتصال CDP انقطع — فشلة واحدة كافية للتأكد ومحاولة إعادة الاتصال مباشرة قبل # اللجوء لإعادة تشغيل العملية بالكامل. _RECONNECT_THRESHOLD = 1 _FULL_RESET_THRESHOLD = 3 _MAX_ROUNDS = 4 async def handle_action(ws: WebSocket, msg: dict, sess: dict): action = msg.get("action", "") data = msg.get("data", {}) async def send(obj): try: await ws.send_text(json.dumps(obj, ensure_ascii=False)) except Exception: pass async def do_screenshot(label: str = "", auto: bool = False): """ يلتقط ويرسل لقطة شاشة، مع تعافي تلقائي بسيط: فشلة 1 → أعد الاتصال بـ CDP فقط (Chromium غالباً حي، الاتصال فقط سقط) فشلة 3 → أعد تشغيل Chromium بالكامل (reset_session) بعد ذلك → أرسل computer_reset صراحة للعميل بدل الصمت """ for _round in range(_MAX_ROUNDS): ok_browser = await ensure_browser_running(sess) if not ok_browser: sess["consecutive_failures"] += 1 else: async with sess["shot_lock"]: result = await capture_full(sess, quality=75) if result.get("data") and not result.get("is_placeholder"): sess["consecutive_failures"] = 0 fh = _frame_hash(result["data"]) if not auto or fh != sess.get("last_frame_hash", ""): sess["last_frame_hash"] = fh await send({ "type": "screenshot", "data": result["data"], "data_grid": result.get("data_grid", ""), "data_grid2": result.get("data_grid2", ""), "data_grid3": result.get("data_grid3", ""), "ts": int(time.time() * 1000), "auto": auto, "label": label, "screen_width": result["width"], "screen_height": result["height"], "mouse_x": result["mouse_x"], "mouse_y": result["mouse_y"], "has_grid": True, }) return sess["consecutive_failures"] += 1 fails = sess["consecutive_failures"] if fails >= _FULL_RESET_THRESHOLD: await send({"type": "computer_reset", "msg": "🔄 إعادة تشغيل تلقائية للكمبيوتر بعد فشل متكرر — لحظات ويعود طبيعياً", "auto": True}) await reset_session(sess["user_id"]) await asyncio.sleep(1.0) continue elif fails >= _RECONNECT_THRESHOLD: # أعد الاتصال بـ CDP فقط (أخف بكثير من إعادة تشغيل Chromium بالكامل) sess["cdp"] = None await asyncio.sleep(0.3) continue await asyncio.sleep(0.3) # استنفدنا كل المحاولات — أعلم العميل صراحة بدل الصمت الأبدي await send({"type": "computer_reset", "msg": "⚠️ استمرت مشكلة الالتقاط — الرجاء طلب لقطة شاشة جديدة", "auto": True}) def bg_shot(label: str = "", delay: float = 0.5): async def _run(): await asyncio.sleep(delay) await do_screenshot(label, auto=True) asyncio.create_task(_run()) # ── reset_computer ─────────────────────────────────────── if action == "reset_computer": await send({"type": "ack", "action": "reset_computer", "status": "resetting"}) await reset_session(sess["user_id"]) await asyncio.sleep(1.0) await do_screenshot("after reset") await send({"type": "computer_reset", "msg": "✅ تم إعادة ضبط الكمبيوتر — الشاشة جديدة تماماً"}) return # ── screenshot ──────────────────────────────────────────── if action == "screenshot": _task = asyncio.create_task(do_screenshot("explicit screenshot")) def _on_done(t: asyncio.Task): exc = t.exception() if not t.cancelled() else None if exc: print(f"[screenshot-task:{sess['user_id']}] ❌ {exc}") _task.add_done_callback(_on_done) return # ── terminal ───────────────────────────────────────────── if action == "terminal": cmd = data.get("cmd", "") if not cmd: await send({"type": "terminal_result", "stdout": "", "stderr": "no cmd", "returncode": -1}) return res = await run_cmd_smart(cmd, int(data.get("timeout", 60))) await send({"type": "terminal_result", "cmd": cmd, "stdout": res["stdout"], "stderr": res.get("stderr", ""), "returncode": res["returncode"]}) return # ── يحتاج المتصفح شغالاً؟ تأكد أولاً لبقية الأوامر ── engine = sess.get("engine", "chromium") if engine == "chromium": await ensure_browser_running(sess) else: cam = await get_camoufox_session(sess["user_id"]) if not cam: # فشل بدء Camoufox (مكتبة غير مثبَّتة/فشل إقلاع) → ارجع تلقائياً لـ Chromium print(f"[engine] ⚠️ فشل Camoufox لـ '{sess['user_id']}' — رجوع تلقائي لـ Chromium") sess["engine"] = "chromium" engine = "chromium" await ensure_browser_running(sess) # ── switch_engine — تبديل محرك المتصفح للمستخدم (chromium ⇄ camoufox) ── if action == "switch_engine": target = data.get("engine", "chromium") if target not in ("chromium", "camoufox"): await send({"type": "error", "msg": f"Unknown engine: '{target}'"}) return if target == "camoufox" and not _CAMOUFOX_LIB_AVAILABLE: await send({"type": "error", "msg": "Camoufox غير مثبَّت على هذا السيرفر"}) return sess["engine"] = target sess["last_frame_hash"] = "" await send({"type": "ack", "action": "switch_engine", "engine": target, "msg": f"✅ تم التبديل لمحرك '{target}'" + (" — متصفح مضاد للكشف، أبطأ قليلاً لكن يتجنب كابتشا المواقع المحمية بشدة" if target == "camoufox" else " — المحرك الافتراضي السريع")}) bg_shot(f"after switch to {target}", delay=0.8) return if action == "mouse_move": x, y = int(data.get("x", 0)), int(data.get("y", 0)) if engine == "camoufox": await cam.mouse_move(x, y) else: await cdp_mouse_move(sess, x, y) await send({"type": "ack", "action": "mouse_move", "x": x, "y": y}) elif action == "mouse_click": x, y = int(data.get("x", 0)), int(data.get("y", 0)) btn = {"left": "left", "middle": "middle", "right": "right"}.get(data.get("button", "left"), "left") double = bool(data.get("double", False)) if engine == "camoufox": await cam.mouse_click(x, y, btn, double) else: await cdp_mouse_click(sess, x, y, btn, double) await send({"type": "ack", "action": "mouse_click", "x": x, "y": y, "button": btn}) bg_shot("after click", delay=0.5) elif action == "mouse_drag": x1, y1 = int(data.get("x1", 0)), int(data.get("y1", 0)) x2, y2 = int(data.get("x2", 0)), int(data.get("y2", 0)) if engine == "camoufox": await cam.mouse_drag(x1, y1, x2, y2) else: await cdp_mouse_drag(sess, x1, y1, x2, y2) await send({"type": "ack", "action": "mouse_drag"}) bg_shot("after drag", delay=0.4) elif action == "keyboard_type": text = _safe_search(data.get("text", "")) if text: if engine == "camoufox": await cam.type_text(text) else: await cdp_type_text(sess, text) await send({"type": "ack", "action": "keyboard_type", "method": engine}) elif action == "keyboard_hotkey": keys = data.get("keys", []) if keys: if engine == "camoufox": await cam.hotkey(keys) else: await cdp_hotkey(sess, keys) await send({"type": "ack", "action": "keyboard_hotkey", "keys": keys}) elif action == "keyboard_press": key = data.get("key", "") if key: if engine == "camoufox": await cam.key_press(key) else: await cdp_key_press(sess, key) await send({"type": "ack", "action": "keyboard_press"}) elif action == "scroll": x, y = int(data.get("x", SCREEN_W // 2)), int(data.get("y", SCREEN_H // 2)) clicks = max(-5, min(5, int(data.get("clicks", 3)))) if engine == "camoufox": await cam.scroll(x, y, delta_y=-clicks * 100) else: await cdp_scroll(sess, x, y, delta_y=-clicks * 100) await send({"type": "ack", "action": "scroll", "clicks": clicks}) elif action == "clipboard_write": text = data.get("text", "") if engine != "camoufox": await cdp_clipboard_write(sess, text) # في Camoufox، النسخ الفعلي يُحاكى عبر paste مباشرة (لا حاجة لتخزين منفصل) await send({"type": "ack", "action": "clipboard_write", "length": len(text)}) elif action == "clipboard_read": text = await cdp_clipboard_read(sess) if engine != "camoufox" else "" await send({"type": "clipboard_content", "text": text}) elif action == "paste": text = data.get("text", "") if text: if engine == "camoufox": await cam.type_text(text) else: await cdp_clipboard_write(sess, text) await cdp_type_text(sess, text) await send({"type": "ack", "action": "paste"}) elif action == "open_app": # في هذه البنية Chromium/Camoufox هو "التطبيق" الوحيد؛ أي أمر يحوي # firefox/chromium/chrome يُفهم كطلب فتح صفحة فارغة جديدة في نفس المتصفح. cmd = _safe_search(data.get("cmd", "")) url_match = re.search(r"https?://\S+", cmd) target_url = url_match.group(0) if url_match else "about:blank" if engine == "camoufox": await cam.navigate(target_url) else: await cdp_navigate(sess, target_url) await send({"type": "ack", "action": "open_app", "cmd": cmd}) bg_shot("after open_app", delay=1.0) elif action == "open_tab": url = _safe_search(data.get("url", "") or "about:blank") if engine == "camoufox": await cam.navigate(url) else: await cdp_navigate(sess, url) await send({"type": "ack", "action": "open_tab", "url": url}) bg_shot("after open_tab", delay=1.2) elif action == "close_tab": if engine == "camoufox": await cam.navigate("about:blank") else: await cdp_navigate(sess, "about:blank") await send({"type": "ack", "action": "close_tab"}) bg_shot("after close_tab", delay=0.4) elif action == "browser_back": if engine == "camoufox": await cam.go_back() else: cdp = sess.get("cdp") if cdp and cdp.alive: hist = await cdp.send("Page.getNavigationHistory", timeout=4.0) if hist: entries = hist.get("result", {}).get("entries", []) idx = hist.get("result", {}).get("currentIndex", 0) if idx > 0: await cdp.send("Page.navigateToHistoryEntry", {"entryId": entries[idx - 1]["id"]}, timeout=5.0) await send({"type": "ack", "action": "browser_back"}) bg_shot("after back", delay=0.6) elif action == "browser_forward": if engine == "camoufox": await cam.go_forward() else: cdp = sess.get("cdp") if cdp and cdp.alive: hist = await cdp.send("Page.getNavigationHistory", timeout=4.0) if hist: entries = hist.get("result", {}).get("entries", []) idx = hist.get("result", {}).get("currentIndex", 0) if idx < len(entries) - 1: await cdp.send("Page.navigateToHistoryEntry", {"entryId": entries[idx + 1]["id"]}, timeout=5.0) await send({"type": "ack", "action": "browser_forward"}) bg_shot("after forward", delay=0.6) elif action == "browser_search": url = _safe_search(data.get("url", "") or data.get("query", "")) if not re.match(r"^https?://", url): url = "https://duckduckgo.com/?q=" + urllib.parse.quote_plus(url) if engine == "camoufox": await cam.navigate(url) else: await cdp_navigate(sess, url) await send({"type": "ack", "action": "browser_search"}) bg_shot("after search", delay=1.0) elif action == "screen_info": info = await cdp_get_screen_info(sess) await send({"type": "screen_info", "width": info["width"], "height": info["height"], "mouse_x": info["mouse_x"], "mouse_y": info["mouse_y"], "browser": BROWSER_BIN if engine == "chromium" else "camoufox (Firefox anti-detect)", "engine": engine}) else: await send({"type": "error", "msg": f"Unknown action: '{action}'"}) # ════════════════════════════════════════════════════════════════ # ── WebSocket Endpoint ──────────────────────────────────────── # ════════════════════════════════════════════════════════════════ @app.websocket("/ws") async def websocket_endpoint(ws: WebSocket, user_id: str = Query(default="anonymous")): await ws.accept() sess = await get_or_create_session(user_id, ws) await ensure_browser_running(sess) async def _heartbeat(): while True: await asyncio.sleep(20) try: await ws.send_text(json.dumps({"type": "ping", "ts": int(time.time() * 1000)})) except Exception: break hb_task = asyncio.create_task(_heartbeat()) try: info = await cdp_get_screen_info(sess) await ws.send_text(json.dumps({ "type": "connected", "screen_width": info["width"], "screen_height": info["height"], "browser": BROWSER_BIN, "display": f"headless:{sess['port']}", "user_id": user_id, "session_id": id(ws), "msg": f"Z Computer Mode v20 | User: {user_id} | Port: {sess['port']} | " f"Browser: {BROWSER_BIN} (headless=new) | Screen: {info['width']}x{info['height']}", }, ensure_ascii=False)) # لقطة شاشة أولية result = await capture_full(sess, quality=72) if result.get("data"): await ws.send_text(json.dumps({ "type": "screenshot", "data": result["data"], "data_grid": result.get("data_grid", ""), "data_grid2": result.get("data_grid2", ""), "data_grid3": result.get("data_grid3", ""), "ts": int(time.time() * 1000), "label": "Initial screen", "screen_width": result["width"], "screen_height": result["height"], "mouse_x": result["mouse_x"], "mouse_y": result["mouse_y"], "has_grid": True, }, ensure_ascii=False)) sess["last_frame_hash"] = _frame_hash(result["data"]) except Exception as e: print(f"[ws:{user_id}] خطأ عند البدء: {e}") try: while True: raw = await ws.receive_text() try: msg = json.loads(raw) if msg.get("type") == "pong": continue await handle_action(ws, msg, sess) except json.JSONDecodeError: pass except WebSocketDisconnect: pass except Exception as e: print(f"[ws:{user_id}] خطأ: {e}") finally: hb_task.cancel() await destroy_session_ws(user_id, ws) # ════════════════════════════════════════════════════════════════ # ── REST Endpoints ──────────────────────────────────────────── # ════════════════════════════════════════════════════════════════ @app.get("/screenshot") async def rest_screenshot(user_id: str = "anonymous"): with _sessions_lock: sess = _sessions.get(user_id) if not sess: return JSONResponse({"error": "no session for this user_id"}, status_code=404) await ensure_browser_running(sess) result = await capture_full(sess, quality=75) return JSONResponse({ "image": result["data"], "image_grid": result.get("data_grid", ""), "image_grid2": result.get("data_grid2", ""), "image_grid3": result.get("data_grid3", ""), "ts": int(time.time() * 1000), "screen_width": result["width"], "screen_height": result["height"], "mouse_x": result["mouse_x"], "mouse_y": result["mouse_y"], "has_grid": True, "user_id": user_id, }) @app.post("/terminal") async def rest_terminal(body: dict): return JSONResponse(await run_cmd_smart(body.get("cmd", ""), body.get("timeout", 60))) @app.get("/health") async def health(): with _sessions_lock: n = len(_sessions) users = [ {"user_id": uid, "cdp_port": s["port"], "connected": bool(s.get("active_ws")), "browser_alive": bool(s.get("cdp") and s["cdp"].alive), "engine": s.get("engine", "chromium")} for uid, s in _sessions.items() ] return {"status": "ok", "version": "v20.1-headless-native+camoufox", "browser": BROWSER_BIN, "brave_search_enabled": bool(BRAVE_API_KEY), "camoufox_available": bool(_CAMOUFOX_LIB_AVAILABLE and CAMOUFOX_ENABLED), "active_sessions": n, "max_sessions": MAX_SESSIONS, "users": users} # ════════════════════════════════════════════════════════════════ # ── Background Cleanup ──────────────────────────────────────── # ════════════════════════════════════════════════════════════════ async def _cleanup_idle_sessions(): """يُزيل جلسات المستخدمين غير النشطين (بدون اتصال WebSocket منذ أكثر من ساعة) — يوقف Chromium الخاص بهم فعلياً لتحرير الذاكرة.""" while True: await asyncio.sleep(1800) now = time.time() with _sessions_lock: idle_uids = [uid for uid, s in _sessions.items() if not s.get("active_ws") and now - s.get("created", now) > 3600] for uid in idle_uids: print(f"[cleanup] 🗑️ إزالة جلسة خاملة '{uid}'") await destroy_session_full(uid) @app.on_event("startup") async def startup(): asyncio.create_task(_cleanup_idle_sessions()) print("✅ Z Computer Mode v20 جاهز — Headless-Native (بدون Xvfb)") print(f" المتصفح المعتمد: {BROWSER_BIN} (--headless=new)") print(f" مرشحو المتصفح المفحوصون: {', '.join(_BROWSER_CANDIDATES)}") print(f" أقصى جلسات متزامنة: {MAX_SESSIONS}") print(f" نطاق منافذ CDP: {CDP_PORT_BASE}+") print(f" websockets lib: {'✅' if _WS_LIB_AVAILABLE else '❌ (pip install websockets)'}") if BRAVE_API_KEY: print(" 🔍 Brave Search API: ✅ مفعّل — البحث سيتم عبره أولاً (بدون كابتشا)") else: print(" 🔍 Brave Search API: ❌ غير مُعرَّف — عيّن BRAVE_API_KEY لتفعيله") print(" (بدونه، البحث سيعتمد على DuckDuckGo/Wikipedia APIs المجانية كبديل)") if _CAMOUFOX_LIB_AVAILABLE and CAMOUFOX_ENABLED: print(" 🦊 Camoufox (متصفح مضاد للكشف): ✅ متاح — عبر action: switch_engine {engine: camoufox}") else: print(" 🦊 Camoufox: ❌ غير متاح — pip install camoufox playwright && playwright install firefox") print(" Connect: wss://your-space.hf.space/ws?user_id=EMAIL") if __name__ == "__main__": port = int(os.environ.get("PORT", 7860)) uvicorn.run("app:app", host="0.0.0.0", port=port, log_level="info")