Data-anonymization / last-correct-progrram
leilaghomashchi's picture
Rename app.py to last-correct-progrram
23e5a8d verified
raw
history blame
30.7 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`)
## قواعد اندیس‌گذاری **CRITICAL**
* **MUST START FROM 01**: هر نوع برچسب از -01 شروع شود: company-01, person-01, amount-01, percent-01
* **MUST BE SEQUENTIAL**: اندیس‌ها باید پیوسته باشند: company-01, company-02, company-03 (نه company-01, company-13)
* **SEPARATE COUNTERS**: شمارنده‌های هر نوع از هم مستقل‌اند
* **CONSISTENCY**: یک موجودیت در متن همیشه همان شناسه را داشته باشد
* **NO GAPS**: هیچ خلاء در شماره‌گذاری نباشد
## چه چیزهایی حفظ می‌شود؟
* تاریخ‌ها/ساعت‌ها (مثل 1404/04/29، 30 آذر 1403) دست‌نخورده
* عناوین شغلی/نقش‌ها (مدیرعامل، بازرس قانونی)
* واحدها و اصطلاحات فنی (میلیارد تومان، درصد، زیان، درآمد، حمل، پردازش)
* شاخص‌های مالی (EPS, P/E) - فقط مقدار ناشناس شود
* کلمات عمومی فارسی و ساختار جمله
* نام مکان‌ها (در صورت عدم حساسیت)
## قواعد جایگزینی **STRICT**
1. شرکت آسان پادرو → company-01
2. سرزمین هوشمند پاد → company-02
3. بانک پاسارگاد → company-03
4. هر عدد (6، 30، 97، 60) → amount-01, amount-02, amount-03, amount-04 (به ترتیب ظهور)
5. هر درصد (97 درصد، 60 درصد) → percent-01, percent-02
6. کلمات انگلیسی اضافی یا غیرضروری نباید اضافه شود
7. کلمات فارسی مانند "مقایسه" نباید به "comparison" تبدیل شود
## مثال دقیق برای متن نمونه:
ورودی: «شرکت آسان پادرو منتهی به 30 آذر 1403 نشان می‌دهد این شرکت سال مالی جاری بیش از 6 میلیارد تومان زیان... سرزمین هوشمند پاد (زیرمجموعه بانک پاسارگاد)»
خروجی صحیح: «company-01 منتهی به 30 آذر 1403 نشان می‌دهد این شرکت سال مالی جاری بیش از amount-01 میلیارد تومان زیان... company-02 (زیرمجموعه company-03)»
## کنترل کیفیت نهایی
- company-01, company-02, company-03 (پیوسته و بدون گپ)
- amount-01, amount-02, amount-03, amount-04 (پیوسته و بدون گپ)
- percent-01, percent-02 (پیوسته و بدون گپ)
- هیچ کلمه انگلیسی اضافی نباشد
- ساختار و معنای فارسی حفظ شود
فقط متن ناشناس‌شده را برگردان، بدون هیچ توضیح اضافی."""
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._validate_anonymized_text(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 _validate_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)
validation_issues = []
# بررسی هر نوع موجودیت
for entity_type, indices in [
("company", companies),
("person", persons),
("amount", amounts),
("percent", percents),
("group", groups)
]:
if indices:
unique_indices = sorted(list(set([int(x) for x in indices])))
# بررسی شروع از 1
if unique_indices[0] != 1:
validation_issues.append(f"اندیس {entity_type} از 01 شروع نشده (شروع: {unique_indices[0]:02d})")
# بررسی پیوستگی
expected = list(range(1, len(unique_indices) + 1))
if unique_indices != expected:
validation_issues.append(f"اندیس‌های {entity_type} پیوسته نیستند: {[f'{x:02d}' for x in unique_indices]}")
# بررسی کلمات انگلیسی غیرضروری
english_words = re.findall(r'\b[a-zA-Z]+\b', text)
unwanted_english = [word for word in english_words if word.lower() not in ['eps', 'p/e', 'arpu', 'npl', 'roe', 'roa']]
if unwanted_english:
validation_issues.append(f"کلمات انگلیسی غیرضروری: {unwanted_english}")
return {
"is_valid": len(validation_issues) == 0,
"issues": validation_issues,
"entity_counts": {
"company": len(set(companies)),
"person": len(set(persons)),
"amount": len(set(amounts)),
"percent": len(set(percents)),
"group": len(set(groups))
}
}
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("is_valid", 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"
entity_counts = quality.get("entity_counts", {})
if entity_counts:
quality_md += f"\n**تعداد موجودیت‌های منحصربه‌فرد:**\n"
for entity_type, count in entity_counts.items():
if count > 0:
quality_md += f"• {entity_type}: {count}\n"
# موجودیت‌های شناسایی شده
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=[
["صورت‌های مالی حسابرسی شده شرکت آسان پادرو منتهی به 30 آذر 1403 نشان می‌دهد این شرکت سال مالی جاری بیش از 6 میلیارد تومان زیان خالص شناسایی کرده است. پادرو سال مالی 1403 حدود 30 میلیارد تومان درآمد عملیاتی کسب کرد که این رقم در مقایسه با سال 1402 رشد تقریباً دوبرابری دارد. 97 درصد درآمد پادرو از «پردازش حمل» است که مربوط به مشتریان متقاضی خدمات پستی بوده و بیش از 60 درصد درآمد آن از سرزمین هوشمند پاد (زیرمجموعه بانک پاسارگاد) و بابت حمل مرسوله کارت‌های بانکی است."],
["بانک پاسارگاد با شناسایی سود خالص 155 هزار میلیارد ریالی در ردۀ دوم سودآورترین بانک‌های کشور قرار گرفت."],
["مهدی اخوان بهابادی، مدیرعامل همراه اول، اعلام کرد درآمد عملیاتی شرکت با رشد 37 درصدی به 70 هزار و 677 میلیارد تومان رسیده است."],
["هزینه لجستیکی بوعلی حدود 100 میلیون دلار بود؛ 40٪ خوراک از خط لوله و 60٪ معادل 1 تا 1.5 میلیون تن در سال."],
["گروه دتا با سود خالص 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 هزار"
### 🔧 مثال صحیح برای متن شما:
**ورودی:** صورت‌های مالی حسابرسی شده شرکت آسان پادرو منتهی به 30 آذر 1403 نشان می‌دهد این شرکت سال مالی جاری بیش از 6 میلیارد تومان زیان خالص شناسایی کرده است...
**خروجی مطلوب:** صورت‌های مالی حسابرسی شده company-01 منتهی به 30 آذر 1403 نشان می‌دهد این شرکت سال مالی جاری بیش از 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
)