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