| 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 |
|
|
| |
| |
| |
| st.set_page_config(page_title="AmaAI V74.0 Sovereign", page_icon="🔱", layout="wide") |
|
|
| |
| st.markdown(""" |
| <style> |
| .stApp { background-color: #010204; color: #e0e0e0; font-family: 'Segoe UI', sans-serif; } |
| .stChatMessage { border-radius: 12px; border: 1px solid #00d4ff33; background: #060b11; margin-bottom:12px; } |
| h1, h2, h3 { color: #00d4ff !important; font-weight: 600; } |
| .stTabs [data-baseweb="tab-list"] { border-bottom: 2px solid #00d4ff33; } |
| .stTabs [data-baseweb="tab"] { background-color: transparent; color: #888; font-weight: bold; } |
| .stTabs [aria-selected="true"] { color: #00d4ff !important; border-bottom: 2px solid #00d4ff; } |
| code { color: #00ffcc !important; background-color: #111b26 !important; padding: 2px 4px; border-radius: 4px; } |
| .agent-thought { border-left: 3px solid #ff007f; padding-left: 10px; color: #ff007f; font-style: italic; font-size: 0.85em; margin: 10px 0; } |
| .report-card { background: #001a2e; border: 1px solid #00d4ff55; padding: 15px; border-radius: 10px; margin: 10px 0; border-right: 5px solid #00d4ff; } |
| </style> |
| """, 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) |
|
|
| |
| MODELS = { |
| "brain": "Qwen/Qwen2.5-72B-Instruct", |
| "coder": "meta-llama/Llama-3.3-70B-Instruct", |
| "vision": "google/vit-base-patch16-224" |
| } |
|
|
| |
| |
| |
| 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: |
| |
| 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() |
|
|
| |
| |
| |
| 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: |
| |
| 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]" |
|
|
| |
| |
| |
| def execute_swarm_loop(user_prompt): |
| |
| 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): |
| 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<div class='agent-thought'>*السرب نفذ أداة {t_name} بنجاح*</div>" |
| 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 |
|
|
| |
| |
| |
| with st.sidebar: |
| st.markdown("## 🔱 قيادة V74.0") |
| st.success("☁️ الحالة: سيادة سحابية مطلقة") |
| |
| |
| st.markdown("### 📱 إشعارات السرب") |
| if st.session_state.brain["reports"]: |
| for r in st.session_state.brain["reports"][-3:]: |
| st.markdown(f"<div class='report-card'><b>🕒 {r['ts']}</b><br>{r['text']}</div>", 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) لتراه حياً هنا.") |
