import streamlit as st
import os, json, re, time, base64, requests
from huggingface_hub import HfApi, HfFileSystem, InferenceClient
from bs4 import BeautifulSoup
from datetime import datetime
# ==========================================
# ⚙️ إعدادات V74.0 (The Absolute Sovereign Core)
# ==========================================
st.set_page_config(page_title="AmaAI V74.0 Sovereign", page_icon="🔱", layout="wide")
# تصميم Twilight UI المطور - مخصص للقيادة من الجوال
st.markdown("""
""", unsafe_allow_html=True)
# --- استدعاء الهوية السحابية والترسانة ---
HF_TOKEN = os.getenv("HF_TOKEN")
SPACE_ID = os.getenv("SPACE_ID")
api = HfApi(token=HF_TOKEN)
fs = HfFileSystem(token=HF_TOKEN)
client = InferenceClient(token=HF_TOKEN)
# نماذج النخبة المختارة (بناءً على تحليلات NotebookLM)
MODELS = {
"brain": "Qwen/Qwen2.5-72B-Instruct", # للتحليل العميق والمنطق
"coder": "meta-llama/Llama-3.3-70B-Instruct", # ملك البرمجة
"vision": "google/vit-base-patch16-224" # (اختياري) للرؤية البسيطة
}
# ==========================================
# 1. الذاكرة الفوتوغرافية (Claude Brain DNA)
# ==========================================
BRAIN_FILE = "sovereign_brain.json"
WORKSPACE = "outputs"
def load_sovereign_brain():
default = {"history": [], "knowledge_graph": {}, "rules": [], "reports": []}
if HF_TOKEN and SPACE_ID and fs.exists(f"spaces/{SPACE_ID}/{BRAIN_FILE}"):
try:
with fs.open(f"spaces/{SPACE_ID}/{BRAIN_FILE}", "r") as f:
return json.load(f)
except: return default
return default
def sync_brain():
if HF_TOKEN and SPACE_ID:
try:
# الحفاظ على آخر 20 رسالة فقط لتوفير الموارد (Deep Context)
if len(st.session_state.brain["history"]) > 20:
st.session_state.brain["history"] = st.session_state.brain["history"][-20:]
content = json.dumps(st.session_state.brain, ensure_ascii=False, indent=2)
api.upload_file(path_or_fileobj=content.encode(), path_in_repo=BRAIN_FILE, repo_id=SPACE_ID, repo_type="space")
except: pass
if "brain" not in st.session_state:
st.session_state.brain = load_sovereign_brain()
# ==========================================
# 2. أدوات السرب الحقيقية (Agentic Tools)
# ==========================================
def get_project_tree():
try:
files = fs.ls(f"spaces/{SPACE_ID}/{WORKSPACE}", detail=False)
return [f.replace(f"spaces/{SPACE_ID}/", "") for f in files]
except: return ["(Workspace Empty)"]
def tool_read(path):
try:
with fs.open(f"spaces/{SPACE_ID}/{path}", "r") as f:
return f"--- CONTENT OF {path} ---\n{f.read()}"
except: return f"[ERROR: File {path} not found]"
def tool_write(path, content):
if HF_TOKEN and SPACE_ID:
try:
clean_path = path.strip().lstrip('/')
if clean_path in ["app.py", BRAIN_FILE, "requirements.txt"]:
clean_path = f"mod_{clean_path}"
api.upload_file(path_or_fileobj=content.encode(), path_in_repo=clean_path, repo_id=SPACE_ID, repo_type="space")
return f"[SUCCESS: {clean_path} forged and synced]"
except Exception as e: return f"[ERROR: {str(e)}]"
return "[ERROR: No HF_TOKEN]"
def tool_web_search(query):
try:
# محاكاة PinchTab عبر استخلاص نصوص الروابط إذا وجدت في الاستعلام
urls = re.findall(r'(https?://[^\s]+)', query)
if not urls: return "[INFO: No URL found to scrape. Search simulation active.]"
resp = requests.get(urls[0], timeout=10, headers={"User-Agent": "AmaAI/74"})
soup = BeautifulSoup(resp.content, "html.parser")
for t in soup(["script", "style", "nav", "footer"]): t.extract()
return soup.get_text(separator=' ', strip=True)[:3500]
except: return "[ERROR: Web access failed]"
# ==========================================
# 3. حلقة السرب الموحد (The Sovereign Agent Loop)
# ==========================================
def execute_swarm_loop(user_prompt):
# الوكيل الآن يفكر بأسلوب ReAct (Reason + Act)
sys_msg = (
"You are AmaAI V74.0, a Sovereign Multi-Agent OS. You build complex, production-ready systems. "
"The user is on a weak mobile device. Offload all compute to your cloud tools.\n"
f"Project Files: {get_project_tree()}\n"
f"Learned Knowledge: {json.dumps(st.session_state.brain['knowledge_graph'], ensure_ascii=False)}\n\n"
"=== ACTION PROTOCOL (JSON ONLY) ===\n"
'{"tool": "READ", "path": "outputs/file.js"}\n'
'{"tool": "WRITE", "path": "outputs/index.html", "content": "..."}\n'
'{"tool": "SEARCH", "query": "url or search term"}\n'
'{"tool": "REPORT", "text": "brief mobile notification"}\n\n'
"=== PROTOCOL ===\n"
"1. THOUGHT: Reason in Arabic about the requirements.\n"
"2. ACTION: Use tools. If final answer is ready, STOP using tools.\n"
"3. PRODUCTION READY ONLY. No placeholders."
)
messages = [{"role": "system", "content": sys_msg}] + st.session_state.brain["history"]
messages.append({"role": "user", "content": user_prompt})
full_output = ""
for cycle in range(4): # 4 دورات تفكير وبناء
with st.spinner(f"🤖 السرب يعمل (دورة {cycle+1})..."):
try:
# استخدام نموذج البرمجة للمهام البرمجية
active_model = MODELS["coder"] if any(k in user_prompt for k in ["ابن", "كود", "build", "code"]) else MODELS["brain"]
response = client.chat_completion(model=active_model, messages=messages, max_tokens=3000, temperature=0.2)
agent_reply = response.choices[0].message.content
messages.append({"role": "assistant", "content": agent_reply})
# استخراج الأداة
tool_match = re.search(r'\{"tool":\s*".*?\}', agent_reply, re.DOTALL)
if tool_match:
call = json.loads(tool_match.group(0))
t_name = call.get("tool")
res = ""
if t_name == "READ": res = tool_read(call['path'])
elif t_name == "WRITE": res = tool_write(call['path'], call['content'])
elif t_name == "SEARCH": res = tool_web_search(call['query'])
elif t_name == "REPORT":
st.session_state.brain["reports"].append({"ts": datetime.now().strftime("%H:%M"), "text": call['text']})
res = "[REPORT SENT TO CHAIRMAN MOBILE]"
messages.append({"role": "system", "content": f"TOOL_RESULT: {res}"})
full_output += f"\n*السرب نفذ أداة {t_name} بنجاح*
"
else:
full_output += f"\n\n{agent_reply}"
break
except Exception as e:
full_output += f"\n\n[خطأ في النواة]: {str(e)}"
break
st.session_state.brain["history"].append({"role": "assistant", "content": full_output})
sync_brain()
return full_output
# ==========================================
# 4. واجهة القيادة المطلقة
# ==========================================
with st.sidebar:
st.markdown("## 🔱 قيادة V74.0")
st.success("☁️ الحالة: سيادة سحابية مطلقة")
# ميزة تقارير الجوال (Reporter Agent)
st.markdown("### 📱 إشعارات السرب")
if st.session_state.brain["reports"]:
for r in st.session_state.brain["reports"][-3:]:
st.markdown(f"🕒 {r['ts']}
{r['text']}
", unsafe_allow_html=True)
else: st.caption("لا توجد تقارير ميدانية بعد.")
with st.expander("📂 خريطة الإمبراطورية"):
st.code("\n".join(get_project_tree()), language="text")
if st.button("🧨 تصفير النواة (مسح الذاكرة)"):
st.session_state.brain = {"history": [], "knowledge_graph": {}, "rules": [], "reports": []}
sync_brain(); st.rerun()
st.title("AmaAI Sovereign Nexus V74.0")
tab_ops, tab_vault, tab_live = st.tabs(["💬 غرفة العمليات", "🛠️ المستودع السحابي", "🌐 معاينة الحقيقة"])
with tab_ops:
for m in st.session_state.brain["history"]:
with st.chat_message(m["role"]): st.markdown(m["content"], unsafe_allow_html=True)
if p := st.chat_input("سيدي، السرب بانتظار أوامرك المعقدة..."):
st.session_state.brain["history"].append({"role": "user", "content": p})
with st.chat_message("user"): st.markdown(p)
with st.chat_message("assistant"):
out = execute_swarm_loop(p)
st.markdown(out, unsafe_allow_html=True)
st.rerun()
with tab_vault:
st.markdown("### 🗄️ محرر الملفات المزامنة")
all_files = get_project_tree()
if all_files and all_files[0] != "(Workspace Empty)":
target = st.selectbox("تصفح أصول المنظومة:", all_files)
st.code(tool_read(target), language="javascript")
else: st.info("الورشة فارغة.")
with tab_live:
st.markdown("### 🖥️ بوابة المعاينة الحية")
html_files = [f for f in get_project_tree() if f.endswith(('.html', '.htm'))]
if html_files:
view_target = st.selectbox("اختر التطبيق:", html_files)
with fs.open(f"spaces/{SPACE_ID}/{view_target}", "r") as f:
st.components.v1.html(f.read(), height=750, scrolling=True)
else: st.info("ابنِ تطبيق ويب (HTML) لتراه حياً هنا.")