diff --git "a/app.py" "b/app.py" --- "a/app.py" +++ "b/app.py" @@ -1,396 +1,256 @@ """ -THE Z AI — Computer Mode Server v7 — SMART CONTROL +THE Z AI — Computer Mode Server v8 — STABLE & FAST ==================================================== -الإصلاحات الجوهرية في v7: - 1. capture_screen_with_grid() — صورة مزدوجة: يمين نظيفة + يسار بشبكة احداثيات شفافة - 2. أوامر بسيطة: open pc, open firefox, mouse go x y, click x y, rclick x y, type TEXT, key ENTER ... - 3. إصلاح الكتابة العربية: xclip clipboard + xdotool key ctrl+v - 4. إرسال احداثيات الشاشة الحقيقية + موقع الماوس مع كل screenshot - 5. auto_shot_with_grid بعد كل خطوة - 6. simple_command parser: يفهم أوامر بسيطة ويحولها لأفعال WebSocket +التحسينات الجوهرية في v8: + 1. Xvfb واحد فقط ثابت على :99 — يُشغَّل مرة واحدة عند بدء السيرفر + بدلاً من Xvfb جديد لكل اتصال (كان يفشل في HuggingFace) + 2. عزل المستخدمين عبر profile منفصل لكل جلسة (firefox --profile /tmp/profile_N) + بدلاً من عزل DISPLAY منفصل (الذي كان غير ممكن في HF) + 3. مجموعة screenshot موثوقة: scrot → import → ffmpeg → PIL Xlib + مع retry تلقائي وتحقق من أن الصورة ليست سوداء تماماً (black_screen_check) + 4. Playwright اختياري — إذا كان متاحاً يُستخدم للتحكم الأسرع + وإن لم يكن متاحاً، يُعود لـ xdotool كالمعتاد + 5. WebSocket heartbeat — ping/pong كل 20 ثانية لمنع قطع الاتصال + 6. Screenshot delta compression — إذا الصورة لم تتغير بأكثر من 3% لا تُرسل + (يوفر bandwidth خصوصاً عند عمليات keyboard_type) + 7. Frame buffer مشترك محمي بـ asyncio.Lock — لا تعارض بين الجلسات + 8. Rate limiting لـ screenshot: لا أكثر من 1 كل 0.3 ثانية لنفس الجلسة + 9. auto_shot_grid يُنفَّذ في background task (لا يُجمّد الـ action loop) + 10. أوامر terminal تُنفَّذ في asyncio.Semaphore(4) لمنع تزاحم العمليات + 11. SafeSearch مفروضة على كل URL في كل مكان + 12. تنظيف /tmp تلقائي كل 5 دقائق (يمنع امتلاء القرص في HF المجاني) """ import asyncio import base64 import contextvars +import hashlib import io import json import os import re import subprocess +import tempfile import time import urllib.parse +import uuid +from pathlib import Path from fastapi import FastAPI, WebSocket, WebSocketDisconnect from fastapi.middleware.cors import CORSMiddleware -from fastapi.responses import JSONResponse +from fastapi.responses import JSONResponse, HTMLResponse import uvicorn -# ─── DISPLAY الأساسي (احتياطي فقط — لا يُستخدم فعلياً بعد تفعيل العزل الكامل) ── -_BASE_DISPLAY = os.environ.get("DISPLAY", ":1") -os.environ["DISPLAY"] = _BASE_DISPLAY +# ════════════════════════════════════════════════════════════════ +# ── الإعداد الأساسي ────────────────────────────────────────── +# ════════════════════════════════════════════════════════════════ -# ─── عزل كامل بين المستخدمين: كل اتصال WebSocket يحصل على Xvfb + Firefox خاصين به ── -# يُخزَّن DISPLAY الحالي في contextvar بحيث كل الدوال (xdo, capture_screen_with_grid, ...) -# التي تُستدعى من داخل معالجة رسائل هذا الاتصال تقرأ تلقائياً display الخاص بجلسته، -# دون الحاجة لتمرير المعامل يدوياً عبر كل سلسلة الاستدعاءات. -_current_display: contextvars.ContextVar[str] = contextvars.ContextVar( - "_current_display", default=_BASE_DISPLAY -) +# DISPLAY واحد ثابت — يُشغَّل مرة واحدة فقط +SHARED_DISPLAY = os.environ.get("DISPLAY", ":99") +os.environ["DISPLAY"] = SHARED_DISPLAY + +# semaphore للتحكم في عدد عمليات terminal المتوازية +_terminal_sem = asyncio.Semaphore(4) + +# lock لحماية كتابة ملفات /tmp من التعارض +_tmp_lock = asyncio.Lock() -def DISPLAY_OF() -> str: - """يُعيد DISPLAY الخاص بالجلسة الحالية (الاتصال الذي تجري معالجته الآن).""" - return _current_display.get() +# ── contextvar للـ profile الخاص بالجلسة ── +_current_profile: contextvars.ContextVar[str] = contextvars.ContextVar( + "_current_profile", default="" +) -# نطاق أرقام الشاشات الافتراضية الديناميكية المخصصة لكل جلسة — تبدأ من 100 لتجنب التعارض -# مع أي DISPLAY أساسي ثابت (:0, :1, :99) قد يكون مستخدَماً خارجياً. -_next_display_num = 100 -_display_lock = asyncio.Lock() -# session_id (id(ws)) → {"display": ":N", "xvfb_proc": Popen, "firefox_proc": Popen, "num": N} +# ── تخزين الجلسات: session_id → {profile_dir, last_screenshot_ts, last_frame_hash} ── _active_sessions: dict[int, dict] = {} -_freed_display_nums: list[int] = [] # أرقام شاشات تحررت يمكن إعادة استخدامها - -async def _allocate_display_num() -> int: - """يحجز رقم شاشة فريد لجلسة جديدة — يعيد استخدام أرقام محرَّرة من جلسات منتهية أولاً.""" - global _next_display_num - async with _display_lock: - if _freed_display_nums: - return _freed_display_nums.pop() - n = _next_display_num - _next_display_num += 1 - return n - -async def _start_session_display(num: int) -> tuple[str, "subprocess.Popen | None"]: - """يُشغّل Xvfb جديداً تماماً على الرقم المحدد ويُعيد (display_string, process).""" - disp = f":{num}" - - def _spawn(): - try: - return subprocess.Popen( - ["Xvfb", disp, "-screen", "0", "1920x1080x24", "-nolisten", "tcp"], - stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL - ) - except FileNotFoundError: - return None # Xvfb غير مثبت في هذه البيئة - - proc = await asyncio.to_thread(_spawn) - if proc is None: - # Xvfb غير متاح — رجوع آمن لـ DISPLAY الأساسي (بلا عزل، لكن بلا انهيار) - print("[session] ⚠️ Xvfb binary not found — falling back to shared base display (no isolation)") - return _BASE_DISPLAY, None - # انتظار قصير حتى يصبح X server جاهزاً فعلياً قبل استخدامه - def _check_ready(): - try: - return subprocess.run( - ["xdpyinfo", "-display", disp], capture_output=True, timeout=2 - ).returncode - except FileNotFoundError: - return None # xdpyinfo غير مثبت — لا يمكن التحقق، سننتظر مهلة ثابتة فقط - for _ in range(30): - await asyncio.sleep(0.1) - rc = await asyncio.to_thread(_check_ready) - if rc is None: - await asyncio.sleep(0.5) # مهلة احتياطية ثابتة بدل التحقق الدقيق - break - if rc == 0: - break - return disp, proc - -async def _start_session_browser(disp: str) -> "subprocess.Popen | None": - """يُشغّل متصفح Firefox خاص بهذه الجلسة على الشاشة الافتراضية الخاصة بها.""" - env = {**os.environ, "DISPLAY": disp} - - def _spawn(): - try: - # حذف ملفات قفل Firefox المتبقية أولاً (حتى لو كل جلسة على display منفصل، - # الـ profile على القرص قد يكون مشترَكاً ويحمل قفلاً من جلسة سابقة لم تُغلق نظيفاً) - try: - subprocess.run( - "rm -f ~/.mozilla/firefox/*/lock ~/.mozilla/firefox/*/.parentlock 2>/dev/null", - shell=True, executable="/bin/bash", timeout=3, capture_output=True - ) - except Exception: - pass - return subprocess.Popen( - [BROWSER, "--new-window", "--no-remote", "about:blank"], - env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL - ) - except Exception as e: - print(f"[session] ⚠️ failed to launch browser on {disp}: {e}") - return None - proc = await asyncio.to_thread(_spawn) - return proc +_sessions_lock = asyncio.Lock() + +# ════════════════════════════════════════════════════════════════ +# ── إعداد Xvfb المشترك الثابت ──────────────────────────────── +# ════════════════════════════════════════════════════════════════ + +_xvfb_proc: "subprocess.Popen | None" = None -async def create_session_for(ws: "WebSocket") -> str: +def _ensure_xvfb() -> bool: """ - يُنشئ بيئة كمبيوتر افتراضي معزولة تماماً (Xvfb + متصفح خاصين) لهذا الاتصال، - ويُعيد سلسلة DISPLAY الخاصة بها (مثل ':103'). يُستدعى عند فتح أي اتصال WebSocket جديد. + يتأكد أن Xvfb يعمل على SHARED_DISPLAY. + إذا كان يعمل بالفعل (من Docker أو systemd): لا يفعل شيئاً. + إذا لم يكن يعمل: يُشغّله ويُعيد True عند النجاح. """ - num = await _allocate_display_num() - disp, xvfb_proc = await _start_session_display(num) - if xvfb_proc is None: - # رجعنا لـ DISPLAY الأساسي المشترك (Xvfb غير متاح) — لا حاجة لمتصفح جديد منفصل، - # المتصفح الأساسي (إن وُجد خارجياً) يكفي، ولا داعي لتتبع جلسة فعلية لهذا الاتصال. - _freed_display_nums.append(num) - return disp - firefox_proc = await _start_session_browser(disp) - _active_sessions[id(ws)] = { - "display": disp, "num": num, - "xvfb_proc": xvfb_proc, "firefox_proc": firefox_proc, - } - print(f"[session] ✅ created isolated session display={disp} for ws={id(ws)} (total active: {len(_active_sessions)})") - return disp + global _xvfb_proc -async def destroy_session_for(ws: "WebSocket"): - """يُغلق ويُحرّر بيئة الكمبيوتر الافتراضي الخاصة بهذا الاتصال بالكامل عند انقطاعه.""" - sess = _active_sessions.pop(id(ws), None) - if not sess: - return + # تحقق أولاً: هل DISPLAY موجود بالفعل؟ + try: + r = subprocess.run( + ["xdpyinfo", "-display", SHARED_DISPLAY], + capture_output=True, timeout=3 + ) + if r.returncode == 0: + print(f"[xvfb] ✅ Display {SHARED_DISPLAY} already active (external)") + return True + except FileNotFoundError: + pass # xdpyinfo غير متاح — سنكمل بدون تحقق + except Exception: + pass - def _kill_all(): - for key in ("firefox_proc", "xvfb_proc"): - p = sess.get(key) - if p: - try: - p.terminate() - try: - p.wait(timeout=3) - except Exception: - p.kill() - except Exception: - pass - # تأكيد إضافي: قتل أي عملية متبقية مرتبطة بهذا DISPLAY بالتحديد فقط - disp = sess.get("display", "") - if disp: + # شغّل Xvfb + try: + _xvfb_proc = subprocess.Popen( + ["Xvfb", SHARED_DISPLAY, "-screen", "0", "1920x1080x24", + "-nolisten", "tcp", "-ac"], + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL + ) + # انتظر حتى يصبح جاهزاً (max 5 ثواني) + for _ in range(50): + time.sleep(0.1) try: - subprocess.run( - ["pkill", "-9", "-f", f"DISPLAY={disp}"], - capture_output=True, timeout=3 + r = subprocess.run( + ["xdpyinfo", "-display", SHARED_DISPLAY], + capture_output=True, timeout=2 ) + if r.returncode == 0: + print(f"[xvfb] ✅ Xvfb started on {SHARED_DISPLAY}") + return True except Exception: - pass + continue + print(f"[xvfb] ⚠️ Xvfb may not be ready yet — continuing anyway") + return True + except FileNotFoundError: + print(f"[xvfb] ❌ Xvfb binary not found — screenshots will fail") + return False + except Exception as e: + print(f"[xvfb] ❌ Failed to start: {e}") + return False - await asyncio.to_thread(_kill_all) - _freed_display_nums.append(sess["num"]) - print(f"[session] 🗑️ destroyed session display={sess['display']} for ws={id(ws)} (remaining active: {len(_active_sessions)})") - -# ─── تشخيص أدوات الـ capture عند البداية ───────────── -def _check_capture_tools(): - tools = ["scrot", "import", "xwd", "convert", "ffmpeg", "xdotool", "Xvfb", "xdpyinfo"] - print("─── Capture Tools Check ───") - for t in tools: - r = subprocess.run(["which", t], capture_output=True, text=True) - status = "✅" if r.returncode == 0 else "❌" - print(f" {status} {t}: {r.stdout.strip() or 'not found'}") - try: - r2 = subprocess.run(["xdpyinfo", "-display", _BASE_DISPLAY], - capture_output=True, text=True, timeout=3) - print(f" {'✅' if r2.returncode==0 else '❌'} DISPLAY={_BASE_DISPLAY}: {'active' if r2.returncode==0 else 'not active'}") - except Exception as ex: - print(f" ❌ DISPLAY={_BASE_DISPLAY}: {ex}") - print("───────────────────────────") - -try: - _check_capture_tools() -except Exception as _e: - print(f"[diag] {_e}") - -# ─── اكتشاف المتصفح ────────���───────────────────────── +# اكتشاف المتصفح def _detect_browser() -> str: - candidates = ["firefox", "firefox-esr", "chromium-browser", "chromium", "google-chrome"] - found = None - for c in candidates: + for c in ["firefox", "firefox-esr", "chromium-browser", "chromium", "google-chrome"]: r = subprocess.run(["which", c], capture_output=True, text=True) if r.returncode == 0 and r.stdout.strip(): - found = c - break - if not found: - return "firefox" - # تأكد من وجود firefox-esr كـ symlink - esr_check = subprocess.run(["which", "firefox-esr"], capture_output=True, text=True) - if esr_check.returncode != 0: - real_path = subprocess.run(["which", found], capture_output=True, text=True).stdout.strip() - if real_path: - try: - subprocess.run(["ln", "-sf", real_path, "/usr/local/bin/firefox-esr"], check=True) - print(f"✅ Created firefox-esr symlink → {real_path}") - except Exception as e: - print(f"⚠️ Could not create firefox-esr symlink: {e}") - return found + return c + return "firefox" BROWSER = _detect_browser() -print(f"🌐 Browser detected: {BROWSER}") - -app = FastAPI(title="Z-Computer-Mode API v7") -app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_credentials=True, - allow_methods=["*"], allow_headers=["*"]) - -from fastapi.responses import HTMLResponse - -@app.get("/", response_class=HTMLResponse) -async def root(): - w, h = _get_screen_size() - return f""" - -Z Computer Mode Server - - -

🖥️ THE Z AI — Computer Mode Server v7

-

✅ Server is RUNNING

-

🌐 Browser: {BROWSER}

-

📐 Screen: {w}x{h}

-

🔌 WebSocket: wss://thezyzstudio-pcservercomp.hf.space/ws

-
-

Endpoints: /health · /screenshot · /ws (WebSocket)

-""" - -active_connections: list[WebSocket] = [] +print(f"🌐 Browser: {BROWSER}") -# ─── CURL HEADERS ──────────────────────────────────── -CURL_HEADERS = ( - '-H "User-Agent: Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 ' - '(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36" ' - '-H "Accept: text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8" ' - '-H "Accept-Language: en-US,en;q=0.9,ar;q=0.8" ' - '-H "Accept-Encoding: gzip, deflate" ' - '--compressed ' - '--max-time 20 ' - '-L ' - '-s ' -) - -# ─── مصادر البحث 8+ ────────────────────────────────── -def _build_search_sources(query: str) -> list[dict]: - q = urllib.parse.quote_plus(query) - q_raw = query.replace(' ', '+') - return [ - { - "name": "DuckDuckGo Instant", - "cmd": f"curl -s --max-time 20 'https://api.duckduckgo.com/?q={q}&format=json&no_html=1&skip_disambig=1' | python3 -c \"import sys,json; d=json.load(sys.stdin); ans=d.get('AbstractText',''); rels=d.get('RelatedTopics',[]); print('ANSWER:',ans if ans else 'no direct answer'); [print('-',r.get('Text','')[:250]) for r in rels if isinstance(r,dict) and r.get('Text')]\"" - }, - { - "name": "Google News RSS", - "cmd": f"curl -sL {CURL_HEADERS} 'https://news.google.com/rss/search?q={q_raw}&hl=en&gl=US&ceid=US:en' | python3 -c \"import sys,re; xml=sys.stdin.read(); titles=re.findall(r'<!\\[CDATA\\[(.*?)\\]\\]>|(.*?)',xml); clean=lambda s:re.sub('<[^>]+>','',s); results=[(a or b).strip() for a,b in titles if (a or b).strip()][1:9]; [print(str(i+1)+'. '+t[:180]) for i,t in enumerate(results)]\"" - }, - { - "name": "Wikipedia English", - "cmd": f"curl -s --max-time 15 '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','')[:1500])\" 2>/dev/null || curl -s --max-time 15 'https://en.wikipedia.org/w/api.php?action=query&list=search&srsearch={q}&format=json&srlimit=5' | python3 -c \"import sys,json,re; d=json.load(sys.stdin); [print(str(i+1)+'. '+r['title']+': '+re.sub('<[^>]+>','',r.get('snippet',''))[:200]) for i,r in enumerate(d.get('query',{{}}).get('search',[]))]\" 2>/dev/null" - }, - { - "name": "DuckDuckGo HTML", - "cmd": f"curl -sL {CURL_HEADERS} 'https://html.duckduckgo.com/html/?q={q}' | python3 -c \"import sys,re; h=sys.stdin.read(); snippets=re.findall(r'class=.result__snippet[^>]*>(.*?)',h,re.DOTALL); titles=re.findall(r'class=.result__title[^>]*>.*?]*>(.*?)',h,re.DOTALL); clean=lambda s:re.sub('<[^>]+>','',s).strip(); [print(str(i+1)+'. '+clean(titles[i] if i(.*?)',xml,re.DOTALL); [print(str(i+1)+'. '+re.sub('<[^>]+>','',re.search(r'(.*?)',it).group(1) if re.search(r'',it) else '')+'\\n '+re.sub('<[^>]+>','',re.search(r'<description>(.*?)</description>',it).group(1) if re.search(r'<description>',it) else '')[:150]) for i,it in enumerate(items[:6])]\"" - }, - { - "name": "Reddit Search", - "cmd": f"curl -sL {CURL_HEADERS} -H 'Accept: application/json' 'https://www.reddit.com/search.json?q={q}&sort=new&limit=8&type=link' | python3 -c \"import sys,json; d=json.load(sys.stdin); posts=d.get('data',{{}}).get('children',[]); [print(str(i+1)+'. '+p['data'].get('title','')+'\\n r/'+p['data'].get('subreddit','')+' Score:'+str(p['data'].get('score',''))+'\\n '+p['data'].get('selftext','')[:200]) for i,p in enumerate(posts[:6])]\"" - }, - { - "name": "HackerNews", - "cmd": f"curl -s --max-time 15 'https://hn.algolia.com/api/v1/search?query={q}&hitsPerPage=8&tags=story' | python3 -c \"import sys,json; d=json.load(sys.stdin); hits=d.get('hits',[]); [print(str(i+1)+'. '+h.get('title','')+'\\n Points:'+str(h.get('points','0'))+' | '+h.get('url','')[:80]) for i,h in enumerate(hits[:6])]\"" - }, - { - "name": "ArXiv Academic", - "cmd": f"curl -s --max-time 15 'https://export.arxiv.org/api/query?search_query=all:{q}&start=0&max_results=5' | python3 -c \"import sys,re; xml=sys.stdin.read(); titles=re.findall(r'<title>(.*?)',xml)[1:]; summaries=re.findall(r'(.*?)',xml,re.DOTALL); [print(str(i+1)+'. '+t.strip()+'\\n '+summaries[i].strip()[:200] if i str: - patterns = [ - r'[?&]q=([^&\'"]+)', - r'[?&]query=([^&\'"]+)', - r'[?&]search_query=all:([^&\'"]+)', - r"search\?q=([^&'\"\s]+)", - ] - for p in patterns: - m = re.search(p, cmd) - if m: - q = m.group(1).replace('+', ' ').replace('%20', ' ') - return urllib.parse.unquote(q) - return "" +# ════════════════════════════════════════════════════════════════ +# ── إدارة الجلسات (Profile-based isolation) ───────────────── +# ════════════════════════════════════════════════════════════════ +async def create_session(ws: WebSocket) -> dict: + """ + ينشئ profile firefox منفصل لهذا المستخدم. + كل مستخدم يحصل على profile نظيف خاص به في /tmp/zpc_XXXXX/ + لمنع التداخل بين جلسات متعددة على نفس DISPLAY. + """ + sid = id(ws) + profile_dir = f"/tmp/zpc_{sid}" + os.makedirs(profile_dir, exist_ok=True) + + sess = { + "id": sid, + "profile": profile_dir, + "last_shot_ts": 0.0, + "last_frame_hash": "", + "browser_proc": None, + "created": time.time(), + } + async with _sessions_lock: + _active_sessions[sid] = sess + print(f"[session] ✅ Created session {sid} profile={profile_dir}") + return sess -def _is_empty_result(stdout: str) -> bool: - if not stdout or len(stdout.strip()) < 10: - return True - empty_signals = [ - "no direct answer", "no results", "0 results", - "no items", "[]", "{}", "error", "not found", "answer: no", - ] - s = stdout.strip().lower() - meaningful_lines = [l for l in s.split('\n') if l.strip() and not any(sig in l for sig in empty_signals)] - return len(meaningful_lines) < 1 +async def destroy_session(ws: WebSocket): + """يحذف profile الجلسة وينظّف الموارد.""" + sid = id(ws) + async with _sessions_lock: + sess = _active_sessions.pop(sid, None) + if not sess: + return -# ─── Screen Capture ───────────────────────────────── + def _cleanup(): + # أغلق المتصفح إذا كان مفتوحاً + bp = sess.get("browser_proc") + if bp: + try: + bp.terminate() + bp.wait(timeout=3) + except Exception: + try: + bp.kill() + except Exception: + pass + # احذف profile + import shutil + try: + shutil.rmtree(sess["profile"], ignore_errors=True) + except Exception: + pass + + await asyncio.to_thread(_cleanup) + print(f"[session] 🗑️ Destroyed session {sid}") + + +def get_browser_cmd(url: str = "", profile: str = "") -> list: + """يبني أمر فتح المتصفح مع profile منفصل لكل مستخدم.""" + cmd = [BROWSER] + if profile: + cmd += ["--profile", profile] + cmd += ["--no-remote"] + if url: + cmd += [url] + else: + cmd += ["about:blank"] + return cmd -def _get_screen_size() -> tuple[int, int]: - """يُعيد حجم الشاشة الحقيقي (لشاشة الجلسة الحالية)""" - r = subprocess.run(["xdotool", "getdisplaygeometry"], - env={**os.environ, "DISPLAY": DISPLAY_OF()}, - capture_output=True, text=True, timeout=5) - try: - parts = r.stdout.strip().split() - return int(parts[0]), int(parts[1]) - except: - return 1920, 1080 +# ════════════════════════════════════════════════════════════════ +# ── Screenshot Engine (Multi-method + Black Screen Check) ──── +# ════════════════════════════════════════════════════════════════ -def _get_mouse_pos() -> tuple[int, int]: - """يُعيد موقع الماوس الحالي (لشاشة الجلسة الحالية)""" - r = subprocess.run(["xdotool", "getmouselocation"], - env={**os.environ, "DISPLAY": DISPLAY_OF()}, - capture_output=True, text=True, timeout=5) +def _is_black_screen(img) -> bool: + """ + يكتشف ما إذا كانت الصورة سوداء تقريباً (Xvfb يعمل لكن لا شيء فيه). + يأخذ عينة من 100 pixel ويحسب متوسط الـ brightness. + """ try: - mx = int(re.search(r"x:(\d+)", r.stdout).group(1)) - my = int(re.search(r"y:(\d+)", r.stdout).group(1)) - return mx, my - except: - return 0, 0 + small = img.resize((100, 100)) + pixels = list(small.getdata()) + avg = sum(sum(p[:3]) for p in pixels) / (len(pixels) * 3 * 255) + return avg < 0.04 # أقل من 4% brightness = أسود فعلياً + except Exception: + return False -def capture_screen_raw() -> tuple[object, int, int]: +def _capture_raw() -> tuple: """ - يلتقط الشاشة بـ 5 طرق متسلسلة — تضمن نجاح واحدة. - كل الملفات المؤقتة (ناجحة أو فاشلة) تُحذف دائماً لمنع تراكمها في /tmp - وامتلاء القرص/الذاكرة بعد عدة لقطات شاشة متتالية. - 1. scrot (الأسرع) - 2. import (ImageMagick) - 3. xwd + convert - 4. ffmpeg + x11grab - 5. python-xlib / Xlib مباشرة + يلتقط الشاشة بـ 4 طرق متسلسلة. + يُعيد (PIL.Image | None, width, height). + ملاحظة: كل ملفات /tmp تُحذف دائماً (finally block). """ from PIL import Image - disp = DISPLAY_OF() - env = {**os.environ, "DISPLAY": disp} - tmp = f"/tmp/zs_{int(time.time()*1000)}" - _tmp_files_to_clean = [] # كل الملفات المؤقتة التي أُنشئت في هذه المحاولة + env = {**os.environ, "DISPLAY": SHARED_DISPLAY} + ts = int(time.time() * 1000) + tmp_base = f"/tmp/zss_{ts}" + created_files = [] - def _load(path) -> tuple: - """تحميل الصورة إذا وُجدت وحجمها > 0 — لا يحذف الملف هنا، الحذف مركزي في النهاية""" + def _try_load(path) -> tuple: if not path or not os.path.exists(path): return None, 0, 0 - size = os.path.getsize(path) - if size < 512: # صورة فارغة/تالفة + if os.path.getsize(path) < 1024: return None, 0, 0 try: img = Image.open(path).convert("RGB") w, h = img.size - if w < 10 or h < 10: + if w < 100 or h < 100: return None, 0, 0 return img, w, h except Exception as ex: - print(f"[_load] {ex}") + print(f"[cap] load error: {ex}") return None, 0, 0 - def _cleanup(): - """يحذف كل الملفات المؤقتة المتبقية من كل المحاولات (ناجحة أو فاشلة)""" - for p in _tmp_files_to_clean: + def _cleanup_files(): + for p in created_files: try: if p and os.path.exists(p): os.unlink(p) @@ -398,612 +258,441 @@ def capture_screen_raw() -> tuple[object, int, int]: pass try: - # ── طريقة 1: scrot ────────────────────────────────── + # ── Method 1: scrot (fastest) ────────────────────── + p1 = f"{tmp_base}_scrot.png" + created_files.append(p1) try: - p = tmp + "_1.png" - _tmp_files_to_clean.append(p) r = subprocess.run( - ["scrot", "-q", "95", p], + ["scrot", "-q", "95", p1], env=env, timeout=8, capture_output=True ) - img, w, h = _load(p) - if img: - print("[capture] ✅ scrot") + img, w, h = _try_load(p1) + if img and not _is_black_screen(img): + print("[cap] ✅ scrot") return img, w, h - print(f"[capture] scrot failed rc={r.returncode} err={r.stderr[:80]}") - except Exception as e: - print(f"[capture] scrot ex: {e}") - - # ── طريقة 2: ImageMagick import ───────────────────── - try: - p = tmp + "_2.png" - _tmp_files_to_clean.append(p) - r = subprocess.run( - ["import", "-window", "root", "-silent", p], - env=env, timeout=10, capture_output=True - ) - img, w, h = _load(p) if img: - print("[capture] ✅ import (ImageMagick)") - return img, w, h - print(f"[capture] import failed rc={r.returncode}") + print("[cap] ⚠️ scrot got black screen — trying next") except Exception as e: - print(f"[capture] import ex: {e}") + print(f"[cap] scrot: {e}") - # ── طريقة 3: xwd + convert ────────────────────────── + # ── Method 2: ImageMagick import ──────────────────── + p2 = f"{tmp_base}_im.png" + created_files.append(p2) try: - xwd_p = tmp + "_3.xwd" - png_p = tmp + "_3.png" - _tmp_files_to_clean.append(xwd_p) - _tmp_files_to_clean.append(png_p) - r1 = subprocess.run( - ["xwd", "-root", "-silent", "-out", xwd_p], + r = subprocess.run( + ["import", "-window", "root", "-silent", p2], env=env, timeout=10, capture_output=True ) - r2 = subprocess.run( - ["convert", xwd_p, png_p], - timeout=10, capture_output=True - ) - img, w, h = _load(png_p) - if img: - print("[capture] ✅ xwd+convert") + img, w, h = _try_load(p2) + if img and not _is_black_screen(img): + print("[cap] ✅ ImageMagick import") return img, w, h - print(f"[capture] xwd rc={r1.returncode} convert rc={r2.returncode}") except Exception as e: - print(f"[capture] xwd ex: {e}") + print(f"[cap] import: {e}") - # ── طريقة 4: ffmpeg x11grab ───────────────────────── + # ── Method 3: ffmpeg x11grab ──────────────────────── + p3 = f"{tmp_base}_ff.png" + created_files.append(p3) try: - p = tmp + "_4.png" - _tmp_files_to_clean.append(p) sw, sh = _get_screen_size() r = subprocess.run([ - "ffmpeg", "-y", - "-f", "x11grab", + "ffmpeg", "-y", "-f", "x11grab", "-video_size", f"{sw}x{sh}", - "-i", disp, - "-vframes", "1", - "-q:v", "2", - p + "-i", SHARED_DISPLAY, + "-vframes", "1", "-q:v", "1", p3 ], env=env, timeout=12, capture_output=True) - img, w, h = _load(p) - if img: - print("[capture] ✅ ffmpeg x11grab") + img, w, h = _try_load(p3) + if img and not _is_black_screen(img): + print("[cap] ✅ ffmpeg") return img, w, h - print(f"[capture] ffmpeg rc={r.returncode} err={r.stderr[-120:]}") except Exception as e: - print(f"[capture] ffmpeg ex: {e}") + print(f"[cap] ffmpeg: {e}") - # ── طريقة 5: python-xlib (Xlib مباشرة) ────────────── + # ── Method 4: python-xlib ──────────────────────────── try: - from Xlib import display as Xdisplay - from Xlib.ext.xtest import fake_input - xdisp = Xdisplay.Display(disp) - root = xdisp.screen().root + from Xlib import display as Xdisp, X + import struct + xd = Xdisp.Display(SHARED_DISPLAY) + root = xd.screen().root geom = root.get_geometry() w, h = geom.width, geom.height - raw = root.get_image(0, 0, w, h, - Xdisplay.X.ZPixmap, - 0xFFFFFFFF) - import struct + raw = root.get_image(0, 0, w, h, X.ZPixmap, 0xFFFFFFFF) data = raw.data - # BGRA → RGB - pixels = [] - for i in range(0, len(data), 4): - b, g, r_, a = struct.unpack_from('BBBB', data, i) - pixels.extend([r_, g, b]) - img = Image.frombytes("RGB", (w, h), bytes(pixels)) - print("[capture] ✅ python-xlib") - return img, w, h + pixels = bytearray(len(data) // 4 * 3) + for i in range(0, len(data) - 3, 4): + b, g, r_c = data[i], data[i+1], data[i+2] + j = (i // 4) * 3 + pixels[j], pixels[j+1], pixels[j+2] = r_c, g, b + from PIL import Image as PILImg + img = PILImg.frombytes("RGB", (w, h), bytes(pixels)) + if not _is_black_screen(img): + print("[cap] ✅ xlib") + return img, w, h except Exception as e: - print(f"[capture] xlib ex: {e}") + print(f"[cap] xlib: {e}") + + # ── Fallback: placeholder ──────────────────────────── + print("[cap] ⚠️ All methods failed or black — returning placeholder") + from PIL import Image as PILImg, ImageDraw + sw, sh = _get_screen_size() + img = PILImg.new("RGB", (sw or 1920, sh or 1080), (15, 15, 25)) + draw = ImageDraw.Draw(img) + draw.rectangle([(0, 0), (sw, 50)], fill=(30, 30, 60)) + draw.text((10, 15), f"⚠️ Screenshot failed — DISPLAY={SHARED_DISPLAY}", fill=(255, 100, 100)) + draw.text((10, 35), f"Methods tried: scrot, import, ffmpeg, xlib", fill=(120, 120, 120)) + return img, sw or 1920, sh or 1080 - # ── طريقة 6: صورة placeholder واضحة ──────────────── - print("[capture] ⚠️ ALL methods failed — returning placeholder") - try: - from PIL import ImageDraw - sw, sh = _get_screen_size() - img = Image.new("RGB", (sw or 1280, sh or 720), (20, 20, 30)) - draw = ImageDraw.Draw(img) - draw.rectangle([(0, 0), (sw, 40)], fill=(30, 30, 50)) - draw.text((10, 10), f"⚠️ Screenshot failed — DISPLAY={disp}", fill=(255, 100, 100)) - draw.text((10, 50), "Methods tried: scrot, import, xwd, ffmpeg, xlib", fill=(150, 150, 150)) - return img, sw or 1280, sh or 720 - except: - return None, 0, 0 finally: - # ── يحذف كل الملفات المؤقتة دائماً، نجحت المحاولة أو فشلت ── - # هذا يمنع تراكم ملفات /tmp التي كانت تمتلئ بعد عدة لقطات وتتسبب - # بقتل العمليات (SIGTERM/OOM) على بيئات مثل Hugging Face المجانية. - _cleanup() + _cleanup_files() -def capture_screen(scale=0.6, quality=70) -> str: - """يلتقط الشاشة العادية بدون grid""" - img, w, h = capture_screen_raw() - if img is None: - return "" +def _get_screen_size() -> tuple: try: - if scale < 1.0: - img = img.resize((int(w * scale), int(h * scale)), img.LANCZOS) - buf = io.BytesIO() - img.save(buf, format="JPEG", quality=quality, optimize=True) - return base64.b64encode(buf.getvalue()).decode() - except Exception as e: - print(f"[capture] {e}") - return "" - + r = subprocess.run( + ["xdotool", "getdisplaygeometry"], + env={**os.environ, "DISPLAY": SHARED_DISPLAY}, + capture_output=True, text=True, timeout=5 + ) + parts = r.stdout.strip().split() + return int(parts[0]), int(parts[1]) + except Exception: + return 1920, 1080 -def capture_screen_with_grid(scale=0.65, quality=72, - force_mx: int | None = None, - force_my: int | None = None) -> dict: - """ - يلتقط الشاشة ويُعيد صورة واحدة فقط مع شبكة إحداثيات واضحة. - الإحداثيات في الصورة تطابق إحداثيات الشاشة الحقيقية (1:1 mapping). - الخطوط كل 100px — الأرقام تُظهر الـ X/Y الحقيقي للنقر. - force_mx / force_my: إذا مُرِّرا يُرسم الـ cursor في هذا الموضع مباشرةً - بدون استعلام xdotool — يُعالج مشكلة race condition بعد mouse_move. - """ +def _get_mouse_pos() -> tuple: try: - from PIL import Image, ImageDraw - img, orig_w, orig_h = capture_screen_raw() - if img is None: - return {"data": "", "width": 1920, "height": 1080, "mouse_x": 0, "mouse_y": 0} - - # استخدم الإحداثيات المُمرَّرة إذا وُجدت (بعد mouse_move/click مباشرة) - # وإلا اقرأ من X server - if force_mx is not None and force_my is not None: - mx, my = force_mx, force_my - else: - mx, my = _get_mouse_pos() - - # resize للعرض فقط — الإحداثيات تبقى للشاشة الأصلية - sw = int(orig_w * scale) - sh = int(orig_h * scale) - grid_img = img.resize((sw, sh), Image.LANCZOS) - draw = ImageDraw.Draw(grid_img, "RGBA") - - # شبكة كل 100px بإحداثيات الشاشة الحقيقية - step_orig = 100 - step_scaled_x = int(step_orig * sw / orig_w) - step_scaled_y = int(step_orig * sh / orig_h) - - grid_color = (255, 255, 255, 45) # خطوط بيضاء شفافة - label_color = (0, 255, 180, 200) # أرقام خضراء واضحة - - # خطوط عمودية + أرقام X الحقيقية - x_sc = step_scaled_x - x_real = step_orig - while x_sc < sw: - draw.line([(x_sc, 0), (x_sc, sh)], fill=grid_color, width=1) - draw.rectangle([(x_sc + 1, 2), (x_sc + 32, 14)], fill=(0, 0, 0, 160)) - draw.text((x_sc + 2, 3), str(x_real), fill=label_color) - x_sc += step_scaled_x - x_real += step_orig - - # خطوط أفقية + أرقام Y الحقيقية - y_sc = step_scaled_y - y_real = step_orig - while y_sc < sh: - draw.line([(0, y_sc), (sw, y_sc)], fill=grid_color, width=1) - draw.rectangle([(2, y_sc + 1), (36, y_sc + 13)], fill=(0, 0, 0, 160)) - draw.text((3, y_sc + 2), str(y_real), fill=label_color) - y_sc += step_scaled_y - y_real += step_orig - - # موقع الماوس الحقيقي — دائرة حمراء - mouse_sx = int(mx * sw / orig_w) - mouse_sy = int(my * sh / orig_h) - r = 10 - draw.ellipse([(mouse_sx-r, mouse_sy-r), (mouse_sx+r, mouse_sy+r)], - outline=(255, 60, 60, 240), width=2) - draw.line([(mouse_sx-16, mouse_sy), (mouse_sx+16, mouse_sy)], - fill=(255, 60, 60, 200), width=1) - draw.line([(mouse_sx, mouse_sy-16), (mouse_sx, mouse_sy+16)], - fill=(255, 60, 60, 200), width=1) - - # شريط معلومات في الأعلى - final = grid_img.convert("RGB") - draw2 = ImageDraw.Draw(final) - draw2.rectangle([(0, 0), (sw, 18)], fill=(0, 0, 0)) - hdr = (f"SCREEN {orig_w}x{orig_h} | MOUSE:({mx},{my}) | " - f"GRID=100px | CLICK COORDS = numbers on grid lines") - draw2.text((4, 2), hdr, fill=(0, 220, 160)) - - # شريط معلومات في الأسفل - draw2.rectangle([(0, sh-18), (sw, sh)], fill=(0, 0, 0)) - draw2.text((4, sh-16), - f"USE REAL COORDS: e.g. click x=500 y=300 means the '500' vertical line + '300' horizontal line", - fill=(180, 180, 80)) - - buf = io.BytesIO() - final.save(buf, format="JPEG", quality=quality, optimize=True) - data = base64.b64encode(buf.getvalue()).decode() - - return { - "data": data, - "width": orig_w, - "height": orig_h, - "mouse_x": mx, - "mouse_y": my, - } - - except Exception as e: - print(f"[capture_grid] {e}") - plain = capture_screen(scale=scale, quality=quality) - mx, my = _get_mouse_pos() - w, h = _get_screen_size() - return {"data": plain, "width": w, "height": h, "mouse_x": mx, "mouse_y": my} - - -# ─── Command Runner ────────────────────────────────── - -_PKILL_SELF_MATCH_RE = re.compile(r"\b(pkill|killall)\s+(-9\s+)?-f\s+(['\"]?)([a-zA-Z0-9_./-]+)\3") - -def _sanitize_pkill_self_match(cmd: str) -> str: - """ - يحوّل أنماط pkill -f الخطيرة (مثل: pkill -f firefox) إلى صيغة آمنة - لا تطابق سطر الأمر الخاص بـ pkill نفسه — لأن pkill -f يبحث في كامل سطر - الأمر لكل العمليات، فيُطابق سطره الخاص ويقتل نفسه، فيُعيد returncode=-15 - وكأن الأمر "فشل" بينما هو فعلياً نجح (أو لم يكن هناك شيء لقتله من الأساس). - """ - def _fix(m): - tool, dash9, _q, pattern = m.group(1), m.group(2) or "", m.group(3), m.group(4) - # استبعد سطر أمر pkill/killall نفسه من المطابقة بإضافة [^k] - safe_pattern = f"[{pattern[0]}]{pattern[1:]}" if len(pattern) > 1 else pattern - return f"{tool} {dash9}-f '{safe_pattern}'" - return _PKILL_SELF_MATCH_RE.sub(_fix, cmd) - + r = subprocess.run( + ["xdotool", "getmouselocation"], + env={**os.environ, "DISPLAY": SHARED_DISPLAY}, + capture_output=True, text=True, timeout=5 + ) + mx = int(re.search(r"x:(\d+)", r.stdout).group(1)) + my = int(re.search(r"y:(\d+)", r.stdout).group(1)) + return mx, my + except Exception: + return 0, 0 -_BING_URL_RE = re.compile(r"https?://(?:www\.)?bing\.com[^\s'\"]*") -_DDG_URL_RE = re.compile(r"https?://(?:www\.)?duckduckgo\.com[^\s'\"]*") -_GOOGLE_URL_RE = re.compile(r"https?://(?:www\.)?google\.[a-z.]+[^\s'\"]*") -def _enforce_safesearch(text: str) -> str: +def capture_with_grid(scale: float = 0.65, quality: int = 72, + force_mx: int | None = None, + force_my: int | None = None) -> dict: """ - يفرض معاملات SafeSearch الصارمة تلقائياً على أي رابط Bing أو DuckDuckGo أو Google - موجود في النص، بغض النظر عمّا كتبه الذكاء — حماية إلزامية على مستوى السيرفر - لمنع ظهور محتوى غير لائق في نتائج البحث، مستقلة عن تعليمات الـ prompt. - تغطي أيضاً حالة الوصول لـ Google بشكل غير مباشر (عبر رابط مكتوب صريح في الأمر). + يلتقط الشاشة ويضيف Grid overlay للذكاء الاصطناعي. + يُعيد dict مع base64 JPEG وإحداثيات الشاشة والماوس. """ - def _fix_bing(m): - url = m.group(0) - if "adlt=" in url: - return re.sub(r"adlt=\w+", "adlt=strict", url) - sep = "&" if "?" in url else "?" - return url + f"{sep}adlt=strict" - - def _fix_ddg(m): - url = m.group(0) - if "kp=" in url: - return re.sub(r"kp=\d", "kp=1", url) - sep = "&" if "?" in url else "?" - return url + f"{sep}kp=1" - - def _fix_google(m): - url = m.group(0) - if "safe=" in url: - return re.sub(r"safe=\w+", "safe=strict", url) - sep = "&" if "?" in url else "?" - return url + f"{sep}safe=strict" - - text = _BING_URL_RE.sub(_fix_bing, text) - text = _DDG_URL_RE.sub(_fix_ddg, text) - text = _GOOGLE_URL_RE.sub(_fix_google, text) - return text - - -def run_raw_command(cmd: str, timeout: int = 60) -> dict: - cmd = _sanitize_pkill_self_match(cmd) - cmd = _enforce_safesearch(cmd) - env = {**os.environ, "DISPLAY": DISPLAY_OF(), - "PYTHONIOENCODING": "utf-8", "LANG": "en_US.UTF-8"} - try: - result = subprocess.run(cmd, shell=True, capture_output=True, - text=True, timeout=timeout, env=env, - executable="/bin/bash") - return { - "stdout": result.stdout[-15000:], - "stderr": result.stderr[-3000:], - "returncode": result.returncode, - } - except subprocess.TimeoutExpired: - return {"stdout": "", "stderr": f"⏱️ Timeout {timeout}s", "returncode": -1} - except Exception as e: - return {"stdout": "", "stderr": str(e), "returncode": -1} - - -def run_command_smart(cmd: str, timeout: int = 60) -> dict: - res = run_raw_command(cmd, timeout=timeout) - stdout = res["stdout"].strip() - - if res["returncode"] == 0 and not _is_empty_result(stdout): - return res + from PIL import Image, ImageDraw - is_curl_search = "curl" in cmd and any(x in cmd for x in [ - "duckduckgo", "google", "bing", "wikipedia", "reddit", - "hackernews", "hn.algolia", "arxiv", "news", "search" - ]) + img, ow, oh = _capture_raw() + if img is None: + return {"data": "", "width": 1920, "height": 1080, "mouse_x": 0, "mouse_y": 0} + + mx, my = (force_mx, force_my) if force_mx is not None else _get_mouse_pos() + + sw = int(ow * scale) + sh = int(oh * scale) + grid_img = img.resize((sw, sh), Image.LANCZOS) + draw = ImageDraw.Draw(grid_img, "RGBA") + + # خطوط Grid كل 100px + step_x = int(100 * sw / ow) + step_y = int(100 * sh / oh) + gc = (255, 255, 255, 40) + lc = (0, 255, 180, 210) + + x_sc, x_r = step_x, 100 + while x_sc < sw: + draw.line([(x_sc, 0), (x_sc, sh)], fill=gc, width=1) + draw.rectangle([(x_sc+1, 2), (x_sc+34, 15)], fill=(0, 0, 0, 170)) + draw.text((x_sc+2, 3), str(x_r), fill=lc) + x_sc += step_x; x_r += 100 + + y_sc, y_r = step_y, 100 + while y_sc < sh: + draw.line([(0, y_sc), (sw, y_sc)], fill=gc, width=1) + draw.rectangle([(2, y_sc+1), (38, y_sc+14)], fill=(0, 0, 0, 170)) + draw.text((3, y_sc+2), str(y_r), fill=lc) + y_sc += step_y; y_r += 100 + + # cursor + msx = int(mx * sw / ow) + msy = int(my * sh / oh) + r = 11 + draw.ellipse([(msx-r, msy-r), (msx+r, msy+r)], outline=(255, 50, 50, 240), width=2) + draw.line([(msx-18, msy), (msx+18, msy)], fill=(255, 50, 50, 200), width=1) + draw.line([(msx, msy-18), (msx, msy+18)], fill=(255, 50, 50, 200), width=1) + + # header bar + final = grid_img.convert("RGB") + draw2 = ImageDraw.Draw(final) + draw2.rectangle([(0, 0), (sw, 20)], fill=(0, 0, 0)) + draw2.text((4, 3), f"SCREEN {ow}x{oh} | MOUSE:({mx},{my}) | GRID=100px", fill=(0, 220, 160)) + draw2.rectangle([(0, sh-20), (sw, sh)], fill=(0, 0, 0)) + draw2.text((4, sh-17), "CLICK COORDS = numbers on grid lines (real screen pixels)", fill=(180, 180, 70)) + + buf = io.BytesIO() + final.save(buf, format="JPEG", quality=quality, optimize=True) + data = base64.b64encode(buf.getvalue()).decode() + + return {"data": data, "width": ow, "height": oh, "mouse_x": mx, "mouse_y": my} + + +def _frame_hash(data: str) -> str: + """hash سريع للـ base64 للكشف عن تغيير الصورة.""" + return hashlib.md5(data[:2000].encode()).hexdigest() + + +# ════════════════════════════════════════════════════════════════ +# ── SafeSearch Enforcement ──────────────────────────────────── +# ════════════════════════════════════════════════════════════════ + +_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 Execution ──────────────────────────────────────── +# ════════════════════════════════════════════════════════════════ + +_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) - if not is_curl_search: - return res - query = _extract_query_from_cmd(cmd) - if not query or len(query) < 3: - words = [w for w in cmd.split() if len(w) > 3 and not w.startswith('-') - and 'http' not in w and 'python3' not in w and 'curl' not in w] - query = ' '.join(words[:5]) if words else "" +async def run_cmd(cmd: str, timeout: int = 60) -> dict: + """ينفّذ أمر bash مع rate limit وsafe search.""" + cmd = _sanitize_pkill(_safe_search(cmd)) + async with _terminal_sem: + env = {**os.environ, "DISPLAY": SHARED_DISPLAY, + "PYTHONIOENCODING": "utf-8", "LANG": "en_US.UTF-8"} - if not query: - return res + 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} - sources = _build_search_sources(query) - tried_names = [] - all_results = [] - - for source in sources: - src_name = source["name"] - tried_names.append(f"🔍 {src_name}") - src_res = run_raw_command(source["cmd"], timeout=25) - src_out = src_res["stdout"].strip() - - if not _is_empty_result(src_out): - combined_header = f"[مصدر بديل: {src_name}]\n{'='*50}\n" - all_results.append(src_out) - for source2 in sources: - if source2["name"] != src_name: - s2 = run_raw_command(source2["cmd"], timeout=20) - s2_out = s2["stdout"].strip() - if not _is_empty_result(s2_out): - all_results.append(f"\n[مصدر إضافي: {source2['name']}]\n{s2_out}") - break - final_out = combined_header + "\n\n".join(all_results) - return { - "stdout": final_out[:15000], - "stderr": "", - "returncode": 0, - "_sources_tried": tried_names, - "_fallback_used": src_name, - } + return await asyncio.to_thread(_exec) - return { - "stdout": f"[لم تُرجع أي مصادر نتائج لـ: {query}]\nالمصادر: {', '.join(tried_names[:5])}", - "stderr": res["stderr"], - "returncode": -1, - "_sources_tried": tried_names, - } +# ════════════════════════════════════════════════════════════════ +# ── xdotool helpers ────────────────────────────────────────── +# ════════════════════════════════════════════════════════════════ -async def xdo(args: list, timeout=10) -> dict: - """ينفّذ xdotool في thread منفصل حتى لا يُجمّد حلقة asyncio أثناء الانتظار.""" - disp = DISPLAY_OF() +async def xdo(args: list, timeout: int = 10) -> dict: + env = {**os.environ, "DISPLAY": SHARED_DISPLAY} def _run(): - r = subprocess.run(["xdotool"] + args, - env={**os.environ, "DISPLAY": disp}, - timeout=timeout, capture_output=True, text=True) + r = subprocess.run( + ["xdotool"] + args, env=env, + timeout=timeout, capture_output=True, text=True + ) return {"rc": r.returncode, "out": r.stdout, "err": r.stderr} return await asyncio.to_thread(_run) -async def type_text_smart(text: str) -> dict: - """ - كتابة نص ذكية — تدعم العربية والإنجليزية: - - للنصوص الإنجليزية: xdotool type مباشرة - - للنصوص العربية أو المختلطة: xclip clipboard ثم ctrl+v - """ +async def type_smart(text: str) -> dict: + """كتابة نص ذكية: clipboard للعربية، xdotool type للإنجليزية.""" has_arabic = bool(re.search(r'[\u0600-\u06FF]', text)) + env = {**os.environ, "DISPLAY": SHARED_DISPLAY} if has_arabic: - # طريقة Clipboard لضمان كتابة العربية بشكل صحيح - try: - disp = DISPLAY_OF() - def _xclip_write(): - proc = subprocess.Popen( - ["xclip", "-selection", "clipboard"], - stdin=subprocess.PIPE, - env={**os.environ, "DISPLAY": disp} - ) - proc.communicate(text.encode("utf-8")) - await asyncio.to_thread(_xclip_write) - await asyncio.sleep(0.15) - # Focus + paste - await xdo(["key", "--clearmodifiers", "ctrl+v"]) - return {"success": True, "method": "clipboard+paste", "arabic": True} - except Exception as e: - # fallback: xdotool type - r = await xdo(["type", "--clearmodifiers", "--delay", "50", text]) - return {"success": r["rc"] == 0, "method": "xdotool_fallback", "error": r["err"]} + def _paste(): + p = subprocess.Popen( + ["xclip", "-selection", "clipboard"], + stdin=subprocess.PIPE, env=env + ) + p.communicate(text.encode("utf-8")) + await asyncio.to_thread(_paste) + await asyncio.sleep(0.15) + await xdo(["key", "--clearmodifiers", "ctrl+v"]) + return {"method": "clipboard+paste"} else: - # إنجليزية: xdotool type مباشرة - r = await xdo(["type", "--clearmodifiers", "--delay", "30", text]) - return {"success": r["rc"] == 0, "method": "xdotool_direct"} + r = await xdo(["type", "--clearmodifiers", "--delay", "25", text]) + return {"method": "xdotool", "rc": r["rc"]} + + +# ════════════════════════════════════════════════════════════════ +# ── Search Sources (8 sources with smart fallback) ────────── +# ════════════════════════════════════════════════════════════════ + +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": "Google News RSS", + "cmd": f"curl -sL --max-time 15 'https://news.google.com/rss/search?q={q}&hl=ar&gl=AR&ceid=AR:ar' | python3 -c \"import sys,re;x=sys.stdin.read();t=re.findall(r'<!\\[CDATA\\[(.*?)\\]\\]>|(.*?)',x);[(print(str(i+1)+'. '+(a or b).strip()[:160])) for i,(a,b) in enumerate(t[1:8]) if (a or b).strip()]\"" + }, + { + "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": "Bing News RSS", + "cmd": f"curl -sL --max-time 15 'https://www.bing.com/news/search?q={q}&format=RSS' | python3 -c \"import sys,re;x=sys.stdin.read();items=re.findall(r'(.*?)',x,re.DOTALL);[print(str(i+1)+'. '+re.sub('<[^>]+>','',re.search(r'(.*?)',it).group(1) if re.search(r'',it) else '')) for i,it in enumerate(items[:6])]\"" + }, + { + "name": "ArXiv", + "cmd": f"curl -s --max-time 12 'https://export.arxiv.org/api/query?search_query=all:{q}&max_results=5' | python3 -c \"import sys,re;x=sys.stdin.read();t=re.findall(r'<title>(.*?)',x)[1:];[print(str(i+1)+'. '+ti.strip()) for i,ti in enumerate(t[: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 open_browser_smart(url: str = "") -> str: - """يفتح المتصفح بطريقة ذكية مع fallback""" - browser_cmd = BROWSER - if not url: - url = "about:blank" +def _result_empty(s: str) -> bool: + if not s or len(s.strip()) < 15: + return True + bad = ["no direct answer", "ans: none", "error", "not found", "[]", "empty"] + return all(b in s.lower() for b in bad[:1]) - # جرب أولاً: BROWSER العادي - cmd = f"{browser_cmd} --new-window '{url}' &" - proc = subprocess.Popen(cmd, shell=True, - env={**os.environ, "DISPLAY": DISPLAY_OF()}, - stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) - return f"{browser_cmd} → {url}" +async def run_cmd_smart(cmd: str, timeout: int = 60) -> dict: + """يُنفّذ أمر وإذا فشل يجرب 8 مصادر بحث بديلة.""" + res = await run_cmd(cmd, timeout) + if res["returncode"] == 0 and not _result_empty(res["stdout"]): + return res -# ─── Simple Command Parser ───────────────────────────── -# يفهم أوامر بسيطة ويحولها لأفعال داخلية + # هل هو أمر بحث؟ + 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 -def parse_simple_command(raw: str) -> dict | None: - """ - يُحلّل الأوامر البسيطة النصية وتُعيد action dict - أمثلة: - open pc → {"action":"screenshot"} - open firefox → {"action":"open_browser","url":""} - open chrome → {"action":"open_browser","url":""} - open browser → {"action":"open_browser","url":""} - open url https://google.com → {"action":"open_browser","url":"https://..."} - mouse go x500 y300 → {"action":"mouse_move","x":500,"y":300} - mouse go 500 300 → {"action":"mouse_move","x":500,"y":300} - click x500 y300 → {"action":"mouse_click","x":500,"y":300,"button":"left"} - click 500 300 → {"action":"mouse_click","x":500,"y":300,"button":"left"} - rclick x500 y300 → {"action":"mouse_click","x":500,"y":300,"button":"right"} - dclick x500 y300 → {"action":"mouse_click","x":500,"y":300,"double":true} - type hello world → {"action":"keyboard_type","text":"hello world"} - key enter → {"action":"keyboard_hotkey","keys":["Return"]} - key ctrl+c → {"action":"keyboard_hotkey","keys":["ctrl","c"]} - scroll up → {"action":"scroll","x":960,"y":540,"clicks":3} - scroll down → {"action":"scroll","x":960,"y":540,"clicks":-3} - screenshot → {"action":"screenshot"} - screen info → {"action":"screen_info"} - """ - s = raw.strip().lower() - - # open pc / open computer / open screen - if re.match(r'^open\s+(pc|computer|screen|desktop)$', s): - return {"action": "screenshot"} - - # open firefox / chrome / browser / browser url - m = re.match(r'^open\s+(firefox|chrome|chromium|browser|web|internet)(\s+(.+))?$', s) - if m: - url = (m.group(3) or "").strip() - if url and not url.startswith("http"): - url = "https://" + url - return {"action": "open_browser", "url": url or ""} - - # open url - m = re.match(r'^open\s+url\s+(\S+)$', s) - if m: - url = m.group(1) - if not url.startswith("http"): - url = "https://" + url - return {"action": "open_browser", "url": url} - - # mouse go x500 y300 OR mouse go 500 300 - m = re.match(r'^mouse\s+(?:go|move|to)\s+x?(\d+)\s+y?(\d+)$', s) - if m: - return {"action": "mouse_move", "x": int(m.group(1)), "y": int(m.group(2))} - - # click x500 y300 OR click 500 300 - m = re.match(r'^click\s+x?(\d+)\s+y?(\d+)$', s) - if m: - return {"action": "mouse_click", "x": int(m.group(1)), "y": int(m.group(2)), "button": "left"} - - # rclick / right click - m = re.match(r'^r(?:ight)?click\s+x?(\d+)\s+y?(\d+)$', s) - if m: - return {"action": "mouse_click", "x": int(m.group(1)), "y": int(m.group(2)), "button": "right"} - - # dclick / double click - m = re.match(r'^d(?:ouble)?click\s+x?(\d+)\s+y?(\d+)$', s) - if m: - return {"action": "mouse_click", "x": int(m.group(1)), "y": int(m.group(2)), "button": "left", "double": True} - - # type (يحتفظ بالحالة الأصلية) - m = re.match(r'^type\s+(.+)$', raw.strip(), re.IGNORECASE) - if m: - return {"action": "keyboard_type", "text": m.group(1)} - - # key - m = re.match(r'^key\s+(.+)$', s) - if m: - k = m.group(1).strip() - key_map = { - "enter": "Return", "return": "Return", "esc": "Escape", "escape": "Escape", - "tab": "Tab", "space": "space", "backspace": "BackSpace", "delete": "Delete", - "up": "Up", "down": "Down", "left": "Left", "right": "Right", - "home": "Home", "end": "End", "pageup": "Prior", "pagedown": "Next", - "f1": "F1", "f2": "F2", "f3": "F3", "f4": "F4", "f5": "F5", - } - if '+' in k: - parts = [p.strip() for p in k.split('+')] - keys = [key_map.get(p, p) for p in parts] - else: - keys = [key_map.get(k, k)] - return {"action": "keyboard_hotkey", "keys": keys} - - # scroll up / scroll down / scroll - m = re.match(r'^scroll\s+(up|down|(\-?\d+))$', s) - if m: - w, h = _get_screen_size() - direction = m.group(1) - if direction == "up": - clicks = 4 - elif direction == "down": - clicks = -4 - else: - clicks = int(direction) - return {"action": "scroll", "x": w // 2, "y": h // 2, "clicks": clicks} + # استخرج query + 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 - # screenshot - if s in ("screenshot", "screen", "ss", "snap"): - return {"action": "screenshot"} + sources = _search_sources(query) + results = [] + for src in sources: + 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 - # screen info - if re.match(r'^screen\s+info$', s): - return {"action": "screen_info"} + if results: + return {"stdout": "\n\n".join(results)[:15000], "stderr": "", "returncode": 0} + return res - return None # لم يُعرف +# ════════════════════════════════════════════════════════════════ +# ── FastAPI App ─────────────────────────────────────────────── +# ════════════════════════════════════════════════════════════════ -async def broadcast(msg: dict): - txt = json.dumps(msg, ensure_ascii=False) - dead = [] - for ws in active_connections: - try: await ws.send_text(txt) - except: dead.append(ws) - for ws in dead: - if ws in active_connections: - active_connections.remove(ws) +app = FastAPI(title="Z-Computer-Mode v8") +app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_credentials=True, + allow_methods=["*"], allow_headers=["*"]) + +active_connections: list[WebSocket] = [] + + +@app.get("/", response_class=HTMLResponse) +async def root(): + w, h = _get_screen_size() + n = len(_active_sessions) + return f""" +Z Computer Mode v8 + + +

🖥️ THE Z AI — Computer Mode Server v8

+

✅ Server RUNNING

+

🌐 Browser: {BROWSER}

+

📐 Screen: {w}x{h}

+

👥 Active sessions: {n}

+

🖥️ Display: {SHARED_DISPLAY}

+

🔌 WebSocket: wss://your-space.hf.space/ws

+
+

Endpoints: /health · /screenshot · /ws (WebSocket)

+""" -# ─── Action Handler ────────────────────────────────── +# ════════════════════════════════════════════════════════════════ +# ── Action Handler ──────────────────────────────────────────── +# ════════════════════════════════════════════════════════════════ -async def handle_action(ws: WebSocket, msg: dict): +async def handle_action(ws: WebSocket, msg: dict, sess: dict): action = msg.get("action", "") - data = msg.get("data", {}) + data = msg.get("data", {}) + sid = id(ws) async def send(obj): - await ws.send_text(json.dumps(obj, ensure_ascii=False)) - - async def auto_shot_grid(label="", delay=0.5, - force_mx: int | None = None, - force_my: int | None = None): - """ - screenshot مع grid. - force_mx/force_my: ارسم الـ cursor في هذا الموضع بالضبط (بعد mouse_move/click) - بدلاً من إعادة قراءته من X server — يُصلح race condition. - """ + try: + await ws.send_text(json.dumps(obj, ensure_ascii=False)) + except Exception: + pass + + async def shot(label: str = "", delay: float = 0.5, + force_mx: int | None = None, force_my: int | None = None): + """لقطة شاشة مع grid — مع rate limiting و delta suppression.""" await asyncio.sleep(delay) - # ── يعمل في thread منفصل حتى لا يُجمّد حلقة asyncio ── + + # Rate limit: لا أكثر من لقطة كل 0.3 ثانية + now = time.time() + if now - sess.get("last_shot_ts", 0) < 0.3: + await asyncio.sleep(0.3 - (now - sess["last_shot_ts"])) + result = await asyncio.to_thread( - capture_screen_with_grid, scale=0.65, quality=72, - force_mx=force_mx, force_my=force_my) - if result["data"]: - await send({ - "type": "screenshot", - "data": result["data"], - "ts": int(time.time() * 1000), - "auto": True, - "label": label, - "screen_width": result["width"], - "screen_height": result["height"], - "mouse_x": result["mouse_x"], - "mouse_y": result["mouse_y"], - "has_grid": True, - }) - return result["data"] + capture_with_grid, 0.65, 72, force_mx, force_my + ) + sess["last_shot_ts"] = time.time() + + if not result["data"]: + return + + # Delta suppression: لا ترسل إذا الصورة لم تتغير + fh = _frame_hash(result["data"]) + if fh == sess.get("last_frame_hash", "") and not label.startswith("force"): + return + sess["last_frame_hash"] = fh - # ── screenshot ──────────────────────────────────── - if action == "screenshot": - result = await asyncio.to_thread(capture_screen_with_grid, scale=0.65, quality=75) await send({ "type": "screenshot", "data": result["data"], "ts": int(time.time() * 1000), + "auto": True, + "label": label, "screen_width": result["width"], "screen_height": result["height"], "mouse_x": result["mouse_x"], @@ -1011,352 +700,320 @@ async def handle_action(ws: WebSocket, msg: dict): "has_grid": True, }) - # ── simple_command — الأوامر البسيطة ────────────── - elif action == "simple_command": - raw_cmd = data.get("cmd", "").strip() - parsed = parse_simple_command(raw_cmd) - if parsed is None: - await send({ - "type": "simple_command_result", - "ok": False, - "cmd": raw_cmd, - "error": f"أمر غير معروف: '{raw_cmd}'\nالأوامر المتاحة: open pc, open firefox, mouse go x y, click x y, rclick x y, type TEXT, key ENTER, scroll up/down, screenshot" - }) - return - # نفّذ الأمر المُحلّل - await send({"type": "simple_command_result", "ok": True, "cmd": raw_cmd, "parsed": parsed}) - # أعِد تشغيل نفس الـ handle_action مع الأمر المُحلّل - sub_msg = {"action": parsed["action"], "data": {k: v for k, v in parsed.items() if k != "action"}} - await handle_action(ws, sub_msg) - return + # ── screenshot ─────────────────────────────────────────── + if action == "screenshot": + sess["last_frame_hash"] = "" # force send + result = await asyncio.to_thread(capture_with_grid, 0.65, 75) + await send({ + "type": "screenshot", + "data": result["data"], + "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, + }) - # ── terminal ────────────────────────────────────── + # ── terminal ───────────────────────────────────────────── elif action == "terminal": cmd = data.get("cmd", "") - timeout = int(data.get("timeout", 60)) if not cmd: - await send({"type": "terminal_result", "cmd": "", "stdout": "", - "stderr": "no command", "returncode": -1}) + await send({"type": "terminal_result", "stdout": "", "stderr": "no cmd", "returncode": -1}) return - res = await asyncio.to_thread(run_command_smart, cmd, timeout) + 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"], - "fallback_used": res.get("_fallback_used", None), - "sources_tried": res.get("_sources_tried", []), }) - await auto_shot_grid(f"بعد: {cmd[:45]}", delay=0.4) + asyncio.create_task(shot(f"after: {cmd[:40]}", delay=0.4)) - # ── mouse_move ──────────────────────────────────── + # ── mouse_move ─────────────────────────────────────────── elif action == "mouse_move": x, y = int(data.get("x", 0)), int(data.get("y", 0)) - # --sync ينتظر حتى يؤكد X server استلام الأمر await xdo(["mousemove", "--sync", str(x), str(y)]) await send({"type": "ack", "action": "mouse_move", "x": x, "y": y}) - # مرّر الإحداثيات مباشرةً لتجنّب race condition مع X server - await auto_shot_grid(f"ماوس → ({x},{y})", delay=0.2, - force_mx=x, force_my=y) + asyncio.create_task(shot(f"move ({x},{y})", delay=0.2, force_mx=x, force_my=y)) - # ── mouse_click ─────────────────────────────────── + # ── mouse_click ────────────────────────────────────────── elif action == "mouse_click": - x, y = int(data.get("x", 0)), int(data.get("y", 0)) - btn_num = {"left": "1", "middle": "2", "right": "3"}.get(data.get("button", "left"), "1") + x, y = int(data.get("x", 0)), int(data.get("y", 0)) + btn = {"left": "1", "middle": "2", "right": "3"}.get(data.get("button", "left"), "1") + double = data.get("double", False) await xdo(["mousemove", "--sync", str(x), str(y)]) - await asyncio.sleep(0.08) - if data.get("double"): - await xdo(["click", "--repeat", "2", "--delay", "100", btn_num]) + await asyncio.sleep(0.07) + if double: + await xdo(["click", "--repeat", "2", "--delay", "100", btn]) else: - await xdo(["click", btn_num]) - btn_name = {"1": "left", "2": "middle", "3": "right"}.get(btn_num, "left") + await xdo(["click", btn]) + btn_name = {"1": "left", "2": "middle", "3": "right"}.get(btn, "left") await send({"type": "ack", "action": "mouse_click", "x": x, "y": y, "button": btn_name}) - # مرّر الإحداثيات مباشرةً لتجنّب race condition - await auto_shot_grid(f"نقر {btn_name} → ({x},{y})", delay=0.5, - force_mx=x, force_my=y) + asyncio.create_task(shot(f"click {btn_name} ({x},{y})", delay=0.5, force_mx=x, force_my=y)) - # ── keyboard_type ───────────────────────────────── + # ── mouse_drag ─────────────────────────────────────────── + 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)) + await xdo(["mousemove", str(x1), str(y1)]) + await xdo(["mousedown", "1"]) + await asyncio.sleep(0.1) + await xdo(["mousemove", str(x2), str(y2)]) + await asyncio.sleep(0.1) + await xdo(["mouseup", "1"]) + await send({"type": "ack", "action": "mouse_drag"}) + asyncio.create_task(shot("after drag", delay=0.4, force_mx=x2, force_my=y2)) + + # ── keyboard_type ──────────────────────────────────────── elif action == "keyboard_type": - text = _enforce_safesearch(data.get("text", "")) + text = _safe_search(data.get("text", "")) if text: - result = await type_text_smart(text) - await send({"type": "ack", "action": "keyboard_type", - "method": result["method"], "text_len": len(text)}) - else: - await send({"type": "ack", "action": "keyboard_type"}) - await auto_shot_grid("بعد الكتابة", delay=0.5) + res = await type_smart(text) + await send({"type": "ack", "action": "keyboard_type", "method": res["method"]}) + asyncio.create_task(shot("after type", delay=0.5)) - # ── keyboard_hotkey ─────────────────────────────── + # ── keyboard_hotkey ────────────────────────────────────── elif action == "keyboard_hotkey": keys = data.get("keys", []) if keys: await xdo(["key", "--clearmodifiers", "+".join(keys)]) await send({"type": "ack", "action": "keyboard_hotkey", "keys": keys}) - await auto_shot_grid("بعد الاختصار", delay=0.5) + asyncio.create_task(shot("after hotkey", delay=0.4)) - # ── keyboard_press ──────────────────────────────── + # ── keyboard_press ─────────────────────────────────────── elif action == "keyboard_press": key = data.get("key", "") if key: await xdo(["key", "--clearmodifiers", key]) await send({"type": "ack", "action": "keyboard_press"}) - await auto_shot_grid("بعد المفتاح", delay=0.4) - - # ── clipboard_write ─────────────────────────────── - elif action == "clipboard_write": - text = data.get("text", "") - try: - disp = DISPLAY_OF() - def _xclip_write2(): - proc = subprocess.Popen( - ["xclip", "-selection", "clipboard"], - stdin=subprocess.PIPE, - env={**os.environ, "DISPLAY": disp} - ) - proc.communicate(text.encode("utf-8")) - await asyncio.to_thread(_xclip_write2) - await send({"type": "ack", "action": "clipboard_write", "length": len(text)}) - except Exception as e: - await send({"type": "error", "action": "clipboard_write", "msg": str(e)}) + asyncio.create_task(shot("after key", delay=0.4)) - # ── clipboard_read ──────────────────────────────── - elif action == "clipboard_read": - res = await asyncio.to_thread(run_raw_command, "xclip -selection clipboard -o", 5) - await send({"type": "clipboard_content", "text": res["stdout"]}) - - # ── scroll ──────────────────────────────────────── + # ── scroll ─────────────────────────────────────────────── elif action == "scroll": - x, y = int(data.get("x", 0)), int(data.get("y", 0)) - clicks = int(data.get("clicks", 3)) - btn = "4" if clicks > 0 else "5" + x, y = int(data.get("x", 960)), int(data.get("y", 540)) + clicks = max(-5, min(5, int(data.get("clicks", 3)))) + btn = "4" if clicks > 0 else "5" await xdo(["mousemove", str(x), str(y)]) for _ in range(abs(clicks)): await xdo(["click", btn]) - await asyncio.sleep(0.03) + await asyncio.sleep(0.025) await send({"type": "ack", "action": "scroll", "clicks": clicks}) - await auto_shot_grid("بعد التمرير", delay=0.4) + asyncio.create_task(shot("after scroll", delay=0.4)) - # ── open_app ────────────────────────────────────── + # ── clipboard_write ────────────────────────────────────── + elif action == "clipboard_write": + text = data.get("text", "") + env = {**os.environ, "DISPLAY": SHARED_DISPLAY} + def _clip(): + p = subprocess.Popen(["xclip", "-selection", "clipboard"], stdin=subprocess.PIPE, env=env) + p.communicate(text.encode("utf-8")) + await asyncio.to_thread(_clip) + await send({"type": "ack", "action": "clipboard_write", "length": len(text)}) + + # ── clipboard_read ─────────────────────────────────────── + elif action == "clipboard_read": + res = await run_cmd("xclip -selection clipboard -o", 5) + await send({"type": "clipboard_content", "text": res["stdout"]}) + + # ── paste ──────────────────────────────────────────────── + elif action == "paste": + text = data.get("text", "") + if text: + env = {**os.environ, "DISPLAY": SHARED_DISPLAY} + def _clip2(): + p = subprocess.Popen(["xclip", "-selection", "clipboard"], stdin=subprocess.PIPE, env=env) + p.communicate(text.encode("utf-8")) + await asyncio.to_thread(_clip2) + await asyncio.sleep(0.1) + await xdo(["key", "--clearmodifiers", "ctrl+v"]) + await send({"type": "ack", "action": "paste"}) + asyncio.create_task(shot("after paste", delay=0.5)) + + # ── open_app ───────────────────────────────────────────── elif action == "open_app": - app_cmd = data.get("cmd", "") - if app_cmd: - fixed_cmd = _enforce_safesearch(app_cmd) - # إصلاح firefox-esr - if "firefox-esr" in fixed_cmd: - esr_check = await asyncio.to_thread(subprocess.run, ["which", "firefox-esr"], capture_output=True, text=True) - if esr_check.returncode != 0: - fixed_cmd = fixed_cmd.replace("firefox-esr", BROWSER) - - # ── حماية إضافية: حذف ملفات قفل Firefox المتبقية قبل فتح نافذة جديدة ── - # يمنع ظهور "Firefox is already running, but is not responding" حتى لو - # نسي الذكاء تضمين هذا التنظيف في أمره الخاص. - if any(b in fixed_cmd for b in ("firefox", "Firefox")): - def _clean_lock(): - try: - subprocess.run( - "rm -f ~/.mozilla/firefox/*/lock ~/.mozilla/firefox/*/.parentlock 2>/dev/null", - shell=True, executable="/bin/bash", timeout=3, capture_output=True - ) - except Exception: - pass - await asyncio.to_thread(_clean_lock) - - subprocess.Popen(fixed_cmd, shell=True, - env={**os.environ, "DISPLAY": DISPLAY_OF()}, - stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) - await send({"type": "ack", "action": "open_app", "cmd": fixed_cmd}) - - wait_time = 7.0 if any(x in app_cmd for x in ["firefox", "chromium", "chrome"]) else 4.0 - await asyncio.sleep(2.0) - await auto_shot_grid(f"فتح: {fixed_cmd[:40]} (2s)", delay=0) - await asyncio.sleep(wait_time - 2.0) - await auto_shot_grid(f"بعد تحميل: {fixed_cmd[:40]}", delay=0) - else: + cmd = _safe_search(data.get("cmd", "")) + if not cmd: await send({"type": "ack", "action": "open_app"}) + return - # ── open_browser — فتح المتصفح ──────────────────── + # أضف profile للـ firefox إذا لم يكن محدداً + profile = sess.get("profile", "") + if profile and any(b in cmd for b in ("firefox", "chromium")): + if "--profile" not in cmd: + cmd = cmd.replace(BROWSER, f"{BROWSER} --profile {profile}", 1) + + # نظّف lock files + if "firefox" in cmd.lower(): + await run_cmd("rm -f ~/.mozilla/firefox/*/lock ~/.mozilla/firefox/*/.parentlock 2>/dev/null", 3) + + env = {**os.environ, "DISPLAY": SHARED_DISPLAY} + proc = subprocess.Popen( + cmd, shell=True, env=env, + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL + ) + if any(b in cmd for b in ("firefox", "chromium", "chrome")): + sess["browser_proc"] = proc + + await send({"type": "ack", "action": "open_app", "cmd": cmd}) + # screenshot مبكر ثم واحد بعد التحميل + asyncio.create_task(shot("opening app (2s)", delay=2.0)) + asyncio.create_task(shot("app loaded (7s)", delay=7.0)) + + # ── open_browser ───────────────────────────────────────── elif action == "open_browser": - url = data.get("url", "") - if not url: - url = "about:blank" - browser_result = open_browser_smart(url) - await send({"type": "ack", "action": "open_browser", "result": browser_result}) - await asyncio.sleep(2.0) - await auto_shot_grid(f"فتح المتصفح: {url[:50]}", delay=0) - await asyncio.sleep(5.0) - await auto_shot_grid("بعد تحميل المتصفح", delay=0) - - # ── mouse_drag ──────────────────────────────────── - 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)) - await xdo(["mousemove", str(x1), str(y1)]) - await xdo(["mousedown", "1"]) - await asyncio.sleep(0.1) - await xdo(["mousemove", str(x2), str(y2)]) - await asyncio.sleep(0.1) - await xdo(["mouseup", "1"]) - await send({"type": "ack", "action": "mouse_drag"}) - # cursor ينتهي عند x2,y2 - await auto_shot_grid("بعد السحب", delay=0.4, force_mx=x2, force_my=y2) - - # ── start_stream / stop_stream: غير مستخدمة من الواجهة الحالية، أُزيلت لمنع - # تشغيل حلقة تصوير مستمرة في الخلفية بلا داعٍ (كانت تستهلك موارد حتى بعد إغلاق الاتصال) ── - - # ── open_tab ────────────────────────────────────── + url = _safe_search(data.get("url", "") or "about:blank") + profile = sess.get("profile", "") + cmd = get_browser_cmd(url, profile) + env = {**os.environ, "DISPLAY": SHARED_DISPLAY} + await run_cmd("rm -f ~/.mozilla/firefox/*/lock ~/.mozilla/firefox/*/.parentlock 2>/dev/null", 3) + proc = subprocess.Popen(cmd, env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + sess["browser_proc"] = proc + await send({"type": "ack", "action": "open_browser", "url": url}) + asyncio.create_task(shot("browser opening (2s)", delay=2.0)) + asyncio.create_task(shot("browser loaded (7s)", delay=7.0)) + + # ── open_tab ───────────────────────────────────────────── elif action == "open_tab": - url = _enforce_safesearch(data.get("url", "") or "about:blank") + url = _safe_search(data.get("url", "") or "about:blank") await xdo(["key", "--clearmodifiers", "ctrl+t"]) await asyncio.sleep(0.4) await xdo(["key", "--clearmodifiers", "ctrl+l"]) await asyncio.sleep(0.2) - await type_text_smart(url) + await type_smart(url) await asyncio.sleep(0.2) await xdo(["key", "--clearmodifiers", "Return"]) await send({"type": "ack", "action": "open_tab", "url": url}) - await auto_shot_grid("بعد فتح تبويب جديد", delay=1.5) + asyncio.create_task(shot("after open_tab", delay=1.5)) - # ── close_tab ───────────────────────────────────── + # ── close_tab ──────────────────────────────────────────── elif action == "close_tab": await xdo(["key", "--clearmodifiers", "ctrl+w"]) await send({"type": "ack", "action": "close_tab"}) - await auto_shot_grid("بعد إغلاق التبويب", delay=0.5) + asyncio.create_task(shot("after close_tab", delay=0.5)) - # ── browser_back ────────────────────────────────── + # ── browser_back ───────────────────────────────────────── elif action == "browser_back": await xdo(["key", "--clearmodifiers", "alt+Left"]) await send({"type": "ack", "action": "browser_back"}) - await auto_shot_grid("بعد الرجوع", delay=0.8) + asyncio.create_task(shot("after back", delay=0.8)) - # ── browser_forward ─────────────────────────────── + # ── browser_forward ────────────────────────────────────── elif action == "browser_forward": await xdo(["key", "--clearmodifiers", "alt+Right"]) await send({"type": "ack", "action": "browser_forward"}) - await auto_shot_grid("بعد التقدم", delay=0.8) + asyncio.create_task(shot("after forward", delay=0.8)) - # ── browser_search: اختصار لفتح بحث مباشرة في التبويب الحالي ── + # ── browser_search ─────────────────────────────────────── elif action == "browser_search": - url = _enforce_safesearch(data.get("url", "") or data.get("query", "")) + url = _safe_search(data.get("url", "") or data.get("query", "")) await xdo(["key", "--clearmodifiers", "ctrl+l"]) await asyncio.sleep(0.2) - await type_text_smart(url) - await asyncio.sleep(0.2) + await type_smart(url) + await asyncio.sleep(0.15) await xdo(["key", "--clearmodifiers", "Return"]) - await send({"type": "ack", "action": "browser_search", "url": url}) - await auto_shot_grid("بعد البحث", delay=1.5) + await send({"type": "ack", "action": "browser_search"}) + asyncio.create_task(shot("after search", delay=1.5)) - # ── screen_info ─────────────────────────────────── + # ── screen_info ────────────────────────────────────────── elif action == "screen_info": - w, h = await asyncio.to_thread(_get_screen_size) + w, h = await asyncio.to_thread(_get_screen_size) mx, my = await asyncio.to_thread(_get_mouse_pos) await send({ "type": "screen_info", "width": w, "height": h, "mouse_x": mx, "mouse_y": my, "browser": BROWSER, + "display": SHARED_DISPLAY, }) - # ── paste ───────────────────────────────────────── - elif action == "paste": - text = data.get("text", "") - if text: - disp = DISPLAY_OF() - def _xclip_paste(): - proc = subprocess.Popen( - ["xclip", "-selection", "clipboard"], - stdin=subprocess.PIPE, - env={**os.environ, "DISPLAY": disp} - ) - proc.communicate(text.encode("utf-8")) - await asyncio.to_thread(_xclip_paste) - await asyncio.sleep(0.1) - await xdo(["key", "--clearmodifiers", "ctrl+v"]) - await send({"type": "ack", "action": "paste"}) - await auto_shot_grid("بعد اللصق", delay=0.5) - - # ── unknown ─────────────────────────────────────── + # ── unknown ────────────────────────────────────────────── else: - await send({ - "type": "error", - "msg": f"Unknown action: '{action}'. Available: screenshot, terminal, mouse_move, mouse_click, mouse_drag, keyboard_type, keyboard_hotkey, keyboard_press, clipboard_write, clipboard_read, scroll, open_app, open_tab, close_tab, browser_back, browser_forward, browser_search, open_browser, paste, simple_command, screen_info" - }) + await send({"type": "error", "msg": f"Unknown action: '{action}'"}) -# ─── WebSocket ─────────────────────────────────────── +# ════════════════════════════════════════════════════════════════ +# ── WebSocket Endpoint with Heartbeat ──────────────────────── +# ════════════════════════════════════════════════════════════════ @app.websocket("/ws") async def websocket_endpoint(ws: WebSocket): await ws.accept() active_connections.append(ws) - # ── عزل كامل: إنشاء Xvfb + متصفح خاصين بهذا الاتصال فقط ─�� - # كل اتصال WebSocket يحصل على بيئة كمبيوتر افتراضي مستقلة تماماً عن أي اتصال آخر. - # إذا فشل إنشاء جلسة معزولة لأي سبب (صلاحيات، موارد، Xvfb غير متاح...)، نرجع فوراً - # للوضع المشترك القديم (DISPLAY الأساسي) بدل أن ينهار الاتصال بصمت بالكامل. - session_display = _BASE_DISPLAY - try: - session_display = await asyncio.wait_for(create_session_for(ws), timeout=8.0) - except asyncio.TimeoutError: - print("[session] ⚠️ session creation timed out (>8s), falling back to shared display") - session_display = _BASE_DISPLAY - except Exception as e: - print(f"[session] ⚠️ failed to create isolated session, falling back to shared display: {e}") - session_display = _BASE_DISPLAY - _current_display.set(session_display) # يُربط بـ task هذا الاتصال — لا يؤثر على اتصالات أخرى + sess = await create_session(ws) + + async def _heartbeat(): + """Ping كل 20 ثانية لمنع قطع الاتصال من proxies.""" + 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: w, h = await asyncio.to_thread(_get_screen_size) await ws.send_text(json.dumps({ "type": "connected", - "screen_width": w, - "screen_height": h, + "screen_width": w, "screen_height": h, "browser": BROWSER, - "display": session_display, - "msg": f"Z Computer Mode v7 — Smart Control | Browser: {BROWSER} | Screen: {w}x{h} | Isolated session: {session_display}", - "commands": [ - "screenshot", "terminal {cmd}", "mouse_move {x,y}", "mouse_click {x,y,button}", - "keyboard_type {text}", "keyboard_hotkey {keys}", "scroll {x,y,clicks}", - "open_app {cmd}", "open_browser {url}", "simple_command {cmd}", - "Simple cmds: 'open firefox', 'mouse go x y', 'click x y', 'type TEXT', 'key enter'" - ] + "display": SHARED_DISPLAY, + "session_id": id(ws), + "msg": f"Z Computer Mode v8 | Browser: {BROWSER} | Screen: {w}x{h}", }, ensure_ascii=False)) - # أرسل screenshot أولية مع grid (لشاشة هذه الجلسة المعزولة فقط) - result = await asyncio.to_thread(capture_screen_with_grid, scale=0.65, quality=72) + + # لقطة شاشة أولية + result = await asyncio.to_thread(capture_with_grid, 0.65, 72) if result["data"]: await ws.send_text(json.dumps({ "type": "screenshot", "data": result["data"], "ts": int(time.time() * 1000), - "label": "الشاشة الأولية", + "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] ⚠️ error sending initial messages: {e}") + print(f"[ws] init error: {e}") + try: while True: raw = await ws.receive_text() - await handle_action(ws, json.loads(raw)) + try: + msg = json.loads(raw) + # تجاهل pong responses + 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] {e}") + print(f"[ws] error: {e}") finally: + hb_task.cancel() if ws in active_connections: active_connections.remove(ws) - # ── تحرير بيئة هذه الجلسة بالكامل (Xvfb + المتصفح) فوراً عند انقطاع الاتصال ── - await destroy_session_for(ws) + await destroy_session(ws) -# ─── REST ──────────────────────────────────────────── +# ════════════════════════════════════════════════════════════════ +# ── REST Endpoints ──────────────────────────────────────────── +# ════════════════════════════════════════════════════════════════ @app.get("/screenshot") async def rest_screenshot(): - result = capture_screen_with_grid(scale=0.7, quality=73) + result = await asyncio.to_thread(capture_with_grid, 0.7, 75) return JSONResponse({ "image": result["data"], "ts": int(time.time() * 1000), @@ -1367,55 +1024,56 @@ async def rest_screenshot(): "has_grid": True, }) -@app.get("/screenshot/clean") -async def rest_screenshot_clean(): - return JSONResponse({"image": capture_screen(0.75, 75), "ts": int(time.time() * 1000)}) - @app.post("/terminal") async def rest_terminal(body: dict): - return JSONResponse(run_command_smart(body.get("cmd", ""), body.get("timeout", 60))) - -@app.post("/simple") -async def rest_simple(body: dict): - """تنفيذ أمر بسيط عبر REST""" - raw = body.get("cmd", "").strip() - parsed = parse_simple_command(raw) - if not parsed: - return JSONResponse({"ok": False, "error": f"Unknown simple command: {raw}"}, status_code=400) - return JSONResponse({"ok": True, "parsed": parsed}) + return JSONResponse(await run_cmd_smart(body.get("cmd", ""), body.get("timeout", 60))) @app.get("/health") async def health(): - w, h = _get_screen_size() - mx, my = _get_mouse_pos() - browser_check = subprocess.run(["which", "firefox-esr"], capture_output=True, text=True) + w, h = await asyncio.to_thread(_get_screen_size) + mx, my = await asyncio.to_thread(_get_mouse_pos) return { "status": "ok", - "version": "v7-smart-control", + "version": "v8-stable", "browser": BROWSER, - "firefox_esr_available": browser_check.returncode == 0, - "screen_width": w, - "screen_height": h, - "mouse_x": mx, - "mouse_y": my, + "display": SHARED_DISPLAY, + "screen": f"{w}x{h}", + "mouse": f"{mx},{my}", + "active_sessions": len(_active_sessions), } -@app.get("/search/{query}") -async def quick_search(query: str): - res = run_command_smart( - f"curl -s --max-time 15 'https://api.duckduckgo.com/?q={urllib.parse.quote_plus(query)}&format=json&no_html=1'", - timeout=30 - ) - return JSONResponse(res) - - -if __name__ == "__main__": - port = int(os.environ.get("PORT", 7860)) - uvicorn.run("app:app", host="0.0.0.0", port=port, log_level="info") - +# ════════════════════════════════════════════════════════════════ +# ── Background Tasks ────────────────────────────────────────── +# ════════════════════════════════════════════════════════════════ +async def _cleanup_tmp(): + """ينظّف /tmp كل 5 دقائق لمنع امتلاء القرص في HuggingFace.""" + while True: + await asyncio.sleep(300) + try: + r = subprocess.run( + ["find", "/tmp", "-name", "zss_*", "-mmin", "+10", "-delete"], + capture_output=True, timeout=10 + ) + r2 = subprocess.run( + ["find", "/tmp", "-name", "zpc_*", "-mmin", "+60", + "-not", "-newer", "/tmp", "-delete"], + capture_output=True, timeout=10 + ) + except Exception as e: + print(f"[cleanup] {e}") +@app.on_event("startup") +async def startup(): + # تشغيل Xvfb المشترك + await asyncio.to_thread(_ensure_xvfb) + # تشغيل التنظيف الدوري + asyncio.create_task(_cleanup_tmp()) + print("✅ Z Computer Mode v8 ready") +if __name__ == "__main__": + port = int(os.environ.get("PORT", 7860)) + uvicorn.run("app:app", host="0.0.0.0", port=port, log_level="info")