leilaghomashchi/Benchmark-data-anonymization

1import pandas as pd2import numpy as np3import re4from typing import Dict, List, Tuple, Set5import gradio as gr6from datetime import datetime7import io8import tempfile9import os10 11class AnonymizationEvaluator:12    """ابزار ارزیابی ناشناس‌سازی با استفاده از متن مرجع"""13    14    def __init__(self):15        self.results_df = None16        17    def extract_entities_from_text(self, text: str) -> Dict[str, Set[str]]:18        """استخراج موجودیت‌ها از متن با debugging"""19        if pd.isna(text) or not isinstance(text, str):20            return {'companies': set(), 'persons': set(), 'amounts': set(), 'percents': set(), 'groups': set()}21        22        # تمیز کردن متن23        text = str(text).strip()24        25        # الگوهای مختلف برای موجودیت‌ها26        patterns = {27            'companies': [r'company-(\d+)', r'Company-(\d+)', r'COMPANY-(\d+)'],28            'persons': [r'person-(\d+)', r'Person-(\d+)', r'PERSON-(\d+)'],29            'amounts': [r'amount-(\d+)', r'Amount-(\d+)', r'AMOUNT-(\d+)'],30            'percents': [r'percent-(\d+)', r'Percent-(\d+)', r'PERCENT-(\d+)'],31            'groups': [r'group-(\d+)', r'Group-(\d+)', r'GROUP-(\d+)']32        }33        34        entities = {}35        for entity_type, pattern_list in patterns.items():36            found = set()37            for pattern in pattern_list:38                matches = re.findall(pattern, text)39                found.update(matches)40            entities[entity_type] = found41            42        return entities43    44    def debug_text_analysis(self, reference_text: str, predicted_text: str, row_num: int = 0) -> str:45        """تابع debugging برای تحلیل متن‌ها"""46        debug_info = f"\n--- Debug Row {row_num + 1} ---\n"47        debug_info += f"Reference: '{reference_text[:100]}...'\n"48        debug_info += f"Predicted: '{predicted_text[:100]}...'\n"49        50        ref_entities = self.extract_entities_from_text(reference_text)51        pred_entities = self.extract_entities_from_text(predicted_text)52        53        debug_info += f"Reference entities: {dict(ref_entities)}\n"54        debug_info += f"Predicted entities: {dict(pred_entities)}\n"55        56        return debug_info57    58    def calculate_precision_recall_f1(self, reference_entities: Dict[str, Set[str]], 59                                    predicted_entities: Dict[str, Set[str]]) -> Tuple[float, float, float]:60        """محاسبه Precision, Recall و F1-Score"""61        62        # ترکیب همه موجودیت‌ها63        ref_all = set()64        pred_all = set()65        66        for entity_type in ['companies', 'persons', 'amounts', 'percents', 'groups']:67            # اضافه کردن prefix برای جلوگیری از تداخل68            ref_entities = {f"{entity_type}:{e}" for e in reference_entities.get(entity_type, set())}69            pred_entities = {f"{entity_type}:{e}" for e in predicted_entities.get(entity_type, set())}70            71            ref_all.update(ref_entities)72            pred_all.update(pred_entities)73        74        if len(pred_all) == 0 and len(ref_all) == 0:75            return 1.0, 1.0, 1.0  # هر دو خالی هستند76        elif len(pred_all) == 0:77            return 0.0, 0.0, 0.0  # predicted خالی ولی reference دارد78        elif len(ref_all) == 0:79            return 0.0, 1.0, 0.0  # reference خالی ولی predicted دارد80        81        # محاسبه True Positive82        true_positive = len(ref_all.intersection(pred_all))83        84        # محاسبه Precision, Recall85        precision = true_positive / len(pred_all) if len(pred_all) > 0 else 0.086        recall = true_positive / len(ref_all) if len(ref_all) > 0 else 0.087        88        # محاسبه F1-Score89        if precision + recall == 0:90            f1 = 0.091        else:92            f1 = 2 * (precision * recall) / (precision + recall)93        94        return precision, recall, f195    96    def calculate_accuracy(self, reference_text: str, predicted_text: str) -> float:97        """محاسبه Accuracy بر اساس تطابق کامل موجودیت‌ها"""98        ref_entities = self.extract_entities_from_text(reference_text)99        pred_entities = self.extract_entities_from_text(predicted_text)100        101        # شمارش کل موجودیت‌ها102        ref_total = sum(len(entities) for entities in ref_entities.values())103        104        if ref_total == 0:105            return 1.0 if sum(len(entities) for entities in pred_entities.values()) == 0 else 0.0106        107        # شمارش موجودیت‌های صحیح108        correct = 0109        for entity_type in ref_entities.keys():110            correct += len(ref_entities[entity_type].intersection(pred_entities[entity_type]))111        112        return correct / ref_total113    114    def evaluate_single_row(self, reference_text: str, predicted_text: str) -> Tuple[float, float, float]:115        """ارزیابی یک سطر"""116        try:117            # استخراج موجودیت‌ها118            ref_entities = self.extract_entities_from_text(reference_text)119            pred_entities = self.extract_entities_from_text(predicted_text)120            121            # محاسبه متریک‌ها122            precision, recall, f1 = self.calculate_precision_recall_f1(ref_entities, pred_entities)123            124            return precision, recall, f1125            126        except Exception as e:127            print(f"خطا در ارزیابی: {str(e)}")128            return 0.0, 0.0, 0.0129    130    def evaluate_dataset(self, file_path: str) -> Tuple[bool, str, pd.DataFrame]:131        """ارزیابی کل دیتاست با debugging"""132        try:133            # بارگذاری فایل134            df = pd.read_csv(file_path)135            136            # بررسی ستون‌های مورد نیاز137            required_columns = ['original_text', 'Reference_text', 'anonymized_text']138            missing_columns = [col for col in required_columns if col not in df.columns]139            140            if missing_columns:141                return False, f"ستون‌های مفقود: {', '.join(missing_columns)}", pd.DataFrame()142            143            # تشخیص مشکل - بررسی نمونه‌ای از داده‌ها144            debug_info = "\n=== Debug Information ===\n"145            debug_info += f"تعداد سطرها: {len(df)}\n"146            debug_info += f"ستون‌ها: {list(df.columns)}\n\n"147            148            # بررسی چند سطر اول149            for i in range(min(3, len(df))):150                ref_text = str(df.iloc[i]['Reference_text'])151                anon_text = str(df.iloc[i]['anonymized_text'])152                153                debug_info += self.debug_text_analysis(ref_text, anon_text, i)154            155            print(debug_info)  # نمایش در console156            157            # محاسبه متریک‌ها برای هر سطر158            precisions = []159            recalls = []160            f1_scores = []161            162            total_entities_found = 0  # شمارنده کل موجودیت‌های یافت شده163            164            for index, row in df.iterrows():165                precision, recall, f1 = self.evaluate_single_row(166                    row['Reference_text'], 167                    row['anonymized_text']168                )169                170                precisions.append(round(precision, 4))171                recalls.append(round(recall, 4))172                f1_scores.append(round(f1, 4))173                174                # شمارش موجودیت‌ها برای debugging175                ref_entities = self.extract_entities_from_text(str(row['Reference_text']))176                pred_entities = self.extract_entities_from_text(str(row['anonymized_text']))177                total_entities_found += sum(len(entities) for entities in ref_entities.values())178                total_entities_found += sum(len(entities) for entities in pred_entities.values())179            180            # اضافه کردن ستون‌های جدید181            df['Precision'] = precisions182            df['Recall'] = recalls183            df['F1_Score'] = f1_scores184            185            # ذخیره نتایج186            self.results_df = df187            188            # پیام وضعیت شامل اطلاعات debugging189            status_message = f"ارزیابی انجام شد. کل موجودیت‌های یافت شده: {total_entities_found}"190            if total_entities_found == 0:191                status_message += "\n⚠️ هیچ موجودیتی تشخیص داده نشد! لطفاً فرمت داده‌ها را بررسی کنید."192            193            return True, status_message, df194            195        except Exception as e:196            return False, f"خطا در پردازش فایل: {str(e)}", pd.DataFrame()197    198    def generate_summary_report(self, df: pd.DataFrame) -> str:199        """تولید گزارش خلاصه"""200        if df.empty:201            return "هیچ داده‌ای برای گزارش یافت نشد"202        203        # محاسبه آمار کلی204        avg_precision = df['Precision'].mean()205        avg_recall = df['Recall'].mean() 206        avg_f1 = df['F1_Score'].mean()207        208        # محاسبه آمار تفصیلی209        total_rows = len(df)210        high_precision_count = len(df[df['Precision'] >= 0.8])211        high_recall_count = len(df[df['Recall'] >= 0.8])212        high_f1_count = len(df[df['F1_Score'] >= 0.8])213        214        # بهترین و بدترین نتایج215        best_f1_idx = df['F1_Score'].idxmax()216        worst_f1_idx = df['F1_Score'].idxmin()217        218        report = f"""219        ## 📊 گزارش جامع ارزیابی220        221        ### آمار کلی:222        - **تعداد کل سطرها:** {total_rows}223        - **میانگین Precision:** {avg_precision:.4f}224        - **میانگین Recall:** {avg_recall:.4f}  225        - **میانگین F1-Score:** {avg_f1:.4f}226        227        ### توزیع عملکرد (امتیاز ≥ 0.8):228        - **Precision بالا:** {high_precision_count} سطر ({high_precision_count/total_rows*100:.1f}%)229        - **Recall بالا:** {high_recall_count} سطر ({high_recall_count/total_rows*100:.1f}%)230        - **F1-Score بالا:** {high_f1_count} سطر ({high_f1_count/total_rows*100:.1f}%)231        232        ### نمونه‌های برتر و ضعیف:233        - **بهترین F1-Score:** {df.loc[best_f1_idx, 'F1_Score']:.4f} (سطر {best_f1_idx + 1})234        - **ضعیف‌ترین F1-Score:** {df.loc[worst_f1_idx, 'F1_Score']:.4f} (سطر {worst_f1_idx + 1})235        """236        237        return report238    239    def create_downloadable_csv(self) -> bytes:240        """ایجاد محتوای CSV برای دانلود مستقیم"""241        if self.results_df is None or self.results_df.empty:242            return None243            244        try:245            # تولید محتوای CSV در حافظه246            csv_buffer = io.StringIO()247            self.results_df.to_csv(csv_buffer, index=False, encoding='utf-8')248            csv_content = csv_buffer.getvalue()249            csv_buffer.close()250            251            # تبدیل به bytes برای دانلود252            return csv_content.encode('utf-8-sig')253            254        except Exception as e:255            print(f"خطا در ایجاد محتوای CSV: {str(e)}")256            return None257 258def create_evaluation_interface():259    """ایجاد رابط کاربری ارزیابی"""260    evaluator = AnonymizationEvaluator()261    262    with gr.Blocks(263        title="ارزیابی ناشناس‌سازی",264        theme=gr.themes.Soft(),265        css="""266        .gradio-container {267            font-family: 'Tahoma', 'Arial', sans-serif !important;268            direction: rtl;269            max-width: 1200px;270            margin: 0 auto;271        }272        .upload-area {273            border: 2px dashed #4CAF50;274            border-radius: 15px;275            padding: 30px;276            text-align: center;277            background: linear-gradient(145deg, #f8f9fa, #e9ecef);278            margin: 20px 0;279        }280        .results-table {281            direction: ltr;282            font-family: monospace;283            font-size: 12px;284        }285        .summary-box {286            background-color: #e3f2fd;287            border: 1px solid #2196F3;288            border-radius: 10px;289            padding: 20px;290            margin: 15px 0;291        }292        """293    ) as interface:294        295        gr.Markdown("""296        # 📊 ابزار ارزیابی ناشناس‌سازی با متن مرجع297        ### آپلود فایل CSV شامل ستون‌های: original_text, Reference_text, anonymized_text298        """)299        300        with gr.Row():301            with gr.Column(scale=1):302                gr.Markdown("### 📁 بارگذاری فایل")303                304                file_input = gr.File(305                    label="انتخاب فایل CSV",306                    file_types=[".csv"],307                    elem_classes=["upload-area"]308                )309                310                evaluate_btn = gr.Button(311                    "🚀 شروع ارزیابی",312                    variant="primary",313                    size="lg",314                    interactive=False315                )316                317                download_btn = gr.Button(318                    "💾 دانلود نتایج CSV",319                    variant="secondary",320                    visible=False321                )322            323            with gr.Column(scale=2):324                status_output = gr.Markdown("وضعیت: آماده بارگذاری فایل...")325                326                summary_output = gr.Markdown(327                    visible=False,328                    elem_classes=["summary-box"]329                )330        331        # جدول نتایج332        results_table = gr.Dataframe(333            label="نتایج تفصیلی (نمایش 10 سطر اول)",334            visible=False,335            elem_classes=["results-table"],336            wrap=True337        )338        339        # فایل دانلود340        download_file = gr.File(341            visible=False,342            label="فایل نتایج"343        )344        345        def on_file_upload(file):346            if file is None:347                return "❌ لطفاً فایل را انتخاب کنید", gr.Button(interactive=False)348            349            return "✅ فایل بارگذاری شد، آماده ارزیابی", gr.Button(interactive=True)350        351        def evaluate_file(file):352            if file is None:353                return (354                    "❌ هیچ فایلی انتخاب نشده",355                    gr.Markdown(visible=False),356                    gr.Dataframe(visible=False),357                    gr.Button(visible=False),358                    gr.File(visible=False)359                )360            361            try:362                success, message, df = evaluator.evaluate_dataset(file.name)363                364                if not success:365                    return (366                        f"❌ {message}",367                        gr.Markdown(visible=False),368                        gr.Dataframe(visible=False),369                        gr.Button(visible=False),370                        gr.File(visible=False)371                    )372                373                # تولید گزارش خلاصه374                summary = evaluator.generate_summary_report(df)375                376                # نمایش 10 سطر اول برای نمونه در رابط377                display_df = df.head(10)378                379                # پیام اطلاع‌رسانی380                status_message = f"✅ {message} - {len(df)} سطر پردازش شد. نمایش: 10 سطر اول، دانلود: همه سطرها"381                382                return (383                    status_message,384                    gr.Markdown(value=summary, visible=True),385                    gr.Dataframe(value=display_df, visible=True),386                    gr.Button(visible=True),387                    gr.File(visible=False)388                )389                390            except Exception as e:391                return (392                    f"❌ خطای غیرمنتظره: {str(e)}",393                    gr.Markdown(visible=False),394                    gr.Dataframe(visible=False),395                    gr.Button(visible=False),396                    gr.File(visible=False)397                )398        399        def download_results():400            try:401                if evaluator.results_df is None or evaluator.results_df.empty:402                    return (403                        "❌ هیچ داده‌ای برای دانلود وجود ندارد. ابتدا ارزیابی را انجام دهید.",404                        gr.File(visible=False)405                    )406                407                # ایجاد محتوای CSV408                csv_content = evaluator.create_downloadable_csv()409                if csv_content:410                    # تولید نام فایل411                    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")412                    filename = f"evaluation_results_{timestamp}.csv"413                    414                    # ذخیره در فایل موقت برای دانلود415                    with tempfile.NamedTemporaryFile(mode='wb', delete=False, 416                                                   suffix='.csv', prefix='eval_') as temp_file:417                        temp_file.write(csv_content)418                        temp_filename = temp_file.name419                    420                    return (421                        f"✅ فایل نتایج آماده شد: {filename} ({len(evaluator.results_df)} سطر)",422                        gr.File(value=temp_filename, visible=True)423                    )424                else:425                    return (426                        "❌ خطا در ایجاد محتوای CSV",427                        gr.File(visible=False)428                    )429            except Exception as e:430                return (431                    f"❌ خطا در دانلود: {str(e)}",432                    gr.File(visible=False)433                )434        435        # اتصال رویدادها436        file_input.change(437            fn=on_file_upload,438            inputs=[file_input],439            outputs=[status_output, evaluate_btn]440        )441        442        evaluate_btn.click(443            fn=evaluate_file,444            inputs=[file_input],445            outputs=[status_output, summary_output, results_table, download_btn, download_file]446        )447        448        download_btn.click(449            fn=download_results,450            outputs=[status_output, download_file]451        )452        453        # راهنمای استفاده454        with gr.Accordion("📖 راهنمای استفاده", open=False):455            gr.Markdown("""456            ### فرمت فایل CSV مورد نیاز:457            458            فایل شما باید حاوی دقیقاً این سه ستون باشد:459            - **original_text**: متن اصلی460            - **Reference_text**: متن ناشناس‌شده مرجع (Ground Truth)  461            - **anonymized_text**: متن ناشناس‌شده مورد ارزیابی462            463            ### متریک‌های محاسبه شده:464            465            - **Precision**: دقت = (تعداد موجودیت‌های صحیح شناسایی شده) / (کل موجودیت‌های شناسایی شده)466            - **Recall**: بازیابی = (تعداد موجودیت‌های صحیح شناسایی شده) / (کل موجودیت‌های مرجع)467            - **F1-Score**: میانگین هارمونیک Precision و Recall468            469            ### مراحل کار:470            471            1. فایل CSV را آپلود کنید472            2. روی "شروع ارزیابی" کلیک کنید  473            3. گزارش خلاصه و جدول نمونه (10 سطر اول) را مشاهده کنید474            4. **فایل نتایج کامل (همه سطرها) را دانلود کنید**475            476            ### نکات مهم:477            478            - **نمایش رابط**: فقط 10 سطر اول نمایش داده می‌شود479            - **فایل دانلود**: شامل تمام سطرهای پردازش شده + متریک‌ها480            - فایل خروجی شامل ستون‌های اصلی + سه ستون متریک خواهد بود481            - متریک‌ها برای هر سطر جداگانه محاسبه می‌شوند482            - آمار کلی در گزارش خلاصه نمایش داده می‌شود483            484            ### مشکل در دانلود؟485            486            اگر فایل دانلود نمی‌شود:487            1. مرورگر خود را رفرش کنید488            2. مجدداً ارزیابی را انجام دهید489            3. اطمینان حاصل کنید که popup blocker غیرفعال است490            """)491        492        # تست دانلود493        with gr.Accordion("🧪 تست دانلود", open=False):494            gr.Markdown("برای تست عملکرد دانلود:")495            test_download_btn = gr.Button("تست دانلود فایل نمونه")496            test_file_output = gr.File(label="فایل تست", visible=False)497            498            def create_test_file():499                """ایجاد فایل تست برای بررسی دانلود"""500                try:501                    test_data = {502                        'original_text': ['متن تست 1', 'متن تست 2'],503                        'Reference_text': ['company-01 amount-01', 'person-01 amount-02'],504                        'anonymized_text': ['company-01 amount-01', 'person-01 amount-02'],505                        'Precision': [1.0, 1.0],506                        'Recall': [1.0, 1.0],507                        'F1_Score': [1.0, 1.0]508                    }509                    test_df = pd.DataFrame(test_data)510                    511                    # ایجاد محتوای CSV در حافظه512                    csv_buffer = io.StringIO()513                    test_df.to_csv(csv_buffer, index=False)514                    csv_content = csv_buffer.getvalue()515                    csv_buffer.close()516                    517                    # تبدیل به bytes و ذخیره در فایل موقت518                    csv_bytes = csv_content.encode('utf-8-sig')519                    with tempfile.NamedTemporaryFile(mode='wb', delete=False, 520                                                   suffix='.csv', prefix='test_') as temp_file:521                        temp_file.write(csv_bytes)522                        temp_filename = temp_file.name523                    524                    return gr.File(value=temp_filename, visible=True)525                except Exception as e:526                    print(f"خطا در ایجاد فایل تست: {str(e)}")527                    return gr.File(visible=False)528            529            test_download_btn.click(530                fn=create_test_file,531                outputs=[test_file_output]532            )533    534    return interface535 536# اجرای برنامه537if __name__ == "__main__":538    interface = create_evaluation_interface()539    interface.launch()