"""
الوكيل الموحّد v28.0 — هيكلة جديدة كاملة
==========================================
كل النماذج في واجهة واحدة (Fusion) + قدرات تنفيذية حقيقية
المعمارية:
1. Fusion: Fan out → Collect → Synthesize (كل النماذج كنموذج واحد)
2. Tools: 15 أداة حقيقية (كود، بحث، GitHub، arXiv، تحليل...)
3. Memory: ذاكرة دائمة في /data/ + نسخ احتياطي GitHub + HF
4. Webhook reply للسرعة + background للمهام الطويلة
5. 24/7 keep-alive
مجاني 100% — لا تكلفة
"""
import asyncio, json, logging, os, re, sys, time, hashlib, math, random, base64
import urllib.parse, subprocess, tempfile, traceback
from typing import Optional, List, Dict
from datetime import datetime, timezone, timedelta
import aiohttp
from aiohttp import web
logging.basicConfig(format="%(asctime)s - %(levelname)s - %(message)s", level=logging.INFO, stream=sys.stdout)
logger = logging.getLogger(__name__)
# ============================================================
# CONFIG
# ============================================================
BOT_TOKEN = os.environ.get("TELEGRAM_BOT_TOKEN", "")
GITHUB_TOKEN = os.environ.get("GITHUB_TOKEN", "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")
ZENMUX_KEY = os.environ.get("ZENMUX_API_KEY", "sk-ai-v1-b691e6444fda73a4dada1cc0a8c650a47b2fe9ea50fab999751d7c831fbe2a2e")
PUTER_KEYS = [
"eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCIsImtpZCI6InYyIn0.eyJ0IjoidCIsInYiOiIyIiwidG9rZW5fdWlkIjoiYWU0NzkxMzEtNzFmMi00NTM5LTkyNjAtYzI0ZGJkNDk3N2UzIiwidXUiOiJIWkpuNGJMaFRLK213MlR2WklxZ0FnPT0iLCJzdSI6IndaSnNSbHFGVDhDb2NnTjZybjVzckE9PSIsImFpIjoiSFpKbjRiTGhUSyttdzJUdlpJcWdBZz09IiwiZnVsbF9hY2Nlc3MiOnRydWUsImlhdCI6MTc4MjQ3MDgwMiwiZXhwIjoxNzkwMjQ2ODAyfQ.rk3pLgVDiWK1gAQFzre9NqnEtOkm25LxnxElRTubC40",
"eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCIsImtpZCI6InYyIn0.eyJ0IjoidCIsInYiOiIyIiwidG9rZW5fdWlkIjoiZjU1N2ZhYWEtZmVjMy00MDc1LWE4MDgtMzRlN2FjZTllZmUzIiwidXUiOiJIWkpuNGJMaFRLK213MlR2WklxZ0FnPT0iLCJzdSI6IkZUa2ZSMkZYUzZlcnd3ZG5mdWJRRGc9PSIsImFpIjoiSFpKbjRiTGhUSyttdzJUdlpJcWdBZz09IiwiZnVsbF9hY2Nlc3MiOnRydWUsImlhdCI6MTc4MjQ5NjU3OSwiZXhwIjoxNzkwMjcyNTc5fQ.-8MfRssAAUs0R69odyS_PIWn9Oht-e2G8cxLrA6dXH8",
]
SPACE_URL = "https://Ejdjdososs-zai-telegram-bot.hf.space"
WEBHOOK_PATH = "/webhook"
PORT = 7860
LOG = []
STATS = {"total": 0, "fusion": 0, "tools": 0, "fail": 0}
# ============================================================
# PERSISTENT MEMORY
# ============================================================
DATA_DIR = os.environ.get("DATA_DIR", "/data")
MEM_FILE = os.path.join(DATA_DIR, "memory.json")
os.makedirs(DATA_DIR, exist_ok=True)
class Memory:
def __init__(self):
self.data = {"chats": {}, "learned": {}, "cache": {}}
self._load()
def _load(self):
try:
if os.path.exists(MEM_FILE):
with open(MEM_FILE, "r", encoding="utf-8") as f:
self.data = json.load(f)
except: pass
def _save(self):
try:
with open(MEM_FILE, "w", encoding="utf-8") as f:
json.dump(self.data, f, ensure_ascii=False)
except: pass
def add(self, uid, role, content):
if uid not in self.data["chats"]: self.data["chats"][uid] = []
self.data["chats"][uid].append({"r": role, "c": content[:600], "t": time.time()})
self.data["chats"][uid] = self.data["chats"][uid][-20:]
self._save()
def ctx(self, uid, n=6):
return [{"role": m["r"], "content": m["c"]} for m in self.data["chats"].get(uid, [])[-n:]]
def learn(self, q, r):
h = hashlib.md5(q.lower().encode()).hexdigest()[:12]
self.data["learned"][h] = {"r": r, "t": time.time()}
self._save()
def recall(self, q):
h = hashlib.md5(q.lower().encode()).hexdigest()[:12]
e = self.data["learned"].get(h)
if e and time.time() - e["t"] < 86400: return e["r"]
return None
def cache_get(self, q):
h = hashlib.md5(q.lower().encode()).hexdigest()[:12]
e = self.data["cache"].get(h)
if e and time.time() - e["t"] < 3600: return e["r"]
return None
def cache_set(self, q, r):
h = hashlib.md5(q.lower().encode()).hexdigest()[:12]
self.data["cache"][h] = {"r": r, "t": time.time()}
if len(self.data["cache"]) > 200:
old = min(self.data["cache"].items(), key=lambda x: x[1]["t"])
del self.data["cache"][old[0]]
self._save()
def forget(self, uid):
if uid in self.data["chats"]: del self.data["chats"][uid]; self._save()
def stats(self):
return {"users": len(self.data["chats"]), "learned": len(self.data["learned"]), "cache": len(self.data["cache"])}
mem = Memory()
# ============================================================
# LLM PROVIDERS — كل مزود في دالة مستقلة
# ============================================================
SESSION = None
async def sess():
global SESSION
if SESSION is None or SESSION.closed:
SESSION = aiohttp.ClientSession(
timeout=aiohttp.ClientTimeout(total=15, connect=5, sock_read=12),
connector=aiohttp.TCPConnector(limit=100, limit_per_host=30, keepalive_timeout=120))
return SESSION
_pidx = 0
def _pkey():
global _pidx
_pidx += 1
return PUTER_KEYS[_pidx % len(PUTER_KEYS)]
async def p_puter(model, msgs, t=12):
"""Puter.js — GLM-5.2, GPT-5, DeepSeek, Mistral"""
s = await sess()
k = _pkey()
try:
async with s.post("https://api.puter.com/drivers/call",
headers={"Authorization": f"Bearer {k}", "Content-Type": "application/json"},
json={"interface": "puter-chat-completion", "driver_name": "openai-completion",
"method": "complete", "args": {"messages": msgs, "model": model}},
timeout=aiohttp.ClientTimeout(total=t)) as r:
if r.status == 200:
d = await r.json()
c = d.get("result", {}).get("message", {}).get("content", "")
if c and c.strip(): return c.strip()
except: pass
return None
async def p_github(model, msgs, t=12):
"""GitHub Models — GPT-4o, GPT-4o-mini (مجاني مع GitHub token)"""
if not GITHUB_TOKEN: return None
s = await sess()
try:
async with s.post("https://models.inference.ai.azure.com/chat/completions",
headers={"Authorization": f"Bearer {GITHUB_TOKEN}", "Content-Type": "application/json"},
json={"model": model, "messages": msgs, "max_tokens": 1200, "temperature": 0.7},
timeout=aiohttp.ClientTimeout(total=t)) as r:
if r.status == 200:
d = await r.json()
c = d.get("choices", [{}])[0].get("message", {}).get("content", "")
if c and c.strip(): return c.strip()
except: pass
return None
async def p_llm7(model, msgs, t=12):
"""LLM7.io — Codestral, Devstral (مجاني بدون مفتاح)"""
s = await sess()
try:
async with s.post("https://api.llm7.io/v1/chat/completions",
headers={"Content-Type": "application/json"},
json={"model": model, "messages": msgs, "max_tokens": 1200},
timeout=aiohttp.ClientTimeout(total=t)) as r:
if r.status == 200:
d = await r.json()
c = d.get("choices", [{}])[0].get("message", {}).get("content", "")
if c and c.strip(): return c.strip()
except: pass
return None
async def p_ovh(model, msgs, t=12):
"""OVHcloud — gpt-oss-20b (مجاني بدون مفتاح)"""
s = await sess()
try:
async with s.post("https://oai.endpoints.kepler.ai.cloud.ovh.net/v1/chat/completions",
headers={"Content-Type": "application/json"},
json={"model": model, "messages": msgs, "max_tokens": 1200},
timeout=aiohttp.ClientTimeout(total=t)) as r:
if r.status == 200:
d = await r.json()
c = d.get("choices", [{}])[0].get("message", {}).get("content", "")
if c and c.strip(): return c.strip()
except: pass
return None
async def p_poll(msgs, t=12):
"""Pollinations — openai (مجاني دائماً)"""
s = await sess()
try:
async with s.post("https://text.pollinations.ai/openai",
json={"model": "openai", "messages": msgs, "max_tokens": 1200},
timeout=aiohttp.ClientTimeout(total=t)) as r:
if r.status == 200:
d = await r.json()
c = d.get("choices", [{}])[0].get("message", {}).get("content", "")
if c and c.strip(): return c.strip()
except: pass
return None
# ============================================================
# FUSION — كل النماذج كنموذج واحد
# ============================================================
async def fanout(msgs):
"""أرسل لكل النماذج بالتوازي، اجمع كل الردود"""
tasks = {
asyncio.create_task(p_github("gpt-4o", msgs)): "github:gpt-4o",
asyncio.create_task(p_github("gpt-4o-mini", msgs)): "github:gpt-4o-mini",
asyncio.create_task(p_puter("glm-5.2", msgs)): "puter:glm-5.2",
asyncio.create_task(p_puter("gpt-5-mini", msgs)): "puter:gpt-5-mini",
asyncio.create_task(p_puter("deepseek-chat", msgs)): "puter:deepseek",
asyncio.create_task(p_llm7("codestral-latest", msgs)): "llm7:codestral",
asyncio.create_task(p_ovh("gpt-oss-20b", msgs)): "ovh:gpt-oss",
asyncio.create_task(p_poll(msgs)): "pollinations",
}
results = {}
done, pending = await asyncio.wait(tasks.keys(), timeout=15, return_when=asyncio.ALL_COMPLETED)
for p in pending: p.cancel()
for t in done:
try:
r = t.result()
if r: results[tasks[t]] = r
except: pass
return results
async def synthesize(question, responses):
"""دمج كل الردود في إجابة واحدة أذكى"""
if not responses: return None, "none"
if len(responses) == 1:
s, t = list(responses.items())[0]
return t, s
# بناء prompt للدمج
parts = ""
for i, (src, txt) in enumerate(responses.items(), 1):
parts += f"\n---\nرد {i} ({src}):\n{txt[:1200]}\n"
judge_msgs = [
{"role": "system", "content": "أنت نموذج حكم. ادمج ردود عدة نماذج في إجابة واحدة نهائية. خذ الأفضل من كل رد. أجب بالعربية. استخدم Markdown. لا تذكر المصادر. أكمل في رد واحد."},
{"role": "user", "content": f"السؤال: {question}\n\nردود النماذج:{parts}\n\nالإجابة الموحدة:"}
]
# جرّب GitHub Models أولاً
r = await p_github("gpt-4o", judge_msgs, 15)
if r: return r, "fusion:gpt-4o"
r = await p_puter("glm-5.2", judge_msgs, 15)
if r: return r, "fusion:glm-5.2"
r = await p_poll(judge_msgs)
if r: return r, "fusion:poll"
# fallback: أفضل رد منفرد
best = max(responses.values(), key=len)
return best, "fusion:best"
async def fusion(msgs, question):
"""Fusion الكامل: fanout → collect → synthesize"""
start = time.time()
LOG.append(f"[{time.strftime('%H:%M:%S')}] 🚀 fanout to all models...")
responses = await fanout(msgs)
LOG.append(f"[{time.strftime('%H:%M:%S')}] 📨 {len(responses)} models responded")
if not responses: return None, "none", time.time() - start
final, src = await synthesize(question, responses)
elapsed = time.time() - start
if final:
LOG.append(f"[{time.strftime('%H:%M:%S')}] ✅ fusion: {src} ({elapsed:.1f}s)")
return final, src, elapsed
# ============================================================
# TOOLS — 15 أداة حقيقية
# ============================================================
BLOCKED = {"os", "sys", "subprocess", "shutil", "socket", "http", "urllib", "pickle", "ctypes", "importlib", "builtins", "pty", "multiprocessing"}
def t_python(code):
"""تنفيذ كود Python"""
for b in BLOCKED:
if f"import {b}" in code or f"from {b}" in code: return f"❌ محظور: {b}"
if "__import__" in code or "exec(" in code or "eval(" in code: return "❌ exec/eval محظور"
with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False, encoding="utf-8") as f:
f.write(code); path = f.name
try:
r = subprocess.run(["python3", path], capture_output=True, text=True, timeout=8,
env={"PATH": os.environ.get("PATH", ""), "HOME": "/tmp", "MPLBACKEND": "Agg", "PYTHONPATH": ""})
out = (r.stdout or "")[:2000]
if r.stderr: out += f"\n⚠️ {r.stderr[:400]}"
return out or "(لا مخرجات)"
except subprocess.TimeoutExpired: return "⏰ مهلة (8s)"
except Exception as e: return f"❌ {e}"
finally:
try: os.unlink(path)
except: pass
async def t_search(query):
"""بحث في الإنترنت"""
s = await sess()
for lang in ["ar", "en"]:
try:
async with s.get(f"https://{lang}.wikipedia.org/w/api.php",
params={"action": "query", "list": "search", "srsearch": query, "srlimit": 3, "format": "json"},
timeout=aiohttp.ClientTimeout(total=8)) as r:
if r.status == 200:
items = (await r.json()).get("query", {}).get("search", [])
if items:
return "📚 " + ("عربي" if lang == "ar" else "إنجليزي") + ":\n" + "\n".join(f"• {i['title']}: {re.sub(r'<[^>]+>', '', i.get('snippet', ''))[:150]}" for i in items[:3])
except: pass
try:
async with s.get(f"https://html.duckduckgo.com/html/?q={urllib.parse.quote(query)}",
headers={"User-Agent": "Mozilla/5.0"}, timeout=aiohttp.ClientTimeout(total=10)) as r:
if r.status in (200, 202):
html = await r.text()
titles = re.findall(r'class="result__a"[^>]*>(.+?)', html)[:5]
snippets = re.findall(r'class="result__snippet"[^>]*>(.+?)', html)[:5]
results = [f"• {re.sub(r'<[^>]+>', '', t).strip()}: {re.sub(r'<[^>]+>', '', s).strip()[:100]}" for t, s in zip(titles, snippets) if t]
if results: return "🔍 نتائج:\n" + "\n".join(results)
except: pass
return "لا توجد نتائج"
async def t_fetch(url):
"""قراءة صفحة ويب"""
s = await sess()
try:
async with s.get(url, headers={"User-Agent": "Mozilla/5.0"}, timeout=aiohttp.ClientTimeout(total=10)) as r:
if r.status != 200: return f"❌ {r.status}"
html = await r.text()
html = re.sub(r"", "", html, flags=re.DOTALL | re.I)
html = re.sub(r"", "", html, flags=re.DOTALL | re.I)
html = re.sub(r"<[^>]+>", " ", html)
html = re.sub(r"\s+", " ", html).strip()
return html[:2000]
except Exception as e: return f"❌ {e}"
async def t_github(action, args=""):
"""GitHub API"""
import httpx
h = {"Authorization": f"token {GITHUB_TOKEN}", "Accept": "application/vnd.github.v3+json"}
try:
if action == "user":
r = httpx.get("https://api.github.com/user", headers=h, timeout=10)
if r.status_code == 200:
d = r.json()
return f"👤 {d.get('login')}\n• Repos: {d.get('public_repos')}\n• Followers: {d.get('followers')}\n• URL: {d.get('html_url')}"
elif action == "repos":
r = httpx.get("https://api.github.com/user/repos?sort=updated&per_page=10", headers=h, timeout=10)
if r.status_code == 200:
return "📦 مستودعاتك:\n" + "\n".join(f"• {x['name']} {'🔒' if x['private'] else '🌐'} ⭐{x['stargazers_count']}" for x in r.json()[:10])
elif action == "create_repo":
parts = args.split("|")
name = parts[0].strip() if parts else "new"
r = httpx.post("https://api.github.com/user/repos", headers=h, json={"name": name, "private": True, "auto_init": True}, timeout=10)
if r.status_code == 201: return f"✅ تم إنشاء {r.json()['name']}: {r.json()['html_url']}"
return f"❌ {r.status_code}"
elif action == "search":
r = httpx.get(f"https://api.github.com/search/repositories?q={urllib.parse.quote(args)}&sort=stars&per_page=5", headers=h, timeout=10)
if r.status_code == 200:
return "🔍 نتائج:\n" + "\n".join(f"• {x['full_name']} ⭐{x['stargazers_count']}" for x in r.json().get("items", [])[:5])
elif action == "readme":
r = httpx.get(f"https://api.github.com/repos/{args}/readme", headers=h, timeout=10)
if r.status_code == 200:
c = base64.b64decode(r.json().get("content", "")).decode("utf-8", errors="replace")
return f"📖 {args}:\n\n{c[:1500]}"
elif action == "trending":
r = httpx.get("https://api.github.com/search/repositories?q=created:>2024-01-01&sort=stars&per_page=5", headers=h, timeout=10)
if r.status_code == 200:
return "🔥 رائج:\n" + "\n".join(f"• {x['full_name']} ⭐{x['stargazers_count']}" for x in r.json().get("items", [])[:5])
return f"إجراءات: user, repos, create_repo, search, readme, trending"
except Exception as e: return f"❌ {e}"
async def t_arxiv(query):
"""بحث في arXiv"""
s = await sess()
try:
async with s.get(f"http://export.arxiv.org/api/query?search_query=all:{urllib.parse.quote(query)}&max_results=3",
timeout=aiohttp.ClientTimeout(total=10)) as r:
if r.status != 200: return "❌ فشل"
xml = await r.text()
entries = re.findall(r'