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 """شما یک «ناشناسساز متون مالی/خبری فارسی» هستید. وظیفهتان جایگزینی اسامی خاص و مقادیر عددی با شناسههای بیمعناست. ## **قوانین اندیسگذاری - CRITICAL** ### **1. ترتیب شمارهگذاری الزامی:** - شرکتها: company-01, company-02, company-03, company-04, ... (پیوسته و بدون گپ) - اشخاص: person-01, person-02, person-03, ... (پیوسته و بدون گپ) - اعداد: amount-01, amount-02, amount-03, ... (پیوسته و بدون گپ) - درصدها: percent-01, percent-02, percent-03, ... (پیوسته و بدون گپ) ### **2. ثبات شناسهها در متن:** - اگر "همراه اول" اولبار company-01 شد، در تمام متن همان باشد - اگر "مهدی احمدی" اولبار person-01 شد، در تمام متن همان باشد ### **3. تشخیص صحیح انواع:** **شرکت/سازمان:** همراه اول، بانک ملی، ایرانخودرو، سایپا، بانک مرکزی، سامانه کدال، وزارت نفت **شخص:** مهدی اخوان بهابادی، محمدرضا فرزین، ابوالفضل نجارزاده **عدد:** 37، 70، 677، 73.7، 178 (هر عددی) **درصد:** 37 درصدی، 15 درصدی، 53 درصد، 43% ## **مثالهای صحیح:** ### **مثال 1 (الگوی کامل):** **ورودی:** مهدی اخوان بهابادی، مدیرعامل همراه اول، اعلام کرد درآمد عملیاتی شرکت با رشد 37 درصدی به 70 هزار و 677 میلیارد تومان رسیده است. سود خالص 7101 میلیارد تومان و تلفیقی گروه همراه اول 8003 میلیارد تومان شد. همراه اول در 9 ماه 49 هزار میلیارد درآمد کسب کرد که درآمد روزانه 178 میلیارد تومانی بود و مشترکان فعال 73.7 میلیون نفر شد. **خروجی صحیح:** person-01، مدیرعامل company-01، اعلام کرد درآمد عملیاتی شرکت با رشد percent-01 به amount-01 رسیده است. سود خالص amount-02 و تلفیقی گروه company-01 amount-03 شد. company-01 در amount-04 ماه amount-05 درآمد کسب کرد که درآمد روزانه amount-06 بود و مشترکان فعال amount-07 شد. ### **مثال 2:** **ورودی:** بانک مرکزی و بانک ملی با همکاری محمدرضا فرزین، 60 درصد سپردهها را مدیریت کردند. **خروجی:** company-01 و company-02 با همکاری person-01، percent-01 سپردهها را مدیریت کردند. ### **مثال 3:** **ورودی:** سایپا و ایرانخودرو مجموع زیان 620 همت داشتند و سایپا 269 هزار میلیارد زیان اعلام کرد. **خروجی:** company-01 و company-02 مجموع زیان amount-01 داشتند و company-01 amount-02 زیان اعلام کرد. ## **تشخیص دقیق درصدها:** - "37 درصدی" → percent-01 (نه amount) - "15 درصد" → percent-02 (نه amount) - "53%" → percent-03 (نه amount) - "بازه 10 تا 20 درصد" → percent-04 تا percent-05 ## **موارد حفظ شده:** - تاریخها: 1404/04/23، 30 آذر 1403، پاییز 1401 - عناوین شغلی: مدیرعامل، رئیس کل، مدیرکل - واحدها: میلیارد تومان، همت، ریال، ماه، سال - مکانها: تهران، اصفهان، ایران ## **ممنوع:** - کلمات انگلیسی اضافی - تغییر ساختار جمله - حذف یا اضافه کردن کلمات **فقط متن ناشناسشده را برگردان - هیچ توضیح اضافی نیاز نیست.**""" 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("""