KashefTech's picture
Rename app.py to app3.py
4e18c7d verified
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
)