term_extract.py

#!/usr/bin/env python3
"""
Multi-concurrent bilingual term extraction script - implements high concurrency without changing existing Agent architecture
"""
import json
import argparse
import asyncio
import logging
import time
from typing import List, Dict, Any, Optional, Tuple
from pathlib import Path
import sys
import os
import aiofiles
from dataclasses import dataclass
from tqdm import tqdm
from tqdm.asyncio import tqdm as atqdm

# Add project root to Python path
project_root = Path(__file__).parent.parent
sys.path.insert(0, str(project_root))

from src.agents.bilingual_term_extract import BilingualTermExtractAgent
from src.agents.bilingual_term_quality_check import BilingualTermQualityCheckAgent
from src.agents.bilingual_term_normalization import TermNormalizationAgent
from src.agents.bilingual_term_standardization import BilingualTermStandardizationAgent

# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)


@dataclass
class ConcurrentConfig:
    """Concurrent processing configuration"""
    batch_size: int = 10   # Number of entries per batch
    max_concurrent_requests: int = 40  # Maximum concurrent requests
    timeout: int = 300    # Timeout in seconds
    save_interval: int = 2  # Save checkpoint every N batches
    checkpoint_file: str = "checkpoint.json"  # Checkpoint file path
    stage_dir: Optional[str] = None  # Stage snapshot output directory (default: same as output file)
    
    # Agent batch processing configuration
    extraction_batch_size: int = 5      # Term extraction: number of article pairs per batch (recommended: 5-10)
    quality_check_batch_size: int = 10  # Quality check: number of term per batch (recommended: 8-15)
    normalization_batch_size: int = 10  # Normalization: number of term per batch (recommended: 10-20)
    
    # Standardization configuration
    max_targets_per_source: int = 3     # Maximum target term to keep per source term
    confidence_weight: float = 0.4      # Confidence weight
    quality_weight: float = 0.6         # Quality score weight


def round_floats(obj, decimals=2):
    """Recursively round all floating point numbers to specified decimal places"""
    if isinstance(obj, float):
        return round(obj, decimals)
    elif isinstance(obj, dict):
        return {k: round_floats(v, decimals) for k, v in obj.items()}
    elif isinstance(obj, list):
        return [round_floats(item, decimals) for item in obj]
    else:
        return obj


class ConcurrentBilingualTermExtractor:
    """Multi-concurrent bilingual term extractor (without changing existing Agent architecture)"""
    
    def __init__(self, config: ConcurrentConfig):
        self.config = config
        self.semaphore = asyncio.Semaphore(config.max_concurrent_requests)
        self.stats = {
            'total_entries': 0,         # Number of input entries
            'total_term': 0,           # Number of final term
            'extracted_term': 0,       # Number of extracted term
            'filtered_term': 0,        # Number of filtered term
            'normalized_term': 0,      # Number of normalized term
            'start_time': None,
            'end_time': None,
            'processed_batches': 0,
            'last_save_time': None,
            # Backward compatibility
            'total_processed': 0,
            'successful': 0
        }
        self.checkpoint_data = {
            'processed_batches': [],
            'all_term': [],
            'all_extracted_term': [],
            'all_filtered_term': [],
            'all_normalized_term': [],
            'all_standardized_term': [],  # Stage 4: Standardized term
            'stats': self.stats
        }
    
    def save_checkpoint(self):
        """Save checkpoint"""
        try:
            self.checkpoint_data['stats'] = self.stats
            # Format all floats to 2 decimal places
            formatted_data = round_floats(self.checkpoint_data, decimals=2)
            with open(self.config.checkpoint_file, 'w', encoding='utf-8') as f:
                json.dump(formatted_data, f, ensure_ascii=False, indent=2)
            # Also write stage snapshots for quick recovery/inspection
            base = Path(self.config.checkpoint_file)
            # Calculate stage output directory
            stage_dir = Path(self.config.stage_dir) if self.config.stage_dir else base.parent
            stage_dir.mkdir(parents=True, exist_ok=True)
            try:
                with open(stage_dir / f"{base.stem}_stage1_extracted.json", 'w', encoding='utf-8') as f1:
                    json.dump(round_floats(self.checkpoint_data.get('all_extracted_term', [])), f1, ensure_ascii=False, indent=2)
                with open(stage_dir / f"{base.stem}_stage2_filtered.json", 'w', encoding='utf-8') as f2:
                    json.dump(round_floats(self.checkpoint_data.get('all_filtered_term', [])), f2, ensure_ascii=False, indent=2)
                with open(stage_dir / f"{base.stem}_stage3_normalized.json", 'w', encoding='utf-8') as f3:
                    json.dump(round_floats(self.checkpoint_data.get('all_normalized_term', [])), f3, ensure_ascii=False, indent=2)
                with open(stage_dir / f"{base.stem}_stage4_standardized.json", 'w', encoding='utf-8') as f4:
                    json.dump(round_floats(self.checkpoint_data.get('all_standardized_term', [])), f4, ensure_ascii=False, indent=2)
                with open(stage_dir / f"{base.stem}_final_term.json", 'w', encoding='utf-8') as f5:
                    json.dump(round_floats(self.checkpoint_data.get('all_term', [])), f5, ensure_ascii=False, indent=2)
            except Exception as se:
                logger.warning(f"Failed to write stage checkpoints: {se}")
            self.stats['last_save_time'] = time.time()
            logger.info(f"Checkpoint saved: {self.config.checkpoint_file}")
        except Exception as e:
            logger.error(f"Failed to save checkpoint: {e}")
    
    def load_checkpoint(self):
        """Load checkpoint"""
        try:
            if os.path.exists(self.config.checkpoint_file):
                with open(self.config.checkpoint_file, 'r', encoding='utf-8') as f:
                    self.checkpoint_data = json.load(f)
                self.stats = self.checkpoint_data.get('stats', self.stats)
                logger.info(f"Checkpoint loaded: {self.config.checkpoint_file}")
                logger.info(f"Processed batches: {len(self.checkpoint_data.get('processed_batches', []))}")
                return True
        except Exception as e:
            logger.error(f"Failed to load checkpoint: {e}")
        return False
    
    def add_batch_result(self, batch_result: Dict[str, Any], batch_index: int):
        """Add batch result to checkpoint data"""
        if batch_result:
            self.checkpoint_data['all_term'].extend(batch_result.get('final_term', []))
            self.checkpoint_data['all_extracted_term'].extend(batch_result.get('extracted_term', []))
            self.checkpoint_data['all_filtered_term'].extend(batch_result.get('filtered_term', []))
            self.checkpoint_data['all_normalized_term'].extend(batch_result.get('normalized_term', []))
            self.checkpoint_data['all_standardized_term'].extend(batch_result.get('standardized_term', []))
            self.checkpoint_data['processed_batches'].append(batch_index)
            self.stats['processed_batches'] += 1
            self.stats['successful'] = len(self.checkpoint_data['all_term'])
    
    async def extract_term_concurrent(self, json_file_path: str, max_entries: Optional[int] = None, resume: bool = True) -> Dict[str, Any]:
        """Multi-concurrent term extraction (supports resume from checkpoint)"""
        logger.info(f"Starting multi-concurrent processing for file: {json_file_path}")
        self.stats['start_time'] = time.time()
        
        # Try to load checkpoint
        if resume and self.load_checkpoint():
            logger.info("Resuming from checkpoint...")
            processed_batches = set(self.checkpoint_data.get('processed_batches', []))
        else:
            processed_batches = set()
        
        # Read JSON file
        async with aiofiles.open(json_file_path, 'r', encoding='utf-8') as f:
            content = await f.read()
            data = json.loads(content)
        
        # Detect language pair from metadata
        metadata = data.get('metadata', {})
        lang_pair = metadata.get('pair', 'zh-en')  # Default: Chinese-English
        src_lang, tgt_lang = lang_pair.split('-')
        logger.info(f"Detected language pair: {src_lang} -> {tgt_lang}")
        
        # Save language pair info to configuration
        self.src_lang = src_lang
        self.tgt_lang = tgt_lang
        
        entries = data.get('entries', [])
        if max_entries:
            entries = entries[:max_entries]
        
        logger.info(f"Found {len(entries)} bilingual entries")
        
        # Split into batches
        batches = [entries[i:i + self.config.batch_size] 
                  for i in range(0, len(entries), self.config.batch_size)]
        
        logger.info(f"Split into {len(batches)} batches, {self.config.batch_size} entries per batch")
        
        # Filter processed batches (for stage 1)
        remaining_batches = [(i, batch) for i, batch in enumerate(batches) if i not in processed_batches]
        logger.info(f"Remaining batches to process: {len(remaining_batches)}/{len(batches)}")

        # =========================
        # Stage 1: Term Extraction (process all inputs)
        # =========================
        extracted_term_all: List[Dict[str, Any]] = self.checkpoint_data.get('all_extracted_term', [])
        if remaining_batches:
            logger.info("Starting Stage 1: Term Extraction (processing all remaining batches)")
            stage1_results = await self._stage1_extract_concurrent(remaining_batches)
            # Flatten and merge results
            for batch_term in stage1_results:
                if batch_term:
                    extracted_term_all.extend(batch_term)
            # Write checkpoint
            self.checkpoint_data['all_extracted_term'] = extracted_term_all
            self.save_checkpoint()
        else:
            logger.info("Stage 1 skipped: All batches already processed")

        # =========================
        # Stage 2: Quality Check (based on all results from Stage 1)
        # =========================
        filtered_term_all: List[Dict[str, Any]] = self.checkpoint_data.get('all_filtered_term', [])
        
        # Check if Stage 2 is already completed
        if filtered_term_all:
            logger.info(f"Stage 2 skipped: Quality check already completed, have {len(filtered_term_all)} filtered term")
        else:
            remaining_for_qc = extracted_term_all if extracted_term_all else self.checkpoint_data.get('all_extracted_term', [])
            if remaining_for_qc:
                logger.info("Starting Stage 2: Quality Check (processing all entries, grouped by entry)")
                stage2_results = await self._stage2_quality_concurrent(batches, remaining_for_qc)
                # ❌ Do not extend and save here again!
                # _stage2_quality_concurrent already deduplicates and saves to checkpoint_data['all_filtered_term']
                filtered_term_all = self.checkpoint_data.get('all_filtered_term', [])
            else:
                logger.info("Stage 2 skipped: No extraction results available")

        # =========================
        # Stage 3: Normalization (based on all results from Stage 2)
        # =========================
        normalized_term_all: List[Dict[str, Any]] = self.checkpoint_data.get('all_normalized_term', [])
        
        # Check if Stage 3 is already completed
        if normalized_term_all:
            logger.info(f"Stage 3 skipped: Normalization already completed, have {len(normalized_term_all)} normalized term")
        else:
            remaining_for_norm = filtered_term_all if filtered_term_all else self.checkpoint_data.get('all_filtered_term', [])
            if remaining_for_norm:
                logger.info("Starting Stage 3: Normalization (processing all filtered term in batches)")
                stage3_results = await self._stage3_normalize_concurrent(remaining_for_norm) 
                # Results already saved in checkpoint_data, just read them
                normalized_term_all = self.checkpoint_data.get('all_normalized_term', [])
                logger.info(f"Stage 3 completed, obtained {len(normalized_term_all)} normalized term")
            else:
                logger.info("Stage 3 skipped: No filtered results available")

        # =========================
        # Stage 4: Standardization (based on all results from Stage 3)
        # =========================
        standardized_term_all: List[Dict[str, Any]] = self.checkpoint_data.get('all_standardized_term', [])
        
        # Check if Stage 4 is already completed
        if standardized_term_all:
            logger.info(f"Stage 4 skipped: Standardization already completed, have {len(standardized_term_all)} standardized term")
        else:
            remaining_for_std = normalized_term_all if normalized_term_all else self.checkpoint_data.get('all_normalized_term', [])
            if remaining_for_std:
                logger.info("Starting Stage 4: Standardization (deduplication, sorting, cleaning)")
                standardized_term_all = await self._stage4_standardize(remaining_for_std)
                self.checkpoint_data['all_standardized_term'] = standardized_term_all
                self.save_checkpoint()
            else:
                logger.info("Stage 4 skipped: No normalization results available")

        # Generate final term (using standardized results)
        all_term = self.checkpoint_data.get('all_term', [])
        if standardized_term_all:
            all_term = standardized_term_all  # Use standardized results directly
            self.checkpoint_data['all_term'] = all_term
            self.save_checkpoint()
        elif normalized_term_all:
            final_term = []
            for term in normalized_term_all:
                # term is already a dict (dictionary saved from normalization stage)
                final_term.append({
                    'source_term': term.get('source_term', ''),
                    'target_term': term.get('target_term', ''),
                    'normalized_source': term.get('normalized_source'),
                    'normalized_target': term.get('normalized_target'),
                    'confidence': term.get('confidence', 0.0),
                    'category': term.get('category', ''),
                    'source_context': term.get('source_context', ''),
                    'target_context': term.get('target_context', ''),
                    'quality_score': term.get('quality_score', 0.0),
                    'is_valid': term.get('is_valid', False),
                    'law': term.get('law', ''),
                    'domain': term.get('domain', ''),
                    'year': term.get('year', ''),
                    'entry_id': term.get('entry_id', ''),
                    'normalization_notes': term.get('normalization_notes')
                })
            all_term.extend(final_term)
            self.checkpoint_data['all_term'] = all_term
            self.save_checkpoint()
        
        self.stats['end_time'] = time.time()
        self.stats['total_entries'] = len(entries)  # Number of input entries
        self.stats['total_term'] = len(all_term)  # Number of final term
        self.stats['extracted_term'] = len(self.checkpoint_data.get('all_extracted_term', []))
        self.stats['filtered_term'] = len(self.checkpoint_data.get('all_filtered_term', []))
        self.stats['normalized_term'] = len(self.checkpoint_data.get('all_normalized_term', []))
        # Fix: No longer calculate "failed" because term and entries are not the same unit
        
        # Calculate elapsed time (ensure start_time is not None)
        if self.stats['start_time'] is not None and self.stats['end_time'] is not None:
            elapsed_time = self.stats['end_time'] - self.stats['start_time']
        else:
            elapsed_time = 0.0
        
        logger.info(f"Multi-concurrent extraction completed, obtained {len(all_term)} final term pairs")
        logger.info(f"Processing statistics: {self.stats.get('total_entries', 0)} entries → {self.stats.get('total_term', 0)} final term, elapsed time: {elapsed_time:.2f} seconds")
        logger.info(f"Terms by stage: Extraction {self.stats.get('extracted_term', 0)} → Filtered {self.stats.get('filtered_term', 0)} → Normalized {self.stats.get('normalized_term', 0)} → Final {self.stats.get('total_term', 0)}")
        
        return {
            'final_term': all_term,
            'extracted_term': self.checkpoint_data.get('all_extracted_term', []),
            'filtered_term': self.checkpoint_data.get('all_filtered_term', []),
            'normalized_term': self.checkpoint_data.get('all_normalized_term', []),
            'standardized_term': self.checkpoint_data.get('all_standardized_term', []),
            'stats': self.stats
        }
    
    async def _process_batches_concurrent(self, batches: List[List[Dict[str, Any]]]) -> List[Dict[str, Any]]:
        """Use async concurrent batch processing"""
        async def process_batch_with_semaphore(batch, batch_index):
            async with self.semaphore:
                return await self._process_single_batch_async(batch, batch_index)
        
        # Create tasks with progress bar
        tasks = [process_batch_with_semaphore(batch, i) for i, batch in enumerate(batches)]
        
        # Use tqdm to display progress
        results = []
        with tqdm(total=len(tasks), desc="Concurrent batch processing", unit="batch") as pbar:
            for coro in asyncio.as_completed(tasks):
                result = await coro
                results.append(result)
                pbar.update(1)
                pbar.set_postfix({
                    'completed': len(results),
                    'total batches': len(tasks)
                })
        
        # Filter exception results
        valid_results = []
        for i, result in enumerate(results):
            if isinstance(result, Exception):
                logger.error(f"batch {i} processing failed: {result}")
            else:
                valid_results.append(result)
        
        return valid_results
    
    async def _process_batches_concurrent_with_checkpoint(self, remaining_batches: List[Tuple[int, List[Dict[str, Any]]]]) -> List[Dict[str, Any]]:
        """Use async concurrent batch processing (with checkpoint)"""
        async def process_batch_with_semaphore(batch_info):
            batch_index, batch = batch_info
            async with self.semaphore:
                result = await self._process_single_batch_async(batch, batch_index)
                # Save to checkpoint immediately
                self.add_batch_result(result, batch_index)
                return result
        
        # Create tasks with progress bar
        tasks = [process_batch_with_semaphore(batch_info) for batch_info in remaining_batches]
        
        # Use tqdm to display progress
        results = []
        with tqdm(total=len(tasks), desc="Concurrent batch processing", unit="batch") as pbar:
            for coro in asyncio.as_completed(tasks):
                result = await coro
                results.append(result)
                pbar.update(1)
                
                # Save checkpoint periodically
                if len(results) % self.config.save_interval == 0:
                    self.save_checkpoint()
                
                pbar.set_postfix({
                    'completed': len(results),
                    'total batches': len(tasks),
                    'total term': len(self.checkpoint_data['all_term'])
                })
        
        # Final checkpoint save
        self.save_checkpoint()
        
        # Filter exception results
        valid_results = []
        for i, result in enumerate(results):
            if isinstance(result, Exception):
                logger.error(f"batch {i} processing failed: {result}")
            else:
                valid_results.append(result)
        
        return valid_results

    def _term_to_dict(self, term: Any) -> Dict[str, Any]:
        """Convert term object or dictionary to unified dictionary format."""
        if isinstance(term, dict):
            return term
        # Object, use __dict__
        return getattr(term, '__dict__', {})

    def _get_attr(self, term: Any, key: str, default: Any = '') -> Any:
        """Safely get field from term object or dictionary."""
        if isinstance(term, dict):
            return term.get(key, default)
        return getattr(term, key, default)

    async def _stage1_extract_concurrent(self, remaining_batches: List[Tuple[int, List[Dict[str, Any]]]]) -> List[List[Dict[str, Any]]]:
        """Stage 1: Execute term extraction on all remaining batches, parallel execution, periodic checkpoint saving."""
        async def run_extract(batch_index: int, batch: List[Dict[str, Any]]):
            async with self.semaphore:
                extracted = await self._extract_term_batch(batch)
                # ❌ Do not save to checkpoint here! Will cause duplicates
                # Should save and deduplicate after all batches complete
                self.checkpoint_data['processed_batches'].append(batch_index)
                self.stats['processed_batches'] += 1
                return extracted

        tasks = [run_extract(i, batch) for i, batch in remaining_batches]
        results: List[List[Dict[str, Any]]] = []
        total_extracted = 0
        
        with tqdm(total=len(tasks), desc="Stage 1: Term extraction", unit="batch") as pbar:
            completed = 0
            for coro in asyncio.as_completed(tasks):
                res = await coro
                results.append(res)
                total_extracted += len(res)
                completed += 1
                pbar.update(1)
                pbar.set_postfix({'completed': completed, 'total term': total_extracted})
        
        # After all batches complete, save results uniformly (deduplicate)
        all_extracted = []
        seen_pairs = set()
        for batch_result in results:
            for term in batch_result:
                term_dict = self._term_to_dict(term)
                pair = (term_dict.get('source_term', ''), term_dict.get('target_term', ''))
                if pair not in seen_pairs:
                    seen_pairs.add(pair)
                    all_extracted.append(term_dict)
        
        logger.info(f"Extraction deduplication: {total_extracted} -> {len(all_extracted)} term")
        
        # Save to checkpoint
        self.checkpoint_data['all_extracted_term'] = all_extracted
        self.save_checkpoint()
        
        return results

    async def _stage2_quality_concurrent(self, batches: List[List[Dict[str, Any]]], extracted_term_all: List[Any]) -> List[List[Dict[str, Any]]]:
        """Stage 2: Execute quality check on all term by fixed batch size, parallel execution, periodic checkpoint saving.
        
        Note: Process all term by fixed batch size here, ignoring entry boundaries to ensure uniform batch sizes.
        """
        if not extracted_term_all:
            return []
        
        # Convert all term to dict format
        term_dicts = [self._term_to_dict(t) for t in extracted_term_all]
        
        # Split into chunks by fixed batch size
        quality_batch_size = self.config.quality_check_batch_size
        term_chunks: List[List[Dict[str, Any]]] = [
            term_dicts[i:i + quality_batch_size] 
            for i in range(0, len(term_dicts), quality_batch_size)
        ]
        
        logger.info(f"Quality check: Total {len(term_dicts)} term, split into {len(term_chunks)} batches, {quality_batch_size} term per batch")
        
        async def run_qc_batch(chunk_index: int, chunk: List[Dict[str, Any]]):
            """Perform quality check on a batch of term"""
            async with self.semaphore:
                # Since term come from different entries, we need to find corresponding original text for each term
                # Use a simplified strategy here: use term context or representative text from entire corpus
                # More precise method would be to find corresponding entry text for each term separately, but more complex
                
                # Collect all entry IDs involved in this batch of term
                entry_ids = set()
                for term in chunk:
                    entry_id = term.get('entry_id', '')
                    if entry_id:
                        entry_ids.add(entry_id)
                
                # Build a comprehensive context text (from related entries)
                source_context = ""
                target_context = ""
                
                # Find corresponding entries from batches
                for batch in batches:
                    for entry in batch:
                        if entry.get('id', '') in entry_ids:
                            source_context += entry.get(self.src_lang, '') + " "
                            target_context += entry.get(self.tgt_lang, '') + " "
                            if len(source_context) > 5000:  # Limit context length
                                break
                    if len(source_context) > 5000:
                        break
                
                # If no context found, use default text
                if not source_context:
                    source_context = "Legal text"
                    target_context = "Legal text"
                
                quality_check_agent = BilingualTermQualityCheckAgent(locale='zh')
                quality_input_data = {
                    'terms': chunk,  # Fixed: changed 'term' to 'terms' to match agent expectation
                    'source_text': source_context[:5000],  # Limit length
                    'target_text': target_context[:5000],
                    'src_lang': 'zh',
                    'tgt_lang': 'en',
                    'batch_mode': True,
                    'batch_size': quality_batch_size
                }
                
                filtered_term = await quality_check_agent.run(quality_input_data, None)
                filtered_dicts = [self._term_to_dict(t) for t in (filtered_term or [])]
                
                # ❌ Do not save to checkpoint here! Will cause duplicates
                # Should save after all batches complete
                
                logger.info(f"Quality check batch {chunk_index+1}/{len(term_chunks)}: "
                          f"Processing {len(chunk)} term，passed {len(filtered_dicts)} per batch")
                
                return filtered_dicts
        
        tasks = [run_qc_batch(i, chunk) for i, chunk in enumerate(term_chunks)]
        results: List[List[Dict[str, Any]]] = []
        
        with tqdm(total=len(tasks), desc="Stage 2: Quality check", unit="batch") as pbar:
            completed = 0
            total_filtered = 0
            for coro in asyncio.as_completed(tasks):
                res = await coro
                results.append(res)
                total_filtered += len(res)
                completed += 1
                pbar.update(1)
                pbar.set_postfix({'completed': completed, 'total filtered': total_filtered})
        
        # After all batches complete, save results uniformly (deduplicate)
        all_filtered = []
        seen_pairs = set()
        for batch_result in results:
            for term in batch_result:
                pair = (term.get('source_term', ''), term.get('target_term', ''))
                if pair not in seen_pairs:
                    seen_pairs.add(pair)
                    all_filtered.append(term)
        
        logger.info(f"Quality check deduplication: {total_filtered} -> {len(all_filtered)} term")
        
        # Save to checkpoint
        self.checkpoint_data['all_filtered_term'] = all_filtered
        self.save_checkpoint()
        
        return results

    async def _stage3_normalize_concurrent(self, filtered_term_all: List[Any]) -> List[List[Dict[str, Any]]]:
        """Stage 3: Normalize all filtered term in chunks, parallel execution, periodic checkpoint saving."""
        if not filtered_term_all:
            return []
        # Convert filtered term (may be objects or dicts) to unified dict list
        filtered_dicts: List[Dict[str, Any]] = [self._term_to_dict(t) for t in filtered_term_all]
        
        # Sort by source_term to group identical or similar term together for normalization to identify duplicates and variants
        logger.info(f"Pre-normalization sorting: Sorting {len(filtered_dicts)} term")
        filtered_dicts.sort(key=lambda x: x.get('source_term', ''))
        logger.info(f"Sorting completed")
        
        # Deduplicate: Keep top 3 highest quality term (deduplicate by source_term + target_term combination)
        logger.info(f"Pre-normalization deduplication: Starting deduplication, keeping top 3 highest quality versions for each term pair...")
        term_groups = {}
        for term in filtered_dicts:
            key = (term.get('source_term', ''), term.get('target_term', ''))
            if key not in term_groups:
                term_groups[key] = []
            term_groups[key].append(term)
        
        # Sort each group by quality score and keep top 3
        top_term = []
        for key, term in term_groups.items():
            # Sort by quality score descending
            term.sort(key=lambda x: x.get('quality_score', 0.0), reverse=True)
            # Keep top 3
            top_3 = term[:3]
            # Merge entry_id metadata
            all_entry_ids = set()
            for term in term:
                entry_id = term.get('entry_id', '')
                if entry_id:
                    # Ensure entry_id is string
                    entry_id_str = str(entry_id)
                    all_entry_ids.update(entry_id_str.split(','))
            
            # Add merged entry_id to top 3 term
            for term in top_3:
                term['entry_id'] = ','.join(filter(None, all_entry_ids))
            
            top_term.extend(top_3)
        
        original_count = len([self._term_to_dict(t) for t in filtered_term_all])
        filtered_dicts = top_term
        logger.info(f"Deduplication completed:{len(filtered_dicts)} term（Keep top 3, original: {original_count} per batch，unique term pairs:{len(term_groups)} per batch）")
        
        # Sort again (maintain order)
        filtered_dicts.sort(key=lambda x: x.get('source_term', ''))
        
        # Split into chunks - use normalization batch size
        chunk_size = max(1, self.config.normalization_batch_size)
        chunks: List[List[Dict[str, Any]]] = [
            filtered_dicts[i:i + chunk_size] for i in range(0, len(filtered_dicts), chunk_size)
        ]

        async def run_norm(chunk: List[Dict[str, Any]]):
            async with self.semaphore:
                normalization_agent = TermNormalizationAgent(locale='zh')
                normalization_input_data = {
                    'terms': chunk,  # Fixed: changed 'term' to 'terms' to match agent expectation
                    'src_lang': self.src_lang,
                    'tgt_lang': self.tgt_lang,
                    'batch_size': self.config.normalization_batch_size
                }
                normalized_term = await normalization_agent.run(normalization_input_data, None)
                # Save as dict and backfill entry metadata from input chunk
                normalized_list = (normalized_term or [])
                normalized_dicts = [self._term_to_dict(t) for t in normalized_list]
                
                # Create metadata mapping of original term (based on source_term + target_term)
                metadata_map = {}
                for src_term in chunk:
                    key = (src_term.get('source_term', ''), src_term.get('target_term', ''))
                    metadata_map[key] = {
                        'entry_id': src_term.get('entry_id', ''),
                        'law': src_term.get('law', ''),
                        'domain': src_term.get('domain', ''),
                        'year': src_term.get('year', '')
                    }
                
                # Match metadata based on term content, not index
                for nt in normalized_dicts:
                    key = (nt.get('source_term', ''), nt.get('target_term', ''))
                    if key in metadata_map:
                        meta = metadata_map[key]
                        for k, v in meta.items():
                            if k not in nt or not nt[k]:
                                nt[k] = v
                    else:
                        # If exact match not found, log warning
                        logger.debug(f"Normalized term cannot find metadata match: {key}")
                
                # ❌ Do not save to checkpoint here! Will cause duplicates
                # Should save after all batches complete
                return normalized_dicts

        tasks = [run_norm(chunk) for chunk in chunks]
        results: List[List[Dict[str, Any]]] = []
        total_normalized = 0
        
        with tqdm(total=len(tasks), desc="Stage 3: Normalization", unit="chunks") as pbar:
            completed = 0
            for coro in asyncio.as_completed(tasks):
                res = await coro
                results.append(res)
                total_normalized += len(res)
                completed += 1
                pbar.update(1)
                pbar.set_postfix({'completed': completed, 'total normalized': total_normalized})
        
        # After all batches complete, save results uniformly (deduplicate)
        all_normalized = []
        seen_pairs = set()
        for batch_result in results:
            for term in batch_result:
                pair = (term.get('source_term', ''), term.get('target_term', ''))
                if pair not in seen_pairs:
                    seen_pairs.add(pair)
                    all_normalized.append(term)
        
        logger.info(f"Normalization deduplication: {total_normalized} -> {len(all_normalized)} term")
        
        # Save to checkpoint
        self.checkpoint_data['all_normalized_term'] = all_normalized
        self.save_checkpoint()
        
        return results
    
    async def _stage4_standardize(self, normalized_term_all: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        """Stage 4: Standardization (deduplication, sorting, cleaning) - pure logic processing, no LLM calls"""
        if not normalized_term_all:
            return []
        
        logger.info(f"Standardization: Starting to process {len(normalized_term_all)} normalized term")
        
        try:
            # Create standardization agent
            standardization_agent = BilingualTermStandardizationAgent(locale='zh')
            
            # Execute standardization (pure logic processing)
            standardization_input_data = {
                'terms': normalized_term_all,  # Fixed: changed 'term' to 'terms' to match agent expectation
                'max_targets_per_source': self.config.max_targets_per_source,
                'confidence_weight': self.config.confidence_weight,
                'quality_weight': self.config.quality_weight
            }
            
            standardized_term = await standardization_agent.execute(standardization_input_data, None)
            
            logger.info(f"Standardization completed:{len(normalized_term_all)} -> {len(standardized_term)} term")
            logger.info(f"Reduced by {len(normalized_term_all) - len(standardized_term)} term "
                       f"({(1 - len(standardized_term)/len(normalized_term_all))*100:.2f}% compression rate)")
            
            return standardized_term
            
        except Exception as e:
            logger.error(f"Error during standardization: {e}")
            # If standardization fails, return original normalization results
            logger.warning("Standardization failed, using normalization results")
            return normalized_term_all
    
    
    async def _process_single_batch_async(self, batch: List[Dict[str, Any]], batch_index: int = 0) -> Dict[str, Any]:
        """Async process single batch - stage-by-stage processing"""
        batch_results = {
            'extracted_term': [],
            'filtered_term': [],
            'normalized_term': [],
            'final_term': []
        }
        
        # Stage 1: batch term extraction
        logger.info(f"batch {batch_index}: Starting term extraction...")
        extracted_term_all = await self._extract_term_batch(batch)
        batch_results['extracted_term'] = [term.__dict__ for term in extracted_term_all]
        logger.info(f"batch {batch_index}: Extracted {len(extracted_term_all)} term")
        
        # Stage 2: batch quality check
        if extracted_term_all:
            logger.info(f"batch {batch_index}: Starting quality check...")
            filtered_term_all = await self._quality_check_batch(extracted_term_all, batch)
            batch_results['filtered_term'] = [term.__dict__ for term in filtered_term_all]
            logger.info(f"batch {batch_index}: Remaining after quality check: {len(filtered_term_all)} term")
        else:
            filtered_term_all = []
            batch_results['filtered_term'] = []
        
        # Stage 3: batch normalization
        if filtered_term_all:
            logger.info(f"batch {batch_index}: Starting normalization...")
            normalized_term_all = await self._normalize_term_batch(filtered_term_all)
            batch_results['normalized_term'] = [term.__dict__ for term in normalized_term_all]
            logger.info(f"batch {batch_index}: obtained after normalization {len(normalized_term_all)} term")
            
            # Convert to final format
            final_term = []
            for term in normalized_term_all:
                term_dict = {
                    'source_term': term.source_term,
                    'target_term': term.target_term,
                    'normalized_source': term.normalized_source,
                    'normalized_target': term.normalized_target,
                    'confidence': term.confidence,
                    'category': term.category,
                    'source_context': term.source_context,
                    'target_context': term.target_context,
                    'quality_score': term.quality_score,
                    'is_valid': term.is_valid,
                    'normalization_notes': term.normalization_notes
                }
                final_term.append(term_dict)
            batch_results['final_term'] = final_term
        else:
            batch_results['normalized_term'] = []
            batch_results['final_term'] = []
        
        return batch_results
    
    async def _extract_term_batch(self, batch: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        """Stage 1: batch term extraction; returns dict list with entry metadata.
        
        Use extraction_batch_size to control number of article pairs per batch.
        Example: extraction_batch_size=3 means sending 3 article pairs to LLM for batch extraction each time.
        """
        all_extracted_term: List[Dict[str, Any]] = []
        
        # Group entries by extraction_batch_size
        extraction_batch_size = self.config.extraction_batch_size
        
        # Split entries into mini-batches by extraction_batch_size
        for i in range(0, len(batch), extraction_batch_size):
            mini_batch = batch[i:i + extraction_batch_size]
            
            # Filter out entries with empty text
            # Support two formats: (1) Using language code keys zh/en/ja (2) Using generic keys source/target
            valid_entries = []
            for entry in mini_batch:
                src_text = entry.get(self.src_lang, '') or entry.get('source', '')
                tgt_text = entry.get(self.tgt_lang, '') or entry.get('target', '')
                if src_text and tgt_text:
                    # Ensure entry contains standardized language keys (for subsequent processing)
                    entry[self.src_lang] = src_text
                    entry[self.tgt_lang] = tgt_text
                    valid_entries.append(entry)
            
            if not valid_entries:
                continue
            
            try:
                bi_extract_agent = BilingualTermExtractAgent(locale='zh')
                
                # If only one entry, use single mode
                if len(valid_entries) == 1:
                    entry = valid_entries[0]
                    extract_input_data = {
                        'source_text': entry.get(self.src_lang, ''),
                        'target_text': entry.get(self.tgt_lang, ''),
                        'src_lang': self.src_lang,
                        'tgt_lang': self.tgt_lang
                    }
                    extracted_term = await bi_extract_agent.run(extract_input_data, None)
                    
                    # Convert to dict with entry metadata
                    for t in (extracted_term or []):
                        t_dict = self._term_to_dict(t)
                        t_dict.setdefault('entry_id', entry.get('id', ''))
                        t_dict.setdefault('law', entry.get('law', ''))
                        t_dict.setdefault('domain', entry.get('domain', ''))
                        t_dict.setdefault('year', entry.get('year', ''))
                        all_extracted_term.append(t_dict)
                
                # If multiple entries, use batch mode
                else:
                    text_pairs = [
                        {
                            'source_text': entry.get(self.src_lang, ''),
                            'target_text': entry.get(self.tgt_lang, ''),
                            'entry_id': entry.get('id', ''),
                            'law': entry.get('law', ''),
                            'domain': entry.get('domain', ''),
                            'year': entry.get('year', '')
                        }
                        for entry in valid_entries
                    ]
                    
                    extract_input_data = {
                        'text_pairs': text_pairs,
                        'src_lang': self.src_lang,
                        'tgt_lang': self.tgt_lang,
                        'batch_mode': True,
                        'batch_size': extraction_batch_size
                    }
                    
                    extracted_term = await bi_extract_agent.run(extract_input_data, None)
                    
                    # Convert to dict with entry metadata
                    # Since batch extraction cannot precisely match each term to specific entry, use simple strategy:
                    # Add first entry metadata to all extracted term (can be optimized via text matching later)
                    for t in (extracted_term or []):
                        t_dict = self._term_to_dict(t)
                        # Try to match specific entry based on source term
                        matched_entry = None
                        source_term = t_dict.get('source_term', '')
                        for entry in valid_entries:
                            if source_term in entry.get(self.src_lang, ''):
                                matched_entry = entry
                                break
                        
                        # If no match, use first entry metadata
                        if not matched_entry:
                            matched_entry = valid_entries[0]
                        
                        t_dict.setdefault('entry_id', matched_entry.get('id', ''))
                        t_dict.setdefault('law', matched_entry.get('law', ''))
                        t_dict.setdefault('domain', matched_entry.get('domain', ''))
                        t_dict.setdefault('year', matched_entry.get('year', ''))
                        all_extracted_term.append(t_dict)
                    
                    logger.info(f"batch extraction: Processing {len(valid_entries)} article pairs, extracted {len(extracted_term or [])} term")
                
            except Exception as e:
                logger.error(f"Error during batch term extraction: {e}")
                continue
        
        return all_extracted_term
    
    async def _quality_check_batch(self, extracted_term: List, batch: List[Dict[str, Any]]) -> List:
        """Stage 2: batch quality check
        
        Use quality_check_batch_size to control number of term per batch check.
        Example: quality_check_batch_size=10 means sending up to 10 term to LLM for quality check each time.
        """
        if not extracted_term:
            return []
        
        all_filtered_term = []
        quality_check_batch_size = self.config.quality_check_batch_size
        
        # Group term by entry
        term_by_entry = {}
        for i, entry in enumerate(batch):
            src_text = entry.get(self.src_lang, '')
            tgt_text = entry.get(self.tgt_lang, '')
            if src_text and tgt_text:
                # Find term belonging to this entry
                entry_term = [term for term in extracted_term if self._is_term_from_entry(term, src_text, tgt_text)]
                term_by_entry[i] = entry_term
        
        # Perform quality check on term for each entry
        for i, entry_term in term_by_entry.items():
            if not entry_term:
                continue
                
            try:
                entry = batch[i]
                src_text = entry.get(self.src_lang, '')
                tgt_text = entry.get(self.tgt_lang, '')
                
                # If number of term exceeds quality_check_batch_size, process in batches
                for batch_start in range(0, len(entry_term), quality_check_batch_size):
                    batch_end = min(batch_start + quality_check_batch_size, len(entry_term))
                    term_chunk = entry_term[batch_start:batch_end]
                    
                    quality_check_agent = BilingualTermQualityCheckAgent(locale='zh')
                    quality_input_data = {
                        'term': [term.__dict__ if hasattr(term, '__dict__') else term for term in term_chunk],
                        'source_text': src_text,
                        'target_text': tgt_text,
                        'src_lang': self.src_lang,
                        'tgt_lang': self.tgt_lang,
                        'batch_mode': True,  # Enable batch mode
                        'batch_size': quality_check_batch_size
                    }
                    filtered_term = await quality_check_agent.run(quality_input_data, None)
                    
                    if filtered_term:
                        all_filtered_term.extend(filtered_term)
                    
                    logger.info(f"Quality check: Entry {i+1}, batch {batch_start//quality_check_batch_size + 1}, "
                              f"checking {len(term_chunk)} term，passed {len(filtered_term or [])} per batch")
                
            except Exception as e:
                logger.error(f"Error during quality check: {e}")
                continue
        
        return all_filtered_term
    
    async def _normalize_term_batch(self, filtered_term: List) -> List:
        """Stage 3: batch normalization
        
        Use normalization_batch_size to control number of term per normalization.
        Example: normalization_batch_size=10 means sending up to 10 term to LLM for normalization each time.
        Note: This method is usually called by _stage3_normalize_concurrent, which already handles chunking.
        """
        if not filtered_term:
            return []
        
        try:
            normalization_agent = TermNormalizationAgent(locale='zh')
            normalization_input_data = {
                'term': [term.__dict__ for term in filtered_term],
                'src_lang': 'zh',
                'tgt_lang': 'en',
                'batch_size': self.config.normalization_batch_size
            }
            normalized_term = await normalization_agent.run(normalization_input_data, None)
            logger.info(f"Normalization: Processing {len(filtered_term)} term，obtained after normalization {len(normalized_term or [])} per batch")
            return normalized_term or []
        except Exception as e:
            logger.error(f"Error during normalization: {e}")
            return []
    
    def _is_term_from_entry(self, term, zh_text: str, en_text: str) -> bool:
        """Determine if term comes from specific entry"""
        t = self._term_to_dict(term)
        source = t.get('source_term', '')
        target = t.get('target_term', '')
        return (source in zh_text and target in en_text)
    
    
    def save_results(self, results: Dict[str, Any], output_file: str):
        """Save results to file"""
        timestamp = int(time.time())
        
        # Ensure output file path is a file, not a directory
        output_path = Path(output_file)
        if output_path.is_dir():
            # If directory, create default filename in directory
            output_file = output_path / f"extracted_term_{timestamp}.json"
            logger.warning(f"Output path is directory, using default filename: {output_file}")
        
        # Ensure output directory exists
        output_path = Path(output_file)
        output_path.parent.mkdir(parents=True, exist_ok=True)
        
        base_name = output_path.stem
        base_dir = output_path.parent
        
        # Save final results
        final_data = {
            'metadata': {
                'source_lang': 'zh',
                'target_lang': 'en',
                'total_term': len(results['final_term']),
                'extraction_time': timestamp,
                'processing_config': {
                    'batch_size': self.config.batch_size,
                    'max_concurrent_requests': self.config.max_concurrent_requests,
                    'save_interval': self.config.save_interval
                },
                'stats': results['stats']
            },
            'term': results['final_term']
        }
        
        with open(output_file, 'w', encoding='utf-8') as f:
            json.dump(final_data, f, ensure_ascii=False, indent=2)
        
        logger.info(f"Results saved to: {output_file}")
        
        # Save intermediate results
        self._save_intermediate_results(results, base_dir, base_name, timestamp)
    
    def _save_intermediate_results(self, results: Dict[str, Any], base_dir: Path, base_name: str, timestamp: int):
        """Save intermediate results"""
        stages = [
            ('extracted', results['extracted_term']),
            ('filtered', results['filtered_term']),
            ('normalized', results['normalized_term']),
            ('standardized', results['standardized_term'])
        ]
        
        for stage_name, term in stages:
            stage_file = base_dir / f"{base_name}_{stage_name}_{timestamp}.json"
            stage_data = {
                'metadata': {
                    'source_lang': 'zh',
                    'target_lang': 'en',
                    'total_term': len(term),
                    'stage': stage_name,
                    'extraction_time': timestamp
                },
                'term': term
            }
            
            with open(stage_file, 'w', encoding='utf-8') as f:
                json.dump(stage_data, f, ensure_ascii=False, indent=2)
            
            logger.info(f"{stage_name} Results saved to: {stage_file}")




async def main():
    """Main function"""
    parser = argparse.ArgumentParser(
        description='Multi-concurrent bilingual term extraction (4-stage pipeline)',
        epilog="""
Usage examples:
  # Run all stages completely
  python %(prog)s input.json -o output.json
  
  # Re-run starting from stage 3 (normalization)
  python %(prog)s input.json -o output.json --start-from-stage 3
  
  # Start from stage 4 (standardization) with adjusted parameters
  python %(prog)s input.json -o output.json --start-from-stage 4 --max-targets-per-source 5

Stage descriptions:
  Stage 1: Term Extraction
  Stage 2: Quality Check
  Stage 3: Normalization - normalize target_term grouped by source_term
  Stage 4: Standardization - deduplication, sorting, cleaning
        """,
        formatter_class=argparse.RawDescriptionHelpFormatter
    )
    parser.add_argument('input_file', help='Input bilingual JSON file path')
    parser.add_argument('-o', '--output', help='Output file path', default='concurrent_extracted_term_v2.json')
    parser.add_argument('--max-entries', type=int, help='Maximum number of entries to process (for testing)')
    parser.add_argument('--batch-size', type=int, default=20, help='batch processing size')
    parser.add_argument('--max-concurrent', type=int, default=40, help='Maximum concurrent LLM requests')
    parser.add_argument('--timeout', type=int, default=300, help='Timeout in seconds')
    parser.add_argument('--save-interval', type=int, default=5, help='Save checkpoint every N batches')
    parser.add_argument('--checkpoint', type=str, default='outputs/checkpoint.json', help='Checkpoint file path (default: outputs/)')
    parser.add_argument('--stage-dir', type=str, help='Stage snapshot output directory (default: same as checkpoint)')
    parser.add_argument('--no-resume', action='store_true', help='Do not resume from checkpoint, start fresh')
    parser.add_argument('--clean-checkpoint', action='store_true', help='Clean checkpoint file')
    parser.add_argument('--start-from-stage', type=int, choices=[1, 2, 3, 4], 
                       help='Start from specified stage (1=extraction, 2=quality check, 3=normalization, 4=standardization), will clear this stage and subsequent data')
    
    # Agent batch processing configuration
    parser.add_argument('--extraction-batch-size', type=int, default=3, help='Term extraction batch size (number of article pairs per batch, recommended: 3-5, default: 5)')
    parser.add_argument('--quality-check-batch-size', type=int, default=30, help='Quality check batch size (number of term per batch, recommended: 30-50, default: 50)')
    parser.add_argument('--normalization-batch-size', type=int, default=50, help='Normalization batch size (number of term per batch, recommended: 30-50, default: 50)')
    
    # Standardization configuration
    parser.add_argument('--max-targets-per-source', type=int, default=3, help='Max target term to keep per source term (default: 3)')
    parser.add_argument('--confidence-weight', type=float, default=0.4, help='Confidence weight (default: 0.4)')
    parser.add_argument('--quality-weight', type=float, default=0.6, help='Quality score weight (default: 0.6)')
    
    args = parser.parse_args()
    
    # Check if input file exists
    if not os.path.exists(args.input_file):
        logger.error(f"Input file does not exist: {args.input_file}")
        return
    
    # Clean checkpoint file
    if args.clean_checkpoint:
        if os.path.exists(args.checkpoint):
            os.remove(args.checkpoint)
            logger.info(f"Checkpoint file cleaned: {args.checkpoint}")
        else:
            logger.info("Checkpoint file does not exist, no need to clean")
        return
    
    # Create concurrent configuration
    config = ConcurrentConfig(
        batch_size=args.batch_size,
        timeout=args.timeout,
        max_concurrent_requests=args.max_concurrent,
        save_interval=args.save_interval,
        checkpoint_file=args.checkpoint,
        stage_dir=args.stage_dir,
        extraction_batch_size=args.extraction_batch_size,
        quality_check_batch_size=args.quality_check_batch_size,
        normalization_batch_size=args.normalization_batch_size,
        max_targets_per_source=args.max_targets_per_source,
        confidence_weight=args.confidence_weight,
        quality_weight=args.quality_weight
    )
    
    # Display configuration info
    logger.info(f"Agent batch processing configuration:")
    logger.info(f"  - Term extraction batch size: {config.extraction_batch_size} article pairs")
    logger.info(f"  - Quality check batch size: {config.quality_check_batch_size} term")
    logger.info(f"  - Normalization batch size: {config.normalization_batch_size} term")
    logger.info(f"Standardization configuration:")
    logger.info(f"  - Max target term per source: {config.max_targets_per_source} target term")
    logger.info(f"  - Confidence weight: {config.confidence_weight}")
    logger.info(f"  - Quality weight: {config.quality_weight}")
    
    # Create extractor
    extractor = ConcurrentBilingualTermExtractor(config)
    
    # Handle --start-from-stage parameter
    if args.start_from_stage:
        stage = args.start_from_stage
        logger.info(f"⚠️  Force starting from stage{stage}execution, will clear stage{stage}and subsequent data")
        
        # Try to load existing checkpoint
        if extractor.load_checkpoint():
            stage_names = {
                1: "Extraction",
                2: "Quality Check",
                3: "Normalization",
                4: "Standardization"
            }
            
            # Clear data based on starting stage
            if stage <= 1:
                extractor.checkpoint_data['processed_batches'] = []
                extractor.checkpoint_data['all_extracted_term'] = []
                extractor.checkpoint_data['all_filtered_term'] = []
                extractor.checkpoint_data['all_normalized_term'] = []
                extractor.checkpoint_data['all_standardized_term'] = []
                extractor.checkpoint_data['all_term'] = []
                logger.info(f"  ✓ Cleared all stage data")
            elif stage == 2:
                extractor.checkpoint_data['all_filtered_term'] = []
                extractor.checkpoint_data['all_normalized_term'] = []
                extractor.checkpoint_data['all_standardized_term'] = []
                extractor.checkpoint_data['all_term'] = []
                logger.info(f"  ✓ Kept stage 1, cleared stages 2/3/4")
            elif stage == 3:
                extractor.checkpoint_data['all_normalized_term'] = []
                extractor.checkpoint_data['all_standardized_term'] = []
                extractor.checkpoint_data['all_term'] = []
                logger.info(f"  ✓ Kept stages 1/2, cleared stages 3/4")
            elif stage == 4:
                extractor.checkpoint_data['all_standardized_term'] = []
                extractor.checkpoint_data['all_term'] = []
                logger.info(f"  ✓ Kept stages 1/2/3, cleared stage 4")
            
            # Save modified checkpoint
            extractor.save_checkpoint()
            logger.info(f"  ✓ Checkpoint updated")
        else:
            logger.warning(f"  Checkpoint file not found, will start from beginning")
    
    try:
        # Extract term (supports resume from checkpoint)
        resume = not args.no_resume
        results = await extractor.extract_term_concurrent(args.input_file, args.max_entries, resume=resume)
        
        # Save results
        extractor.save_results(results, args.output)
        
        # Keep checkpoint and stage snapshots for later viewing/manual summary
        logger.info(f"Checkpoint and stage snapshots saved to: {args.checkpoint} （stage directory: {args.stage_dir or Path(args.checkpoint).parent}）")
        
        # Display statistics
        stats = results['stats']
        logger.info(f"Multi-concurrent extraction completed!")
        logger.info(f"Input entries: {stats.get('total_entries', stats.get('total_processed', 0))}")
        logger.info(f"Final term: {stats.get('total_term', stats.get('successful', 0))}")
        logger.info(f"\nStage statistics:")
        logger.info(f"  Stage 1 (Term Extraction): {stats.get('extracted_term', 0)} per batch")
        logger.info(f"  Stage 2 (Quality Check): {stats.get('filtered_term', 0)} per batch")
        logger.info(f"  Stage 3 (Normalization): {stats.get('normalized_term', 0)} per batch")
        logger.info(f"  Stage 4 (Standardization): {stats.get('total_term', 0)} per batch(Final)")
        
        # Calculate elapsed time and speed (ensure time values are not None)
        if stats.get('start_time') and stats.get('end_time'):
            elapsed = stats['end_time'] - stats['start_time']
            logger.info(f"Total elapsed time: {elapsed:.2f} seconds")
            total_entries = stats.get('total_entries', stats.get('total_processed', 0))
            if elapsed > 0 and total_entries > 0:
                logger.info(f"Average speed: {total_entries / elapsed:.2f} entries/sec")
                logger.info(f"Term generation speed: {stats.get('total_term', 0) / elapsed:.2f} term/sec")
        else:
            logger.info(f"Total elapsed time: N/A (resumed from checkpoint)")
        
        # Display first few term as examples
        if results['final_term']:
            logger.info("Top 5 final term examples:")
            for i, term in enumerate(results['final_term'][:5]):
                normalized_info = f" (Normalization: {term.get('normalized_source', 'N/A')} -> {term.get('normalized_target', 'N/A')})" if 'normalized_source' in term else ""
                logger.info(f"  {i+1}. {term['source_term']} -> {term['target_term']}{normalized_info} (confidence: {term['confidence']:.2f}, quality score: {term.get('quality_score', 'N/A')})")
    
    except KeyboardInterrupt:
        logger.info("User interrupted, Checkpoint saved, can resume later with --resume")
        extractor.save_checkpoint()
    except Exception as e:
        logger.error(f"Error during multi-concurrent extraction: {e}")
        logger.info("Checkpoint saved, can resume later with --resume")
        extractor.save_checkpoint()
        raise


if __name__ == '__main__':
    asyncio.run(main())