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, AVAILABLE_MODELS 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, llm_api_key: str = None, entities_to_anonymize: List[str] = None # ✅ اضافه شد ): self.cerebras_key = cerebras_key or os.getenv("CEREBRAS_API_KEY") self.llm_provider = llm_provider self.llm_model = llm_model self.llm_api_key = llm_api_key self.entities_to_anonymize = entities_to_anonymize or ["person", "company", "amount", "percent"] self.mapping_table = {} self.reverse_mapping = {} # ایجاد LLM sender self._create_llm_sender() logger.info(f"✅ Anonymizer Advanced مقداردهی شد با {llm_provider}") def _create_llm_sender(self): """ایجاد LLM sender مناسب""" try: # ✅ تصمیم‌گیری برای API key if self.llm_provider == "chatgpt" and self.llm_model == "gpt-4o-mini": # فقط برای gpt-4o-mini از secret بخوان api_key = os.getenv("OPENAI_API_KEY") logger.info("🔑 استفاده از API key از Secret برای gpt-4o-mini") else: # برای بقیه مدل‌ها از input کاربر api_key = self.llm_api_key logger.info("🔑 استفاده از API key ورودی کاربر") # ایجاد 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, api_key: str = None, entities: List[str] = None): """تغییر provider و مدل LLM و موجودیت‌های ناشناس‌سازی""" self.llm_provider = provider self.llm_model = model self.llm_api_key = api_key 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 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}. مقادیر پولی → 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: ناشناس‌سازی متن prompt1 = f"""متن زیر را ناشناس کنید. قوانین: {instructions_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": 2048, "temperature": 0.1 }, timeout=60 ) if response2.status_code == 200: mapping_text = response2.json()['choices'][0]['message']['content'].strip() mapping_text = mapping_text.replace('```json', '').replace('```', '').strip() try: self.mapping_table = json.loads(mapping_text) self._fix_percent_mapping() self.reverse_mapping = {v: k for k, v in self.mapping_table.items()} logger.info(f"✅ Mapping استخراج شد: {len(self.mapping_table)} موجودیت") except json.JSONDecodeError: logger.warning("⚠️ خطا در parse کردن JSON mapping - استفاده از روش fallback") self._extract_mapping_from_text(text, anonymized_text) else: logger.warning("⚠️ خطا در دریافت mapping - استفاده از روش fallback") self._extract_mapping_from_text(text, anonymized_text) return anonymized_text, self.mapping_table except Exception as e: logger.error(f"❌ Cerebras Exception: {e}") raise def _fix_percent_mapping(self): """اصلاح mapping برای درصدها""" for token, value in self.mapping_table.items(): value_str = str(value).strip() if token.startswith('percent-'): if not re.search(r'(درصد|%|درصدی)', value_str): self.mapping_table[token] = f"{value_str} درصد" logger.info(f"✅ اصلاح {token}: '{value_str}' → '{value_str} درصد'") elif token.startswith('amount-'): if not re.search(r'(میلیارد|میلیون|هزار|تومان|ریال|دلار|یورو|تن)', value_str): logger.warning(f"⚠️ {token}: فقط عدد '{value_str}' - واحد مشخص نیست") def _extract_mapping_from_text(self, original: str, anonymized: str): """استخراج mapping از متن‌های اصلی و ناشناس شده - فقط برای موجودیت‌های انتخابی""" # ✅ استخراج فقط توکن‌های انتخابی all_tokens = [] for entity_type in self.entities_to_anonymize: tokens = re.findall(f'{entity_type}-\\d+', anonymized) all_tokens.extend([(t, entity_type) for t in tokens]) all_tokens = sorted(set(all_tokens), key=lambda x: (x[1], int(x[0].split('-')[1]))) # ✅ الگوهای موجودیت - فقط برای انتخابی‌ها patterns = {} if "person" in self.entities_to_anonymize: patterns['person'] = r'\b[ء-ي]+\s+[ء-ي]+(?:\s+[ء-ي]+)*\b' if "company" in self.entities_to_anonymize: patterns['company'] = r'(?:شرکت|بانک|سازمان|گروه|هلدینگ)\s+[ء-ي]+(?:\s+[ء-ي]+)*' if "amount" in self.entities_to_anonymize: patterns['amount'] = r'\d+(?:\.\d+)?\s*(?:میلیارد|میلیون|هزار|تومان|ریال|دلار|یورو|تن)' if "percent" in self.entities_to_anonymize: patterns['percent'] = r'\d+(?:\.\d+)?\s*(?:درصد|%|درصدی)' original_entities = {} for entity_type, pattern in patterns.items(): matches = list(re.finditer(pattern, original)) original_entities[entity_type] = [m.group().strip() for m in matches] for token, entity_type in all_tokens: if entity_type in original_entities and original_entities[entity_type]: token_num = int(token.split('-')[1]) - 1 if token_num < len(original_entities[entity_type]): original_text = original_entities[entity_type][token_num] self.mapping_table[token] = original_text self.reverse_mapping[original_text] = token else: original_text = original_entities[entity_type][-1] if token not in self.mapping_table: self.mapping_table[token] = original_text self.reverse_mapping[original_text] = token def analyze_with_llm(self, anonymized_text: str, analysis_prompt: str = None) -> str: """استفاده از LLM یکپارچه""" logger.info(f"🤖 {self.llm_provider.upper()} اجرای پرامپت...") if not analysis_prompt or not analysis_prompt.strip(): logger.info("⚠️ پرامپت خالی - بدون تحلیل") return "⚠️ هیچ دستور تحلیل داده نشده است" combined_text = f"""متن ناشناس‌سازی شده: {anonymized_text} دستورات: {analysis_prompt} توجه: در پاسخ از همان کدهای ناشناس (person-XX, company-XX, amount-XX, percent-XX) استفاده کن.""" try: response = self.llm_sender.send_simple(combined_text, lang='fa') logger.info(f"✅ {self.llm_provider.upper()}: {len(response)} کاراکتر") return response except Exception as e: logger.error(f"❌ {self.llm_provider.upper()} Exception: {e}") return f"❌ خطا در ارتباط با {self.llm_provider.upper()}: {str(e)}" def restore_text(self, anonymized_text: str) -> str: """بازگردانی متن""" logger.info("🔄 بازگردانی متن...") if not self.mapping_table: logger.warning("⚠️ جدول نگاشت خالی است") return anonymized_text restored = anonymized_text for placeholder, original in sorted(self.mapping_table.items()): restored = restored.replace(placeholder, original) logger.info("✅ بازگردانی کامل") return restored def get_mapping_table_md(self) -> str: """تبدیل جدول نگاشت به Markdown""" if not self.mapping_table: return "### 📋 جدول نگاشت\n\nهیچ موجودیتی شناسایی نشد" table = "### 📋 جدول نگاشت\n\n" table += "| شناسه | متن اصلی |\n" table += "|-------|----------|\n" for token, original in sorted(self.mapping_table.items()): table += f"| **{token}** | {original} |\n" return table # متغیر سراسری anonymizer = None def process( input_text: str, analysis_prompt: str, llm_provider: str, llm_model: str, api_key_input: str, # ✅ checkboxها anonymize_all: bool, anonymize_person: bool, anonymize_company: bool, anonymize_amount: bool, anonymize_percent: bool ): """پردازش متن - 4 مرحله""" global anonymizer if not input_text.strip(): return "", "", "", "" # ✅ ساخت لیست موجودیت‌های انتخابی 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 "", "❌ لطفاً حداقل یک موجودیت برای ناشناس‌سازی انتخاب کنید", "", "" cerebras_key = os.getenv("CEREBRAS_API_KEY") # ایجاد یا آپدیت anonymizer if not anonymizer: anonymizer = AnonymizerAdvanced( cerebras_key, llm_provider=llm_provider, llm_model=llm_model, llm_api_key=api_key_input, entities_to_anonymize=entities # ✅ ارسال لیست موجودیت‌ها ) else: anonymizer.set_llm_provider(llm_provider, llm_model, api_key_input, entities) anonymizer.mapping_table = {} anonymizer.reverse_mapping = {} try: logger.info("=" * 70) logger.info(f"🚀 شروع پردازش - LLM: {llm_provider} ({llm_model})") logger.info(f"🎯 موجودیت‌های انتخابی: {', '.join(entities)}") logger.info("=" * 70) # مرحله 1: ناشناس‌سازی logger.info("📝 مرحله 1: ناشناس‌سازی...") anonymized_text, _ = anonymizer.anonymize_with_cerebras(input_text) logger.info(f"✅ ناشناس‌سازی: {len(anonymized_text)} کاراکتر") # مرحله 2: LLM logger.info(f"🤖 مرحله 2: {llm_provider.upper()}...") llm_response = anonymizer.analyze_with_llm(anonymized_text, analysis_prompt) logger.info(f"✅ {llm_provider.upper()}: {len(llm_response)} کاراکتر") # مرحله 3: بازگردانی logger.info("🔄 مرحله 3: بازگردانی...") restored_text = anonymizer.restore_text(llm_response) logger.info("✅ بازگردانی کامل") # مرحله 4: جدول نگاشت logger.info("📋 مرحله 4: جدول نگاشت...") mapping_str = anonymizer.get_mapping_table_md() logger.info(f"✅ {len(anonymizer.mapping_table)} موجودیت") logger.info("=" * 70) logger.info("✅ تمام مراحل کامل!") logger.info("=" * 70) return restored_text, llm_response, anonymized_text, mapping_str except Exception as e: logger.error(f"❌ خطا: {str(e)}", exc_info=True) return "", f"❌ خطا: {str(e)}", "", "" def clear_all(): """پاک کردن همه""" return "", "", "", "", "", "", "", True, False, False, False, False # ✅ اضافه شد: checkboxها def update_model_choices(provider: str): """آپدیت لیست مدل‌ها بر اساس provider""" models = AVAILABLE_MODELS.get(provider, []) return gr.Dropdown(choices=models, value=models[0] if models else None) def update_api_key_visibility(provider: str, model: str): """نمایش/مخفی کردن textbox API key""" # ✅ فقط برای gpt-4o-mini مخفی کن if provider == "chatgpt" and model == "gpt-4o-mini": return gr.Textbox(visible=False, value="") else: return gr.Textbox(visible=True, value="") # Gradio Interface css_rtl = """ .input-box { direction: rtl; text-align: right; } .textbox textarea { direction: rtl; text-align: right; font-family: 'Tahoma', serif; } """ with gr.Blocks(title="سیستم ناشناس‌سازی متون", theme=gr.themes.Soft(), css=css_rtl) 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") llm_provider = gr.Dropdown( choices=["chatgpt", "grok"], value="chatgpt", label="🤖 انتخاب LLM", interactive=True ) llm_model = gr.Dropdown( choices=AVAILABLE_MODELS["chatgpt"], value="gpt-4o-mini", label="📦 انتخاب مدل", interactive=True ) # ✅ textbox برای API key (مخفی برای gpt-4o-mini) api_key_input = gr.Textbox( label="🔑 API Key", placeholder="فقط برای مدل‌های غیر از gpt-4o-mini", type="password", visible=False, # پیش‌فرض مخفی (چون gpt-4o-mini انتخاب شده) elem_classes="textbox" ) gr.Markdown( "💡 **نکته:** gpt-4o-mini از Secret خوانده می‌شود. برای بقیه مدل‌ها API key وارد کنید.", elem_classes="input-box" ) gr.Markdown("---") # ✅ انتخاب موجودیت‌ها برای ناشناس‌سازی with gr.Group(): gr.Markdown("### 🎯 انتخاب موجودیت‌ها", elem_classes="input-box") anonymize_all = gr.Checkbox( label="✅ همه موجودیت‌ها", value=True, elem_classes="input-box" ) anonymize_person = gr.Checkbox( label="👤 اسامی اشخاص", value=False, elem_classes="input-box" ) anonymize_company = gr.Checkbox( label="🏢 نام شرکت‌ها", value=False, elem_classes="input-box" ) anonymize_amount = gr.Checkbox( label="💰 ارقام مالی", value=False, elem_classes="input-box" ) anonymize_percent = gr.Checkbox( label="📊 درصدها", value=False, elem_classes="input-box" ) gr.Markdown( "💡 اگر 'همه' را انتخاب کنید، بقیه نادیده گرفته می‌شوند", elem_classes="input-box" ) gr.Markdown("---") analysis_prompt = gr.Textbox( lines=6, placeholder="مثال: این متن را خلاصه کن\nیا: نکات کلیدی را استخراج کن", label="📋 دستورات LLM (اختیاری)", elem_classes="textbox" ) gr.Markdown("---") with gr.Column(): process_btn = gr.Button( "▶️ پردازش", variant="primary", size="lg" ) clear_btn = gr.Button( "🗑️ پاک کردن", variant="stop", size="lg" ) # سمت چپ: متن ورودی with gr.Column(scale=3): input_text = gr.Textbox( lines=22, placeholder="متن مالی/خبری را وارد کنید...", label="📝 متن ورودی", elem_classes="textbox" ) # نتایج gr.Markdown("---") 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" ) gr.Markdown("---") mapping_table = gr.Markdown( value="### 📋 جدول نگاشت\n\nهنوز پردازشی انجام نشده", label="📋 جدول نگاشت", elem_classes="input-box" ) # Event Handlers # ✅ آپدیت مدل‌ها و نمایش API key def handle_provider_change(provider): models = AVAILABLE_MODELS.get(provider, []) default_model = models[0] if models else None # چک کن آیا باید API key نمایش داده بشه show_api = not (provider == "chatgpt" and default_model == "gpt-4o-mini") return ( gr.Dropdown(choices=models, value=default_model), gr.Textbox(visible=show_api, value="") ) llm_provider.change( fn=handle_provider_change, inputs=[llm_provider], outputs=[llm_model, api_key_input] ) # ✅ آپدیت نمایش API key وقتی مدل عوض میشه def handle_model_change(provider, model): show_api = not (provider == "chatgpt" and model == "gpt-4o-mini") return gr.Textbox(visible=show_api, value="") llm_model.change( fn=handle_model_change, inputs=[llm_provider, llm_model], outputs=[api_key_input] ) # ✅ وقتی "همه" انتخاب میشه، بقیه رو غیرفعال کن def handle_select_all(select_all): if select_all: # همه انتخاب شده، بقیه رو غیرفعال کن return ( gr.Checkbox(value=False, interactive=False), # person gr.Checkbox(value=False, interactive=False), # company gr.Checkbox(value=False, interactive=False), # amount gr.Checkbox(value=False, interactive=False) # percent ) else: # همه غیرفعال، بقیه رو فعال کن return ( gr.Checkbox(value=False, interactive=True), gr.Checkbox(value=False, interactive=True), gr.Checkbox(value=False, interactive=True), gr.Checkbox(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, api_key_input, # ✅ checkboxها anonymize_all, anonymize_person, anonymize_company, anonymize_amount, anonymize_percent ], outputs=[restored_text, llm_analysis, anonymized_text, mapping_table] ) # پاک کردن clear_btn.click( fn=clear_all, outputs=[ input_text, analysis_prompt, api_key_input, restored_text, llm_analysis, anonymized_text, mapping_table, # ✅ checkboxها anonymize_all, anonymize_person, anonymize_company, anonymize_amount, anonymize_percent ] ) if __name__ == "__main__": print("=" * 70) print("🚀 سیستم ناشناس‌سازی متون در حال راه‌اندازی...") print("=" * 70) print("\n📋 نحوه استفاده:\n") print("1. کلیدهای API را تنظیم کنید:") print(" - CEREBRAS_API_KEY (ضروری)") print(" - OPENAI_API_KEY (فقط برای gpt-4o-mini)") print("2. http://localhost:7860 را باز کنید") print("3. LLM و مدل را انتخاب کنید") print("4. برای مدل‌های غیر از gpt-4o-mini، API key وارد کنید") print("5. متن را وارد کنید") print("6. 'پردازش' را کلیک کنید\n") print("💡 فقط gpt-4o-mini از Secret می‌خواند") print(" بقیه مدل‌ها نیاز به API key دارند") print("=" * 70 + "\n") app.launch( server_name="0.0.0.0", server_port=7860, share=False, show_error=True )