new-last.py

import gradio as gr
import pandas as pd
import re

class AnonymizationEvaluator:
    def __init__(self):
        # فهرست بانک‌های ایران
        self.iranian_banks = [
            'بانک ملی', 'بانک صادرات', 'بانک پاسارگاد', 'بانک کشاورزی',
            'بانک ملت', 'بانک تجارت', 'بانک صنعت و معدن', 'بانک رسالت',
            'بانک دی', 'بانک پارسیان', 'بانک کارآفرین', 'بانک سامان',
            'بانک اقتصاد نوین', 'بانک مهر اقتصاد', 'بانک آینده', 'بانک کشاورزی'
        ]
        
        # فهرست سازمان‌های دولتی
        self.government_orgs = [
            'بانک مرکزی جمهوری اسلامی ایران',
            'دفتر اسناد رسمی',
            'اداره کل مالیات',
            'تامین اجتماعی',
            'وزارت دادگستری'
        ]
        
        self.patterns = {
            'person_names': {
                'pattern': re.compile(r'(?:آقای|خانم)\s+[\u0600-\u06FF\s]+?(?=\s+با|$)', re.UNICODE),
                'replacement': re.compile(r'person_\d+'),
                'name': 'اسامی اشخاص'
            },
            'national_ids': {
                'pattern': re.compile(r'\b\d{10,11}\b'),
                'replacement': re.compile(r'id_number_\d+'),
                'name': 'کدهای ملی'
            },
            'phone_numbers': {
                'pattern': re.compile(r'(?:09\d{9}|021-\d{8}|0\d{2,3}-\d{8})'),
                'replacement': re.compile(r'phone_\d+'),
                'name': 'شماره تلفن‌ها'
            },
            'account_numbers': {
                'pattern': re.compile(r'\b\d{3}-\d{3}-\d{3}-\d{1}\b'),
                'replacement': re.compile(r'account_\d+'),
                'name': 'شماره حساب‌ها'
            },
            'card_numbers': {
                'pattern': re.compile(r'\b\d{4}-\d{4}-\d{4}-\d{4}\b'),
                'replacement': re.compile(r'card_number_\d+'),
                'name': 'شماره کارت‌ها'
            },
            'amounts': {
                'pattern': re.compile(r'\b\d+\s*تومان'),
                'replacement': re.compile(r'amount_\d+'),
                'name': 'مبالغ مالی'
            },
            'dates': {
                'pattern': re.compile(r'(?:\d{4}\/\d{2}\/\d{2}|۳۰\s*اسفند\s*۱۴۰۳|\b\d{4}\b(?=\s*سال))'),
                'replacement': re.compile(r'date_\d+'),
                'name': 'تاریخ‌ها'
            },
            'iranian_banks': {
                'pattern': re.compile(f"({'|'.join(self.iranian_banks)})", re.UNICODE),
                'replacement': re.compile(r'company_\d+'),
                'name': 'بانک‌های ایران'
            },
            'government_orgs': {
                'pattern': re.compile(f"({'|'.join(self.government_orgs)})", re.UNICODE),
                'replacement': re.compile(r'company_\d+'),
                'name': 'سازمان‌های دولتی'
            },
            'other_companies': {
                'pattern': re.compile(r'شرکت\s+[\u0600-\u06FF\s]+?(?=\s|$|،|\.)', re.UNICODE),
                'replacement': re.compile(r'company_\d+'),
                'name': 'سایر شرکت‌ها'
            },
            'addresses': {
                'pattern': re.compile(r'(?:تهران|کرج|اصفهان|بندر\s*ماهشهر)[،\s]+[^،\.]+', re.UNICODE),
                'replacement': re.compile(r'(?:full_address_\d+|location_\d+)'),
                'name': 'آدرس‌ها'
            },
            'documents': {
                'pattern': re.compile(r'(?:INV-\d{4}-\d{4}|RPT-\d{4}-\d{4}|\b\d{4}(?=\s+تهران)|\b\d{7}\b)'),
                'replacement': re.compile(r'(?:invoice_\d+|report_\d+|contract_\d+|cheque_\d+)'),
                'name': 'شماره اسناد'
            }
        }
    
    def analyze_entities(self, original_text, anonymized_text):
        results = {}
        for entity_type, config in self.patterns.items():
            original_matches = config['pattern'].findall(original_text)
            replacement_matches = config['replacement'].findall(anonymized_text)
            
            anonymized_count = 0
            for match in original_matches:
                if not anonymized_text.count(match.strip()):
                    anonymized_count += 1
            
            if len(replacement_matches) > anonymized_count:
                anonymized_count = min(len(replacement_matches), len(original_matches))
            
            percentage = (anonymized_count / len(original_matches) * 100) if original_matches else 0
            
            results[entity_type] = {
                'name': config['name'],
                'total': len(original_matches),
                'anonymized': anonymized_count,
                'percentage': round(percentage, 1),
                'samples': original_matches[:3] if original_matches else []
            }
        return results
    
    def evaluate_csv(self, csv_file):
        try:
            if csv_file is None:
                return "لطفاً یک فایل CSV آپلود کنید."
            
            try:
                df = pd.read_csv(csv_file.name, encoding='utf-8')
            except:
                df = pd.read_csv(csv_file.name, encoding='utf-8', sep='\t')
            
            if 'original_text' not in df.columns or 'anonymized_text' not in df.columns:
                return "فایل CSV باید شامل ستون‌های 'original_text' و 'anonymized_text' باشد."
            
            overall_stats = {}
            total_entities = 0
            total_anonymized = 0
            
            for _, row in df.iterrows():
                if pd.isna(row['original_text']) or pd.isna(row['anonymized_text']):
                    continue
                
                row_analysis = self.analyze_entities(str(row['original_text']), str(row['anonymized_text']))
                
                for entity_type, data in row_analysis.items():
                    if entity_type not in overall_stats:
                        overall_stats[entity_type] = {
                            'name': data['name'],
                            'total': 0,
                            'anonymized': 0,
                            'samples': []
                        }
                    
                    overall_stats[entity_type]['total'] += data['total']
                    overall_stats[entity_type]['anonymized'] += data['anonymized']
                    overall_stats[entity_type]['samples'].extend(data['samples'])
                    
                    total_entities += data['total']
                    total_anonymized += data['anonymized']
            
            for entity_type in overall_stats:
                stats = overall_stats[entity_type]
                stats['percentage'] = round((stats['anonymized'] / stats['total'] * 100) if stats['total'] > 0 else 0, 1)
                stats['samples'] = list(set(stats['samples']))[:3]
            
            return self.generate_report(overall_stats, total_entities, total_anonymized, len(df))
            
        except Exception as e:
            return f"خطا در پردازش فایل: {str(e)}"
    
    def generate_report(self, stats, total_entities, total_anonymized, total_rows):
        report = f"""# گزارش ارزیابی ناشناس‌سازی متن

## خلاصه کلی
- **تعداد ردیف‌های پردازش شده**: {total_rows:,} ردیف
- **تعداد موجودیت‌های حساس شناسایی شده**: {total_entities:,} مورد
- **تعداد موجودیت‌های ناشناس شده**: {total_anonymized:,} مورد  
- **درصد پوشش کلی**: {(total_anonymized/total_entities*100) if total_entities > 0 else 0:.1f}%

## تحلیل تفصیلی دسته‌بندی موجودیت‌ها

"""
        
        excellent = []
        good = []
        poor = []
        not_found = []
        
        for entity_type, data in stats.items():
            if data['total'] == 0:
                not_found.append((entity_type, data))
            elif data['percentage'] == 100:
                excellent.append((entity_type, data))
            elif data['percentage'] >= 80:
                good.append((entity_type, data))
            else:
                poor.append((entity_type, data))
        
        if excellent:
            report += "### ✅ عملکرد عالی (100% موفقیت)\n"
            for entity_type, data in excellent:
                report += f"- **{data['name']}**: {data['anonymized']}/{data['total']} (100%)\n"
            report += "\n"
        
        if good:
            report += "### 🟡 عملکرد خوب (80-99% موفقیت)\n"
            for entity_type, data in good:
                report += f"- **{data['name']}**: {data['anonymized']}/{data['total']} ({data['percentage']}%)\n"
            report += "\n"
        
        if poor:
            report += "### 🔴 عملکرد ضعیف (<80% موفقیت)\n"
            for entity_type, data in poor:
                missed = data['total'] - data['anonymized']
                report += f"- **{data['name']}**: {data['anonymized']}/{data['total']} ({data['percentage']}%) - {missed} مورد جا مانده\n"
            report += "\n"
        
        if not_found:
            report += "### ⚪ موجودیت‌های یافت نشده\n"
            for entity_type, data in not_found:
                report += f"- **{data['name']}**: هیچ موجودیتی یافت نشد\n"
            report += "\n"
        
        report += "## جدول خلاصه متریک‌ها\n\n"
        report += "| دسته موجودیت | یافته شده | ناشناس شده | درصد موفقیت | موارد جا مانده |\n"
        report += "|---------------|-----------|-------------|-------------|----------------|\n"
        
        for entity_type, data in stats.items():
            if data['total'] > 0:
                missed = data['total'] - data['anonymized']
                report += f"| {data['name']} | {data['total']} | {data['anonymized']} | {data['percentage']}% | {missed} |\n"
        
        major_issues = [(k, v) for k, v in stats.items() if v['total'] > 0 and v['percentage'] < 80]
        major_issues.sort(key=lambda x: x[1]['total'] - x[1]['anonymized'], reverse=True)
        
        if major_issues:
            report += "\n## 🚨 مشکلات اصلی شناسایی شده\n\n"
            for i, (entity_type, data) in enumerate(major_issues, 1):
                missed = data['total'] - data['anonymized']
                impact = round(missed / total_entities * 100, 1) if total_entities > 0 else 0
                report += f"### {i}. {data['name']}\n"
                report += f"- **وضعیت**: {data['percentage']}% موفقیت\n"
                report += f"- **موارد جا مانده**: {missed} مورد از {data['total']} مورد\n"
                report += f"- **تاثیر بر کل**: {impact}% از کل موجودیت‌ها\n\n"
        
        precision = round((total_anonymized / total_entities * 100) if total_entities > 0 else 0, 1)
        
        report += f"""## 📊 آمار نهایی

- **کل موجودیت‌های شناسایی شده**: {total_entities:,}
- **کل موجودیت‌های ناشناس شده**: {total_anonymized:,}
- **موجودیت‌های جا مانده**: {total_entities - total_anonymized:,}
- **دقت (Precision)**: {precision}%
- **پوشش (Recall)**: {precision}%
- **امتیاز F1**: {precision}%
"""
        
        return report

def create_interface():
    evaluator = AnonymizationEvaluator()
    
    def process_file(csv_file):
        if csv_file is None:
            return "لطفاً یک فایل CSV آپلود کنید."
        return evaluator.evaluate_csv(csv_file)
    
    with gr.Blocks(title="ارزیاب متریک‌های ناشناس‌سازی") as demo:
        gr.Markdown("# ارزیاب متریک‌های ناشناس‌سازی متن فارسی")
        
        file_input = gr.File(label="آپلود فایل CSV", file_types=[".csv"])
        analyze_btn = gr.Button("محاسبه متریک‌ها", variant="primary")
        output = gr.Markdown(value="فایل CSV خود را آپلود کنید.")
        
        analyze_btn.click(fn=process_file, inputs=[file_input], outputs=[output])
    
    return demo

if __name__ == "__main__":
    app = create_interface()
    app.launch()