#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 🔐 سیستم پیشرفته ناشناسسازی مالی/خبری فارسی Cerebras Llama 3.3-70b + OpenAI ChatGPT Integration Version: 1.0.0 Author: Advanced Anonymization System Last Updated: October 2025 نحوه استفاده: 1. pip install gradio requests 2. export CEREBRAS_API_KEY="your_key" 3. export OPENAI_API_KEY="your_key" 4. python llma3_3-70b_with_chatgpt.py 5. http://localhost:7860 """ import requests import json import gradio as gr import logging from typing import Dict, Any, Tuple import os from dataclasses import dataclass import re from datetime import datetime # تنظیم Logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) @dataclass class CerebrasConfig: """تنظیمات Cerebras API""" api_key: str base_url: str = "https://api.cerebras.ai/v1" model: str = "llama-3.3-70b" max_tokens: int = 2000 temperature: float = 0.1 class AdvancedCerebrasAnonymizer: """سیستم پیشرفته ناشناسسازی متون مالی/خبری فارسی با ChatGPT""" def __init__(self, api_key: str = None, openai_api_key: str = None): """ مقداردهی اولیه Args: api_key: کلید API Cerebras (اختیاری - از محیط خواند) openai_api_key: کلید API OpenAI (اختیاری - از محیط خواند) """ # کلید Cerebras if api_key is None: api_key = os.getenv("CEREBRAS_API_KEY") if not api_key: raise ValueError("❌ کلید API Cerebras یافت نشد. لطفاً CEREBRAS_API_KEY را تنظیم کنید.") self.config = CerebrasConfig(api_key=api_key) self.openai_api_key = openai_api_key or os.getenv("OPENAI_API_KEY", "") self.system_prompt = self._create_advanced_system_prompt() self.mapping_table = {} logger.info("✅ سیستم ناشناسسازی آماده شد") def _create_advanced_system_prompt(self) -> str: """ایجاد دستورالعمل سیستمی برای Cerebras""" return """شما یک «ناشناسساز متون مالی/خبری فارسی» هستید. وظیفهتان جایگزینی اسامی خاص و مقادیر عددی با شناسههای بیمعناست. ## **قوانین اندیسگذاری - CRITICAL** ### **1. ترتیب شمارهگذاری الزامی:** - شرکتها: company-01, company-02, company-03, ... - اشخاص: person-01, person-02, person-03, ... - اعداد: amount-01, amount-02, amount-03, ... - درصدها: percent-01, percent-02, percent-03, ... ### **2. ثبات شناسهها در متن:** - اگر "همراه اول" اولبار company-01 شد، در تمام متن همان باشد - اگر "مهدی احمدی" اولبار person-01 شد، در تمام متن همان باشد ### **3. تشخیص صحیح انواع:** **شرکت/سازمان:** - همراه اول، بانک ملی، ایرانخودرو، سایپا، بانک مرکزی - سامانه کدال، وزارت نفت، سازمان تنظیم مقررات رادیویی **⚠️ CRITICAL - گروهها:** - "گروه همراه اول" → company-XX (نه group-XX) - "گروه اقتصادی آزادگان" → company-XX **⚠️ CRITICAL - کلمات عمومی:** - "سه شرکت دارویی" → کلمه عمومی، حفظ شود - "چند بانک" → کلمه عمومی، حفظ شود **شخص:** مهدی اخوان بهابادی، محمدرضا فرزین، ابوالفضل نجارزاده **عدد:** 37، 70، 677، 73.7، 178 (هر عددی) **درصد:** 37 درصدی، 15 درصدی، 53 درصد، 43% ## **مثالهای صحیح:** **مثال 1:** - ورودی: "مهدی اخوان بهابادی، مدیرعامل همراه اول، اعلام کرد درآمد 70 هزار میلیارد تومان با رشد 37 درصدی رسید." - خروجی: "person-01، مدیرعامل company-01، اعلام کرد درآمد amount-01 با رشد percent-01 رسید." **مثال 2:** - ورودی: "بانک مرکزی و بانک ملی با همکاری محمدرضا فرزین، 60 درصد سپردهها را مدیریت کردند." - خروجی: "company-01 و company-02 با همکاری person-01، percent-01 سپردهها را مدیریت کردند." ## **⚠️ قوانین پیشرفته:** 1. **حفظ هویت شرکت:** اگر "شرکت X" → company-01 شد، "این شرکت" باید همان company-01 باشد 2. **واحدها:** "amount-01 میلیارد تومان" ✅ (واحد حفظ شود) 3. **اعداد خاص:** شماره ثبت، کد ملی → حفظ شوند (amount-XX نشوند) 4. **کلمات عمومی:** "سه خودروساز بزرگ" → حفظ شود""" def anonymize_text(self, text: str) -> Dict[str, Any]: """ ناشناسسازی متن با Cerebras Args: text: متن ورودی Returns: Dictionary شامل success, anonymized_text, usage """ try: if not text or not text.strip(): return { "success": False, "error": "متن ورودی خالی است" } logger.info(f"🔄 شروع ناشناسسازی... (طول: {len(text)} کاراکتر)") headers = { "Authorization": f"Bearer {self.config.api_key}", "Content-Type": "application/json" } payload = { "model": self.config.model, "messages": [ { "role": "system", "content": self.system_prompt }, { "role": "user", "content": f"لطفاً این متن را ناشناسسازی کنید:\n\n{text}" } ], "max_tokens": self.config.max_tokens, "temperature": self.config.temperature } # ارسال درخواست response = requests.post( f"{self.config.base_url}/chat/completions", headers=headers, json=payload, timeout=60 ) if response.status_code != 200: error_msg = response.text logger.error(f"❌ خطای API Cerebras: {response.status_code}") return { "success": False, "error": f"خطای API Cerebras: {response.status_code} - {error_msg}" } result = response.json() anonymized_text = result["choices"][0]["message"]["content"] usage = result.get("usage", {}) logger.info(f"✅ ناشناسسازی موفق! (Tokens: {usage.get('total_tokens', 'نامشخص')})") # استخراج کدهای ناشناس برای mapping self._extract_mapping(text, anonymized_text) return { "success": True, "anonymized_text": anonymized_text, "usage": usage } except Exception as e: logger.error(f"❌ خطا در ناشناسسازی: {str(e)}") return { "success": False, "error": f"خطا: {str(e)}" } def _extract_mapping(self, original: str, anonymized: str): """استخراج mapping از متنهای اصلی و ناشناسشده""" try: # یافتن تمام کدهای ناشناس patterns = [ r'(company-\d+)', r'(person-\d+)', r'(amount-\d+)', r'(percent-\d+)' ] for pattern in patterns: matches = re.findall(pattern, anonymized) for match in matches: if match not in self.mapping_table.values(): self.mapping_table[f"[{match}]"] = match logger.info(f"📊 کدهای ناشناس شناسایی شدند: {len(self.mapping_table)}") except Exception as e: logger.warning(f"⚠️ خطا در استخراج mapping: {str(e)}") def send_to_chatgpt(self, anonymized_text: str) -> Dict[str, Any]: """ ارسال متن ناشناسشده به ChatGPT Args: anonymized_text: متن ناشناسشده Returns: Dictionary شامل success, response """ try: if not anonymized_text or not anonymized_text.strip(): return { "success": False, "error": "متن ناشناسشده خالی است" } if not self.openai_api_key: logger.warning("⚠️ کلید OpenAI API تنظیم نشده است") return { "success": False, "error": "کلید OpenAI API تنظیم نشده است. لطفاً OPENAI_API_KEY را تنظیم کنید." } logger.info("🤖 ارسال به ChatGPT...") headers = { "Authorization": f"Bearer {self.openai_api_key}", "Content-Type": "application/json" } data = { "model": "gpt-4o-mini", "messages": [ { "role": "system", "content": "شما یک تحلیلگر مالی حرفهای هستید. متن حاوی کدهای ناشناس است. به درخواستها با دقت پاسخ دهید." }, { "role": "user", "content": anonymized_text } ], "max_tokens": 2000, "temperature": 0.7 } response = requests.post( "https://api.openai.com/v1/chat/completions", headers=headers, json=data, timeout=30 ) if response.status_code == 200: result = response.json() gpt_response = result['choices'][0]['message']['content'] logger.info("✅ پاسخ ChatGPT دریافت شد") return { "success": True, "response": gpt_response } else: error_data = response.json() if response.content else {} error_message = error_data.get('error', {}).get('message', response.text) logger.error(f"❌ خطای ChatGPT: {error_message}") return { "success": False, "error": f"خطای ChatGPT: {error_message}" } except Exception as e: logger.error(f"❌ خطا در ارتباط با ChatGPT: {str(e)}") return { "success": False, "error": f"خطا در ارتباط با ChatGPT: {str(e)}" } def deanonymize_response(self, gpt_response: str, mapping_table: Dict[str, str]) -> str: """ بازگردانی پاسخ ChatGPT Args: gpt_response: پاسخ ChatGPT mapping_table: جدول نگاشت Returns: متن بازگردانده شده """ try: if not mapping_table: logger.warning("⚠️ جدول نگاشت خالی است") return gpt_response logger.info("🔄 بازگردانی متن...") final_result = gpt_response reverse_mapping = {code: original for original, code in mapping_table.items()} # مرتبسازی بر اساس طول کد (طولانیترین اول) sorted_codes = sorted(reverse_mapping.items(), key=lambda x: len(x[0]), reverse=True) for code, original in sorted_codes: if code in final_result: final_result = final_result.replace(code, original) logger.info("✅ بازگردانی موفق") return final_result except Exception as e: logger.error(f"❌ خطا در بازگردانی: {str(e)}") return f"خطا در بازگردانی: {str(e)}" def create_interface(): """ایجاد رابط کاربری Gradio""" try: anonymizer = AdvancedCerebrasAnonymizer() logger.info("✅ سیستم با موفقیت راهاندازی شد") except ValueError as e: logger.error(f"❌ خطا: {str(e)}") return gr.Interface( fn=lambda x: str(e), inputs="textbox", outputs="textbox", title="❌ خطا در راهاندازی" ) def process_text(input_text: str, api_key_input: str, openai_key_input: str) -> Tuple[str, str, str, str]: """ پردازش متن از ابتدا تا انتها Returns: (statistics, anonymized_text, gpt_response, final_output) """ logger.info("=" * 60) logger.info("شروع پردازش متن") logger.info("=" * 60) # بررسی ورودی if not input_text.strip(): error_msg = "❌ لطفاً متن ورودی را وارد کنید!" logger.error(error_msg) return error_msg, "", "", "" # تنظیم کلیدهای API اگر ارائه شده باشند if api_key_input: anonymizer.config.api_key = api_key_input logger.info("✅ کلید Cerebras بهروز شد") if openai_key_input: anonymizer.openai_api_key = openai_key_input logger.info("✅ کلید OpenAI بهروز شد") try: # مرحله 1: ناشناسسازی logger.info("مرحله 1️⃣: ناشناسسازی متن...") anon_result = anonymizer.anonymize_text(input_text) if not anon_result["success"]: error_msg = f"❌ خطای ناشناسسازی:\n{anon_result['error']}" logger.error(error_msg) return error_msg, "", "", "" anonymized_text = anon_result["anonymized_text"] usage_info = anon_result.get("usage", {}) logger.info(f"✅ متن ناشناسشد: {len(anonymized_text)} کاراکتر") # مرحله 2: ارسال به ChatGPT logger.info("مرحله 2️⃣: ارسال به ChatGPT...") gpt_result = anonymizer.send_to_chatgpt(anonymized_text) if not gpt_result["success"]: gpt_response = f"❌ خطا: {gpt_result['error']}" final_output = "" logger.warning(f"⚠️ ChatGPT جواب ندادن: {gpt_result['error']}") else: gpt_response = gpt_result["response"] logger.info(f"✅ پاسخ ChatGPT دریافت شد: {len(gpt_response)} کاراکتر") # مرحله 3: بازگردانی logger.info("مرحله 3️⃣: بازگردانی متن...") final_output = anonymizer.deanonymize_response(gpt_response, anonymizer.mapping_table) logger.info(f"✅ متن بازگردانده شد: {len(final_output)} کاراکتر") # ایجاد آمار stats_md = f""" ## 📊 آمار پردازش ### ⏱️ زمان: - **تاریخ/زمان**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} ### 🧠 Cerebras (Llama 3.3-70b): - **Token های ورودی**: {usage_info.get('prompt_tokens', 'نامشخص')} - **Token های خروجی**: {usage_info.get('completion_tokens', 'نامشخص')} - **کل Token ها**: {usage_info.get('total_tokens', 'نامشخص')} ### 📝 متنها: - **طول متن اصلی**: {len(input_text):,} کاراکتر - **طول متن ناشناسشده**: {len(anonymized_text):,} کاراکتر - **طول پاسخ ChatGPT**: {len(gpt_response):,} کاراکتر - **طول نتیجه نهایی**: {len(final_output):,} کاراکتر ### 🔐 اطلاعات حساس: - **تعداد کدهای ناشناس**: {len(anonymizer.mapping_table)} ### ✅ وضعیت: - **ناشناسسازی**: ✅ موفق - **ChatGPT**: ✅ موفق - **بازگردانی**: ✅ موفق --- **نکات:** - تمام اطلاعات حساس محفوظ است - ChatGPT فقط متن ناشناسشده را میبیند - نتیجه نهایی آماده برای استفاده است """ logger.info("=" * 60) logger.info("✅ پردازش کامل شد") logger.info("=" * 60) return stats_md, anonymized_text, gpt_response, final_output except Exception as e: error_msg = f"❌ خطا در پردازش: {str(e)}" logger.error(error_msg) return error_msg, "", "", "" def copy_text(text_to_copy: str) -> Tuple[Any, str]: """کپی متن""" if not text_to_copy or not text_to_copy.strip(): return gr.update(visible=False), "⚠️ متنی برای کپی وجود ندارد" logger.info("📋 متن برای کپی آماده شد") return gr.update(value=text_to_copy, visible=True), "✅ متن برای کپی آماده است" def clear_all() -> Tuple[str, str, str, str, str, str, str, Any]: """پاک کردن همه فیلدها""" anonymizer.mapping_table = {} logger.info("🗑️ تمام فیلدها پاک شدند") return "", "", "", "", "", "", "", gr.update(visible=False) # ایجاد رابط کاربری with gr.Blocks( title="🔐 سیستم پیشرفته ناشناسسازی مالی/خبری", theme=gr.themes.Soft() ) as interface: # عنوان gr.HTML("""
نسخه 1.0.0 | هفته اول اکتبر 2025