RSK World - Named Entity Recognition Dataset - Project Files Browser | RSK World

example_usage.py batch_process.py api_server.py

scripts/batch_process.py

"""
Named Entity Recognition Dataset - Batch Processing Script
Project: Named Entity Recognition Dataset
Author: RSK World
Website: https://rskworld.in
Email: help@rskworld.in
Phone: +91 93305 39277
Description: Advanced batch processing script for NER dataset with parallel processing and progress tracking
"""

import json
import csv
import os
from typing import List, Dict, Any, Optional
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
from tqdm import tqdm
import time
from collections import Counter
import pandas as pd


def load_dataset(file_path: str) -> List[Dict[str, Any]]:
    """Load NER dataset from JSON file."""
    with open(file_path, 'r', encoding='utf-8') as f:
        return json.load(f)


def process_single_sample(sample: Dict[str, Any]) -> Dict[str, Any]:
    """
    Process a single sample and extract statistics.
    
    Args:
        sample: Single data sample with text and entities
        
    Returns:
        Dictionary with processed statistics
    """
    text = sample.get('text', '')
    entities = sample.get('entities', [])
    
    return {
        'id': sample.get('id', 0),
        'text_length': len(text),
        'word_count': len(text.split()),
        'entity_count': len(entities),
        'entity_types': [e.get('label', '') for e in entities],
        'avg_entity_length': sum(len(e.get('text', '')) for e in entities) / len(entities) if entities else 0,
        'entities': entities
    }


def batch_process_dataset(data: List[Dict[str, Any]], 
                         batch_size: int = 10,
                         use_multiprocessing: bool = False,
                         max_workers: int = 4) -> List[Dict[str, Any]]:
    """
    Process dataset in batches with progress tracking.
    
    Args:
        data: List of data samples
        batch_size: Number of samples per batch
        use_multiprocessing: Use multiprocessing instead of threading
        max_workers: Maximum number of worker threads/processes
        
    Returns:
        List of processed samples
    """
    executor_class = ProcessPoolExecutor if use_multiprocessing else ThreadPoolExecutor
    
    results = []
    batches = [data[i:i + batch_size] for i in range(0, len(data), batch_size)]
    
    with executor_class(max_workers=max_workers) as executor:
        with tqdm(total=len(data), desc="Processing samples") as pbar:
            futures = []
            for batch in batches:
                future = executor.submit(process_batch, batch)
                futures.append(future)
            
            for future in futures:
                batch_results = future.result()
                results.extend(batch_results)
                pbar.update(len(batch_results))
    
    return results


def process_batch(batch: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """Process a batch of samples."""
    return [process_single_sample(sample) for sample in batch]


def extract_entity_statistics(processed_data: List[Dict[str, Any]]) -> Dict[str, Any]:
    """
    Extract comprehensive statistics from processed data.
    
    Args:
        processed_data: List of processed samples
        
    Returns:
        Dictionary with statistics
    """
    all_entity_types = []
    entity_type_counts = Counter()
    text_lengths = []
    word_counts = []
    entity_counts = []
    
    for sample in processed_data:
        text_lengths.append(sample['text_length'])
        word_counts.append(sample['word_count'])
        entity_counts.append(sample['entity_count'])
        all_entity_types.extend(sample['entity_types'])
        entity_type_counts.update(sample['entity_types'])
    
    return {
        'total_samples': len(processed_data),
        'total_entities': sum(entity_counts),
        'avg_text_length': sum(text_lengths) / len(text_lengths) if text_lengths else 0,
        'avg_word_count': sum(word_counts) / len(word_counts) if word_counts else 0,
        'avg_entities_per_sample': sum(entity_counts) / len(entity_counts) if entity_counts else 0,
        'entity_type_distribution': dict(entity_type_counts),
        'min_text_length': min(text_lengths) if text_lengths else 0,
        'max_text_length': max(text_lengths) if text_lengths else 0,
        'min_entities': min(entity_counts) if entity_counts else 0,
        'max_entities': max(entity_counts) if entity_counts else 0
    }


def export_statistics(stats: Dict[str, Any], output_file: str):
    """Export statistics to JSON file."""
    with open(output_file, 'w', encoding='utf-8') as f:
        json.dump(stats, f, indent=2, ensure_ascii=False)
    print(f"Statistics exported to {output_file}")


def export_processed_data(processed_data: List[Dict[str, Any]], output_file: str, format: str = 'json'):
    """
    Export processed data to file.
    
    Args:
        processed_data: List of processed samples
        output_file: Output file path
        format: Export format ('json' or 'csv')
    """
    if format == 'json':
        with open(output_file, 'w', encoding='utf-8') as f:
            json.dump(processed_data, f, indent=2, ensure_ascii=False)
    elif format == 'csv':
        rows = []
        for sample in processed_data:
            for entity in sample.get('entities', []):
                rows.append({
                    'id': sample['id'],
                    'text_length': sample['text_length'],
                    'word_count': sample['word_count'],
                    'entity_text': entity.get('text', ''),
                    'entity_type': entity.get('label', ''),
                    'start': entity.get('start', 0),
                    'end': entity.get('end', 0)
                })
        df = pd.DataFrame(rows)
        df.to_csv(output_file, index=False)
    
    print(f"Processed data exported to {output_file}")


if __name__ == '__main__':
    import argparse
    
    parser = argparse.ArgumentParser(description='Batch process NER dataset')
    parser.add_argument('--input', type=str, default='../dataset/train.json', help='Input JSON file')
    parser.add_argument('--output', type=str, default='processed_data.json', help='Output file')
    parser.add_argument('--batch-size', type=int, default=10, help='Batch size')
    parser.add_argument('--workers', type=int, default=4, help='Number of workers')
    parser.add_argument('--multiprocessing', action='store_true', help='Use multiprocessing')
    parser.add_argument('--stats', type=str, default='statistics.json', help='Statistics output file')
    parser.add_argument('--format', type=str, choices=['json', 'csv'], default='json', help='Output format')
    
    args = parser.parse_args()
    
    print("Loading dataset...")
    data = load_dataset(args.input)
    print(f"Loaded {len(data)} samples")
    
    print("\nProcessing dataset...")
    start_time = time.time()
    processed_data = batch_process_dataset(
        data, 
        batch_size=args.batch_size,
        use_multiprocessing=args.multiprocessing,
        max_workers=args.workers
    )
    elapsed_time = time.time() - start_time
    print(f"\nProcessing completed in {elapsed_time:.2f} seconds")
    
    print("\nExtracting statistics...")
    stats = extract_entity_statistics(processed_data)
    print("\nDataset Statistics:")
    print(f"  Total samples: {stats['total_samples']}")
    print(f"  Total entities: {stats['total_entities']}")
    print(f"  Average text length: {stats['avg_text_length']:.2f} characters")
    print(f"  Average word count: {stats['avg_word_count']:.2f} words")
    print(f"  Average entities per sample: {stats['avg_entities_per_sample']:.2f}")
    print(f"\nEntity Type Distribution:")
    for entity_type, count in sorted(stats['entity_type_distribution'].items()):
        print(f"  {entity_type}: {count}")
    
    print(f"\nExporting processed data to {args.output}...")
    export_processed_data(processed_data, args.output, format=args.format)
    
    print(f"Exporting statistics to {args.stats}...")
    export_statistics(stats, args.stats)
    
    print("\nBatch processing complete!")

217 lines•8.1 KB

python

scripts/api_server.py

Raw Download

"""
Named Entity Recognition Dataset - Flask API Server
Project: Named Entity Recognition Dataset
Author: RSK World
Website: https://rskworld.in
Email: help@rskworld.in
Phone: +91 93305 39277
Description: RESTful API server for NER dataset operations
"""

from flask import Flask, request, jsonify
from flask_cors import CORS
import json
import os
from typing import List, Dict, Any
import re
from datetime import datetime

app = Flask(__name__)
CORS(app)  # Enable CORS for all routes

# Load datasets
TRAIN_DATA = None
TEST_DATA = None


def load_datasets():
    """Load training and test datasets."""
    global TRAIN_DATA, TEST_DATA
    try:
        # Try relative path from scripts directory first
        train_path = os.path.join(os.path.dirname(__file__), '..', 'dataset', 'train.json')
        test_path = os.path.join(os.path.dirname(__file__), '..', 'dataset', 'test.json')
        train_path = os.path.normpath(train_path)
        test_path = os.path.normpath(test_path)
        
        # Try alternative paths
        if not os.path.exists(train_path):
            train_path = 'dataset/train.json'
        if not os.path.exists(test_path):
            test_path = 'dataset/test.json'
        
        with open(train_path, 'r', encoding='utf-8') as f:
            TRAIN_DATA = json.load(f)
        with open(test_path, 'r', encoding='utf-8') as f:
            TEST_DATA = json.load(f)
        print("Datasets loaded successfully")
    except FileNotFoundError as e:
        print(f"Warning: Dataset files not found: {e}. Some endpoints may not work.")
    except Exception as e:
        print(f"Error loading datasets: {e}")


# Simple NER patterns for demo
NER_PATTERNS = {
    'PERSON': [
        re.compile(r'\b([A-Z][a-z]+ [A-Z][a-z]+)\b'),
        re.compile(r'\b([A-Z][a-z]+ [A-Z]\. [A-Z][a-z]+)\b')
    ],
    'ORG': [
        re.compile(r'\b([A-Z][a-z]+ (?:Inc\.|Corporation|Corp\.|LLC|Ltd\.|Company|Co\.))\b'),
        re.compile(r'\b([A-Z]{2,})\b'),
        re.compile(r'\b(Apple|Microsoft|Google|Amazon|Facebook|Twitter|Tesla|SpaceX|Netflix|Uber|Airbnb|Spotify|LinkedIn|Zoom|Dropbox|Salesforce|Oracle|Intel|Adobe|NVIDIA|OpenAI|Meta|Alphabet)\b')
    ],
    'LOC': [
        re.compile(r'\b([A-Z][a-z]+(?:, [A-Z][a-z]+)?)\b'),
        re.compile(r'\b(New York|Los Angeles|San Francisco|Chicago|Houston|Phoenix|Philadelphia|San Antonio|San Diego|Dallas|San Jose|Austin|Jacksonville|Fort Worth|Columbus|Charlotte|Seattle|Denver|Washington|Boston|El Paso|Detroit|Nashville|Portland|Oklahoma City|Las Vegas|Memphis|Louisville|Baltimore|Milwaukee|Albuquerque|Tucson|Fresno|Sacramento|Kansas City|Mesa|Atlanta|Omaha|Colorado Springs|Raleigh|Virginia Beach|Miami|Oakland|Minneapolis|Tulsa|Cleveland|Wichita|Arlington)\b'),
        re.compile(r'\b(United States|USA|UK|Canada|Mexico|China|Japan|India|Germany|France|Italy|Spain|Brazil|Australia|Russia|South Korea|Netherlands|Sweden|Switzerland|Belgium|Norway|Denmark|Finland|Poland|Austria|Greece|Portugal|Ireland|Czech Republic|Romania|Hungary)\b')
    ],
    'DATE': [
        re.compile(r'\b(January|February|March|April|May|June|July|August|September|October|November|December) \d{1,2}, \d{4}\b'),
        re.compile(r'\b\d{4}\b'),
        re.compile(r'\b(January|February|March|April|May|June|July|August|September|October|November|December) \d{4}\b')
    ],
    'MONEY': [
        re.compile(r'\$\d+(?:,\d{3})*(?:\.\d{2})?\b'),
        re.compile(r'\b\d+(?:,\d{3})* (?:dollars|USD)\b', re.IGNORECASE)
    ],
    'PERCENT': [
        re.compile(r'\b\d+(?:\.\d+)?%\b')
    ]
}


def extract_entities(text: str) -> List[Dict[str, Any]]:
    """Extract entities from text using pattern matching."""
    entities = []
    entity_map = {}
    
    for entity_type, patterns in NER_PATTERNS.items():
        for pattern in patterns:
            for match in pattern.finditer(text):
                start = match.start()
                end = match.end()
                entity_text = match.group(0)
                
                # Check for overlaps
                overlap = False
                for (existing_start, existing_end), _ in entity_map.items():
                    if not (end <= existing_start or start >= existing_end):
                        overlap = True
                        break
                
                if not overlap:
                    entities.append({
                        'text': entity_text,
                        'label': entity_type,
                        'start': start,
                        'end': end
                    })
                    entity_map[(start, end)] = entity_type
    
    # Sort by start position
    entities.sort(key=lambda x: x['start'])
    return entities


@app.route('/')
def index():
    """API information endpoint."""
    return jsonify({
        'name': 'NER Dataset API',
        'version': '1.0.0',
        'author': 'RSK World',
        'website': 'https://rskworld.in',
        'email': 'help@rskworld.in',
        'phone': '+91 93305 39277',
        'endpoints': {
            '/api/extract': 'POST - Extract entities from text',
            '/api/dataset/train': 'GET - Get training dataset',
            '/api/dataset/test': 'GET - Get test dataset',
            '/api/dataset/stats': 'GET - Get dataset statistics',
            '/api/dataset/sample': 'GET - Get random sample',
            '/health': 'GET - Health check'
        }
    })


@app.route('/health')
def health():
    """Health check endpoint."""
    return jsonify({
        'status': 'healthy',
        'timestamp': datetime.now().isoformat(),
        'datasets_loaded': TRAIN_DATA is not None and TEST_DATA is not None
    })


@app.route('/api/extract', methods=['POST'])
def extract():
    """Extract entities from provided text."""
    try:
        data = request.get_json()
        if not data or 'text' not in data:
            return jsonify({'error': 'Text is required'}), 400
        
        text = data['text']
        entities = extract_entities(text)
        
        return jsonify({
            'text': text,
            'entities': entities,
            'entity_count': len(entities),
            'timestamp': datetime.now().isoformat()
        })
    except Exception as e:
        return jsonify({'error': str(e)}), 500


@app.route('/api/dataset/train', methods=['GET'])
def get_train_dataset():
    """Get training dataset."""
    if TRAIN_DATA is None:
        return jsonify({'error': 'Training dataset not loaded'}), 500
    
    limit = request.args.get('limit', type=int)
    offset = request.args.get('offset', type=int, default=0)
    
    data = TRAIN_DATA[offset:]
    if limit:
        data = data[:limit]
    
    return jsonify({
        'total': len(TRAIN_DATA),
        'returned': len(data),
        'offset': offset,
        'data': data
    })


@app.route('/api/dataset/test', methods=['GET'])
def get_test_dataset():
    """Get test dataset."""
    if TEST_DATA is None:
        return jsonify({'error': 'Test dataset not loaded'}), 500
    
    limit = request.args.get('limit', type=int)
    offset = request.args.get('offset', type=int, default=0)
    
    data = TEST_DATA[offset:]
    if limit:
        data = data[:limit]
    
    return jsonify({
        'total': len(TEST_DATA),
        'returned': len(data),
        'offset': offset,
        'data': data
    })


@app.route('/api/dataset/stats', methods=['GET'])
def get_stats():
    """Get dataset statistics."""
    if TRAIN_DATA is None or TEST_DATA is None:
        return jsonify({'error': 'Datasets not loaded'}), 500
    
    from collections import Counter
    
    train_entities = Counter()
    test_entities = Counter()
    
    for sample in TRAIN_DATA:
        for entity in sample.get('entities', []):
            train_entities[entity['label']] += 1
    
    for sample in TEST_DATA:
        for entity in sample.get('entities', []):
            test_entities[entity['label']] += 1
    
    return jsonify({
        'train': {
            'samples': len(TRAIN_DATA),
            'total_entities': sum(train_entities.values()),
            'entity_distribution': dict(train_entities)
        },
        'test': {
            'samples': len(TEST_DATA),
            'total_entities': sum(test_entities.values()),
            'entity_distribution': dict(test_entities)
        }
    })


@app.route('/api/dataset/sample', methods=['GET'])
def get_sample():
    """Get a random sample from dataset."""
    import random
    
    dataset_type = request.args.get('type', 'train')
    dataset = TRAIN_DATA if dataset_type == 'train' else TEST_DATA
    
    if dataset is None:
        return jsonify({'error': 'Dataset not loaded'}), 500
    
    if not dataset:
        return jsonify({'error': 'Dataset is empty'}), 404
    
    sample = random.choice(dataset)
    return jsonify(sample)


if __name__ == '__main__':
    print("Loading datasets...")
    load_datasets()
    print("\nStarting NER Dataset API Server...")
    print("API available at http://localhost:5000")
    print("API documentation at http://localhost:5000/")
    print("\nPress Ctrl+C to stop the server")
    app.run(debug=True, host='0.0.0.0', port=5000)

272 lines•9.2 KB

python

Theme Settings

Color Scheme

Display Options

Font Size

Theme Settings

Color Scheme

Display Options

Font Size

🚀 Support RSK World