Spaces:
Sleeping
Sleeping
| import gradio as gr | |
| import re | |
| import os | |
| import requests | |
| import logging | |
| from typing import Dict, List, Tuple, Set | |
| import json | |
| logging.basicConfig(level=logging.INFO) | |
| logger = logging.getLogger(__name__) | |
| class AnonymizerCerebrasEnhanced: | |
| def __init__(self, api_key: str = None): | |
| self.api_key = api_key or os.getenv("CEREBRAS_API_KEY") | |
| self.mapping_table = {} | |
| self.counters = { | |
| 'company': 0, 'person': 0, 'amount': 0, 'phone': 0, | |
| 'email': 0, 'id_number': 0, 'date': 0, 'location': 0, | |
| 'percent': 0 | |
| } | |
| self.seen_entities = {} # برای ثبات نگاشت | |
| if not self.api_key: | |
| raise ValueError("❌ کلید API Cerebras یافت نشد!") | |
| logger.info("✅ Anonymizer Enhanced مقداردهی شد") | |
| def get_system_prompt(self) -> str: | |
| """ایجاد دستورالعمل سیستمی بهینه شده""" | |
| return """شما یک سیستم ناشناسسازی متون مالی فارسی هستید. | |
| ⚠️ CRITICAL: در پاسخ نهایی خود، فقط و فقط متن ناشناسسازی شده را برگردانید، بدون هیچ توضیح، تحلیل، یا تگ اضافی. | |
| ## قوانین اندیسگذاری: | |
| 1. **ترتیب پیوسته**: company-01, company-02, ... | person-01, person-02, ... | amount-01, amount-02, ... | percent-01, percent-02, ... | |
| 2. **ثبات**: اگر "همراه اول" → company-01 شد، در تمام متن همان باشد | |
| 3. **نام مستعار**: "فاما" = "فولاد مبارکه" → هر دو company-01 | |
| 4. **اشاره ضمنی**: "این شرکت" اگر به company-01 اشاره دارد → company-01 (نه company-02) | |
| ## انواع موجودیت: | |
| - **company-XX**: شرکتها، بانکها، سازمانها، گروهها | |
| - **person-XX**: نام و نام خانوادگی اشخاص | |
| - **amount-XX**: مبالغ - واحد را حفظ کن | |
| - **percent-XX**: درصدها | |
| - **phone-XX**: شماره تلفن | |
| - **email-XX**: آدرس ایمیل | |
| - **date-XX**: تاریخ و دوره زمانی مشخص (نه "ماهه") | |
| - **location-XX**: شهر، استان، کشور | |
| - **id_number-XX**: شماره شناسایی، کد ملی | |
| ## قوانین کلیدی: | |
| 1. **بازرس = شرکت**: "بازرس شرکت X" → بازرس حفظ، X = company-XX | |
| 2. **واحدها**: "amount-01 میلیارد تومان" ✅ (واحد را حفظ کن) | |
| 3. **گروهها**: "گروه X" → company-XX | |
| 4. **کلمات عمومی حفظ**: "سه شرکت" → حفظ (فقط نام شرکت را ناشناس کن) | |
| 5. **دوره زمانی حفظ**: "۵ ماهه" → حفظ (فقط تاریخ مشخص = date-XX) | |
| 6. **بازه = یک entity**: "یک تا 1.5 میلیون" → amount-01 | |
| 7. **درصدها**: شناسایی تمام درصدها (خصوصاً بین 50 تا 70) | |
| 8. **تمام ارقام**: شناسایی تمام ارقام موجود در متن به عنوان amount-XX | |
| ## فرمت خروجی JSON: | |
| [ | |
| {"text": "متن دقیق موجودیت", "type": "company", "original": "نام اصلی"}, | |
| {"text": "...", "type": "person", "original": "..."}, | |
| ... | |
| ] | |
| ✅ فقط متن ناشناسشده را برگردانید.""" | |
| def get_user_prompt(self, text: str) -> str: | |
| """تشکیل پرامپت کاربر""" | |
| return f"""متن مالی فارسی زیر را تجزیه و تحلیل کنید. تمام موجودیتهای حساس را شناسایی کنید و یک JSON Array برگردانید. | |
| متن: | |
| {text} | |
| **مهم**: | |
| - اگر چند بار یک نام تکرار شود، یک id بدهید | |
| - کلمات عمومی را حفظ کنید | |
| - واحدها را حفظ کنید | |
| - فقط JSON برگردانید! | |
| یک JSON Array برگردانید. هر عنصر دارای: | |
| - "text": متن دقیق استخراج شده | |
| - "type": نوع (company, person, amount, percent, phone, email, date, location, id_number) | |
| - "original": توضیح اضافی اگر نام مستعار باشد""" | |
| def call_cerebras(self, text: str) -> List[Dict]: | |
| """فراخوانی Cerebras API با پرامپت بهبود شده""" | |
| logger.info("🔄 فراخوانی Cerebras API با دستورالعمل قوی...") | |
| system_prompt = self.get_system_prompt() | |
| user_prompt = self.get_user_prompt(text) | |
| try: | |
| response = requests.post( | |
| "https://api.cerebras.ai/v1/chat/completions", | |
| headers={ | |
| "Authorization": f"Bearer {self.api_key}", | |
| "Content-Type": "application/json" | |
| }, | |
| json={ | |
| "model": "llama-3.3-70b", | |
| "messages": [ | |
| {"role": "system", "content": system_prompt}, | |
| {"role": "user", "content": user_prompt} | |
| ], | |
| "max_tokens": 4000, | |
| "temperature": 0.1 | |
| }, | |
| timeout=30 | |
| ) | |
| if response.status_code != 200: | |
| logger.error(f"❌ خطای API Cerebras: {response.text}") | |
| return [] | |
| result = response.json() | |
| content = result['choices'][0]['message']['content'] | |
| try: | |
| # تمیز کردن محتوا از markdown اگر وجود داشته باشد | |
| content = content.replace("```json", "").replace("```", "").strip() | |
| entities = json.loads(content) | |
| if not isinstance(entities, list): | |
| entities = [] | |
| logger.info(f"✅ {len(entities)} موجودیت استخراج شد") | |
| return entities | |
| except json.JSONDecodeError: | |
| logger.error(f"❌ خطا در JSON parsing: {content[:200]}") | |
| return [] | |
| except Exception as e: | |
| logger.error(f"❌ خطا Cerebras: {e}") | |
| return [] | |
| def get_placeholder(self, entity_type: str) -> str: | |
| """تولید placeholder با format جدید""" | |
| type_lower = entity_type.lower() | |
| if type_lower not in self.counters: | |
| type_lower = 'amount' | |
| self.counters[type_lower] += 1 | |
| return f"{type_lower}-{self.counters[type_lower]:02d}" | |
| def anonymize(self, text: str) -> Tuple[str, List]: | |
| """ناشناسسازی متن با قوانین ثبات""" | |
| logger.info("🚀 شروع ناشناسسازی متن...") | |
| # تنظیف | |
| self.mapping_table = {} | |
| self.seen_entities = {} | |
| for key in self.counters: | |
| self.counters[key] = 0 | |
| # دریافت موجودیتها | |
| entities = self.call_cerebras(text) | |
| if not entities: | |
| logger.warning("⚠️ موجودیتی شناسایی نشد") | |
| return text, [] | |
| logger.info("🔄 Processing entities...") | |
| # جایگزینی با قانون ثبات | |
| anonymized = text | |
| replacements = [] | |
| for entity in entities: | |
| entity_type = entity.get('type', 'amount').lower() | |
| entity_text = entity.get('text', '').strip() | |
| original_info = entity.get('original', '') | |
| if not entity_text: | |
| continue | |
| # بررسی اگر این موجودیت قبلاً دیده شده است | |
| entity_key = (entity_type, entity_text.lower()) | |
| if entity_key in self.seen_entities: | |
| token = self.seen_entities[entity_key] | |
| logger.info(f"🔄 موجودیت تکراری: {entity_text} → {token}") | |
| else: | |
| token = self.get_placeholder(entity_type) | |
| self.seen_entities[entity_key] = token | |
| self.mapping_table[token] = { | |
| 'original': entity_text, | |
| 'type': entity_type, | |
| 'note': original_info | |
| } | |
| logger.info(f"✅ جایگزینی: {entity_text} → {token}") | |
| # جایگزینی دقیق (case-sensitive اول، سپس case-insensitive) | |
| idx = anonymized.find(entity_text) | |
| if idx != -1: | |
| anonymized = anonymized[:idx] + token + anonymized[idx + len(entity_text):] | |
| replacements.append({ | |
| 'original': entity_text, | |
| 'placeholder': token, | |
| 'type': entity_type | |
| }) | |
| logger.info(f"✅ ناشناسسازی کامل - {len(self.mapping_table)} نگاشت") | |
| return anonymized, entities | |
| def get_mapping_table_str(self) -> str: | |
| """جدول نگاشت جزئی""" | |
| if not self.mapping_table: | |
| return "❌ موجودیتی شناسایی نشد" | |
| result = "## 📊 جدول نگاشت\n\n" | |
| result += "| توکن | اطلاعات اصلی | نوع |\n" | |
| result += "|------|--------|------|\n" | |
| for token, info in sorted(self.mapping_table.items()): | |
| entity_type = info.get('type', 'unknown') | |
| original = info.get('original', '') | |
| note = info.get('note', '') | |
| note_str = f" ({note})" if note else "" | |
| result += f"| `{token}` | {original}{note_str} | {entity_type} |\n" | |
| return result | |
| def restore(self, text: str) -> str: | |
| """بازگردانی اطلاعات اصلی""" | |
| logger.info("🔄 بازگردانی اطلاعات...") | |
| restored = text | |
| for token, info in self.mapping_table.items(): | |
| original = info.get('original', '') | |
| restored = restored.replace(token, original) | |
| logger.info("✅ بازگردانی کامل") | |
| return restored | |
| # متغیرهای global | |
| anonymizer = None | |
| def process(input_text: str) -> Tuple[str, str, str, str, str]: | |
| """ | |
| روند کامل: | |
| 1. ناشناسسازی با Cerebras (llama-3.3-70b) + پرامپت قوی | |
| 2. ارسال به ChatGPT (حتما!) | |
| 3. بازگردانی پاسخ ChatGPT | |
| """ | |
| global anonymizer | |
| try: | |
| if not input_text.strip(): | |
| return "", "", "", "", "" | |
| # دریافت API Keys | |
| api_key_cerebras = os.getenv("CEREBRAS_API_KEY") | |
| api_key_gpt = os.getenv("OPENAI_API_KEY") | |
| if not api_key_gpt: | |
| logger.error("❌ OPENAI_API_KEY یافت نشد") | |
| return "", "", "", "", "" | |
| if not api_key_cerebras: | |
| logger.error("❌ CEREBRAS_API_KEY یافت نشد") | |
| return "", "", "", "", "" | |
| # ============================================ | |
| # مرحله 1: مقداردهی | |
| # ============================================ | |
| if not anonymizer: | |
| logger.info("Initializing anonymizer...") | |
| anonymizer = AnonymizerCerebrasEnhanced() | |
| # ============================================ | |
| # مرحله 2: ناشناسسازی با پرامپت قوی | |
| # ============================================ | |
| logger.info("Step 1: Anonymizing text with Cerebras...") | |
| anonymized_text, entities = anonymizer.anonymize(input_text) | |
| if not entities: | |
| logger.warning("⚠️ موجودیتی شناسایی نشد - متن ناشناس نشد") | |
| return input_text, "", "", "", "" | |
| # ============================================ | |
| # مرحله 3: جدول نگاشت | |
| # ============================================ | |
| logger.info("Step 2: Creating mapping table") | |
| mapping = anonymizer.get_mapping_table_str() | |
| logger.info(f"📋 {len(anonymizer.mapping_table)} نگاشت ایجاد شد") | |
| # ============================================ | |
| # مرحله 4: ارسال به ChatGPT (حتما!) | |
| # ============================================ | |
| logger.info("Step 3: Sending to ChatGPT...") | |
| prompt = f"""متن ناشناسشده زیر (متن مالی) را تحلیل و خلاصه کنید. | |
| متن: | |
| {anonymized_text} | |
| لطفاً: | |
| 1. خلاصهای مختصر و معنادار ارائه دهید | |
| 2. نکات اصلی را مشخص کنید | |
| 3. تمام توکنهای ناشناس (مثل company-01، amount-02) را حفظ کنید | |
| 4. تنها اطلاعات موجود در متن را بیان کنید""" | |
| logger.info(f"📤 ارسال به ChatGPT (gpt-4o-mini)...") | |
| try: | |
| gpt_response_obj = requests.post( | |
| "https://api.openai.com/v1/chat/completions", | |
| headers={"Authorization": f"Bearer {api_key_gpt}"}, | |
| json={ | |
| "model": "gpt-4o-mini", | |
| "messages": [ | |
| { | |
| "role": "system", | |
| "content": "شما دستیار تحلیل متون مالی فارسی هستید. متنهای ناشناسشده را دقیق تحلیل کنید. تمام توکنهای ناشناس را حفظ کنید." | |
| }, | |
| {"role": "user", "content": prompt} | |
| ], | |
| "max_tokens": 1500, | |
| "temperature": 0.7 | |
| }, | |
| timeout=30 | |
| ) | |
| if gpt_response_obj.status_code == 200: | |
| gpt_response = gpt_response_obj.json()['choices'][0]['message']['content'] | |
| logger.info("✅ پاسخ دریافت شد") | |
| else: | |
| error_text = gpt_response_obj.json().get('error', {}).get('message', gpt_response_obj.text) | |
| logger.error(f"❌ خطای ChatGPT: {error_text}") | |
| return input_text, anonymized_text, "", "", mapping | |
| except Exception as e: | |
| logger.error(f"❌ خطا در ارسال به ChatGPT: {e}") | |
| return input_text, anonymized_text, "", "", mapping | |
| # ============================================ | |
| # مرحله 5: بازگردانی پاسخ ChatGPT | |
| # ============================================ | |
| logger.info("Step 4: Restoring original text...") | |
| restored_text = anonymizer.restore(gpt_response) | |
| logger.info(f"✅ بازگردانی کامل") | |
| logger.info(f"Done. Input: {len(input_text)} | Anonymized: {len(anonymized_text)} | Entities: {len(entities)}") | |
| return input_text, anonymized_text, gpt_response, restored_text, mapping | |
| except Exception as e: | |
| logger.error(f"❌ خطا عمومی: {e}", exc_info=True) | |
| return "", "", "", "", "" | |
| def clear(): | |
| """پاک کردن""" | |
| return "", "", "", "", "" | |
| # رابط Gradio | |
| with gr.Blocks(title="Text Anonymization", theme=gr.themes.Soft()) as app: | |
| with gr.Row(): | |
| with gr.Column(scale=2): | |
| input_text = gr.Textbox( | |
| lines=12, | |
| placeholder="متن را وارد کنید...", | |
| label="Input" | |
| ) | |
| with gr.Column(scale=1): | |
| process_btn = gr.Button("Process", variant="primary", size="lg") | |
| clear_btn = gr.Button("Clear", variant="stop") | |
| with gr.Row(): | |
| with gr.Column(): | |
| anonymized_text = gr.Textbox( | |
| lines=10, | |
| label="Anonymized", | |
| interactive=False | |
| ) | |
| with gr.Column(): | |
| gpt_response = gr.Textbox( | |
| lines=10, | |
| label="GPT Response", | |
| interactive=False | |
| ) | |
| with gr.Column(): | |
| restored_text = gr.Textbox( | |
| lines=10, | |
| label="Restored", | |
| interactive=False | |
| ) | |
| with gr.Row(): | |
| with gr.Column(): | |
| mapping = gr.Textbox( | |
| lines=10, | |
| label="Mapping", | |
| interactive=False | |
| ) | |
| # Event handlers | |
| process_btn.click( | |
| fn=process, | |
| inputs=[input_text], | |
| outputs=[input_text, anonymized_text, gpt_response, restored_text, mapping] | |
| ) | |
| clear_btn.click( | |
| fn=clear, | |
| outputs=[input_text, anonymized_text, gpt_response, restored_text, mapping] | |
| ) | |
| if __name__ == "__main__": | |
| print("Starting Text Anonymization System...") | |
| app.launch( | |
| server_name="0.0.0.0", | |
| server_port=7860, | |
| share=False, | |
| show_error=True | |
| ) | |