📊 إحصائيات استهلاك المفاتيح
إجمالي الطلبات الناجحة: {self.total_requests}
| المفتاح |
الطلبات |
مؤشر الحمل |
{rows}
"""
return html
# --- دوال مساعدة ---
def read_docx_and_chunk(file_obj, chunk_size=4000):
try:
if not file_obj: return []
doc = Document(file_obj.name)
full_text = "\n".join([p.text for p in doc.paragraphs if p.text.strip()])
chunks = []
current_chunk = ""
for paragraph in full_text.split('\n'):
if len(current_chunk) + len(paragraph) < chunk_size:
current_chunk += paragraph + "\n"
else:
chunks.append(current_chunk)
current_chunk = paragraph + "\n"
if current_chunk: chunks.append(current_chunk)
return chunks
except: return []
def clean_json_text(text):
text = text.strip()
if text.startswith('```json'): text = text[7:]
if text.startswith('```'): text = text[3:]
if text.endswith('```'): text = text[:-3]
return text.strip()
def parse_api_keys(api_text, api_file):
keys = []
if api_file is not None:
try:
with open(api_file.name, 'r', encoding='utf-8') as f:
content = f.read()
keys = [line.split(',')[0].strip() for line in content.split('\n') if line.strip()]
except: pass
if not keys and api_text.strip():
keys = [k.strip() for k in api_text.split('\n') if k.strip()]
return keys
def generate_fiqh_system_instruction(author, madhab, field, book_type):
base_prompt = f"""
أنت باحث شرعي وخبير متخصص في تحويل التراث الإسلامي إلى بيانات هيكلية.
النص من مذهب: ({madhab})، فن: ({field}). المؤلف: ({author}). النوع: ({book_type}).
"""
if "متن" in book_type:
specific = 'استخرج القواعد والمسائل (المنطوق) بدقة.'
else:
specific = 'اربط الشرح بالمتن واستخرج المسألة مع التعليل والدليل.'
output_format = """
المطلوب: JSON List فقط.
كل عنصر: {"instruction": "السؤال/المسألة", "input": "السياق (اختياري)", "output": "الجواب المحرر", "source_meta": "المصدر"}
"""
return base_prompt + specific + output_format
# --- المحرك الرئيسي ---
def process_fiqh_book(file_obj, api_text, api_file, selected_model, author, madhab, field, book_type, delay):
# 1. إعداد المفاتيح
keys_list = parse_api_keys(api_text, api_file)
if not keys_list: return None, None, "⚠️ يجب إدخال مفاتيح API."
# إنشاء مدير المفاتيح
key_manager = KeyManager(keys_list)
if not file_obj: return None, key_manager.get_stats_html(), "⚠️ المرجو رفع ملف الكتاب."
chunks = read_docx_and_chunk(file_obj)
if not chunks: return None, key_manager.get_stats_html(), "⚠️ الملف فارغ أو معطوب."
system_instruction = generate_fiqh_system_instruction(author, madhab, field, book_type)
log_msg = f"🚀 تم تحميل {len(keys_list)} مفتاح.\nبدء المعالجة (توزيع الحمل المتساوي)...\n"
yield None, key_manager.get_stats_html(), log_msg
processed_data = []
for i, chunk in enumerate(chunks):
# الحصول على المفتاح التالي في الدور
current_key, key_index = key_manager.get_next_key()
try:
genai.configure(api_key=current_key)
model = genai.GenerativeModel(
model_name=selected_model,
system_instruction=system_instruction,
generation_config={"response_mime_type": "application/json"}
)
response = model.generate_content(chunk)
if not response.text: raise ValueError("رد فارغ")
data = json.loads(clean_json_text(response.text))
for item in data:
item['author'] = author
item['madhab'] = madhab
processed_data.extend(data)
# تسجيل النجاح وتحديث الإحصائيات
key_manager.increment_usage(key_index)
log_msg += f"✅ جزء {i+1} تم ({len(data)} مسألة) -> مفتاح {current_key[:5]}...\n"
# تحديث الواجهة (الجدول واللوج)
yield None, key_manager.get_stats_html(), log_msg
time.sleep(delay)
except Exception as e:
log_msg += f"❌ خطأ في جزء {i+1}: {str(e)}\n"
yield None, key_manager.get_stats_html(), log_msg
if processed_data:
fname = "fiqh_data.json"
with open(fname, 'w', encoding='utf-8') as f:
json.dump(processed_data, f, ensure_ascii=False, indent=4)
yield fname, key_manager.get_stats_html(), log_msg + "\n🎉 اكتملت المهمة."
else:
yield None, key_manager.get_stats_html(), log_msg + "\n⚠️ لم يتم استخراج بيانات."
# --- الواجهة ---
with gr.Blocks(theme=gr.themes.Soft()) as demo:
gr.Markdown("# 📚 محول الفقه (نظام توزيع الحمل المتوازن)")
with gr.Row():
with gr.Column(scale=1):
gr.Markdown("### 1️⃣ المفاتيح والإعدادات")
api_file = gr.File(label="ملف المفاتيح (txt)")
api_text = gr.Textbox(label="أو لصق المفاتيح", lines=3, placeholder="KEY_1\nKEY_2", type="password")
model_in = gr.Dropdown(USER_MODELS, value="models/gemini-2.5-flash", label="الموديل")
with gr.Group():
author_in = gr.Textbox(label="المؤلف")
madhab_in = gr.Dropdown(['المالكي', 'الشافعي', 'الحنفي', 'الحنبلي'], value='المالكي', label="المذهب")
field_in = gr.Textbox(value="الفقه", label="الفن")
type_in = gr.Dropdown(['متن', 'شرح'], value='شرح', label="النوع")
delay_in = gr.Slider(0, 10, 1, label="تأخير (ث)")
with gr.Column(scale=1):
gr.Markdown("### 2️⃣ التشغيل والمراقبة")
file_in = gr.File(label="ملف الكتاب (DOCX)")
btn = gr.Button("🚀 ابدأ المعالجة", variant="primary")
# مكون عرض الإحصائيات الجديد
stats_view = gr.HTML(label="إحصائيات المفاتيح")
logs = gr.Textbox(label="سجل العمليات", lines=8)
f_out = gr.File(label="تحميل النتيجة")
btn.click(
fn=process_fiqh_book,
inputs=[file_in, api_text, api_file, model_in, author_in, madhab_in, field_in, type_in, delay_in],
outputs=[f_out, stats_view, logs]
)
if __name__ == "__main__":
demo.queue().launch()