Update app.py

app.py

@@ -1,12 +1,18 @@
-r"""
-Smart Dev Sandbox - تطبيق بسيط وآمن يدعم ZIP / DOCX / PDF مع واجهة Gradio
-متطلبات:
-  pip install gradio huggingface-hub pyzipper python-docx PyPDF2
-
-متغيرات بيئة:
-  HF_TOKEN (اختياري، لتكامل Hugging Face)
-  HF_MODEL_ID (اختياري، معرف النموذج)
 """
+Smart Dev Sandbox - تطبيق Gradio احترافي للتعديل التفاعلي على ملفات ZIP/DOCX/PDF باستخدام AI
+ملف: smart_dev_sandbox.py
+
+المتطلبات:
+- gradio
+- huggingface-hub
+- pyzipper
+- python-docx
+- PyPDF2
+
+تشغيل:
+    python smart_dev_sandbox.py
+"""
+
 import os
 import re
 import shutil
@@ -17,109 +23,128 @@ from typing import Dict, Tuple, Optional, List
 
 import pyzipper
 import gradio as gr
+from huggingface_hub import InferenceClient
+from docx import Document
+import PyPDF2
 
-# استيراد مكتبات لتحليل docx/pdf
-try:
-    from docx import Document
-except Exception:
-    Document = None
-
-try:
-    from PyPDF2 import PdfReader
-except Exception:
-    PdfReader = None
-
-# إعدادات عامة
-ALLOWED_EXTENSIONS = {
-    '.py', '.js', '.html', '.css', '.txt', '.json',
-    '.md', '.php', '.yml', '.yaml', '.docx', '.pdf'
-}
+# --------------------------- إعدادات عامة ---------------------------
+ALLOWED_EXTENSIONS = {'.py', '.js', '.html', '.css', '.txt', '.json', '.md', '.php', '.yml', '.yaml', '.docx', '.pdf'}
+MAX_FILE_SIZE_BYTES = 200 * 1024        # 200 KB لكل ملف عند القراءة لضم السياق
+MAX_TOTAL_FILES = 500                   # عدد الملفات الأقصى
+MAX_EXTRACTED_BYTES = 20 * 1024 * 1024 # 20 MB إجمالي
 MODEL_ID = os.environ.get('HF_MODEL_ID', 'moonshotai/Kimi-K2-Instruct')
 HF_TOKEN = os.environ.get('HF_TOKEN')
 
 if not HF_TOKEN:
-    print("ملاحظة: HF_TOKEN غير معطى — سيعمل التطبيق لكن بدون اتصال بنموذج Hugging Face.")
+    print("⚠️ تحذير: لم يتم ضبط HF_TOKEN. لتشغيل نموذج Hugging Face ضع متغير البيئة HF_TOKEN.")
+
+# --------------------------- أدوات مساعدة ---------------------------
 
-# ---------- Sandbox بسيط وآمن ----------
+def safe_extract(zip_obj: pyzipper.AESZipFile, dest: str) -> Tuple[bool, str]:
+    total_bytes = 0
+    try:
+        for info in zip_obj.infolist():
+            name = info.filename
+            if name.startswith('__MACOSX') or name.strip() == '':
+                continue
+            if os.path.isabs(name):
+                return False, f"ملف غير صالح داخل الأرشيف: {name}"
+            dest_path = os.path.normpath(os.path.join(dest, name))
+            if not dest_path.startswith(os.path.normpath(dest) + os.sep):
+                return False, f"محاولة اختراق المسار داخل ZIP: {name}"
+            total_bytes += info.file_size
+            if total_bytes > MAX_EXTRACTED_BYTES:
+                return False, "حجم الملفات داخل ZIP يتجاوز الحد المسموح (20MB)"
+            zip_obj.extract(info, path=dest)
+        return True, "تم الاستخراج الآمن"
+    except Exception as e:
+        return False, str(e)
+
+def read_allowed_files(base_dir: str) -> Dict[str, str]:
+    files = {}
+    count = 0
+    for path in Path(base_dir).rglob('*'):
+        if path.is_file():
+            rel = str(path.relative_to(base_dir))
+            ext = path.suffix.lower()
+            if ext not in ALLOWED_EXTENSIONS:
+                continue
+            count += 1
+            if count > MAX_TOTAL_FILES:
+                break
+            try:
+                size = path.stat().st_size
+                if ext == '.docx':
+                    try:
+                        doc = Document(path)
+                        content = "\n".join([p.text for p in doc.paragraphs])
+                    except Exception:
+                        content = "/* خطأ في قراءة DOCX */"
+                elif ext == '.pdf':
+                    try:
+                        with open(path, 'rb') as f:
+                            reader = PyPDF2.PdfReader(f)
+                            content = "\n".join([page.extract_text() or "" for page in reader.pages])
+                    except Exception:
+                        content = "/* خطأ في قراءة PDF */"
+                else:
+                    content = path.read_text(encoding='utf-8', errors='ignore')
+                    if size > MAX_FILE_SIZE_BYTES:
+                        head = content[:MAX_FILE_SIZE_BYTES//2]
+                        tail = content[-MAX_FILE_SIZE_BYTES//2:]
+                        content = f"/* الملف كبير جدًا ({size} بايت). عرض مقتطفات */\n{head}\n...\n{tail}"
+                files[rel] = content
+            except Exception:
+                files[rel] = "/* خطأ في قراءة الملف */"
+    return files
+
+# --------------------------- Sandbox ---------------------------
 class Sandbox:
     def __init__(self):
         self.dir: Optional[str] = None
         self.files_content: Dict[str, str] = {}
-        self.original_file_path: Optional[str] = None
+        self.original_zip_path: Optional[str] = None
+        self.password: Optional[str] = None
 
     def cleanup(self):
         if self.dir and os.path.isdir(self.dir):
-            shutil.rmtree(self.dir, ignore_errors=True)
+            try:
+                shutil.rmtree(self.dir)
+            except Exception:
+                pass
         self.dir = None
         self.files_content = {}
-        self.original_file_path = None
+        self.original_zip_path = None
+        self.password = None
 
-    def reset(self, file_path: str, password: Optional[str] = None) -> Tuple[Optional[str], str]:
-        """
-        يدعم: .zip (استخراج)، .docx (استخراج نص)، .pdf (استخراج نص).
-        يعيد (path, message) أو (None, error_message).
-        """
+    def reset(self, zip_path: str, password: Optional[str] = None) -> Tuple[Optional[str], str]:
         self.cleanup()
-        if not file_path:
-            return None, "⚠️ يرجى رفع ملف ZIP / DOCX / PDF أولاً."
+        if not zip_path:
+            return None, "⚠️ يرجى رفع ملف ZIP أولاً"
         self.dir = tempfile.mkdtemp(prefix='sandbox_')
-        self.original_file_path = file_path
+        self.files_content = {}
+        self.password = password
         try:
-            ext = Path(file_path).suffix.lower()
-            if ext == '.zip':
-                with pyzipper.AESZipFile(file_path, 'r') as zf:
-                    # استخراج آمن: منع zip-slip
-                    for info in zf.infolist():
-                        name = info.filename
-                        if name.startswith('__MACOSX') or name.strip() == '':
-                            continue
-                        # منع المسارات المطلقة أو ../../
-                        safe_name = os.path.normpath(name).lstrip(os.sep)
-                        dest_path = os.path.join(self.dir, safe_name)
-                        os.makedirs(os.path.dirname(dest_path), exist_ok=True)
-                        zf.extract(info, path=self.dir)
-            elif ext == '.docx':
-                if Document is None:
-                    raise RuntimeError("مكتبة python-docx غير مثبتة")
-                doc = Document(file_path)
-                out_txt = os.path.join(self.dir, 'document.txt')
-                os.makedirs(os.path.dirname(out_txt), exist_ok=True)
-                with open(out_txt, 'w', encoding='utf-8') as f:
-                    for para in doc.paragraphs:
-                        f.write(para.text + '\n')
-            elif ext == '.pdf':
-                if PdfReader is None:
-                    raise RuntimeError("مكتبة PyPDF2 غير مثبتة")
-                reader = PdfReader(file_path)
-                out_txt = os.path.join(self.dir, 'document.txt')
-                os.makedirs(os.path.dirname(out_txt), exist_ok=True)
-                with open(out_txt, 'w', encoding='utf-8') as f:
-                    for page in reader.pages:
-                        text = page.extract_text() or ''
-                        f.write(text + '\n')
-            else:
-                self.cleanup()
-                return None, "❌ صيغة الملف غير مدعومة. ادعم: .zip, .docx, .pdf"
-
-            # جمع المحتويات للعرض / للسياق
-            self.files_content = {}
-            for p in Path(self.dir).rglob('*'):
-                if p.is_file():
-                    rel = str(p.relative_to(self.dir))
-                    try:
-                        self.files_content[rel] = p.read_text(encoding='utf-8', errors='ignore')
-                    except Exception:
-                        self.files_content[rel] = "/* لا يمكن قراءة الملف */"
-
-            return file_path, f"✅ تم تجهيز {len(self.files_content)} ملف/ملفات داخل الـ sandbox."
+            with pyzipper.AESZipFile(zip_path, 'r') as zf:
+                if password:
+                    zf.setpassword(password.encode('utf-8'))
+                ok, msg = safe_extract(zf, self.dir)
+                if not ok:
+                    self.cleanup()
+                    return None, f"❌ خطأ أثناء الاستخراج: {msg}"
+            self.files_content = read_allowed_files(self.dir)
+            self.original_zip_path = zip_path
+            return zip_path, f"✅ تم استخراج {len(self.files_content)} ملف/ملفات"
         except Exception as e:
+            tb = traceback.format_exc()
             self.cleanup()
-            return None, f"❌ استثناء أثناء المعالجة: {e}"
+            return None, f"❌ استثناء: {str(e)}\n{tb}"
 
     def get_context(self) -> str:
-        if not self.files_content:
-            return ""
-        return "\n".join(f"--- FILE: {name} ---\n{content}" for name, content in self.files_content.items())
+        parts: List[str] = []
+        for name, content in self.files_content.items():
+            parts.append(f"--- FILE: {name} ---\n{content}")
+        return "\n".join(parts)
 
     def update_file(self, name: str, content: str) -> None:
         if not self.dir:
@@ -134,83 +159,124 @@ class Sandbox:
     def create_zip(self) -> Tuple[Optional[str], str]:
         if not self.dir:
             return None, "❌ لا توجد ملفات لحزمها"
-        fd, temp_zip = tempfile.mkstemp(prefix='modified_', suffix='.zip')
-        os.close(fd)
-        base = temp_zip[:-4]
-        shutil.make_archive(base, 'zip', self.dir)
-        return base + '.zip', "✅ تم إنشاء ZIP المعدل"
+        out_fd, out_path = tempfile.mkstemp(prefix='modified_', suffix='.zip')
+        os.close(out_fd)
+        base_name = out_path[:-4]
+        try:
+            shutil.make_archive(base_name, 'zip', self.dir)
+            return base_name + '.zip', "✅ تم إنشاء ZIP المعدل"
+        except Exception as e:
+            return None, f"❌ خطأ في إنشاء ZIP: {e}"
+
+# --------------------------- نموذج Hugging Face ---------------------------
+client = None
+if HF_TOKEN:
+    client = InferenceClient(api_key=HF_TOKEN)
+
+def build_messages(user_prompt: str, sandbox_ctx: str) -> List[Dict[str, str]]:
+    system = (
+        "أنت مساعد برمجي خبير. سأزودك بمشروع (عدة ملفات).\n"
+        "عند طلب تعديل، قم بإرجاع التعديلات فقط في كتل واضحة وبالتنسيق التالي:\n"
+        "[ملف: relative/path/to/file.ext]```<language>\n<محتوى الملف الجديد>\n```\n"
+        "لا تكتب شروحات أو حوار داخل الكتل.")
+    user = f"المشروع الحالي:\n{sandbox_ctx}\n\nطلب المستخدم: {user_prompt}"
+    return [{"role": "system", "content": system}, {"role": "user", "content": user}]
+
+def stream_chat_completion(model_id: str, messages: List[Dict[str, str]], max_tokens: int = 2048):
+    if client is None:
+        yield {"content": "❗️ نموذج Hugging Face غير مهيأ. ضع HF_TOKEN لتفعيل التكامل."}
+        return
+    try:
+        for chunk in client.chat_completion(model=model_id, messages=messages, max_tokens=max_tokens, stream=True):
+            try:
+                token = chunk.choices[0].delta.content
+            except Exception:
+                token = None
+            if token:
+                yield {"content": token}
+    except Exception as e:
+        yield {"content": f"❌ خطأ عند الاتصال بالنموذج: {str(e)}"}
 
+CHANGE_BLOCK_RE = re.compile(
+    r"\[\s*(?:ملف|file)\s*:\s*(?P<name>[^\]]+?)\s*\]\s*```(?:[\w\-+]+)?\n(?P<code>.*?)\n```",
+    re.IGNORECASE | re.DOTALL,
+)
+
+def extract_changes_from_text(text: str) -> List[Tuple[str, str]]:
+    matches = CHANGE_BLOCK_RE.finditer(text)
+    return [(m.group('name').strip(), m.group('code')) for m in matches]
+
+# --------------------------- واجهة Gradio ---------------------------
 sandbox = Sandbox()
 
-# ---------- واجهة Gradio ----------
-with gr.Blocks(title="Smart Dev Sandbox") as demo:
-    gr.Markdown("# Smart Dev Sandbox — دعم ZIP / DOCX / PDF")
+def api_reset(zip_file, password):
+    if zip_file is None:
+        return None, "⚠️ ارفع ملف ZIP أولاً"
+    zip_path = zip_file.name if hasattr(zip_file, 'name') else zip_file
+    return sandbox.reset(zip_path, password)
+
+def api_send_message(user_message: str, chat_history: List[dict]):
+    chat_history = chat_history + [("user", user_message)]
+    if not sandbox.files_content:
+        # لا توجد ملفات بعد، نرسل تأكيد للطلب فقط
+        chat_history = chat_history + [("assistant", f"✅ تم استلام طلبك: \"{user_message}\".\nيمكنك رفع الملفات لاحقًا لتطبيق التعديلات الفعلية.")]
+        return chat_history
+    # إذا كانت هناك ملفات، نستدعي النموذج
+    sandbox_ctx = sandbox.get_context()
+    messages = build_messages(user_message, sandbox_ctx)
+    full_response = ""
+    for chunk in stream_chat_completion(MODEL_ID, messages):
+        token = chunk.get('content', '')
+        full_response += token
+        if chat_history and chat_history[-1][0] == 'assistant':
+            chat_history[-1] = ('assistant', full_response)
+        else:
+            chat_history = chat_history + [('assistant', full_response)]
+    return chat_history
+
+def api_apply_and_package(chat_history: List[tuple]):
+    last_assistant = None
+    for role, content in reversed(chat_history):
+        if role == 'assistant' and content and content.strip():
+            last_assistant = content
+            break
+    if not last_assistant:
+        return None, "⚠️ لا توجد تعديلات صالحة في آخر رد للمساعد"
+    changes = extract_changes_from_text(last_assistant)
+    if not changes:
+        return None, "⚠️ لم يتم العثور على كتل تعديل بصيغة [ملف: ...]```...```"
+    applied = 0
+    for name, code in changes:
+        try:
+            sandbox.update_file(name, code)
+            applied += 1
+        except Exception as e:
+            print(f"فشل تطبيق {name}: {e}")
+    out_zip, msg = sandbox.create_zip()
+    if not out_zip:
+        return None, msg
+    return out_zip, f"✅ تم تطبيق {applied} ملفات. {msg}"
+
+# --------------------------- بناء الواجهة ---------------------------
+with gr.Blocks(title="Smart Dev Sandbox — المبرمج الذكي PRO") as demo:
+    gr.Markdown("# 🚀 المبرمج الذكي PRO — Sandbox احترافي لتعديل مشاريع ZIP/DOCX/PDF")
     with gr.Row():
         with gr.Column(scale=3):
-            chatbot = gr.Chatbot(label="محادثة")
-            msg = gr.Textbox(placeholder="اطرح سؤالاً أو اطلب تعديلًا...", label="الرسالة")
+            chatbot = gr.Chatbot(label="محادثة المساعد", elem_id="chatbot")
+            msg = gr.Textbox(placeholder="اطلب تعديلًا (مثال: عدل style.css لتغيّر الخلفية إلى #0f172a)", label="طلب التعديل")
             send_btn = gr.Button("إرسال")
         with gr.Column(scale=1):
-            file_input = gr.File(label="اختر ملف (ZIP / DOCX / PDF)", file_types=[".zip", ".docx", ".pdf"])
-            pwd_input = gr.Textbox(label="كلمة مرور ZIP (اختياري)", type="password")
-            analyze_btn = gr.Button("🔍 تجهيز الملف")
-            status_md = gr.Markdown("**الحالة:** جاهز")
+            zip_input = gr.File(label="1) اختر ملف المشروع (ZIP)")
+            pwd_input = gr.Textbox(label="كلمة مرور (اختياري)", type="password")
+            analyze = gr.Button("🔍 2) تحليل الملفات")
+            status = gr.Markdown("**الحالة:** جاهز")
             gr.HTML("<hr>")
-            save_btn = gr.Button("✨ تطبيق التعديلات وإنشاء ZIP")
-            download_file = gr.File(label="تحميل الناتج")
-
-    # دوال الواجهة
-    def api_reset(uploaded_file, password):
-        if uploaded_file is None:
-            return None, "⚠️ ارفع ملفًا أولًا"
-        # gradio يعطي كائن له .name أو .temp_path
-        file_path = getattr(uploaded_file, "name", None) or getattr(uploaded_file, "temp_path", None) or uploaded_file
-        return sandbox.reset(file_path, password)
-
-    def api_send_message(user_message, chat_history):
-        chat_history = chat_history or []
-        # يسمح بالدردشة حتى بدون ملف مرفوع
-        chat_history.append(("user", user_message))
-        if sandbox.files_content:
-            ctx = sandbox.get_context()
-            assistant_reply = f"تم استلام طلب تعديل. (سياق المشروع مرفوع — {len(sandbox.files_content)} ملف)."  # placeholder
-        else:
-            assistant_reply = "يمكنني الإجابة على الأسئلة العامة الآن. لطلب تعديل ملفات، ارفع ZIP/DOCX/PDF ثم اضغط 'تجهيز الملف'."
-        chat_history.append(("assistant", assistant_reply))
-        return chat_history
+            save_btn = gr.Button("✨ 3) تطبيق التعديلات وإنشاء ZIP")
+            download_output = gr.File(label="4) حمل الملف المعدل")
 
-    def api_apply_and_package(chat_history):
-        # يأخذ آخر رد مساعد ويستخرج كتل التعديل بصيغة [ملف: path]```...```
-        if not sandbox.files_content:
-            return None, "⚠️ لا ملفات داخل الـ sandbox للتعديل."
-        if not chat_history:
-            return None, "⚠️ المحادثة فارغة."
-        # إيجاد آخر رد المساعد
-        last = None
-        for role, content in reversed(chat_history):
-            if role == 'assistant' and content and content.strip():
-                last = content
-                break
-        if not last:
-            return None, "⚠️ لا توجد تعديلات في المحادثة."
-        # استخراج كتل التعديل
-        pattern = re.compile(r"\[\s*(?:ملف|file)\s*:\s*([^\]]+?)\s*\]\s*```(?:[\\w\\-+]+)?\\n(.*?)\\n```", re.IGNORECASE | re.DOTALL)
-        matches = pattern.findall(last)
-        if not matches:
-            return None, "⚠️ لم يعثر على كتل تعديل بصيغة [ملف: ...]```...```"
-        applied = 0
-        for name, code in matches:
-            try:
-                sandbox.update_file(name.strip(), code)
-                applied += 1
-            except Exception as e:
-                print(f"فشل تطبيق {name}: {e}")
-        out_zip, msg = sandbox.create_zip()
-        return out_zip, f"✅ تم تطبيق {applied} ملفات. {msg}"
-
-    analyze_btn.click(api_reset, inputs=[file_input, pwd_input], outputs=[file_input, status_md])
+    analyze.click(api_reset, inputs=[zip_input, pwd_input], outputs=[zip_input, status])
     send_btn.click(api_send_message, inputs=[msg, chatbot], outputs=[chatbot])
-    save_btn.click(api_apply_and_package, inputs=[chatbot], outputs=[download_file, status_md])
+    save_btn.click(api_apply_and_package, inputs=[chatbot], outputs=[download_output, status])
 
 if __name__ == "__main__":
     demo.launch()