#!/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("""

🔐 سیستم پیشرفته ناشناس‌سازی مالی/خبری

Cerebras Llama 3.3-70b + OpenAI ChatGPT

نسخه 1.0.0 | هفته اول اکتبر 2025

""") with gr.Row(): # ستون 1: ورودی و تنظیمات with gr.Column(scale=1, min_width=300): gr.HTML("

📥 ورودی و تنظیمات

") api_key_input = gr.Textbox( label="🔑 کلید API Cerebras", type="password", placeholder="اختیاری - از CEREBRAS_API_KEY استفاده می‌شود" ) openai_key_input = gr.Textbox( label="🔑 کلید API OpenAI", type="password", placeholder="اختیاری - از OPENAI_API_KEY استفاده می‌شود" ) input_text = gr.Textbox( lines=12, placeholder="متن خود را اینجا وارد کنید...\n\nمثال:\nشرکت پتروشیمی بوعلی سینا با مدیریت محمد علی فرزین...", label="📝 متن ورودی", rtl=True ) process_btn = gr.Button("🚀 پردازش و تحلیل", variant="primary", size="lg") with gr.Row(): copy_btn = gr.Button("📋 کپی نتیجه نهایی", scale=1) clear_btn = gr.Button("🗑️ پاک کردن همه", variant="stop", scale=1) copy_output = gr.Textbox(visible=False, interactive=True, label="متن برای کپی") # ستون 2: متن ناشناس‌شده with gr.Column(scale=1, min_width=300): gr.HTML("

🎭 متن ناشناس‌شده

") anonymized_output = gr.Textbox( lines=25, placeholder="متن ناشناس‌شده اینجا نمایش داده می‌شود...", label="", interactive=False, rtl=True ) # ستون 3: ChatGPT و نتایج with gr.Column(scale=1, min_width=300): gr.HTML("

🤖 پاسخ ChatGPT و نتیجه

") with gr.Tabs(): with gr.TabItem("📤 پاسخ خام ChatGPT", id=1): gpt_output = gr.Textbox( lines=12, placeholder="پاسخ خام ChatGPT اینجا نمایش داده می‌شود...", label="", interactive=False, rtl=True ) with gr.TabItem("✅ نتیجه نهایی", id=2): final_output = gr.Textbox( lines=12, placeholder="نتیجه نهایی با اطلاعات اصلی اینجا نمایش داده می‌شود...", label="", interactive=False, rtl=True ) with gr.TabItem("📊 آمار", id=3): statistics_output = gr.Markdown( value="### 📊 آمار پردازش\n\nپس از پردازش، آمارها اینجا نمایش داده می‌شوند." ) # اتصالات رویدادها process_btn.click( fn=process_text, inputs=[input_text, api_key_input, openai_key_input], outputs=[statistics_output, anonymized_output, gpt_output, final_output] ) copy_btn.click( fn=copy_text, inputs=[final_output], outputs=[copy_output, statistics_output] ) clear_btn.click( fn=clear_all, outputs=[input_text, anonymized_output, gpt_output, final_output, statistics_output, api_key_input, openai_key_input, copy_output] ) # مثال‌های استفاده gr.Examples( examples=[ "مجمع عمومی عادی سالیانه شرکت پتروشیمی بوعلی سینا برگزار شد. شرکت وانیا نیک تدبیر را به‌عنوان بازرس قانونی و حسابرس انتخاب کردند.", "تحلیل صورت‌های مالی شرکت پالایش نفت اصفهان در سال 1403 این احتمال را مطرح می‌کند که درآمد به 2500 ریال برسد.", "شرکت فولاد مبارکه اصفهان با همکاری شرکت ملی نفت ایران، قرارداد توسعه میدان گازی مدار را امضا کرد و سرمایه خود را افزایش داد." ], inputs=input_text, label="📚 مثال‌های آزمایشی" ) # راهنمای استفاده with gr.Accordion("📖 راهنمای کامل استفاده", open=False): gr.Markdown(""" ## 🎯 نحوه استفاده سیستم ### مراحل کار: 1. **وارد کردن متن**: متن خود را در کادر "متن ورودی" وارد کنید 2. **تنظیم کلیدهای API** (اختیاری): اگر نخواهید از محیط تنظیم شود 3. **کلیک بر "پردازش و تحلیل"**: سیستم متن را پردازش می‌کند 4. **مشاهده نتایج**: - **متن ناشناس‌شده**: متنی که اطلاعات حساس جایگزین شده‌اند - **پاسخ ChatGPT**: پاسخ خام ChatGPT به متن ناشناس‌شده - **نتیجه نهایی**: پاسخ ChatGPT با اطلاعات اصلی بازگردانده شده - **آمار**: تفاصیل فنی پردازش ### ویژگی‌های سیستم: ✅ **ناشناس‌سازی قوی:** - شناسایی اسامی شرکت‌ها - شناسایی نام‌های اشخاص - شناسایی مبالغ مالی - شناسایی درصدها و نسبت‌ها ✅ **پردازش هوشمند با ChatGPT:** - تحلیل متن ناشناس‌شده - پاسخ به سوالات مالی/خبری - حفاظت کامل از اطلاعات حساس ✅ **بازگردانی خودکار:** - جایگزینی کدهای ناشناس به اطلاعات اصلی - حفظ ساختار و معنای متن ### 🔐 نکات امنیتی: - اطلاعات حساس ابتدا ناشناس‌سازی می‌شود - ChatGPT تنها متن ناشناس‌شده را می‌بیند - بازگردانی فقط در پایان انجام می‌شود - تمام داده‌ها در محیط محلی باقی می‌مانند """) return interface def main(): """تابع اصلی برنامه""" print("\n" + "=" * 70) print("🚀 سیستم پیشرفته ناشناس‌سازی مالی/خبری فارسی") print("=" * 70) print("📊 Cerebras Llama 3.3-70b + OpenAI ChatGPT Integration") print("\n✨ ویژگی‌ها:") print(" • ناشناس‌سازی دقیق متون مالی/خبری") print(" • تحلیل هوشمند با ChatGPT") print(" • بازگردانی خودکار نتایج") print(" • رابط کاربری بصری و منطقی") print(" • حفاظت کامل از اطلاعات حساس") print("\n🔐 اطلاعات حساس محفوظ:") print(" • نام‌های اشخاص و شرکت‌ها") print(" • مبالغ مالی و حساب‌های بانکی") print(" • شماره‌های شناسایی و کد‌های ملی") print(" • تمام اطلاعات محرمانه") print("\n⚙️ تنظیمات:") print(" • مدل: Llama 3.3-70b (Cerebras)") print(" • ChatGPT: gpt-4o-mini (OpenAI)") print(" • Max Tokens: 2000") print(" • Temperature: 0.1 (دقت بالا)") print("\n📱 رابط کاربری:") print(" • آدرس: http://localhost:7860") print(" • 3 ستون برای سهولت مشاهده") print(" • Tabs برای نتایج مختلف") print("=" * 70 + "\n") interface = create_interface() interface.launch( server_name="0.0.0.0", server_port=7860, share=False, show_error=True ) if __name__ == "__main__": main()