leilaghomashchi's picture
Rename persian_anonymizer_advanced.py to app.py
9086f5c verified
raw
history blame
30.2 kB
import requests
import json
import gradio as gr
from typing import Dict, Any
import os
from dataclasses import dataclass
import re
@dataclass
class GroqConfig:
"""تنظیمات Groq API"""
api_key: str
base_url: str = "https://api.groq.com/openai/v1"
model: str = "llama-3.1-8b-instant"
max_tokens: int = 2000
temperature: float = 0.1
class AdvancedGroqAnonymizer:
"""سیستم پیشرفته ناشناس‌سازی متون مالی/خبری فارسی"""
def __init__(self, api_key: str = None):
if api_key is None:
api_key = os.getenv("GROQ_API_KEY")
if not api_key:
raise ValueError("کلید API یافت نشد")
self.config = GroqConfig(api_key=api_key)
self.system_prompt = self._create_advanced_system_prompt()
def _create_advanced_system_prompt(self) -> str:
"""ایجاد دستورالعمل سیستمی پیشرفته برای Groq"""
return """شما یک «ناشناس‌ساز متون مالی/خبری فارسی» هستید. وظیفه‌تان جایگزینی اسامی خاص و مقادیر عددی با شناسه‌های بی‌معناست، بی‌آنکه ساختار جمله و معنای تحلیلی متن مخدوش شود.
## طرح برچسب‌ها (Tagging Scheme)
* شرکت/نهاد/برند/سازمان/سامانه → `company-XX`
* شخص حقیقی (نام و نام‌خانوادگی، القابِ مختص فرد) → `person-XX`
* گروه/هلدینگ/کنسرسیوم → `group-XX`
* هر عدد پولی/تعدادی/حجمی/زمانی/شناسه ثبت/شماره/… → `amount-XX` (واحد دقیق بلافاصله پس از برچسب حفظ شود: «amount-03 میلیارد تومان»)
* درصدها (٪) و بازه‌های درصد → `percent-XX` (مثال بازه: `percent-05–percent-07`)
* شاخص‌ها/علائم مالی (EPS, P/E, ARPU, NPL و…) باقی بمانند؛ فقط مقدارشان ناشناس شود (`EPS amount-01 ریال`, `P/E amount-02`)
## قواعد اندیس‌گذاری
* برای هر نوع برچسب، از `-01` شروع و افزایشی تخصیص دهید (شمارنده‌های هر نوع از هم مستقل‌اند).
* پایداری «درون همان قطعه»: یک موجودیت در یک قطعه همیشه همان شناسه را نگه دارد.
* عدم نیاز به پایداری بین قطعات/پاراگراف‌های مستقل.
## چه چیزهایی حفظ می‌شود؟
* تاریخ‌ها/ساعت‌ها (مثل 1404/04/29، 31 اردیبهشت 1404) دست‌نخورده.
* عناوین شغلی/نقش‌ها (مدیرعامل، بازرس قانونی، …).
* واحدها و اصطلاحات فنی (همت، میلیارد ریال، تُن، TEU، EPS، P/E، ARPU، NPL، …).
* نام مکان‌ها/آدرس‌ها/محل برگزاری رویدادها اصولاً حفظ می‌شوند (مگر صریحاً سیاست دیگری داده شود).
* ساختار جمله، روابط علّی/مقایسه‌ای، نتیجه‌گیری‌ها و لحن.
## قواعد جایگزینی
1. هر اسم خاص سازمانی/برندی/رسانه‌ای → `company-XX` (بانک مرکزی، سامانه کدال، بیمه مرکزی، … نیز سازمان‌اند.)
2. هر نام شخص → `person-XX` (عنوان شغلی حفظ شود).
3. هر عدد مستقل یا همراه واحد → `amount-XX` + واحد.
4. هر درصد → `percent-XX`؛ در بازه‌ها با خط تیره بدون فاصله.
5. اعداد چسبیده به واژه («5هزار») → بخش عددی ناشناس شود: `amount-01 هزار`.
6. مدل/سری محصول (مثل G10) معمولاً حفظ می‌شود؛ اگر سیاست پوشش کامل ارقام سری لازم بود: «Gamount-XX».
7. اگر درباره‌ی نوع برچسب مردد بودید، اولویت با `company-XX` (برای موجودیت‌های حقوقی) است.
## کنترل کیفیت (برای هر قطعه)
* هر موجودیت در همان قطعه شناسه ثابت دارد.
* نوع برچسب با ماهیت موجودیت سازگار است (شخص≠شرکت).
* همه‌ی اعداد و درصدها پوشانده و واحدها حفظ شده‌اند.
* تاریخ/ساعت حفظ شده است.
* اندیس‌ها از 01 شروع و پیوسته‌اند (به‌تفکیک نوع).
* هیچ توضیح یا کروشه اضافه به متن ناشناس‌شده افزوده نشده است.
## قالب خروجی
برای هر ورودی، **صرفاً متن ناشناس‌شده** را برگردان. هیچ توضیح، سربرگ، یا نشانه‌گذاری اضافه ننویس. زبان و علائم نگارشی اصلی را حفظ کن.
## مثال‌های کوتاه
**مثال 1:**
ورودی: «بانک پاسارگاد با شناسایی سود خالص 155 هزار میلیارد ریالی…»
خروجی: «company-01 با شناسایی سود خالص amount-01 هزار میلیارد ریالی…»
**مثال 2:**
ورودی: «مهدی اخوان بهابادی، مدیرعامل همراه اول، گفت سود خالص 7101 میلیارد تومان شد و EPS به 936 ریال رسید.»
خروجی: «person-01، مدیرعامل company-01، گفت سود خالص amount-01 میلیارد تومان شد و EPS به amount-02 ریال رسید.»
**مثال 3:**
ورودی: «هزینه لجستیکی بوعلی حدود 100 میلیون دلار بود؛ 40٪ خوراک از خط لوله و 60٪ معادل 1 تا 1.5 میلیون تن.»
خروجی: «هزینه لجستیکی company-01 حدود amount-01 میلیون دلار بود؛ percent-01 خوراک از خط لوله و percent-02 معادل amount-02 تا amount-03 میلیون تن.»
## نکات لبه‌ای
* بازه‌های عددی/درصدی: `amount-01–amount-02` / `percent-01–percent-02`.
* اعداد سریالی سند/ثبت/شماره ملی/شماره ثبت: عدد → `amount-XX`، عنوان/واژه‌ها حفظ.
* اگر یک نام هم‌زمان برند محصول و شرکت است، در متن مالی/حاکمیتی آن را شرکت فرض کنید (`company-XX`).
فقط متن ناشناس‌شده را برگردان، بدون هیچ توضیح اضافی."""
def _make_api_request(self, text: str) -> Dict[str, Any]:
"""ارسال درخواست به Groq API"""
headers = {
"Authorization": f"Bearer {self.config.api_key}",
"Content-Type": "application/json"
}
payload = {
"messages": [
{
"role": "system",
"content": self.system_prompt
},
{
"role": "user",
"content": text
}
],
"model": self.config.model,
"temperature": self.config.temperature,
"max_tokens": self.config.max_tokens
}
try:
response = requests.post(
f"{self.config.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=45
)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
raise Exception(f"خطا در ارتباط با Groq API: {str(e)}")
def anonymize_text(self, text: str) -> Dict[str, Any]:
"""ناشناس‌سازی متن با استفاده از Groq"""
if not text.strip():
return {
"success": False,
"error": "متن ورودی خالی است"
}
try:
response = self._make_api_request(text)
if "choices" not in response or not response["choices"]:
return {
"success": False,
"error": "پاسخ نامعتبر از API"
}
content = response["choices"][0]["message"]["content"]
# پاک کردن markdown اگر وجود دارد
content = self._clean_markdown(content)
# حذف خطوط اضافی و فضاهای خالی
content = content.strip()
# تحلیل نتایج
analysis = self._analyze_anonymized_text(content)
return {
"success": True,
"anonymized_text": content,
"entities": analysis["entities"],
"statistics": analysis["statistics"],
"detailed_analysis": analysis["detailed_analysis"],
"usage": response.get("usage", {}),
"quality_check": self._quality_check(content)
}
except Exception as e:
return {
"success": False,
"error": f"خطا در پردازش: {str(e)}"
}
def _clean_markdown(self, content: str) -> str:
"""پاک کردن markdown از پاسخ"""
if "```" in content:
lines = content.split('\n')
clean_lines = []
skip = False
for line in lines:
if line.strip().startswith('```'):
skip = not skip
continue
if not skip:
clean_lines.append(line)
content = '\n'.join(clean_lines)
return content
def _analyze_anonymized_text(self, text: str) -> Dict[str, Any]:
"""تحلیل متن ناشناس‌سازی شده"""
import re
# شمارش موجودیت‌ها
companies = re.findall(r'company-(\d+)', text)
persons = re.findall(r'person-(\d+)', text)
amounts = re.findall(r'amount-(\d+)', text)
percents = re.findall(r'percent-(\d+)', text)
groups = re.findall(r'group-(\d+)', text)
# آمار کلی
statistics = {
"company": len(set(companies)),
"person": len(set(persons)),
"amount": len(set(amounts)),
"percent": len(set(percents)),
"group": len(set(groups)),
"total_replacements": len(companies) + len(persons) + len(amounts) + len(percents) + len(groups)
}
# جزئیات موجودیت‌ها
entities = {
"companies": sorted(list(set(companies)), key=lambda x: int(x)),
"persons": sorted(list(set(persons)), key=lambda x: int(x)),
"amounts": sorted(list(set(amounts)), key=lambda x: int(x)),
"percents": sorted(list(set(percents)), key=lambda x: int(x)),
"groups": sorted(list(set(groups)), key=lambda x: int(x))
}
# تحلیل دقیق‌تر
detailed_analysis = {
"preserved_dates": len(re.findall(r'\d{4}/\d{1,2}/\d{1,2}|\d{1,2}\s+\w+\s+\d{4}', text)),
"preserved_times": len(re.findall(r'\d{1,2}:\d{2}', text)),
"financial_indicators": len(re.findall(r'\b(EPS|P/E|ARPU|NPL|ROE|ROA)\b', text)),
"units_preserved": len(re.findall(r'(میلیارد|میلیون|هزار|تومان|ریال|درهم|دلار|یورو|تن|کیلوگرم)', text))
}
return {
"statistics": statistics,
"entities": entities,
"detailed_analysis": detailed_analysis
}
def _quality_check(self, text: str) -> Dict[str, Any]:
"""بررسی کیفیت ناشناس‌سازی"""
import re
# بررسی شروع اندیس‌ها از 01
company_indices = [int(x) for x in re.findall(r'company-(\d+)', text)]
person_indices = [int(x) for x in re.findall(r'person-(\d+)', text)]
amount_indices = [int(x) for x in re.findall(r'amount-(\d+)', text)]
percent_indices = [int(x) for x in re.findall(r'percent-(\d+)', text)]
issues = []
# بررسی شروع از 01
for entity_type, indices in [
("company", company_indices),
("person", person_indices),
("amount", amount_indices),
("percent", percent_indices)
]:
if indices and min(indices) != 1:
issues.append(f"اندیس {entity_type} از 01 شروع نشده")
# بررسی پیوستگی اندیس‌ها
for entity_type, indices in [
("company", company_indices),
("person", person_indices),
("amount", amount_indices),
("percent", percent_indices)
]:
if indices:
unique_indices = sorted(list(set(indices)))
expected = list(range(1, len(unique_indices) + 1))
if unique_indices != expected:
issues.append(f"اندیس‌های {entity_type} پیوسته نیستند")
return {
"passed": len(issues) == 0,
"issues": issues,
"total_checks": 8,
"passed_checks": 8 - len(issues)
}
def create_advanced_interface():
"""ایجاد رابط کاربری پیشرفته"""
# بررسی وجود کلید API
api_key_available = bool(os.getenv("GROQ_API_KEY"))
# CSS سفارشی پیشرفته
custom_css = """
.gradio-container {
font-family: 'Tahoma', 'Arial', sans-serif !important;
direction: rtl;
max-width: 1400px;
margin: 0 auto;
}
.result-box {
background-color: #f8f9fa;
border: 2px solid #e9ecef;
border-radius: 12px;
padding: 20px;
margin: 10px 0;
}
.warning-box {
background-color: #fff3cd;
border: 2px solid #ffeaa7;
border-radius: 12px;
padding: 15px;
color: #856404;
margin: 10px 0;
}
.success-box {
background-color: #d4edda;
border: 2px solid #c3e6cb;
border-radius: 12px;
padding: 15px;
color: #155724;
margin: 10px 0;
}
.stats-grid {
display: grid;
grid-template-columns: repeat(auto-fit, minmax(200px, 1fr));
gap: 15px;
margin: 15px 0;
}
.stat-card {
background-color: #ffffff;
border: 1px solid #dee2e6;
border-radius: 8px;
padding: 15px;
text-align: center;
box-shadow: 0 2px 4px rgba(0,0,0,0.1);
}
.quality-badge {
display: inline-block;
padding: 5px 10px;
border-radius: 20px;
font-weight: bold;
margin: 5px;
}
.quality-pass {
background-color: #28a745;
color: white;
}
.quality-fail {
background-color: #dc3545;
color: white;
}
"""
with gr.Blocks(css=custom_css, title="ناشناس‌ساز پیشرفته متن فارسی", theme=gr.themes.Soft()) as interface:
# عنوان
gr.Markdown("""
# 🔒 سیستم پیشرفته ناشناس‌سازی متون مالی/خبری فارسی
### قدرت‌گرفته از Groq AI با الگوریتم‌های هوشمند تشخیص و جایگزینی موجودیت‌ها
""")
# نمایش وضعیت API
if api_key_available:
gr.Markdown("""
<div class="success-box">
✅ <strong>سیستم آماده است</strong> - کلید API تنظیم شده
</div>
""")
api_key_input = gr.Textbox(visible=False, value="")
else:
gr.Markdown("""
<div class="warning-box">
⚠️ <strong>کلید API تنظیم نشده</strong><br>
لطفاً کلید Groq API خود را در زیر وارد کنید
</div>
""")
api_key_input = gr.Textbox(
label="🔑 کلید Groq API",
placeholder="gsk_...",
type="password",
value="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"
)
with gr.Row():
with gr.Column(scale=1):
input_text = gr.Textbox(
label="📝 متن ورودی",
placeholder="متن مالی یا خبری خود را اینجا وارد کنید...",
lines=12,
max_lines=25
)
with gr.Row():
anonymize_btn = gr.Button(
"🔒 ناشناس‌سازی پیشرفته",
variant="primary",
size="lg"
)
clear_btn = gr.Button(
"🗑️ پاک کردن",
variant="secondary"
)
with gr.Column(scale=1):
output_text = gr.Textbox(
label="🎯 متن ناشناس‌سازی شده",
lines=12,
max_lines=25,
elem_classes=["result-box"]
)
# دکمه کپی
copy_btn = gr.Button(
"📋 کپی متن",
variant="secondary",
size="sm"
)
# متن برای کپی
copy_output = gr.Textbox(
label="📋 متن برای کپی (Ctrl+A و Ctrl+C)",
lines=3,
max_lines=10,
visible=False,
interactive=True
)
# نمایش آمار پیشرفته
with gr.Row():
with gr.Column():
statistics_output = gr.Markdown(label="📊 آمار کلی")
with gr.Column():
quality_output = gr.Markdown(label="✅ کنترل کیفیت")
with gr.Row():
with gr.Column():
entities_output = gr.Markdown(label="🏷️ موجودیت‌های شناسایی شده")
with gr.Column():
detailed_analysis_output = gr.Markdown(label="🔍 تحلیل دقیق")
usage_output = gr.Markdown(label="⚡ اطلاعات پردازش")
def process_advanced_text(text: str, api_key_manual: str = ""):
"""پردازش پیشرفته متن"""
# حل مشکل NoneType
if api_key_manual is None:
api_key_manual = ""
# تعیین کلید API
final_api_key = ""
if api_key_manual and api_key_manual.strip():
final_api_key = api_key_manual.strip()
elif os.getenv("GROQ_API_KEY"):
final_api_key = os.getenv("GROQ_API_KEY")
if not final_api_key:
return (
"",
"❌ کلید API وارد نشده است",
"",
"",
"",
""
)
if not text or not text.strip():
return (
"",
"❌ لطفاً متن ورودی را وارد کنید",
"",
"",
"",
""
)
try:
anonymizer = AdvancedGroqAnonymizer(api_key=final_api_key)
result = anonymizer.anonymize_text(text)
if not result["success"]:
return (
"",
f"❌ خطا: {result['error']}",
"",
"",
"",
""
)
# آمار کلی
stats = result.get("statistics", {})
stats_md = "📊 **آمار کلی:**\n\n"
stats_md += f"""
<div class="stats-grid">
<div class="stat-card">
<h3>🏢 شرکت‌ها</h3>
<h2>{stats.get('company', 0)}</h2>
</div>
<div class="stat-card">
<h3>👤 اشخاص</h3>
<h2>{stats.get('person', 0)}</h2>
</div>
<div class="stat-card">
<h3>💰 مبالغ</h3>
<h2>{stats.get('amount', 0)}</h2>
</div>
<div class="stat-card">
<h3>📊 درصدها</h3>
<h2>{stats.get('percent', 0)}</h2>
</div>
<div class="stat-card">
<h3>👥 گروه‌ها</h3>
<h2>{stats.get('group', 0)}</h2>
</div>
<div class="stat-card">
<h3>🔢 کل تغییرات</h3>
<h2>{stats.get('total_replacements', 0)}</h2>
</div>
</div>
"""
# کنترل کیفیت
quality = result.get("quality_check", {})
quality_md = "✅ **کنترل کیفیت:**\n\n"
if quality.get("passed", False):
quality_md += '<span class="quality-badge quality-pass">✅ تمام بررسی‌ها موفق</span>\n\n'
else:
quality_md += '<span class="quality-badge quality-fail">❌ مشکلاتی یافت شد</span>\n\n'
issues = quality.get("issues", [])
if issues:
quality_md += "**مشکلات:**\n"
for issue in issues:
quality_md += f"• {issue}\n"
quality_md += f"\n**نتیجه:** {quality.get('passed_checks', 0)}/{quality.get('total_checks', 0)} بررسی موفق"
# موجودیت‌های شناسایی شده
entities = result.get("entities", {})
entities_md = "🏷️ **موجودیت‌های شناسایی شده:**\n\n"
if entities.get("companies"):
entities_md += f"🏢 **شرکت‌ها:** company-{', company-'.join(entities['companies'])}\n\n"
if entities.get("persons"):
entities_md += f"👤 **اشخاص:** person-{', person-'.join(entities['persons'])}\n\n"
if entities.get("amounts"):
entities_md += f"💰 **مبالغ:** amount-{', amount-'.join(entities['amounts'])}\n\n"
if entities.get("percents"):
entities_md += f"📊 **درصدها:** percent-{', percent-'.join(entities['percents'])}\n\n"
if entities.get("groups"):
entities_md += f"👥 **گروه‌ها:** group-{', group-'.join(entities['groups'])}\n\n"
# تحلیل دقیق
detailed = result.get("detailed_analysis", {})
detailed_md = "🔍 **تحلیل دقیق:**\n\n"
detailed_md += f"📅 **تاریخ‌های حفظ شده:** {detailed.get('preserved_dates', 0)}\n"
detailed_md += f"🕐 **ساعت‌های حفظ شده:** {detailed.get('preserved_times', 0)}\n"
detailed_md += f"📈 **شاخص‌های مالی:** {detailed.get('financial_indicators', 0)}\n"
detailed_md += f"📏 **واحدهای حفظ شده:** {detailed.get('units_preserved', 0)}\n"
# اطلاعات پردازش
usage = result.get("usage", {})
usage_md = "⚡ **اطلاعات پردازش:**\n\n"
if usage:
usage_md += f"🤖 **مدل:** {anonymizer.config.model}\n"
usage_md += f"📥 **Token های ورودی:** {usage.get('prompt_tokens', 'نامشخص')}\n"
usage_md += f"📤 **Token های خروجی:** {usage.get('completion_tokens', 'نامشخص')}\n"
usage_md += f"📊 **کل Token ها:** {usage.get('total_tokens', 'نامشخص')}\n"
else:
usage_md += "✅ پردازش با موفقیت انجام شد"
return (
result["anonymized_text"],
stats_md,
quality_md,
entities_md,
detailed_md,
usage_md
)
except Exception as e:
return (
"",
f"❌ خطای غیرمنتظره: {str(e)}",
"",
"",
"",
""
)
def copy_text(text_to_copy):
"""تابع کپی متن"""
if not text_to_copy or not text_to_copy.strip():
return gr.Textbox(visible=False), "⚠️ متنی برای کپی وجود ندارد"
return gr.Textbox(value=text_to_copy, visible=True), "✅ متن در کادر زیر آماده کپی است"
def clear_all():
"""پاک کردن تمام فیلدها"""
return "", "", "", "", "", "", gr.Textbox(visible=False)
# اتصال رویدادها
anonymize_btn.click(
fn=process_advanced_text,
inputs=[input_text, api_key_input],
outputs=[output_text, statistics_output, quality_output, entities_output, detailed_analysis_output, usage_output]
)
copy_btn.click(
fn=copy_text,
inputs=[output_text],
outputs=[copy_output, statistics_output]
)
clear_btn.click(
fn=clear_all,
outputs=[input_text, output_text, statistics_output, quality_output, entities_output, detailed_analysis_output, usage_output, copy_output]
)
# مثال‌های پیشرفته
gr.Examples(
examples=[
["بانک پاسارگاد با شناسایی سود خالص 155 هزار میلیارد ریالی در ردۀ دوم سودآورترین بانک‌های کشور قرار گرفت."],
["مهدی اخوان بهابادی، مدیرعامل همراه اول، اعلام کرد درآمد عملیاتی شرکت با رشد 37 درصدی به 70 هزار و 677 میلیارد تومان رسیده است."],
["هزینه لجستیکی بوعلی حدود 100 میلیون دلار بود؛ 40٪ خوراک از خط لوله و 60٪ معادل 1 تا 1.5 میلیون تن در سال."],
["مجمع عمومی در تاریخ 1404/04/29 ساعت 10:00 در هتل بزرگ نخل کنگان برگزار شد و علی محمدی، رئیس هیئت مدیره، گزارش عملکرد ارائه داد."],
["گروه دتا با سود خالص 45.3 میلیارد تومان و EPS برابر 2850 ریال، رشد 15.7 درصدی نسبت به مدت مشابه سال قبل داشته است."]
],
inputs=input_text,
label="📚 مثال‌های پیشرفته آزمایشی"
)
# راهنمای کامل
with gr.Accordion("📖 راهنمای کامل استفاده", open=False):
gr.Markdown("""
## 🎯 ویژگی‌های سیستم پیشرفته:
### 🏷️ انواع برچسب‌ها:
- **company-XX:** شرکت‌ها، سازمان‌ها، برندها، نهادها
- **person-XX:** اشخاص حقیقی (نام و نام‌خانوادگی)
- **group-XX:** گروه‌ها، هلدینگ‌ها، کنسرسیوم‌ها
- **amount-XX:** تمام اعداد (پولی، تعدادی، حجمی، زمانی)
- **percent-XX:** درصدها و بازه‌های درصدی
### ✅ موارد حفظ شده:
- 📅 تاریخ‌ها و ساعت‌ها
- 🏢 عناوین شغلی و نقش‌ها
- 📏 واحدها (تومان، ریال، میلیارد، تن، ...)
- 📈 شاخص‌های مالی (EPS, P/E, ARPU, NPL)
- 🗺️ نام مکان‌ها و آدرس‌ها
- 📝 ساختار جمله و لحن
### 🔍 کنترل کیفیت:
- بررسی شروع اندیس‌ها از 01
- بررسی پیوستگی اندیس‌ها
- تضمین ثبات شناسه‌ها در یک متن
- حفظ واحدها و شاخص‌های مالی
### 💡 نکات مهم:
- هر نوع موجودیت شماره‌گذاری مستقل دارد
- در بازه‌های عددی: amount-01–amount-02
- برای درصدها: percent-01–percent-02
- اعداد چسبیده: "5هزار" → "amount-01 هزار"
""")
return interface
# اجرای برنامه
if __name__ == "__main__":
interface = create_advanced_interface()
interface.launch(
server_name="0.0.0.0",
server_port=7860,
share=True,
show_error=True
)