Update upload_app.py

upload_app.py

@@ -1,413 +0,0 @@
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
-
-print("[CHECKPOINT - UPLOAD_APP] >>> بدأ تحميل ملف upload_app.py وتحديد المسارات...", flush=True)
-
-import os
-import re
-import json
-import shutil
-import secrets
-from datetime import datetime as dt
-from datetime import timedelta
-from urllib.parse import unquote
-import gradio as gr
-from huggingface_hub import HfApi, list_repo_files, login, hf_hub_download, upload_file
-
-# استيراد مكاتب التحويل والـ ZIP (افترض أنها موجودة)
-from converter import convert_text, zip_single_file
-
-# ============================================================
-# مسارات التخزين المحلية والسرية للنظام
-# ============================================================
-UPLOAD_DIR = "uploads"
-STATUS_DIR = "system"
-STATUS_FILE_PATH = os.path.join(STATUS_DIR, "status.json")
-
-REPO_ID = "Asem75/aiocr_asistant"
-RAW_DIR = "raw_files"
-WATCH_DIR_1 = "process"
-WATCH_DIR_2 = "process_files"
-FINAL_DIR = "Final_Arabic_Files"
-MESSAGES_PATH = "system/messages.json"
-
-TOKEN = os.getenv("AIOCR_KEY")
-api = None
-if TOKEN:
-    try:
-        TOKEN = TOKEN.strip().split()[0]
-        login(token=TOKEN)
-        api = HfApi(token=TOKEN)
-        print("[CHECKPOINT - UPLOAD_APP] تم تسجيل الدخول في HuggingFace بنجاح.", flush=True)
-    except Exception as e:
-        print(f"[WARNING] فشل تسجيل الدخول بالتوكن: {e}", flush=True)
-
-print(f"[CHECKPOINT - UPLOAD_APP] جاري إنشاء المجلدات المحلية إن لم تكن موجودة...", flush=True)
-os.makedirs(UPLOAD_DIR, exist_ok=True)
-os.makedirs(STATUS_DIR, exist_ok=True)
-print("[CHECKPOINT - UPLOAD_APP] تم إنشاء المجلدات المحلية بنجاح.", flush=True)
-
-# ============================================================
-# 1) دوال مساعدة وسجلات النظام السيادية
-# ============================================================
-
-def sanitize_filename(name: str) -> str:
-    name = name.replace(" ", "_")
-    return re.sub(r"[^\w\.\-]", "", name)
-
-def _write_status(entry: dict):
-    try:
-        with open(STATUS_FILE_PATH, "w", encoding="utf-8") as f:
-            json.dump(entry, f, ensure_ascii=False, indent=4)
-        if TOKEN:
-            upload_file(
-                path_or_fileobj=STATUS_FILE_PATH,
-                path_in_repo="system/status.json",
-                repo_id=REPO_ID,
-                repo_type="dataset",
-                token=TOKEN
-            )
-    except Exception as e:
-        print(f"[CHECKPOINT ERROR] فشل كتابة السجل أو مزامنته سحابياً: {e}", flush=True)
-
-def _read_status() -> dict:
-    if not os.path.exists(STATUS_FILE_PATH):
-        return {}
-    try:
-        with open(STATUS_FILE_PATH, "r", encoding="utf-8") as f:
-            return json.load(f)
-    except Exception as e:
-        print(f"[CHECKPOINT ERROR] فشل قراءة السجل المحلي status.json: {e}", flush=True)
-        return {}
-
-def send_report_to_manager(report_text: str):
-    try:
-        try:
-            local_existing = hf_hub_download(
-                repo_id=REPO_ID,
-                filename=MESSAGES_PATH,
-                repo_type="dataset",
-                token=TOKEN,
-            )
-            with open(local_existing, "r", encoding="utf-8") as f:
-                messages = json.load(f)
-            if not isinstance(messages, list): messages = []
-        except:
-            messages = []
-
-        messages.append({
-            "time": dt.now().strftime("%Y-%m-%d %H:%M:%S"),
-            "from": "upload_interface",
-            "report": report_text,
-        })
-
-        local_path = os.path.join("system", "messages.json")
-        with open(local_path, "w", encoding="utf-8") as f:
-            json.dump(messages, f, ensure_ascii=False, indent=2)
-
-        upload_file(
-            path_or_fileobj=local_path,
-            path_in_repo=MESSAGES_PATH,
-            repo_id=REPO_ID,
-            repo_type="dataset",
-            token=TOKEN,
-        )
-    except Exception as e:
-        print(f"[REPORT SEND ERROR] {e}", flush=True)
-
-# ============================================================
-# 2) رادار فحص القيود للمشترك المجاني (FREE)
-# ============================================================
-def check_free_user_limits(username):
-    try:
-        all_files = list_repo_files(REPO_ID, repo_type="dataset")
-        raw_prefix = f"{RAW_DIR}/{username}_"
-        watch1_prefix = f"{WATCH_DIR_1}/{username}_"
-        watch2_prefix = f"{WATCH_DIR_2}/{username}_"
-        
-        active_files = [
-            f for f in all_files 
-            if f.startswith(raw_prefix) or f.startswith(watch1_prefix) or f.startswith(watch2_prefix)
-        ]
-        
-        if active_files:
-            return False, f"❌ الرادار رصد وجود {len(active_files)} ملفات لك قيد المعالجة حالياً. يرجى ال��نتظار."
-        return True, "مسموح بالرفع"
-    except Exception as e:
-        return False, "⚠️ حدث خطأ غير متوقع أثناء فحص رادار القيود."
-
-# ============================================================
-# 3) دالة عرض/إخفاء لوحة الاشتراك (VIP / Free)
-# ============================================================
-def toggle_subscription_panel(user_session, panel_visible_state):
-    new_state = not panel_visible_state
-    if not new_state:
-        return gr.update(visible=False), False
-
-    rank = (user_session or {}).get("rank", "free")
-    name = (user_session or {}).get("name", "زائر")
-
-    if rank == "vip":
-        html = f"""
-        <div style='background:#1e293b; border:1px solid #d4af37; padding:14px; border-radius:10px;'>
-            <p style='color:#d4af37; margin:0;'>🥇 أنت مشترك VIP يا {name} — لديك صلاحية كاملة لكل خيارات المعالجة المتقدمة.</p>
-        </div>
-        """
-    else:
-        html = f"""
-        <div style='background:#1e293b; border:1px solid #38bdf8; padding:14px; border-radius:10px;'>
-            <p style='color:#9ca3af; margin:0;'>🥈 أنت مشترك مجاني يا {name}. يمكنك الترقية لـ VIP لفتح كل خيارات المعالجة المتقدمة.</p>
-        </div>
-        """
-    return gr.update(visible=True, value=html), True
-
-# ============================================================
-# 4) دالة استلام ومعالجة وحفظ الملفات مباشرة في مجلد raw_files
-# ============================================================
-def handle_upload(files, user_session, enable_table, enable_handwriting,
-                   enable_heavy_prep, enable_high_dpi, enable_fast, enable_tashkeel):
-
-    print(f"[CHECKPOINT - UPLOAD_APP] بدأ استلام طلب رفع ملفات من الجلسة...", flush=True)
-    if not (user_session or {}).get("authenticated"):
-        return "<div style='color:#f87171;'>⚠️ يجب تسجيل الدخول وتوثيق الهوية أولاً قبل رفع أي ملفات.</div>"
-
-    if not files:
-        return "<div style='color:#facc15;'>⚠️ لم يتم اختيار أي ملف للرفع.</div>"
-
-    username = sanitize_filename((user_session.get("name") or "unknown").strip())
-    rank = user_session.get("rank", "free")
-    is_vip = (rank == "vip")
-
-    if not is_vip:
-        allowed, message = check_free_user_limits(username)
-        if not allowed:
-            return f"<div role='alert' style='color:red; font-weight:bold;'>{message}</div>"
-
-    user_folder = os.path.join(UPLOAD_DIR, username)
-    os.makedirs(user_folder, exist_ok=True)
-
-    saved_names = []
-    prefix = f"VIP_{username}_" if is_vip else f"{username}_"
-
-    for file in files:
-        try:
-            fname = sanitize_filename(unquote(os.path.basename(file.name)))
-            local_dest = os.path.join(user_folder, fname)
-            
-            shutil.copy(file.name, local_dest)
-            
-            option_prefix = ""
-            if enable_fast:
-                option_prefix = "fast_"
-            else:
-                if enable_table: option_prefix += "tbl_"
-                if enable_handwriting: option_prefix += "hw_"
-                if enable_heavy_prep: option_prefix += "prep_"
-                if enable_high_dpi: option_prefix += "dpi300_"
-                if enable_tashkeel: option_prefix += "tash_"
-
-            final_filename = f"{prefix}{option_prefix}{fname}"
-
-            if TOKEN:
-                print(f"[CLOUD UPLOAD] جاري إرسال الملف {final_filename} إلى مجلد raw_files السحابي...", flush=True)
-                upload_file(
-                    path_or_fileobj=local_dest,
-                    path_in_repo=f"{RAW_DIR}/{final_filename}",
-                    repo_id=REPO_ID,
-                    repo_type="dataset",
-                    token=TOKEN
-                )
-                saved_names.append(final_filename)
-
-        except Exception as e:
-            print(f"[UPLOAD ERROR] فشل رفع الملف سحابياً: {e}", flush=True)
-            continue
-
-    options = {
-        "table": bool(enable_table),
-        "handwriting": bool(enable_handwriting),
-        "heavy_prep": bool(enable_heavy_prep),
-        "high_dpi": bool(enable_high_dpi),
-        "fast": bool(enable_fast),
-        "tashkeel": bool(enable_tashkeel),
-    }
-
-    _write_status({
-        "user": username,
-        "files": saved_names,
-        "options": options,
-        "status": "🟡 في قائمة الانتظار للمعالجة الفورية...",
-        "queued_at": dt.now().isoformat()
-    })
-
-    if not saved_names:
-        return "<div style='color:#f87171;'>❌ فشل حفظ ورفع الملفات المرفوعة إلى السحابة. حاول مرة أخرى.</div>"
-
-    send_report_to_manager(f"المستخدم {username} ({'VIP' if is_vip else 'FREE'}) رفع {len(saved_names)} ملف/ات.")
-
-    return f"<div role='status' style='color:green; font-weight:bold;'>✅ تم رفع {len(saved_names)} ملف بنجاح وإضافته إلى مجلد raw_files السحابي.</div>"
-
-def read_status_json():
-    status = _read_status()
-    if not status:
-        return "<div role='status' aria-live='polite'>بانتظار تحديث الحالة من system/status.json...</div>"
-    files_list = "، ".join(status.get("files", [])) or "—"
-    return f"""
-    <div role='status' aria-live='polite' style='line-height:1.6;'>
-        👤 المستخدم: <b>{status.get('user', '—')}</b><br>
-        📄 الملفات: {files_list}<br>
-        📌 الحالة: {status.get('status', '—')}<br>
-        🕒 آخر تحديث: {status.get('queued_at', '—')}
-    </div>
-    """
-
-# ============================================================
-# 5) دوال قسم ملفاتي الأصلي والمصحح والمحمي
-# ============================================================
-
-def list_final_files():
-    """جلب قائمة الملفات الجاهزة من Final_Arabic_Files في المستودع"""
-    try:
-        files = list_repo_files(REPO_ID, repo_type="dataset")
-        final_files = [f.split("/", 1)[1] for f in files if f.startswith(FINAL_DIR + "/")]
-        return sorted(final_files) if final_files else []
-    except Exception as e:
-        print(f"[FINAL LIST ERROR] {e}", flush=True)
-        return []
-
-def load_file_content(selected_name: str):
-    if not selected_name:
-        return "", "", ""
-    try:
-        remote_path = f"{FINAL_DIR}/{selected_name}"
-        local_path = hf_hub_download(
-            repo_id=REPO_ID,
-            filename=remote_path,
-            repo_type="dataset",
-            token=TOKEN
-        )
-        with open(local_path, "r", encoding="utf-8") as f:
-            text = f.read()
-        return text, os.path.splitext(selected_name)[0], selected_name
-    except Exception as e:
-        print(f"[FINAL LOAD ERROR] {selected_name} — {e}", flush=True)
-        return "❌ فشل تحميل الملف", "", ""
-
-def convert_and_upload(text, base_name, target_ext, font_name, font_size):
-    if not text.strip(): return "❌ لا يوجد نص", None
-    try:
-        out_path = convert_text(text, base_name, target_ext, font_name, font_size)
-        new_name = os.path.basename(out_path)
-        upload_file(
-            path_or_fileobj=out_path,
-            path_in_repo=f"{FINAL_DIR}/{new_name}",
-            repo_id=REPO_ID, repo_type="dataset", token=TOKEN
-        )
-        return f"✅ تم التحويل والرفع باسم {new_name}", out_path
-    except Exception as e:
-        print(f"[CONVERT/UPLOAD ERROR] {e}", flush=True)
-        return f"⚠️ فشل: {e}", None
-
-def do_zip(text, base_name):
-    if not text.strip(): return "❌ لا يوجد نص", None
-    try:
-        txt_path = convert_text(text, base_name, "txt")
-        return "✅ تم إنشاء ZIP", zip_single_file(txt_path, base_name)
-    except Exception as e:
-        return "⚠️ فشل إنشاء ZIP", None
-
-# ============================================================
-# 6) بناء الواجهة الرادارية الرسمية المحصنة
-# ============================================================
-
-def build_upload_ui():
-    print("[CHECKPOINT - UPLOAD_APP] >>> دخلنا الآن داخل دالة build_upload_ui لبناء عتاد الواجهات...", flush=True)
-    
-    # -------- أ) واجهة الرفع المباشر --------
-    with gr.Column() as upload_layout:
-        gr.Markdown("<h1 style='text-align:center;'>📤 بوابة الرفع ودورة حياة الملفات الموثقة</h1>")
-
-        user_session = gr.State({"authenticated": False, "name": "", "rank": "free"})
-        panel_visible_state = gr.State(False)
-
-        welcome_box = gr.HTML("<p role='status'>🟢 الجلسة السيادية الموحدة نشطة وجاهزة لاستلام العتاد.</p>")
-
-        with gr.Row():
-            rank_icon_btn = gr.Button("🥈 FREE", elem_id="rank_icon_btn", scale=0, min_width=110, visible=True)
-
-        subscription_panel = gr.HTML(visible=False)
-
-        with gr.Group() as upload_area:
-            uploader = gr.File(label="اختر ملفاتك", file_count="multiple")
-            upload_btn = gr.Button("رفع الملفات", variant="primary")
-            status_box = gr.HTML("<div>بانتظار رفع الملفات...</div>")
-
-        with gr.Group() as options_area:
-            gr.Markdown("### ⚙️ خيارات معالجة العتاد المتقدمة")
-            with gr.Row():
-                enable_table = gr.Checkbox(label="📊 كشف الجداول", value=False)
-                enable_handwriting = gr.Checkbox(label="🖐️ خط يدوي", value=False)
-                enable_heavy_prep = gr.Checkbox(label="🔧 معالجة ثقيلة", value=False)
-                enable_high_dpi = gr.Checkbox(label="📐 دقة عالية (300 DPI)", value=False)
-                enable_fast = gr.Checkbox(label="⚡ الوضع السريع (EasyOCR)", value=False)
-                enable_tashkeel = gr.Checkbox(label="🔤 تشكيل النص", value=False)
-
-        gr.Markdown("### 📜 حالة التجميع والمعالجة الفورية")
-        system_status = gr.HTML("<div role='status' aria-live='polite'>بانتظار تحديث الحالة من system/status.json...</div>")
-
-        rank_icon_btn.click(
-            fn=toggle_subscription_panel,
-            inputs=[user_session, panel_visible_state],
-            outputs=[subscription_panel, panel_visible_state]
-        )
-
-        upload_btn.click(
-            fn=handle_upload,
-            inputs=[uploader, user_session, enable_table, enable_handwriting, enable_heavy_prep, enable_high_dpi, enable_fast, enable_tashkeel],
-            outputs=status_box
-        ).then(read_status_json, None, system_status)
-
-    # -------- ب) واجهة "ملفاتي" الكاملة --------
-    with gr.Column() as files_layout:
-        gr.Markdown("<h1 style='text-align:center;'>📁 الملفات الجاهزة للتحويل</h1>")
-        files_list = gr.Dropdown(label="اختر ملفاً من Final_Arabic_Files", choices=[], interactive=True)
-        refresh_btn = gr.Button("تحديث القائمة")
-
-        editor = gr.Textbox(label="مربع التعديل", lines=15)
-        filename_box = gr.Textbox(label="اسم الملف (بدون امتداد)")
-        ext_dropdown = gr.Dropdown(label="الامتداد", choices=["txt", "pdf", "docx", "xlsx", "csv", "html"], value="pdf")
-        font_dropdown = gr.Dropdown(label="الخط", choices=["Arial", "Tahoma", "Amiri-Regular", "Cairo-Regular"], value="Amiri-Regular")
-        font_size_dropdown = gr.Slider(label="حجم الخط", minimum=10, maximum=40, value=18, step=1)
-
-        convert_btn = gr.Button("تحويل")
-        download_file = gr.File(label="تحميل الملف")
-        convert_status = gr.HTML()
-
-        zip_btn = gr.Button("إنشاء ZIP")
-        zip_file = gr.File(label="تحميل ZIP")
-        zip_status = gr.HTML()
-
-        def refresh_files():
-            return gr.update(choices=list_final_files())
-
-        refresh_btn.click(fn=refresh_files, outputs=files_list)
-        files_list.change(
-            fn=load_file_content,
-            inputs=files_list,
-            outputs=[editor, filename_box, files_list]
-        )
-        convert_btn.click(
-            fn=convert_and_upload,
-            inputs=[editor, filename_box, ext_dropdown, font_dropdown, font_size_dropdown],
-            outputs=[convert_status, download_file]
-        )
-        zip_btn.click(
-            fn=do_zip,
-            inputs=[editor, filename_box],
-            outputs=[zip_status, zip_file]
-        )
-
-    print("[CHECKPOINT - UPLOAD_APP] <<< انتهى بناء وتأمين دمج الواجهتين بنجاح وخرجنا من الدالة.", flush=True)
-    return upload_layout, files_layout, user_session, welcome_box, rank_icon_btn, files_list
-
-print("[CHECKPOINT - UPLOAD_APP] <<< انتهى ملف upload_app.py بالكامل وهو جاهز للمطابقة السيادية.", flush=True)