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 )