KashefTech's picture
Rename app_FIXED.py to app.py
ec7c242 verified
import gradio as gr
import re
import os
import requests
import json
import logging
from typing import Dict, List, Tuple, Optional
from llm_sender_unified import create_llm_sender
from qwen_anonymizer import create_qwen_anonymizer # ✅ اضافه شد
# ✅ مدل‌های موجود - به‌روزرسانی نوامبر 2024
AVAILABLE_MODELS = {
"chatgpt": [
# GPT-5 Series (جدیدترین)
"gpt-5.1", # بهترین برای کدنویسی و وظایف agentic
"gpt-5", # مدل reasoning قبلی
# GPT-4 Series
"gpt-4.1", # هوشمندترین non-reasoning
"gpt-4o", # قدرتمند
"gpt-4o-mini", # سریع و ارزان
"gpt-4-turbo", # سریع‌تر از GPT-4
],
"grok": [
# Grok-4 Series (جدیدترین)
"grok-4-fast-reasoning", # سریع با reasoning
"grok-4-fast-non-reasoning", # سریع بدون reasoning
"grok-4-0709", # نسخه پایدار
# Grok-3 Series
"grok-3", # قدرتمند
"grok-3-mini", # سبک
# Grok-2 Series
"grok-2-vision-1212", # با قابلیت بینایی
"grok-2-1212", # نسخه پایدار
"grok-2" # نسخه قدیمی
]
}
# ✅ روش‌های ناشناس‌سازی
ANONYMIZATION_METHODS = {
"cerebras": "🧠 Cerebras (Llama 3.3 70B)",
"qwen": "🤖 Qwen2.5-1.5B (فاین‌تیون شده)" # ✅ جدید
}
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class AnonymizerAdvanced:
"""ناشناس‌ساز پیشرفته با روش‌های متعدد"""
def __init__(
self,
cerebras_key: str = None,
llm_provider: str = "chatgpt",
llm_model: str = None,
entities_to_anonymize: List[str] = None,
anonymization_method: str = "cerebras" # ✅ جدید
):
self.cerebras_key = cerebras_key or os.getenv("CEREBRAS_API_KEY")
self.llm_provider = llm_provider
self.llm_model = llm_model
self.entities_to_anonymize = entities_to_anonymize or ["person", "company", "amount", "percent"]
self.anonymization_method = anonymization_method # ✅ جدید
self.mapping_table = {}
self.reverse_mapping = {}
# ایجاد LLM sender
self._create_llm_sender()
# ✅ ایجاد Qwen Anonymizer
if self.anonymization_method == "qwen":
self.qwen_anonymizer = create_qwen_anonymizer()
logger.info("✅ Qwen Anonymizer مقداردهی شد")
logger.info(f"✅ Anonymizer Advanced مقداردهی شد با {llm_provider}")
def _create_llm_sender(self):
"""ایجاد LLM sender مناسب"""
try:
# ✅ همیشه از Hugging Face Secrets استفاده کن
if self.llm_provider == "chatgpt":
api_key = os.getenv("OPENAI_API_KEY")
logger.info("🔑 استفاده از OPENAI_API_KEY از Secrets")
elif self.llm_provider == "grok":
api_key = os.getenv("XAI_API_KEY")
logger.info("🔑 استفاده از XAI_API_KEY از Secrets")
else:
api_key = None
logger.warning("⚠️ Provider ناشناخته")
# ایجاد sender
self.llm_sender = create_llm_sender(
provider=self.llm_provider,
api_key=api_key,
model=self.llm_model
)
logger.info(f"✅ LLM Sender ایجاد شد: {self.llm_provider} - {self.llm_sender.model}")
except Exception as e:
logger.error(f"❌ خطا در ایجاد LLM Sender: {e}")
# fallback to ChatGPT
self.llm_sender = create_llm_sender("chatgpt")
def set_llm_provider(self, provider: str, model: str = None, entities: List[str] = None):
"""تغییر provider و مدل LLM و موجودیت‌های ناشناس‌سازی"""
self.llm_provider = provider
self.llm_model = model
if entities is not None:
self.entities_to_anonymize = entities
self._create_llm_sender()
logger.info(f"✅ LLM تغییر یافت به: {provider} - {model}")
logger.info(f"✅ موجودیت‌های ناشناس‌سازی: {self.entities_to_anonymize}")
def set_anonymization_method(self, method: str):
"""تغییر روش ناشناس‌سازی"""
self.anonymization_method = method
if method == "qwen" and not hasattr(self, 'qwen_anonymizer'):
self.qwen_anonymizer = create_qwen_anonymizer()
logger.info("✅ Qwen Anonymizer مقداردهی شد")
logger.info(f"✅ روش ناشناس‌سازی تغییر یافت به: {method}")
def anonymize_with_cerebras(self, text: str) -> Tuple[str, Dict]:
"""ناشناس‌سازی با Cerebras - بر اساس موجودیت‌های انتخابی"""
logger.info("🧠 روش Cerebras...")
if not self.cerebras_key:
logger.error("❌ Cerebras API Key موجود نیست")
raise ValueError("Cerebras API Key مورد نیاز است")
# ✅ ساخت دستورات بر اساس موجودیت‌های انتخابی
instructions = []
instruction_number = 1
if "person" in self.entities_to_anonymize:
instructions.append(f"{instruction_number}. اسامی اشخاص → person-01, person-02, ...")
instruction_number += 1
if "company" in self.entities_to_anonymize:
instructions.append(f"{instruction_number}. نام شرکت‌ها/سازمان‌ها → company-01, company-02, ...")
instruction_number += 1
if "amount" in self.entities_to_anonymize:
instructions.append(f"{instruction_number}. اعداد و ارقام و مبالغ (مثل: 50 میلیارد، 100 هزار، 25.5 میلیون، ۳۰۰ دستگاه) → amount-01, amount-02, ...")
instruction_number += 1
if "percent" in self.entities_to_anonymize:
instructions.append(f"{instruction_number}. درصدها → percent-01, percent-02, ...")
instruction_number += 1
# اگه هیچی انتخاب نشده، متن رو همون‌طور برگردون
if not instructions:
logger.warning("⚠️ هیچ موجودیتی برای ناشناس‌سازی انتخاب نشده!")
return text, {}
instructions_text = "\n".join(instructions)
instructions_text += f"\n{instruction_number}. فقط این توکن‌ها استفاده کنید"
instructions_text += f"\n{instruction_number + 1}. شماره‌های نسخه را درست حفظ کنید"
instructions_text += f"\n{instruction_number + 2}. اگر موجودیت تکرار شود از شماره قدیمی استفاده کنید"
try:
# مرحله 1: ناشناس‌سازی متن
# ✅ ساخت مثال برای amount (اگر انتخاب شده)
example_text = ""
if "amount" in self.entities_to_anonymize:
example_text = """
مثال:
متن اصلی: "فروش 50 میلیارد ریال در سال گذشته بود."
متن ناشناس: "فروش amount-01 در سال گذشته بود."
"""
prompt1 = f"""متن زیر را ناشناس کنید. قوانین:
{instructions_text}
{example_text}
متن:
{text}
خروجی: فقط متن ناشناس شده (بدون توضیح اضافی)"""
response1 = requests.post(
"https://api.cerebras.ai/v1/chat/completions",
headers={
"Authorization": f"Bearer {self.cerebras_key}",
"Content-Type": "application/json"
},
json={
"model": "llama-3.3-70b",
"messages": [{"role": "user", "content": prompt1}],
"max_tokens": 4096,
"temperature": 0.1
},
timeout=60
)
if response1.status_code != 200:
logger.error(f"❌ Cerebras Error: {response1.status_code}")
raise Exception(f"Cerebras API Error: {response1.status_code}")
anonymized_text = response1.json()['choices'][0]['message']['content'].strip()
logger.info("✅ Cerebras: ناشناس‌سازی موفق")
# مرحله 2: استخراج mapping - فقط برای موجودیت‌های انتخابی
mapping_instructions = []
json_example = "{\n"
if "person" in self.entities_to_anonymize:
mapping_instructions.append('- برای person-XX: نام کامل شخص (مثلاً "علی احمدی")')
json_example += ' "person-01": "متن اصلی کامل",\n'
if "company" in self.entities_to_anonymize:
mapping_instructions.append('- برای company-XX: نام کامل شرکت/سازمان (مثلاً "شرکت پتروشیمی")')
json_example += ' "company-01": "متن اصلی کامل",\n'
if "amount" in self.entities_to_anonymize:
mapping_instructions.append('- برای amount-XX: عدد + واحد (مثلاً "80 هزار تومان" یا "50 میلیارد ریال")')
json_example += ' "amount-01": "متن اصلی کامل با واحد",\n'
if "percent" in self.entities_to_anonymize:
mapping_instructions.append('- برای percent-XX: عدد + کلمه "درصد" (مثلاً "40 درصد" نه فقط "40")')
json_example += ' "percent-01": "عدد + درصد",\n'
json_example += " ...\n}"
mapping_instructions_text = "\n".join(mapping_instructions)
prompt2 = f"""متن اصلی:
{text}
متن ناشناس شده:
{anonymized_text}
لطفاً یک جدول mapping برای همه توکن‌های ناشناس ایجاد کن.
برای هر توکن، متن اصلی کامل آن را مشخص کن.
**مهم:**
{mapping_instructions_text}
خروجی را به این فرمت JSON بده (فقط JSON، بدون توضیح اضافی):
{json_example}"""
response2 = requests.post(
"https://api.cerebras.ai/v1/chat/completions",
headers={
"Authorization": f"Bearer {self.cerebras_key}",
"Content-Type": "application/json"
},
json={
"model": "llama-3.3-70b",
"messages": [{"role": "user", "content": prompt2}],
"max_tokens": 4096,
"temperature": 0.1
},
timeout=60
)
if response2.status_code != 200:
logger.error(f"❌ Cerebras Mapping Error: {response2.status_code}")
raise Exception(f"Cerebras Mapping Error: {response2.status_code}")
mapping_text = response2.json()['choices'][0]['message']['content'].strip()
# پارس JSON
mapping_dict = self._parse_json_mapping(mapping_text)
logger.info("✅ Cerebras: mapping استخراج شد")
return anonymized_text, mapping_dict
except Exception as e:
logger.error(f"❌ خطا در Cerebras: {e}")
return f"❌ خطا در ناشناس‌سازی: {str(e)}", {}
def anonymize_with_qwen(self, text: str) -> Tuple[str, Dict]:
"""✅ ناشناس‌سازی با مدل فاین‌تیون شده Qwen"""
logger.info("🤖 روش Qwen...")
try:
if not hasattr(self, 'qwen_anonymizer'):
self.qwen_anonymizer = create_qwen_anonymizer()
anonymized_text, mapping_dict = self.qwen_anonymizer.anonymize(
text,
entities_to_anonymize=self.entities_to_anonymize
)
logger.info("✅ Qwen: ناشناس‌سازی موفق")
return anonymized_text, mapping_dict
except Exception as e:
logger.error(f"❌ خطا در Qwen: {e}")
return f"❌ خطا در ناشناس‌سازی: {str(e)}", {}
def _parse_json_mapping(self, text: str) -> Dict:
"""پارس کردن JSON mapping از پاسخ LLM"""
try:
# حذف markdown code blocks
text = re.sub(r'```json\s*', '', text)
text = re.sub(r'```\s*', '', text)
text = text.strip()
# پیدا کردن اولین { و آخرین }
start = text.find('{')
end = text.rfind('}')
if start != -1 and end != -1:
json_text = text[start:end+1]
mapping = json.loads(json_text)
return mapping
else:
logger.error("❌ فرمت JSON نامعتبر")
return {}
except json.JSONDecodeError as e:
logger.error(f"❌ خطا در پارس JSON: {e}")
return {}
except Exception as e:
logger.error(f"❌ خطای غیرمنتظره: {e}")
return {}
def anonymize(self, text: str) -> Tuple[str, Dict]:
"""✅ ناشناس‌سازی با روش انتخابی"""
if self.anonymization_method == "cerebras":
return self.anonymize_with_cerebras(text)
elif self.anonymization_method == "qwen":
return self.anonymize_with_qwen(text)
else:
logger.error(f"❌ روش ناشناخته: {self.anonymization_method}")
return text, {}
def restore_text(self, anonymized_text: str) -> str:
"""بازگردانی متن با استفاده از mapping"""
restored = anonymized_text
for token, original in self.mapping_table.items():
restored = restored.replace(token, original)
return restored
def analyze_with_llm(self, anonymized_text: str, prompt: str) -> str:
"""تحلیل متن ناشناس شده با LLM"""
if not prompt or not prompt.strip():
return "❌ دستور LLM خالی است!"
full_prompt = f"""{prompt}
متن:
{anonymized_text}"""
try:
response = self.llm_sender.send_simple(full_prompt, lang='fa')
return response
except Exception as e:
logger.error(f"❌ خطا در تحلیل LLM: {e}")
return f"❌ خطا: {str(e)}"
# ==================== Gradio Interface ====================
# متغیر global برای نگهداری instance
anonymizer = None
def initialize_anonymizer():
"""مقداردهی اولیه anonymizer"""
global anonymizer
anonymizer = AnonymizerAdvanced(
llm_provider="chatgpt",
llm_model="gpt-4o-mini",
anonymization_method="cerebras" # پیش‌فرض
)
logger.info("✅ Anonymizer مقداردهی شد")
def process(
input_text: str,
analysis_prompt: str,
llm_provider: str,
llm_model: str,
anonymize_all: bool,
anonymize_person: bool,
anonymize_company: bool,
anonymize_amount: bool,
anonymize_percent: bool,
anonymization_method: str # ✅ پارامتر جدید
):
"""پردازش متن"""
global anonymizer
# بررسی اولیه
if not input_text or not input_text.strip():
return "", "", "", "### ⚠️ خطا\n\nمتن ورودی خالی است!"
try:
# تعیین موجودیت‌ها
if anonymize_all:
entities = ["person", "company", "amount", "percent"]
else:
entities = []
if anonymize_person:
entities.append("person")
if anonymize_company:
entities.append("company")
if anonymize_amount:
entities.append("amount")
if anonymize_percent:
entities.append("percent")
if not entities:
return "", "", "", "### ⚠️ خطا\n\nحداقل یک موجودیت را انتخاب کنید!"
# تنظیم anonymizer
anonymizer.set_llm_provider(llm_provider, llm_model, entities)
anonymizer.set_anonymization_method(anonymization_method) # ✅ جدید
# ناشناس‌سازی
anonymized_text, mapping = anonymizer.anonymize(input_text)
if not mapping:
return "", "", anonymized_text, "### ⚠️ هشدار\n\nMapping خالی است!"
anonymizer.mapping_table = mapping
# تحلیل (اختیاری) - باید اول انجام بشه!
has_analysis = analysis_prompt and analysis_prompt.strip()
if has_analysis:
# مرحله 1: تحلیل LLM
llm_analysis = anonymizer.analyze_with_llm(anonymized_text, analysis_prompt)
# مرحله 2: بازگردانی خروجی LLM (نه متن ناشناس اولیه!)
restored_text = anonymizer.restore_text(llm_analysis)
else:
# بدون تحلیل
llm_analysis = "تحلیل درخواست نشده"
# بازگردانی متن ناشناس اولیه
restored_text = anonymizer.restore_text(anonymized_text)
# ساخت جدول mapping
mapping_md = "### 📋 جدول نگاشت\n\n"
mapping_md += "| توکن | مقدار اصلی |\n"
mapping_md += "|------|------------|\n"
for token, value in sorted(mapping.items()):
mapping_md += f"| `{token}` | {value} |\n"
return restored_text, llm_analysis, anonymized_text, mapping_md
except Exception as e:
logger.error(f"❌ خطا در پردازش: {e}")
error_md = f"### ❌ خطا\n\n{str(e)}"
return "", "", "", error_md
def clear_all():
"""پاک کردن همه فیلدها"""
return (
"", # input_text
"", # analysis_prompt
"", # restored_text
"", # llm_analysis
"", # anonymized_text
"### 📋 جدول نگاشت\n\nهنوز پردازشی انجام نشده", # mapping_table
True, # anonymize_all
False, # anonymize_person
False, # anonymize_company
False, # anonymize_amount
False # anonymize_percent
)
# ==================== Gradio App ====================
# مقداردهی اولیه
initialize_anonymizer()
# CSS سفارشی
custom_css = """
.input-box {
direction: rtl;
text-align: right;
}
.textbox {
direction: rtl !important;
text-align: right !important;
}
.thick-divider {
border-top: 3px solid #ccc;
margin: 20px 0;
}
.compact-checkbox label {
padding: 5px 0 !important;
}
"""
with gr.Blocks(css=custom_css, title="سیستم ناشناس‌سازی متون", theme=gr.themes.Soft()) as app:
gr.Markdown(
"""
# 🔒 سیستم ناشناس‌سازی متون مالی و خبری
### با قابلیت تحلیل هوشمند و بازگردانی دقیق
""",
elem_classes="input-box"
)
# ردیف اول: تنظیمات
with gr.Row():
# سمت راست: انتخاب مدل
with gr.Column(scale=1):
with gr.Group():
gr.Markdown("### 🎯 انتخاب مدل", elem_classes="input-box")
# ✅ انتخاب روش ناشناس‌سازی
anonymization_method = gr.Radio(
choices=list(ANONYMIZATION_METHODS.keys()),
value="cerebras",
label="🔧 روش ناشناس‌سازی",
interactive=True,
elem_classes="input-box"
)
llm_provider = gr.Radio(
choices=["chatgpt", "grok"],
value="chatgpt",
label="🤖 انتخاب مدل زبانی (برای تحلیل)",
interactive=True
)
llm_model = gr.Dropdown(
choices=AVAILABLE_MODELS["chatgpt"],
value="gpt-4o-mini",
label="📦 انتخاب نسخه مدل",
interactive=True
)
# سمت چپ: انتخاب موجودیت‌ها
with gr.Column(scale=1):
with gr.Group():
gr.Markdown("### 🎯 انتخاب موجودیت‌ها", elem_classes="input-box")
anonymize_all = gr.Checkbox(
label="✅ همه موجودیت‌ها",
value=True,
elem_classes="input-box compact-checkbox"
)
anonymize_person = gr.Checkbox(
label="👤 اسامی اشخاص",
value=False,
elem_classes="input-box compact-checkbox"
)
anonymize_company = gr.Checkbox(
label="🏢 نام شرکت‌ها",
value=False,
elem_classes="input-box compact-checkbox"
)
anonymize_amount = gr.Checkbox(
label="💰 ارقام مالی",
value=False,
elem_classes="input-box compact-checkbox"
)
anonymize_percent = gr.Checkbox(
label="📊 درصدها",
value=False,
elem_classes="input-box compact-checkbox"
)
# خط جداکننده پررنگ
gr.Markdown("---", elem_classes="thick-divider")
# ردیف دوم: دستورات پردازش و متن ورودی
with gr.Row():
# سمت راست: دستورات پردازش
with gr.Column(scale=1):
gr.Markdown("### 📋 دستورات پردازش", elem_classes="input-box")
analysis_prompt = gr.Textbox(
lines=22,
placeholder="مثال: این متن را خلاصه کن\nیا: نکات کلیدی را استخراج کن",
label="📋 دستورات LLM (اختیاری)",
elem_classes="textbox"
)
# سمت چپ: متن ورودی
with gr.Column(scale=1):
gr.Markdown("### 📝 متن ورودی", elem_classes="input-box")
input_text = gr.Textbox(
lines=22,
placeholder="متن مالی/خبری را وارد کنید...",
label="",
elem_classes="textbox"
)
# دکمه‌های پردازش و پاک کردن
with gr.Row():
process_btn = gr.Button(
"▶️ پردازش",
variant="primary",
size="lg",
scale=2
)
clear_btn = gr.Button(
"🗑️ پاک کردن",
variant="stop",
size="lg",
scale=1
)
# نتایج
gr.Markdown("## 📊 نتایج پردازش", elem_classes="input-box")
with gr.Row():
with gr.Column(scale=1):
restored_text = gr.Textbox(
lines=12,
label="✅ متن بازگردانی شده",
interactive=False,
elem_classes="textbox"
)
with gr.Column(scale=1):
llm_analysis = gr.Textbox(
lines=12,
label="🤖 تحلیل LLM",
interactive=False,
elem_classes="textbox"
)
with gr.Column(scale=1):
anonymized_text = gr.Textbox(
lines=12,
label="🔒 متن ناشناس‌شده",
interactive=False,
elem_classes="textbox"
)
mapping_table = gr.Markdown(
value="### 📋 جدول نگاشت\n\nهنوز پردازشی انجام نشده",
label="📋 جدول نگاشت",
elem_classes="input-box"
)
# Event Handler برای تغییر provider
def handle_provider_change(provider):
models = AVAILABLE_MODELS.get(provider, [])
default_model = models[0] if models else None
return gr.update(choices=models, value=default_model)
llm_provider.change(
fn=handle_provider_change,
inputs=[llm_provider],
outputs=[llm_model]
)
def handle_select_all(select_all):
if select_all:
return (
gr.update(value=False, interactive=False),
gr.update(value=False, interactive=False),
gr.update(value=False, interactive=False),
gr.update(value=False, interactive=False)
)
else:
return (
gr.update(value=False, interactive=True),
gr.update(value=False, interactive=True),
gr.update(value=False, interactive=True),
gr.update(value=False, interactive=True)
)
anonymize_all.change(
fn=handle_select_all,
inputs=[anonymize_all],
outputs=[anonymize_person, anonymize_company, anonymize_amount, anonymize_percent]
)
# پردازش
process_btn.click(
fn=process,
inputs=[
input_text,
analysis_prompt,
llm_provider,
llm_model,
anonymize_all,
anonymize_person,
anonymize_company,
anonymize_amount,
anonymize_percent,
anonymization_method # ✅ جدید
],
outputs=[restored_text, llm_analysis, anonymized_text, mapping_table]
)
# پاک کردن
clear_btn.click(
fn=clear_all,
outputs=[
input_text,
analysis_prompt,
restored_text,
llm_analysis,
anonymized_text,
mapping_table,
anonymize_all,
anonymize_person,
anonymize_company,
anonymize_amount,
anonymize_percent
]
)
if __name__ == "__main__":
print("=" * 70)
print("🚀 سیستم ناشناس‌سازی متون در حال راه‌اندازی...")
print("=" * 70)
print("\n📋 نحوه استفاده:\n")
print("1. API Keyها را در Hugging Face Secrets تنظیم کنید:")
print(" - CEREBRAS_API_KEY (برای روش Cerebras)")
print(" - HF_TOKEN (برای استفاده از مدل Qwen)")
print(" - OPENAI_API_KEY (برای ChatGPT)")
print(" - XAI_API_KEY (برای Grok)")
print("2. روش ناشناس‌سازی را انتخاب کنید (Cerebras یا Qwen)")
print("3. مدل زبانی (ChatGPT/Grok) و نسخه مدل را انتخاب کنید")
print("4. موجودیت‌های مورد نظر برای ناشناس‌سازی را انتخاب کنید")
print("5. متن و دستورات پردازش را وارد کنید")
print("6. 'پردازش' را کلیک کنید\n")
print("=" * 70 + "\n")
app.launch(
server_name="0.0.0.0",
server_port=7860,
share=False,
show_error=True
)