#!/usr/bin/env python3
# -*- coding: utf-8 -*-
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 difflib import SequenceMatcher
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
@dataclass
class CerebrasConfig:
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:
def __init__(self, api_key: str = None, openai_api_key: str = None):
if api_key is None:
api_key = os.getenv("CEREBRAS_API_KEY")
if not api_key:
raise ValueError("❌ کلید API Cerebras یافت نشد")
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_system_prompt()
self.mapping_table = {}
logger.info("✅ سیستم آماده شد")
def _create_system_prompt(self) -> str:
return """شما یک «ناشناسساز متون» هستید. وظیفهتان جایگزینی اسامی خاص و مقادیر عددی با شناسههای استاندارد است.
قوانین:
- شرکتها: company-01, company-02, ...
- اشخاص: person-01, person-02, ...
- اعداد/مقادیر: amount-01, amount-02, ...
- درصدها: percent-01, percent-02, ...
مثال:
- "ایرانخودرو" → "company-01"
- "۴.۵۸" → "percent-01"
- "۳۷ میلیارد" → "amount-01"
فقط متن ناشناسشده را برگردانید. هیچ توضیح اضافی نیاید."""
def anonymize_text(self, text: str) -> Dict[str, Any]:
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:
return {"success": False, "error": f"خطای API: {response.status_code}"}
result = response.json()
anonymized_text = result["choices"][0]["message"]["content"]
usage = result.get("usage", {})
self._extract_mapping_advanced(text, anonymized_text)
logger.info(f"✅ ناشناسسازی موفق ({len(self.mapping_table)} موجودیت)")
return {
"success": True,
"anonymized_text": anonymized_text,
"usage": usage
}
except Exception as e:
logger.error(f"❌ خطا: {str(e)}")
return {"success": False, "error": str(e)}
def _extract_mapping_advanced(self, original: str, anonymized: str):
"""استخراج mapping دقیق با مقایسه متنها"""
self.mapping_table = {}
try:
# تقسیم به جملات
orig_lines = re.split(r'([.!?])', original)
anon_lines = re.split(r'([.!?])', anonymized)
for orig_line, anon_line in zip(orig_lines, anon_lines):
if not orig_line.strip():
continue
# تقسیم به کلمات
orig_words = orig_line.split()
anon_words = anon_line.split()
# یافتن تطابقات
i = 0
j = 0
while i < len(orig_words) and j < len(anon_words):
orig_word = orig_words[i]
anon_word = anon_words[j]
# اگر کدی است
if re.match(r'(company|person|amount|percent)-\d+', anon_word):
# سعی کن تا 3 کلمه قبل را پیدا کن
found = False
for phrase_len in range(min(4, len(orig_words) - i), 0, -1):
phrase = ' '.join(orig_words[i:i+phrase_len])
# بررسی اینکه آیا این عبارت منطقی است
if self._is_valid_entity(phrase, anon_word):
if phrase not in self.mapping_table:
self.mapping_table[phrase] = anon_word
logger.info(f" {phrase} → {anon_word}")
i += phrase_len
found = True
break
if not found:
i += 1
j += 1
else:
i += 1
j += 1
except Exception as e:
logger.warning(f"⚠️ خطا در mapping: {str(e)}")
def _is_valid_entity(self, phrase: str, code: str) -> bool:
"""بررسی صحت موجودیت"""
phrase_lower = phrase.lower()
if code.startswith('company'):
return any(word in phrase_lower for word in ['شرکت', 'بانک', 'سازمان', 'وزارت', 'گروه'])
elif code.startswith('person'):
return len(phrase) > 2
elif code.startswith('amount'):
return bool(re.search(r'[\d]+', phrase))
elif code.startswith('percent'):
return 'درصد' in phrase_lower or '%' in phrase
return False
def send_to_chatgpt(self, anonymized_text: str) -> Dict[str, Any]:
try:
if not anonymized_text or not anonymized_text.strip():
return {"success": False, "error": "متن ناشناسشده خالی است"}
if not self.openai_api_key:
return {"success": False, "error": "کلید OpenAI تنظیم نشده"}
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()
return {"success": True, "response": result['choices'][0]['message']['content']}
else:
error_data = response.json() if response.content else {}
error_message = error_data.get('error', {}).get('message', response.text)
return {"success": False, "error": error_message}
except Exception as e:
logger.error(f"❌ خطا: {str(e)}")
return {"success": False, "error": str(e)}
def _create_reverse_mapping(self, mapping_table: Dict[str, str]) -> Dict[str, str]:
"""ایجاد reverse mapping: code -> original"""
reverse_map = {}
for original, code in mapping_table.items():
reverse_map[code] = original
return reverse_map
def deanonymize_response(self, gpt_response: str, mapping_table: Dict[str, str]) -> str:
"""بازگردانی تمام موجودیتها (placeholderها و کدهای ساده را پشتیبانی میکند)"""
try:
if not mapping_table:
return gpt_response
result = gpt_response
# ایجاد reverse mapping برای تبدیل code به original
reverse_map = self._create_reverse_mapping(mapping_table)
# مرتبسازی بر اساس طول (طولانیترین اول)
sorted_reverse = sorted(reverse_map.items(), key=lambda x: len(x[0]), reverse=True)
logger.info(f"🔄 شروع بازگردانی {len(sorted_reverse)} موجودیت")
for code, original in sorted_reverse:
# الگوهای مختلف برای جستجو:
# 1. [code] (داخل براکت)
pattern_bracket = f"[{code}]"
if pattern_bracket in result:
result = result.replace(pattern_bracket, original)
logger.info(f" [{code}] → {original} ✓")
# 2. code بدون براکت
if code in result:
result = result.replace(code, original)
logger.info(f" {code} → {original} ✓")
# جایگزینی نسخههای دارای درصد و هزار جدا شده
result = self._replace_special_patterns(result, reverse_map)
logger.info(f"✅ بازگردانی موفق ({len(mapping_table)} موجودیت)")
return result
except Exception as e:
logger.error(f"❌ خطا در بازگردانی: {str(e)}")
return gpt_response
def _replace_special_patterns(self, text: str, reverse_map: Dict[str, str]) -> str:
"""جایگزینی الگوهای خاص (درصد-XX، amount-XX و غیره) - شامل placeholderهای بدون براکت"""
try:
result = text
# الگوی 1: Placeholderهای با براکت [code]
pattern_bracket = r'\[((?:company|person|amount|percent|درصد|شرکت|مبلغ)-\d+)\]'
def replacer_bracket(match):
code = match.group(1)
# تبدیل الگوهای فارسی به انگلیسی
code_normalized = code.replace('درصد', 'percent').replace('شرکت', 'company').replace('مبلغ', 'amount')
return reverse_map.get(code_normalized, reverse_map.get(code, match.group(0)))
result = re.sub(pattern_bracket, replacer_bracket, result)
logger.info(f" الگوی 1 (با براکت): جایگزین شد")
# الگوی 2: Placeholderهای داخل پرانتز (code)
pattern_paren = r'\(((?:company|person|amount|percent|درصد|شرکت|مبلغ)-\d+)\)'
def replacer_paren(match):
code = match.group(1)
code_normalized = code.replace('درصد', 'percent').replace('شرکت', 'company').replace('مبلغ', 'amount')
return f"({reverse_map.get(code_normalized, reverse_map.get(code, code))})"
result = re.sub(pattern_paren, replacer_paren, result)
logger.info(f" الگوی 2 (پرانتز): جایگزین شد")
# الگوی 3: Placeholderهای بدون براکت (بسیار مهم!)
# جایگزینی تمام code-XX که با word boundary احاطه شدهاند
pattern_simple = r'\b((?:company|person|amount|percent|درصد|شرکت|مبلغ)-\d+)\b'
def replacer_simple(match):
code = match.group(1)
code_normalized = code.replace('درصد', 'percent').replace('شرکت', 'company').replace('مبلغ', 'amount')
replacement = reverse_map.get(code_normalized, reverse_map.get(code, None))
if replacement:
logger.info(f" {code} → {replacement}")
return replacement
return match.group(0)
result = re.sub(pattern_simple, replacer_simple, result)
logger.info(f" الگوی 3 (بدون براکت): جایگزین شد")
return result
except Exception as e:
logger.warning(f"⚠️ خطا در جایگزینی الگوهای خاص: {str(e)}")
return text
def create_interface():
try:
anonymizer = AdvancedCerebrasAnonymizer()
except ValueError as e:
return gr.Interface(fn=lambda x: str(e), inputs="textbox", outputs="textbox", title="❌ خطا")
def process_text(input_text: str) -> Tuple[str, str, str, str]:
logger.info("=" * 60)
logger.info("شروع پردازش")
logger.info("=" * 60)
if not input_text.strip():
return "❌ متن ورودی خالی است", "", "", ""
try:
# مرحله 1: ناشناسسازی
logger.info("1️⃣ ناشناسسازی...")
anon_result = anonymizer.anonymize_text(input_text)
if not anon_result["success"]:
return f"❌ {anon_result['error']}", "", "", ""
anonymized_text = anon_result["anonymized_text"]
usage_info = anon_result.get("usage", {})
# مرحله 2: ارسال به ChatGPT
logger.info("2️⃣ ارسال به ChatGPT...")
gpt_result = anonymizer.send_to_chatgpt(anonymized_text)
# تعریف اولیه
gpt_response = ""
gpt_response_deanon = ""
if not gpt_result["success"]:
gpt_response = f"❌ {gpt_result['error']}"
gpt_response_deanon = ""
else:
# پاسخ ChatGPT (ناشناس)
gpt_response = gpt_result["response"]
# مرحله 3: بازگردانی
logger.info("3️⃣ بازگردانی...")
gpt_response_deanon = anonymizer.deanonymize_response(gpt_response, anonymizer.mapping_table)
# آمار
stats = f"""Token: {usage_info.get('total_tokens', '?')} | Mapping: {len(anonymizer.mapping_table)}"""
logger.info("=" * 60)
logger.info("✅ پردازش کامل")
logger.info("=" * 60)
# ✅ ترتیب صحیح:
# 1. آمار
# 2. متن ناشناسشده (ورودی)
# 3. پاسخ ChatGPT (ناشناس) ← باکس "🤖 نتایج ChatGPT"
# 4. نتیجه نهایی (بازگردانی شده) ← باکس "✅ نتیجه نهایی"
return stats, anonymized_text, gpt_response, gpt_response_deanon
except Exception as e:
return f"❌ خطا: {str(e)}", "", "", ""
def copy_text(text: str):
if not text or not text.strip():
return gr.update(visible=False), "⚠️ متنی وجود ندارد"
return gr.update(value=text, visible=True), "✅ آماده برای کپی"
def clear_all():
anonymizer.mapping_table = {}
return "", "", "", "", gr.update(visible=False)
with gr.Blocks(title="سیستم ناشناسسازی", theme=gr.themes.Soft()) as interface:
gr.HTML("🔐 سیستم ناشناسسازی
")
with gr.Row():
# ستون 1
with gr.Column(scale=1):
gr.HTML("📥 ورودی
")
input_text = gr.Textbox(lines=20, placeholder="متن را وارد کنید...", 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)
# ستون 2
with gr.Column(scale=1):
gr.HTML("🎭 متن ناشناسشده
")
anonymized_output = gr.Textbox(lines=20, placeholder="", label="", interactive=False, rtl=True)
# ستون 3
with gr.Column(scale=1):
gr.HTML("🤖 نتایج ChatGPT
")
gpt_output = gr.Textbox(lines=10, placeholder="", label="📤 پاسخ", interactive=False, rtl=True)
final_output = gr.Textbox(lines=10, placeholder="", label="✅ نتیجه نهایی", interactive=False, rtl=True)
statistics_output = gr.Textbox(lines=1, label="📊 آمار", interactive=False)
process_btn.click(fn=process_text, inputs=[input_text], 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, copy_output])
return interface
def main():
print("\n🔐 سیستم ناشناسسازی - نسخه 1.0.0\n")
interface = create_interface()
interface.launch(server_name="0.0.0.0", server_port=7860, share=False, show_error=True)
if __name__ == "__main__":
main()