mengzhx/semantica
1
0
Fork 0
mirror of https://github.com/semantica-agi/semantica.git synced 2026-09-13 04:04:09 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(normalize): enhance documentation and code quality for normalize module

Browse Source 
- Add comprehensive module-level docstrings with features and examples
- Enhance class and method docstrings with detailed parameter descriptions
- Improve error handling with specific exception types
- Add type hints for better code clarity
- Enhance logging with debug/info messages
- Improve code organization and maintainability
- Add graceful handling of optional dependencies (BeautifulSoup, langdetect, chardet)

Refactored files:
- __init__.py: Module-level documentation and exports
- data_cleaner.py: Data cleaning, duplicate detection, validation, missing values
- date_normalizer.py: Date/time normalization, timezone handling, relative dates
- encoding_handler.py: Encoding detection, UTF-8 conversion, BOM removal
- entity_normalizer.py: Entity normalization, alias resolution, disambiguation
- language_detector.py: Language detection with optional langdetect support
- number_normalizer.py: Number normalization, unit conversion, currency handling
- text_cleaner.py: Text cleaning, HTML removal, Unicode normalization
- text_normalizer.py: Text normalization, Unicode/whitespace/special character processing
This commit is contained in:
KaifAhmad1
2025-11-06 19:24:17 +05:30
parent c1e3848a6c
commit ea3ef0c65e
9 changed files with 2033 additions and 601 deletions
Download Patch File Download Diff File Expand all files Collapse all files
semantica/normalize/__init__.py
+23 -2
View File
@@ -1,9 +1,20 @@
"""
Data Normalization Module
This module provides comprehensive data normalization and cleaning capabilities.
This module provides comprehensive data normalization and cleaning capabilities
for the Semantica framework, enabling standardization and quality improvement
of various data types.
Exports:
Key Features:
- Text normalization and cleaning (Unicode, whitespace, special characters)
- Entity name normalization (aliases, variants, disambiguation)
- Date and time normalization (formats, timezones, relative dates)
- Number and quantity normalization (formats, units, currency, scientific notation)
- Data cleaning (duplicates, validation, missing values)
- Language detection (multi-language support)
- Encoding handling (detection, conversion, BOM removal)
Main Classes:
- TextNormalizer: Text cleaning and normalization
- EntityNormalizer: Entity name normalization
- DateNormalizer: Date and time normalization
@@ -12,6 +23,16 @@ Exports:
- TextCleaner: Text cleaning utilities
- LanguageDetector: Language detection
- EncodingHandler: Encoding detection and conversion
Example Usage:
>>> from semantica.normalize import TextNormalizer, EntityNormalizer
>>> text_norm = TextNormalizer()
>>> normalized = text_norm.normalize_text("Hello World")
>>> entity_norm = EntityNormalizer()
>>> canonical = entity_norm.normalize_entity("John Doe")
Author: Semantica Contributors
License: MIT
"""
from .text_normalizer import (
semantica/normalize/data_cleaner.py
+408 -103
View File
@@ -1,20 +1,32 @@
"""
Data Cleaning Module
Handles general data cleaning and quality improvement.
This module provides comprehensive data cleaning and quality improvement
capabilities for the Semantica framework, enabling detection and resolution
of data quality issues.
Key Features:
- Data quality assessment
- Duplicate detection and removal
- Data validation and correction
- Missing value handling
- Duplicate detection and removal (fuzzy matching, similarity scoring)
- Data validation and correction (schema validation, type checking)
- Missing value handling (removal, filling, imputation)
- Data consistency checking
- Batch processing support
Main Classes:
- DataCleaner: Main data cleaning class
- DataCleaner: Main data cleaning coordinator
- DuplicateDetector: Duplicate detection engine
- DataValidator: Data validation engine
- MissingValueHandler: Missing value processor
Example Usage:
>>> from semantica.normalize import DataCleaner
>>> cleaner = DataCleaner()
>>> cleaned = cleaner.clean_data(dataset, remove_duplicates=True, validate=True)
>>> duplicates = cleaner.detect_duplicates(dataset, threshold=0.8)
Author: Semantica Contributors
License: MIT
"""
from typing import Any, Dict, List, Optional, Union
@@ -27,7 +39,18 @@ from ..utils.logging import get_logger
@dataclass
class DuplicateGroup:
"""Duplicate record group."""
"""
Duplicate record group dataclass.
This dataclass represents a group of duplicate records identified during
duplicate detection, containing the records, similarity score, and canonical
record.
Attributes:
records: List of duplicate record dictionaries
similarity_score: Average similarity score for the group (0.0 to 1.0)
canonical_record: Canonical/representative record (typically first record)
"""
records: List[Dict[str, Any]]
similarity_score: float
canonical_record: Optional[Dict[str, Any]] = None
@@ -35,7 +58,17 @@ class DuplicateGroup:
@dataclass
class ValidationResult:
"""Data validation result."""
"""
Data validation result dataclass.
This dataclass represents the result of data validation, containing
validation status, errors, and warnings.
Attributes:
valid: Whether the data is valid (True if no errors)
errors: List of error dictionaries (critical validation failures)
warnings: List of warning dictionaries (non-critical issues)
"""
valid: bool
errors: List[Dict[str, Any]] = field(default_factory=list)
warnings: List[Dict[str, Any]] = field(default_factory=list)
@@ -43,23 +76,37 @@ class ValidationResult:
class DataCleaner:
"""
General data cleaning and quality improvement handler.
Data cleaning and quality improvement coordinator.
• Cleans and improves data quality
• Detects and removes duplicates
• Validates data integrity
• Handles missing values
• Ensures data consistency
• Supports various data types
This class provides comprehensive data cleaning capabilities, coordinating
duplicate detection, data validation, and missing value handling to
improve data quality.
Features:
- Data quality improvement
- Duplicate detection and removal
- Data validation against schemas
- Missing value handling (multiple strategies)
- Data consistency checking
- Support for various data types
Example Usage:
>>> cleaner = DataCleaner()
>>> cleaned = cleaner.clean_data(dataset, remove_duplicates=True, validate=True)
>>> duplicates = cleaner.detect_duplicates(dataset)
>>> validation = cleaner.validate_data(dataset, schema)
"""
def __init__(self, config=None, **kwargs):
def __init__(self, config: Optional[Dict[str, Any]] = None, **kwargs):
"""
Initialize data cleaner.
Sets up the cleaner with duplicate detector, data validator, and
missing value handler components.
Args:
config: Configuration dictionary
**kwargs: Additional configuration options
config: Configuration dictionary (optional)
**kwargs: Additional configuration options (merged into config)
"""
self.logger = get_logger("data_cleaner")
self.config = config or {}
@@ -68,37 +115,52 @@ class DataCleaner:
self.duplicate_detector = DuplicateDetector(**self.config)
self.data_validator = DataValidator(**self.config)
self.missing_value_handler = MissingValueHandler(**self.config)
self.logger.debug("Data cleaner initialized")
def clean_data(self, dataset: List[Dict[str, Any]], **options) -> List[Dict[str, Any]]:
def clean_data(
self,
dataset: List[Dict[str, Any]],
remove_duplicates: bool = True,
validate: bool = True,
handle_missing: bool = True,
**options
) -> List[Dict[str, Any]]:
"""
Clean dataset with various cleaning operations.
This method performs comprehensive data cleaning by applying missing
value handling, validation, and duplicate removal in sequence.
Args:
dataset: List of data records
**options: Cleaning options:
- remove_duplicates: Remove duplicates (default: True)
- validate: Validate data (default: True)
- handle_missing: Handle missing values (default: True)
dataset: List of data record dictionaries
remove_duplicates: Whether to remove duplicate records (default: True)
validate: Whether to validate data against schema (default: True)
handle_missing: Whether to handle missing values (default: True)
**options: Additional cleaning options:
- missing_strategy: Strategy for missing values ("remove", "fill", "impute")
- schema: Validation schema dictionary
- duplicate_criteria: Criteria for duplicate detection
Returns:
Cleaned dataset
list: Cleaned dataset (list of record dictionaries)
"""
cleaned = list(dataset)
# Handle missing values
if options.get("handle_missing", True):
if handle_missing:
strategy = options.get("missing_strategy", "remove")
cleaned = self.missing_value_handler.handle_missing_values(cleaned, strategy=strategy)
# Validate data
if options.get("validate", True):
if validate:
schema = options.get("schema")
validation = self.data_validator.validate_dataset(cleaned, schema)
if not validation.valid:
self.logger.warning(f"Validation found {len(validation.errors)} errors")
# Remove duplicates
if options.get("remove_duplicates", True):
if remove_duplicates:
criteria = options.get("duplicate_criteria", {})
duplicates = self.detect_duplicates(cleaned, **criteria)
@@ -114,83 +176,158 @@ class DataCleaner:
return cleaned
def detect_duplicates(self, dataset: List[Dict[str, Any]], **criteria) -> List[DuplicateGroup]:
def detect_duplicates(
self,
dataset: List[Dict[str, Any]],
threshold: Optional[float] = None,
key_fields: Optional[List[str]] = None,
**criteria
) -> List[DuplicateGroup]:
"""
Detect duplicate records in dataset.
This method identifies duplicate records using similarity matching
based on specified criteria and threshold.
Args:
dataset: List of data records
**criteria: Duplicate detection criteria
dataset: List of data record dictionaries
threshold: Similarity threshold for duplicates (0.0 to 1.0, optional,
uses detector's default if not provided)
key_fields: List of field names to use for comparison (optional,
uses all common fields if not provided)
**criteria: Additional duplicate detection criteria
Returns:
List of duplicate groups
list: List of DuplicateGroup objects, each containing duplicate
records with similarity scores
"""
return self.duplicate_detector.detect_duplicates(dataset, **criteria)
return self.duplicate_detector.detect_duplicates(
dataset, threshold=threshold, key_fields=key_fields, **criteria
)
def validate_data(self, dataset: List[Dict[str, Any]], schema=None) -> ValidationResult:
def validate_data(
self,
dataset: List[Dict[str, Any]],
schema: Optional[Dict[str, Any]] = None
) -> ValidationResult:
"""
Validate data against schema or rules.
This method validates all records in the dataset against the provided
schema, checking required fields, data types, and constraints.
Args:
dataset: List of data records
schema: Validation schema
dataset: List of data record dictionaries
schema: Validation schema dictionary (optional) containing:
- fields: Dictionary mapping field names to field schemas with:
- type: Expected data type
- required: Whether field is required (bool)
Returns:
Validation result
ValidationResult: Validation result containing:
- valid: True if no errors, False otherwise
- errors: List of error dictionaries with record_index and field info
- warnings: List of warning dictionaries
"""
return self.data_validator.validate_dataset(dataset, schema)
def handle_missing_values(self, dataset: List[Dict[str, Any]], **strategy) -> List[Dict[str, Any]]:
def handle_missing_values(
self,
dataset: List[Dict[str, Any]],
strategy: str = "remove",
**options
) -> List[Dict[str, Any]]:
"""
Handle missing values in dataset.
This method processes missing values in the dataset using the specified
strategy (remove, fill, or impute).
Args:
dataset: List of data records
**strategy: Missing value handling strategy
dataset: List of data record dictionaries
strategy: Handling strategy:
- "remove": Remove records with missing values (default)
- "fill": Fill missing values with default value
- "impute": Impute missing values using statistical methods
**options: Additional strategy options:
- fill_value: Value to use for filling (for "fill" strategy)
- method: Imputation method ("mean", "median", "mode", "zero")
Returns:
Processed dataset
list: Processed dataset with missing values handled
"""
return self.missing_value_handler.handle_missing_values(dataset, **strategy)
return self.missing_value_handler.handle_missing_values(dataset, strategy=strategy, **options)
class DuplicateDetector:
"""
Duplicate detection engine.
• Detects duplicate records
• Calculates similarity scores
• Handles fuzzy matching
• Manages duplicate resolution
This class provides duplicate detection capabilities using similarity
matching and fuzzy comparison algorithms.
Features:
- Duplicate record detection
- Similarity score calculation
- Fuzzy string matching
- Duplicate group formation
- Duplicate resolution strategies
Example Usage:
>>> detector = DuplicateDetector(similarity_threshold=0.8)
>>> duplicates = detector.detect_duplicates(dataset, threshold=0.85)
>>> resolved = detector.resolve_duplicates(duplicates, strategy="merge")
"""
def __init__(self, **config):
"""
Initialize duplicate detector.
Sets up the detector with similarity threshold and key fields for
comparison.
Args:
**config: Configuration options:
- similarity_threshold: Minimum similarity for duplicates (default: 0.8)
- key_fields: Fields to use for comparison
- similarity_threshold: Minimum similarity for duplicates
(default: 0.8, range: 0.0 to 1.0)
- key_fields: List of field names to use for comparison
(optional, uses all common fields if empty)
"""
self.logger = get_logger("duplicate_detector")
self.config = config
self.similarity_threshold = config.get("similarity_threshold", 0.8)
self.key_fields = config.get("key_fields", [])
self.logger.debug(f"Duplicate detector initialized (threshold={self.similarity_threshold})")
def detect_duplicates(self, dataset: List[Dict[str, Any]], **criteria) -> List[DuplicateGroup]:
def detect_duplicates(
self,
dataset: List[Dict[str, Any]],
threshold: Optional[float] = None,
key_fields: Optional[List[str]] = None,
**criteria
) -> List[DuplicateGroup]:
"""
Detect duplicates in dataset.
This method identifies duplicate records by comparing records pairwise
using similarity matching. Records with similarity above the threshold
are grouped together.
Args:
dataset: List of records
**criteria: Detection criteria
dataset: List of record dictionaries
threshold: Similarity threshold for duplicates (optional, uses
instance threshold if not provided)
key_fields: List of field names for comparison (optional, uses
instance key_fields if not provided)
**criteria: Additional detection criteria (unused)
Returns:
List of duplicate groups
list: List of DuplicateGroup objects containing duplicate records
with similarity scores
"""
threshold = criteria.get("threshold", self.similarity_threshold)
key_fields = criteria.get("key_fields", self.key_fields)
threshold = threshold if threshold is not None else self.similarity_threshold
key_fields = key_fields if key_fields is not None else self.key_fields
duplicate_groups = []
processed = set()
@@ -229,19 +366,31 @@ class DuplicateDetector:
return duplicate_groups
def calculate_similarity(self, record1: Dict[str, Any], record2: Dict[str, Any], **options) -> float:
def calculate_similarity(
self,
record1: Dict[str, Any],
record2: Dict[str, Any],
key_fields: Optional[List[str]] = None,
**options
) -> float:
"""
Calculate similarity between records.
This method calculates a similarity score between two records by
comparing values in key fields. Uses exact matching for non-string
values and string similarity for string values.
Args:
record1: First record
record2: Second record
**options: Similarity calculation options
record1: First record dictionary
record2: Second record dictionary
key_fields: List of field names to compare (optional, uses
instance key_fields or all common fields)
**options: Additional similarity calculation options (unused)
Returns:
Similarity score (0.0 to 1.0)
float: Similarity score between 0.0 and 1.0 (higher is more similar)
"""
key_fields = options.get("key_fields", self.key_fields)
key_fields = key_fields if key_fields is not None else self.key_fields
if not key_fields:
# Use all common fields
@@ -273,7 +422,19 @@ class DuplicateDetector:
return sum(similarities) / len(similarities) if similarities else 0.0
def _string_similarity(self, s1: str, s2: str) -> float:
"""Calculate string similarity using simple ratio."""
"""
Calculate string similarity using character overlap.
This method calculates similarity between two strings using character
set intersection over union (Jaccard-like similarity).
Args:
s1: First string
s2: Second string
Returns:
float: Similarity score between 0.0 and 1.0
"""
if not s1 or not s2:
return 0.0
@@ -296,19 +457,30 @@ class DuplicateDetector:
return intersection / union if union > 0 else 0.0
def resolve_duplicates(self, duplicate_groups: List[DuplicateGroup], **strategy) -> List[Dict[str, Any]]:
def resolve_duplicates(
self,
duplicate_groups: List[DuplicateGroup],
strategy: str = "keep_first",
**options
) -> List[Dict[str, Any]]:
"""
Resolve duplicate groups.
This method resolves duplicate groups by applying a resolution strategy,
returning a single record per group.
Args:
duplicate_groups: List of duplicate groups
**strategy: Resolution strategy
duplicate_groups: List of DuplicateGroup objects
strategy: Resolution strategy:
- "keep_first": Keep first record in group (default)
- "merge": Merge all records into one
**options: Additional resolution options (unused)
Returns:
List of resolved records
list: List of resolved record dictionaries (one per duplicate group)
"""
resolved = []
strategy_type = strategy.get("strategy", "keep_first")
strategy_type = strategy
for group in duplicate_groups:
if strategy_type == "keep_first":
@@ -322,7 +494,18 @@ class DuplicateDetector:
return resolved
def _merge_records(self, records: List[Dict[str, Any]]) -> Dict[str, Any]:
"""Merge multiple records into one."""
"""
Merge multiple records into one.
This method merges multiple records by combining their fields, keeping
the first non-null value for each field.
Args:
records: List of record dictionaries to merge
Returns:
dict: Merged record dictionary
"""
merged = {}
for record in records:
@@ -340,32 +523,56 @@ class DataValidator:
"""
Data validation engine.
• Validates data integrity
• Checks data types and formats
• Validates constraints
• Handles validation errors
This class provides data validation capabilities, checking data integrity,
types, formats, and constraints against schemas.
Features:
- Data integrity validation
- Data type checking
- Format validation
- Constraint validation
- Error and warning reporting
Example Usage:
>>> validator = DataValidator()
>>> result = validator.validate_dataset(dataset, schema)
>>> if not result.valid:
... print(f"Errors: {result.errors}")
"""
def __init__(self, **config):
"""
Initialize data validator.
Sets up the validator with configuration options.
Args:
**config: Configuration options
**config: Configuration options (currently unused)
"""
self.logger = get_logger("data_validator")
self.config = config
self.logger.debug("Data validator initialized")
def validate_dataset(self, dataset: List[Dict[str, Any]], schema: Optional[Dict[str, Any]] = None) -> ValidationResult:
def validate_dataset(
self,
dataset: List[Dict[str, Any]],
schema: Optional[Dict[str, Any]] = None
) -> ValidationResult:
"""
Validate entire dataset.
This method validates all records in the dataset against the provided
schema, collecting errors and warnings for each record.
Args:
dataset: List of records
schema: Validation schema
dataset: List of record dictionaries to validate
schema: Validation schema dictionary (optional) containing:
- fields: Dictionary mapping field names to field schemas
Returns:
Validation result
ValidationResult: Validation result with errors and warnings
aggregated across all records
"""
errors = []
warnings = []
@@ -388,16 +595,26 @@ class DataValidator:
warnings=warnings
)
def validate_record(self, record: Dict[str, Any], schema: Optional[Dict[str, Any]] = None) -> ValidationResult:
def validate_record(
self,
record: Dict[str, Any],
schema: Optional[Dict[str, Any]] = None
) -> ValidationResult:
"""
Validate individual record.
This method validates a single record against the provided schema,
checking required fields and data types.
Args:
record: Record to validate
schema: Validation schema
record: Record dictionary to validate
schema: Validation schema dictionary (optional) containing:
- fields: Dictionary mapping field names to field schemas with:
- type: Expected data type (str, int, float, bool, list, dict)
- required: Whether field is required (bool)
Returns:
Validation result
ValidationResult: Validation result for this record
"""
errors = []
warnings = []
@@ -429,16 +646,27 @@ class DataValidator:
warnings=warnings
)
def check_data_types(self, data: Any, expected_types: Union[type, List[type]]) -> bool:
def check_data_types(
self,
data: Any,
expected_types: Union[type, List[type], str, List[str]]
) -> bool:
"""
Check data types against expected types.
This method validates that data matches one of the expected types.
Supports both type objects and type name strings.
Args:
data: Data to check
expected_types: Expected type(s)
data: Data value to check
expected_types: Expected type(s) - can be:
- Single type object (e.g., str, int)
- List of type objects
- Type name string (e.g., "str", "int")
- List of type name strings
Returns:
True if type matches
bool: True if data type matches one of the expected types, False otherwise
"""
if isinstance(expected_types, type):
expected_types = [expected_types]
@@ -469,33 +697,54 @@ class MissingValueHandler:
"""
Missing value processing engine.
• Identifies missing values
• Applies handling strategies
• Fills missing data
• Removes incomplete records
This class provides missing value handling capabilities, including
identification, removal, filling, and imputation strategies.
Features:
- Missing value identification
- Multiple handling strategies (remove, fill, impute)
- Statistical imputation (mean, median, mode)
- Missing value statistics
Example Usage:
>>> handler = MissingValueHandler()
>>> missing_info = handler.identify_missing_values(dataset)
>>> processed = handler.handle_missing_values(dataset, strategy="impute", method="mean")
"""
def __init__(self, **config):
"""
Initialize missing value handler.
Sets up the handler with configuration and missing value definitions.
Args:
**config: Configuration options:
- missing_values: List of values considered missing (default: [None, "", "N/A", "null"])
- missing_values: List of values considered missing
(default: [None, "", "N/A", "null", "NULL"])
"""
self.logger = get_logger("missing_value_handler")
self.config = config
self.missing_values = config.get("missing_values", [None, "", "N/A", "null", "NULL"])
self.logger.debug("Missing value handler initialized")
def identify_missing_values(self, dataset: List[Dict[str, Any]]) -> Dict[str, Any]:
"""
Identify missing values in dataset.
This method analyzes the dataset to identify missing values across
all fields, providing counts and percentages.
Args:
dataset: List of records
dataset: List of record dictionaries
Returns:
Missing value information
dict: Missing value information containing:
- total_records: Total number of records
- missing_counts: Dictionary mapping field names to missing counts
- missing_percentages: Dictionary mapping field names to
missing percentages (0.0 to 100.0)
"""
missing_info = defaultdict(int)
total_records = len(dataset)
@@ -524,28 +773,55 @@ class MissingValueHandler:
}
}
def handle_missing_values(self, dataset: List[Dict[str, Any]], strategy: str = "remove") -> List[Dict[str, Any]]:
def handle_missing_values(
self,
dataset: List[Dict[str, Any]],
strategy: str = "remove",
**options
) -> List[Dict[str, Any]]:
"""
Handle missing values using specified strategy.
This method processes missing values in the dataset using the specified
strategy: remove records, fill with default values, or impute using
statistical methods.
Args:
dataset: List of records
strategy: Handling strategy ('remove', 'fill', 'impute')
dataset: List of record dictionaries
strategy: Handling strategy:
- "remove": Remove records with any missing values (default)
- "fill": Fill missing values with default value
- "impute": Impute missing values using statistical methods
**options: Strategy-specific options:
- fill_value: Value to use for filling (for "fill" strategy)
- method: Imputation method for "impute" strategy
("mean", "median", "mode", "zero")
Returns:
Processed dataset
list: Processed dataset with missing values handled
"""
if strategy == "remove":
return self._remove_missing(dataset)
elif strategy == "fill":
return self._fill_missing(dataset)
return self._fill_missing(dataset, fill_value=options.get("fill_value", ""))
elif strategy == "impute":
return self.impute_values(dataset)
return self.impute_values(dataset, method=options.get("method", "mean"))
else:
return dataset
def _remove_missing(self, dataset: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""Remove records with missing values."""
"""
Remove records with missing values.
This method filters out records that contain any missing values
(as defined in missing_values list).
Args:
dataset: List of record dictionaries
Returns:
list: Dataset with records containing missing values removed
"""
return [
record for record in dataset
if not any(
@@ -554,8 +830,26 @@ class MissingValueHandler:
)
]
def _fill_missing(self, dataset: List[Dict[str, Any]], fill_value: Any = "") -> List[Dict[str, Any]]:
"""Fill missing values with default value."""
def _fill_missing(
self,
dataset: List[Dict[str, Any]],
fill_value: Optional[Any] = None
) -> List[Dict[str, Any]]:
"""
Fill missing values with default value.
This method replaces all missing values in records with the specified
fill value.
Args:
dataset: List of record dictionaries
fill_value: Value to use for filling missing values (default: "")
Returns:
list: Dataset with missing values filled
"""
if fill_value is None:
fill_value = ""
filled = []
for record in dataset:
filled_record = {}
@@ -567,16 +861,27 @@ class MissingValueHandler:
filled.append(filled_record)
return filled
def impute_values(self, dataset: List[Dict[str, Any]], method: str = "mean") -> List[Dict[str, Any]]:
def impute_values(
self,
dataset: List[Dict[str, Any]],
method: str = "mean"
) -> List[Dict[str, Any]]:
"""
Impute missing values using specified method.
This method imputes missing numeric values using statistical methods
(mean, median, mode, or zero). Only numeric fields are imputed.
Args:
dataset: List of records
method: Imputation method ('mean', 'median', 'mode', 'zero')
dataset: List of record dictionaries
method: Imputation method:
- "mean": Use mean of non-missing values (default)
- "median": Use median of non-missing values
- "zero": Use zero
- "mode": Use mode (most frequent value)
Returns:
Imputed dataset
list: Dataset with missing numeric values imputed
"""
if not dataset:
return dataset
semantica/normalize/date_normalizer.py
+284 -89
View File
@@ -1,20 +1,33 @@
"""
Date and Time Normalization Module
Handles normalization of dates, times, and temporal expressions.
This module provides comprehensive date and time normalization capabilities
for the Semantica framework, enabling standardization of temporal data
across various formats and conventions.
Key Features:
- Date format standardization
- Time zone normalization
- Relative date processing
- Temporal expression parsing
- Date range handling
- Date format standardization (ISO8601, custom formats)
- Time zone normalization and UTC conversion
- Relative date processing ("yesterday", "3 days ago", etc.)
- Temporal expression parsing (natural language dates)
- Date range handling ("from X to Y")
- Support for multiple calendar systems
- Optional dateutil integration for advanced parsing
Main Classes:
- DateNormalizer: Main date normalization class
- TimeZoneNormalizer: Time zone processing
- RelativeDateProcessor: Relative date handling
- DateNormalizer: Main date normalization coordinator
- TimeZoneNormalizer: Time zone processing engine
- RelativeDateProcessor: Relative date handling engine
- TemporalExpressionParser: Temporal expression parser
Example Usage:
>>> from semantica.normalize import DateNormalizer
>>> normalizer = DateNormalizer()
>>> normalized = normalizer.normalize_date("2023-01-15", format="ISO8601")
>>> relative = normalizer.process_relative_date("3 days ago")
Author: Semantica Contributors
License: MIT
"""
import re
@@ -37,23 +50,36 @@ except ImportError:
class DateNormalizer:
"""
Date and time normalization handler.
Date and time normalization coordinator.
• Normalizes dates and times to standard formats
• Handles various date formats and conventions
• Processes time zones and UTC conversion
• Manages relative dates and temporal expressions
• Standardizes date representations
• Supports multiple calendar systems
This class provides comprehensive date and time normalization capabilities,
coordinating timezone normalization, relative date processing, and temporal
expression parsing.
Features:
- Date and time format standardization
- Multiple date format support
- Timezone normalization and UTC conversion
- Relative date processing
- Temporal expression parsing
- Date range handling
Example Usage:
>>> normalizer = DateNormalizer()
>>> normalized = normalizer.normalize_date("2023-01-15", format="ISO8601")
>>> relative = normalizer.process_relative_date("3 days ago")
"""
def __init__(self, config=None, **kwargs):
def __init__(self, config: Optional[Dict[str, Any]] = None, **kwargs):
"""
Initialize date normalizer.
Sets up the normalizer with timezone normalizer, relative date processor,
and temporal expression parser components.
Args:
config: Configuration dictionary
**kwargs: Additional configuration options
config: Configuration dictionary (optional)
**kwargs: Additional configuration options (merged into config)
"""
self.logger = get_logger("date_normalizer")
self.config = config or {}
@@ -62,19 +88,38 @@ class DateNormalizer:
self.timezone_normalizer = TimeZoneNormalizer(**self.config)
self.relative_date_processor = RelativeDateProcessor(**self.config)
self.temporal_parser = TemporalExpressionParser(**self.config)
self.logger.debug("Date normalizer initialized")
def normalize_date(self, date_input: Any, **options) -> str:
def normalize_date(
self,
date_input: Any,
format: str = "ISO8601",
timezone: str = "UTC",
**options
) -> str:
"""
Normalize date to standard format.
This method normalizes date input (string or datetime) to a standard
format, handling timezone conversion and relative date expressions.
Args:
date_input: Date input (string, datetime, or other)
**options: Normalization options:
- format: Output format (default: 'ISO8601')
- timezone: Target timezone (default: 'UTC')
date_input: Date input - can be:
- String (e.g., "2023-01-15", "yesterday")
- datetime object
format: Output format (default: "ISO8601"):
- "ISO8601": ISO 8601 format (e.g., "2023-01-15T10:30:00")
- "date": Date only (e.g., "2023-01-15")
- Custom format string (strftime format)
timezone: Target timezone (default: "UTC")
**options: Additional normalization options (unused)
Returns:
Normalized date string
str: Normalized date string in specified format
Raises:
ValidationError: If date input type is unsupported or parsing fails
"""
if not date_input:
return ""
@@ -96,14 +141,13 @@ class DateNormalizer:
raise ValidationError(f"Unsupported date input type: {type(date_input)}")
# Normalize timezone
target_tz = options.get("timezone", "UTC")
if target_tz != "UTC":
dt = self.timezone_normalizer.normalize_timezone(dt, target_tz)
if timezone != "UTC":
dt = self.timezone_normalizer.normalize_timezone(dt, timezone)
else:
dt = self.timezone_normalizer.convert_to_utc(dt)
# Format output
output_format = options.get("format", "ISO8601")
output_format = format
if output_format == "ISO8601":
return dt.isoformat()
elif output_format == "date":
@@ -115,12 +159,17 @@ class DateNormalizer:
"""
Normalize time to standard format.
This method normalizes time input to ISO format (HH:MM:SS).
Args:
time_input: Time input
**options: Normalization options
time_input: Time input - can be:
- String (e.g., "10:30:00", "10:30 AM")
- datetime object
**options: Additional normalization options (unused)
Returns:
Normalized time string
str: Normalized time string in ISO format (HH:MM:SS), or empty
string if parsing fails
"""
if isinstance(time_input, str):
try:
@@ -138,29 +187,50 @@ class DateNormalizer:
return dt.time().isoformat()
def process_relative_date(self, relative_expression: str, reference_date: Optional[datetime] = None) -> datetime:
def process_relative_date(
self,
relative_expression: str,
reference_date: Optional[datetime] = None
) -> datetime:
"""
Process relative date expressions.
This method processes relative date expressions like "yesterday",
"3 days ago", "2 weeks from now", etc.
Args:
relative_expression: Relative date expression
reference_date: Reference date (default: now)
relative_expression: Relative date expression (e.g., "yesterday",
"3 days ago", "2 weeks from now")
reference_date: Reference date for calculation (default: current
datetime)
Returns:
Calculated date
datetime: Calculated absolute date
"""
return self.relative_date_processor.process_relative_expression(relative_expression, reference_date)
def parse_temporal_expression(self, temporal_text: str, **context) -> Dict[str, Any]:
def parse_temporal_expression(
self,
temporal_text: str,
**context
) -> Dict[str, Any]:
"""
Parse temporal expressions and references.
This method parses natural language temporal expressions, extracting
date, time, and range information.
Args:
temporal_text: Temporal expression text
**context: Context information
temporal_text: Temporal expression text (e.g., "from January to
March", "last week")
**context: Context information (optional)
Returns:
Parsed temporal data
dict: Parsed temporal data containing:
- date: Date string (if found)
- time: Time string (if found)
- range: Range dictionary with start/end (if found)
- relative: Whether expression is relative (bool)
"""
return self.temporal_parser.parse_temporal_expression(temporal_text, **context)
@@ -169,33 +239,55 @@ class TimeZoneNormalizer:
"""
Time zone normalization engine.
• Handles time zone conversion and normalization
• Manages UTC conversion
• Processes time zone abbreviations
• Handles daylight saving time
• Manages time zone databases
This class provides timezone conversion and normalization capabilities,
handling UTC conversion, timezone abbreviations, and daylight saving
time transitions.
Features:
- Timezone conversion and normalization
- UTC conversion
- Timezone abbreviation processing
- Daylight saving time handling
- ZoneInfo integration (Python 3.9+)
Example Usage:
>>> tz_normalizer = TimeZoneNormalizer()
>>> utc_dt = tz_normalizer.convert_to_utc(datetime_obj)
>>> normalized = tz_normalizer.normalize_timezone(dt, "America/New_York")
"""
def __init__(self, **config):
"""
Initialize time zone normalizer.
Sets up the normalizer with configuration options.
Args:
**config: Configuration options
**config: Configuration options (currently unused)
"""
self.logger = get_logger("timezone_normalizer")
self.config = config
self.logger.debug("Time zone normalizer initialized")
def normalize_timezone(self, datetime_obj: datetime, target_timezone: str = "UTC") -> datetime:
def normalize_timezone(
self,
datetime_obj: datetime,
target_timezone: str = "UTC"
) -> datetime:
"""
Normalize datetime to target timezone.
This method converts a datetime object to the specified timezone,
using ZoneInfo if available (Python 3.9+), or falling back to UTC.
Args:
datetime_obj: Datetime object
target_timezone: Target timezone
datetime_obj: Datetime object to normalize
target_timezone: Target timezone string (e.g., "America/New_York",
"UTC", "Europe/London")
Returns:
Normalized datetime
datetime: Normalized datetime in target timezone
"""
try:
from zoneinfo import ZoneInfo
@@ -207,16 +299,25 @@ class TimeZoneNormalizer:
# Fallback if zoneinfo not available
return datetime_obj
def convert_to_utc(self, datetime_obj: datetime, source_timezone: Optional[str] = None) -> datetime:
def convert_to_utc(
self,
datetime_obj: datetime,
source_timezone: Optional[str] = None
) -> datetime:
"""
Convert datetime to UTC.
This method converts a datetime object to UTC timezone. If the datetime
has no timezone info, it assumes UTC or uses the source_timezone if
provided.
Args:
datetime_obj: Datetime object
source_timezone: Source timezone (auto-detect if None)
datetime_obj: Datetime object to convert
source_timezone: Source timezone string (optional, used if datetime
has no timezone info)
Returns:
UTC datetime
datetime: Datetime object in UTC timezone
"""
if datetime_obj.tzinfo is None:
if source_timezone:
@@ -226,16 +327,23 @@ class TimeZoneNormalizer:
return datetime_obj.astimezone(timezone.utc)
def handle_dst_transitions(self, datetime_obj: datetime, timezone_str: str) -> datetime:
def handle_dst_transitions(
self,
datetime_obj: datetime,
timezone_str: str
) -> datetime:
"""
Handle daylight saving time transitions.
This method handles DST transitions by normalizing the datetime to
the specified timezone, which automatically accounts for DST.
Args:
datetime_obj: Datetime object
timezone_str: Timezone string
timezone_str: Timezone string (e.g., "America/New_York")
Returns:
Adjusted datetime
datetime: Adjusted datetime accounting for DST
"""
return self.normalize_timezone(datetime_obj, timezone_str)
@@ -244,18 +352,30 @@ class RelativeDateProcessor:
"""
Relative date processing engine.
• Processes relative date expressions
• Calculates absolute dates from relative terms
• Handles various relative formats
• Manages date arithmetic
This class provides relative date expression processing, converting
natural language relative dates (e.g., "yesterday", "3 days ago")
into absolute datetime objects.
Features:
- Relative date expression processing
- Natural language date parsing
- Date arithmetic (days, weeks, months, years)
- Support for common relative terms
Example Usage:
>>> processor = RelativeDateProcessor()
>>> date = processor.process_relative_expression("3 days ago")
>>> offset = processor.calculate_date_offset("2 weeks from now", ref_date)
"""
def __init__(self, **config):
"""
Initialize relative date processor.
Sets up the processor with relative terms dictionary and configuration.
Args:
**config: Configuration options
**config: Configuration options (currently unused)
"""
self.logger = get_logger("relative_date_processor")
self.config = config
@@ -266,17 +386,29 @@ class RelativeDateProcessor:
"tomorrow": 1,
"now": 0,
}
self.logger.debug("Relative date processor initialized")
def process_relative_expression(self, expression: str, reference_date: Optional[datetime] = None) -> datetime:
def process_relative_expression(
self,
expression: str,
reference_date: Optional[datetime] = None
) -> datetime:
"""
Process relative date expression.
This method processes relative date expressions like "yesterday",
"3 days ago", "2 weeks from now", etc., converting them to absolute
datetime objects.
Args:
expression: Relative date expression
reference_date: Reference date (default: now)
expression: Relative date expression (e.g., "yesterday",
"3 days ago", "2 weeks from now")
reference_date: Reference date for calculation (default: current
datetime)
Returns:
Calculated date
datetime: Calculated absolute date
"""
if reference_date is None:
reference_date = datetime.now()
@@ -331,29 +463,44 @@ class RelativeDateProcessor:
except Exception:
return reference_date
def calculate_date_offset(self, expression: str, reference_date: datetime) -> datetime:
def calculate_date_offset(
self,
expression: str,
reference_date: datetime
) -> datetime:
"""
Calculate date offset from expression.
This method calculates a date offset from a relative expression,
using the provided reference date.
Args:
expression: Offset expression
reference_date: Reference date
expression: Offset expression (e.g., "3 days ago",
"2 weeks from now")
reference_date: Reference date for calculation
Returns:
Calculated date
datetime: Calculated date with offset applied
"""
return self.process_relative_expression(expression, reference_date)
def handle_relative_terms(self, term: str, reference_date: datetime) -> datetime:
def handle_relative_terms(
self,
term: str,
reference_date: datetime
) -> datetime:
"""
Handle specific relative terms.
This method processes specific relative terms like "today", "yesterday",
"tomorrow", using the provided reference date.
Args:
term: Relative term
reference_date: Reference date
term: Relative term (e.g., "today", "yesterday", "tomorrow")
reference_date: Reference date for calculation
Returns:
Calculated date
datetime: Calculated date based on relative term
"""
return self.process_relative_expression(term, reference_date)
@@ -362,32 +509,57 @@ class TemporalExpressionParser:
"""
Temporal expression parsing engine.
• Parses natural language temporal expressions
• Extracts date and time components
• Handles complex temporal references
• Processes temporal ranges and periods
This class provides natural language temporal expression parsing,
extracting date, time, and range information from text.
Features:
- Natural language temporal expression parsing
- Date and time component extraction
- Temporal range processing
- Complex temporal reference handling
Example Usage:
>>> parser = TemporalExpressionParser()
>>> result = parser.parse_temporal_expression("from January to March")
>>> date_components = parser.extract_date_components("2023-01-15")
"""
def __init__(self, **config):
"""
Initialize temporal expression parser.
Sets up the parser with configuration options.
Args:
**config: Configuration options
**config: Configuration options (currently unused)
"""
self.logger = get_logger("temporal_expression_parser")
self.config = config
self.logger.debug("Temporal expression parser initialized")
def parse_temporal_expression(self, text: str, **context) -> Dict[str, Any]:
def parse_temporal_expression(
self,
text: str,
**context
) -> Dict[str, Any]:
"""
Parse temporal expression from text.
This method parses natural language temporal expressions, extracting
date, time, and range information.
Args:
text: Temporal expression text
**context: Context information
text: Temporal expression text (e.g., "from January to March",
"last week", "2023-01-15")
**context: Context information (optional)
Returns:
Parsed temporal data
dict: Parsed temporal data containing:
- date: Date string (if found)
- time: Time string (if found)
- range: Range dictionary with start/end (if found)
- relative: Whether expression is relative (bool)
"""
result = {
"date": None,
@@ -417,11 +589,19 @@ class TemporalExpressionParser:
"""
Extract date components from text.
This method extracts date components (year, month, day) from text
using date parsing.
Args:
text: Input text
text: Input text containing date information
Returns:
Date components
dict: Date components containing:
- date: Date string in ISO format
- year: Year (int)
- month: Month (int, 1-12)
- day: Day (int, 1-31)
Returns empty dict if parsing fails
"""
try:
if HAS_DATEUTIL and date_parser:
@@ -442,11 +622,19 @@ class TemporalExpressionParser:
"""
Extract time components from text.
This method extracts time components (hour, minute, second) from
text using date parsing.
Args:
text: Input text
text: Input text containing time information
Returns:
Time components
dict: Time components containing:
- time: Time string in ISO format (HH:MM:SS)
- hour: Hour (int, 0-23)
- minute: Minute (int, 0-59)
- second: Second (int, 0-59)
Returns empty dict if parsing fails
"""
try:
if HAS_DATEUTIL and date_parser:
@@ -467,11 +655,18 @@ class TemporalExpressionParser:
"""
Process temporal ranges and periods.
This method extracts temporal ranges from text patterns like
"from X to Y" or "between X and Y".
Args:
text: Input text
text: Input text containing temporal range (e.g., "from January
to March", "between 2023-01-01 and 2023-03-31")
Returns:
Range information or None
dict: Range information containing:
- start: Start date string in ISO format
- end: End date string in ISO format
Returns None if no range is found
"""
# Look for range patterns like "from X to Y", "between X and Y"
range_patterns = [
semantica/normalize/encoding_handler.py
+157 -45
View File
@@ -1,8 +1,30 @@
"""
Encoding handling utilities for Semantica framework.
Encoding Handling Module
This module provides encoding detection, conversion, and handling
for UTF-8 conversion and BOM handling.
This module provides comprehensive encoding detection, conversion, and handling
capabilities for the Semantica framework, enabling robust text processing across
various character encodings.
Key Features:
- Encoding detection (chardet integration)
- UTF-8 conversion with fallback support
- BOM (Byte Order Mark) removal
- File encoding detection and conversion
- Encoding validation
- Error handling strategies
Main Classes:
- EncodingHandler: Encoding detection and conversion coordinator
Example Usage:
>>> from semantica.normalize import EncodingHandler
>>> handler = EncodingHandler()
>>> encoding, confidence = handler.detect(data)
>>> utf8_text = handler.convert_to_utf8(data)
>>> content = handler.convert_file_to_utf8("input.txt", "output.txt")
Author: Semantica Contributors
License: MIT
"""
import chardet
@@ -14,30 +36,65 @@ from ..utils.logging import get_logger
class EncodingHandler:
"""Encoding handling utilities."""
"""
Encoding detection and conversion coordinator.
This class provides comprehensive encoding handling capabilities, including
detection, conversion, BOM removal, and validation.
Features:
- Encoding detection using chardet
- UTF-8 conversion with fallback encodings
- BOM removal (UTF-8, UTF-16)
- File encoding detection and conversion
- Encoding validation
- Graceful error handling
Example Usage:
>>> handler = EncodingHandler()
>>> encoding, confidence = handler.detect(data)
>>> utf8_text = handler.convert_to_utf8(data, source_encoding="latin-1")
>>> content = handler.convert_file_to_utf8("input.txt")
"""
def __init__(self, **config):
"""
Initialize encoding handler.
Sets up the handler with default encoding and fallback encodings.
Args:
**config: Configuration options
**config: Configuration options:
- default_encoding: Default encoding (default: "utf-8")
- fallback_encodings: List of fallback encodings to try
(default: ["latin-1", "cp1252", "iso-8859-1"])
"""
self.logger = get_logger("encoding_handler")
self.config = config
self.default_encoding = config.get("default_encoding", "utf-8")
self.fallback_encodings = config.get("fallback_encodings", ["latin-1", "cp1252", "iso-8859-1"])
self.logger.debug(f"Encoding handler initialized (default={self.default_encoding})")
def detect(self, data: Union[str, bytes], **options) -> Tuple[str, float]:
def detect(
self,
data: Union[str, bytes],
**options
) -> Tuple[str, float]:
"""
Detect encoding of data.
This method detects the character encoding of input data using chardet.
If data is a string, it's first encoded to UTF-8 bytes for detection.
Args:
data: Input data (string or bytes)
**options: Detection options
data: Input data - can be string or bytes
**options: Detection options (unused)
Returns:
tuple: (encoding, confidence)
tuple: (encoding_name, confidence_score) where:
- encoding_name: Detected encoding name (e.g., "utf-8", "latin-1")
- confidence_score: Confidence score between 0.0 and 1.0
"""
if isinstance(data, str):
data = data.encode('utf-8')
@@ -57,25 +114,37 @@ class EncodingHandler:
self.logger.warning(f"Failed to detect encoding: {e}")
return (self.default_encoding, 0.0)
def detect_file(self, file_path: Union[str, Path], **options) -> Tuple[str, float]:
def detect_file(
self,
file_path: Union[str, Path],
sample_size: int = 10000,
**options
) -> Tuple[str, float]:
"""
Detect encoding of file.
This method detects the character encoding of a file by reading a sample
of bytes and using chardet for detection.
Args:
file_path: Path to file
**options: Detection options
file_path: Path to file (string or Path object)
sample_size: Number of bytes to read for detection (default: 10000)
**options: Additional detection options (unused)
Returns:
tuple: (encoding, confidence)
tuple: (encoding_name, confidence_score) where:
- encoding_name: Detected encoding name
- confidence_score: Confidence score between 0.0 and 1.0
Raises:
ValidationError: If file does not exist
ProcessingError: If file reading or detection fails
"""
file_path = Path(file_path)
if not file_path.exists():
raise ValidationError(f"File not found: {file_path}")
# Read sample bytes for detection
sample_size = options.get("sample_size", 10000)
try:
with open(file_path, 'rb') as f:
sample = f.read(sample_size)
@@ -83,19 +152,32 @@ class EncodingHandler:
return self.detect(sample, **options)
except Exception as e:
self.logger.error(f"Failed to detect file encoding: {e}")
raise ProcessingError(f"Failed to detect file encoding: {e}")
raise ProcessingError(f"Failed to detect file encoding: {e}") from e
def convert_to_utf8(self, data: Union[str, bytes], source_encoding: Optional[str] = None, **options) -> str:
def convert_to_utf8(
self,
data: Union[str, bytes],
source_encoding: Optional[str] = None,
**options
) -> str:
"""
Convert data to UTF-8.
This method converts input data (string or bytes) to UTF-8 encoding.
If source_encoding is not provided, it's auto-detected. Falls back to
multiple encodings if initial conversion fails.
Args:
data: Input data
source_encoding: Source encoding (auto-detected if None)
**options: Conversion options
data: Input data - can be string or bytes
source_encoding: Source encoding name (optional, auto-detected if None)
**options: Additional conversion options (unused)
Returns:
str: UTF-8 encoded string
Note:
Uses 'replace' error handling strategy to handle invalid characters
gracefully, replacing them with replacement characters.
"""
if isinstance(data, str):
# Already a string, just ensure it's valid UTF-8
@@ -128,13 +210,21 @@ class EncodingHandler:
"""
Convert file to UTF-8.
This method reads a file, detects its encoding, converts it to UTF-8,
and optionally writes it to an output file.
Args:
file_path: Path to input file
output_path: Path to output file (overwrites if None)
**options: Conversion options
file_path: Path to input file (string or Path object)
output_path: Path to output file (optional, if provided, writes
converted content to this file)
**options: Additional conversion options (unused)
Returns:
str: Converted content as string
str: Converted content as UTF-8 string
Raises:
ValidationError: If input file does not exist
ProcessingError: If file reading or conversion fails
"""
file_path = Path(file_path)
@@ -161,7 +251,7 @@ class EncodingHandler:
if content is None:
self.logger.error(f"Failed to read file with any encoding: {e}")
raise ProcessingError(f"Failed to read file: {e}")
raise ProcessingError(f"Failed to read file: {e}") from e
# Ensure content is UTF-8
utf8_content = content.encode('utf-8', errors='replace').decode('utf-8')
@@ -180,11 +270,14 @@ class EncodingHandler:
"""
Remove BOM (Byte Order Mark) from data.
This method removes BOM markers from the beginning of data, supporting
UTF-8, UTF-16 LE, and UTF-16 BE BOMs.
Args:
data: Input data
data: Input data - can be string or bytes
Returns:
Data without BOM
Union[str, bytes]: Data without BOM (same type as input)
"""
if isinstance(data, bytes):
# Remove UTF-8 BOM
@@ -204,18 +297,29 @@ class EncodingHandler:
else:
return data
def handle_encoding_error(self, data: bytes, **options) -> str:
def handle_encoding_error(
self,
data: bytes,
error_strategy: str = "replace",
**options
) -> str:
"""
Handle encoding errors gracefully.
This method attempts to decode bytes data using detected encoding and
fallback encodings, applying the specified error handling strategy.
Args:
data: Input bytes data
**options: Error handling options
error_strategy: Error handling strategy (default: "replace"):
- "replace": Replace invalid characters with replacement char
- "ignore": Ignore invalid characters
- "strict": Raise exception on errors
**options: Additional error handling options (unused)
Returns:
str: Decoded string with errors handled
str: Decoded string with errors handled according to strategy
"""
error_strategy = options.get("error_strategy", "replace")
# Try detected encoding first
encoding, _ = self.detect(data, **options)
@@ -233,17 +337,25 @@ class EncodingHandler:
# Final fallback
return data.decode(self.default_encoding, errors=error_strategy)
def validate_encoding(self, data: Union[str, bytes], encoding: str, **options) -> bool:
def validate_encoding(
self,
data: Union[str, bytes],
encoding: str,
**options
) -> bool:
"""
Validate that data can be decoded with given encoding.
Validate that data can be decoded/encoded with given encoding.
This method validates that the data can be successfully decoded (for bytes)
or encoded (for strings) using the specified encoding without errors.
Args:
data: Input data
encoding: Encoding to validate
**options: Validation options
data: Input data - can be string or bytes
encoding: Encoding name to validate (e.g., "utf-8", "latin-1")
**options: Additional validation options (unused)
Returns:
bool: True if encoding is valid for data
bool: True if encoding is valid for data, False otherwise
"""
if isinstance(data, str):
try:
semantica/normalize/entity_normalizer.py
+273 -86
View File
@@ -1,20 +1,32 @@
"""
Entity Normalization Module
Handles normalization of named entities and proper nouns.
This module provides comprehensive entity normalization capabilities for the
Semantica framework, enabling standardization of named entities and proper nouns
across various formats and naming conventions.
Key Features:
- Entity name standardization
- Alias resolution and mapping
- Entity disambiguation
- Name variant handling
- Entity disambiguation (context-aware)
- Name variant handling (titles, honorifics, formats)
- Entity linking and resolution
- Support for multiple entity types (Person, Organization, etc.)
Main Classes:
- EntityNormalizer: Main entity normalization class
- AliasResolver: Entity alias resolution
- EntityDisambiguator: Entity disambiguation
- NameVariantHandler: Name variant processing
- EntityNormalizer: Main entity normalization coordinator
- AliasResolver: Entity alias resolution engine
- EntityDisambiguator: Entity disambiguation engine
- NameVariantHandler: Name variant processing engine
Example Usage:
>>> from semantica.normalize import EntityNormalizer
>>> normalizer = EntityNormalizer()
>>> normalized = normalizer.normalize_entity("John Doe", entity_type="Person")
>>> canonical = normalizer.resolve_aliases("J. Doe")
Author: Semantica Contributors
License: MIT
"""
import re
@@ -26,23 +38,36 @@ from ..utils.logging import get_logger
class EntityNormalizer:
"""
Entity normalization and standardization handler.
Entity normalization and standardization coordinator.
• Normalizes entity names and proper nouns
• Resolves entity aliases and variants
• Handles entity disambiguation
• Standardizes entity formats
• Links entities to canonical forms
• Supports multiple entity types
This class provides comprehensive entity normalization capabilities, coordinating
alias resolution, disambiguation, and name variant handling.
Features:
- Entity name normalization and standardization
- Alias resolution and mapping
- Entity disambiguation using context
- Name format standardization
- Entity linking to canonical forms
- Support for multiple entity types
Example Usage:
>>> normalizer = EntityNormalizer()
>>> normalized = normalizer.normalize_entity("John Doe", entity_type="Person")
>>> canonical = normalizer.resolve_aliases("J. Doe")
>>> linked = normalizer.link_entities(["John Doe", "J. Doe", "Johnny Doe"])
"""
def __init__(self, config=None, **kwargs):
def __init__(self, config: Optional[Dict[str, Any]] = None, **kwargs):
"""
Initialize entity normalizer.
Sets up the normalizer with alias resolver, disambiguator, and variant
handler components.
Args:
config: Configuration dictionary
**kwargs: Additional configuration options
config: Configuration dictionary (optional)
**kwargs: Additional configuration options (merged into config)
"""
self.logger = get_logger("entity_normalizer")
self.config = config or {}
@@ -51,18 +76,30 @@ class EntityNormalizer:
self.alias_resolver = AliasResolver(**self.config)
self.disambiguator = EntityDisambiguator(**self.config)
self.variant_handler = NameVariantHandler(**self.config)
self.logger.debug("Entity normalizer initialized")
def normalize_entity(self, entity_name: str, entity_type: Optional[str] = None, **options) -> str:
def normalize_entity(
self,
entity_name: str,
entity_type: Optional[str] = None,
resolve_aliases: bool = True,
**options
) -> str:
"""
Normalize entity name to standard form.
This method normalizes an entity name by cleaning whitespace, resolving
aliases, and standardizing the format based on entity type.
Args:
entity_name: Entity name to normalize
entity_type: Entity type (optional)
**options: Normalization options
entity_type: Entity type (optional, e.g., "Person", "Organization")
resolve_aliases: Whether to resolve aliases (default: True)
**options: Additional normalization options (unused)
Returns:
Normalized entity name
str: Normalized entity name in standard form
"""
if not entity_name:
return ""
@@ -74,7 +111,7 @@ class EntityNormalizer:
normalized = normalized.title() if entity_type == "Person" else normalized
# Resolve aliases
if options.get("resolve_aliases", True):
if resolve_aliases:
resolved = self.alias_resolver.resolve_aliases(normalized, entity_type=entity_type)
if resolved:
normalized = resolved
@@ -84,42 +121,67 @@ class EntityNormalizer:
return normalized
def resolve_aliases(self, entity_name: str, **context) -> Optional[str]:
def resolve_aliases(
self,
entity_name: str,
**context
) -> Optional[str]:
"""
Resolve entity aliases and variants.
This method attempts to resolve an entity name to its canonical form
using alias mapping.
Args:
entity_name: Entity name
**context: Context information
entity_name: Entity name to resolve
**context: Context information (e.g., entity_type)
Returns:
Resolved canonical form or None
Optional[str]: Resolved canonical form if found, None otherwise
"""
return self.alias_resolver.resolve_aliases(entity_name, **context)
def disambiguate_entity(self, entity_name: str, **context) -> Dict[str, Any]:
def disambiguate_entity(
self,
entity_name: str,
**context
) -> Dict[str, Any]:
"""
Disambiguate entity when multiple candidates exist.
This method disambiguates an entity name when multiple candidates exist,
using context information to select the most likely candidate.
Args:
entity_name: Entity name
**context: Context information
entity_name: Entity name to disambiguate
**context: Context information (e.g., entity_type, context text)
Returns:
Disambiguation result
dict: Disambiguation result containing:
- entity_name: Original entity name
- entity_type: Detected entity type
- confidence: Confidence score (0.0 to 1.0)
- candidates: List of candidate entity names
"""
return self.disambiguator.disambiguate(entity_name, **context)
def link_entities(self, entities: List[str], **options) -> Dict[str, str]:
def link_entities(
self,
entities: List[str],
**options
) -> Dict[str, str]:
"""
Link entities to canonical forms.
This method links a list of entity names to their canonical forms,
creating a mapping from original names to normalized names.
Args:
entities: List of entity names
**options: Linking options
entities: List of entity names to link
**options: Linking options (passed to normalize_entity)
Returns:
Dictionary mapping entities to canonical forms
dict: Dictionary mapping original entity names to canonical forms
"""
linked = {}
@@ -134,33 +196,54 @@ class AliasResolver:
"""
Entity alias resolution engine.
• Resolves entity aliases and nicknames
• Maps name variations to canonical forms
• Handles different naming conventions
• Processes cultural and linguistic variations
This class provides alias resolution capabilities, mapping entity name
variations and aliases to canonical forms.
Features:
- Entity alias and nickname resolution
- Name variation mapping
- Support for different naming conventions
- Cultural and linguistic variation handling
Example Usage:
>>> resolver = AliasResolver(alias_map={"j. doe": "John Doe"})
>>> canonical = resolver.resolve_aliases("J. Doe")
"""
def __init__(self, **config):
"""
Initialize alias resolver.
Sets up the resolver with alias mapping dictionary.
Args:
**config: Configuration options
**config: Configuration options:
- alias_map: Dictionary mapping aliases to canonical forms
"""
self.logger = get_logger("alias_resolver")
self.config = config
self.alias_map = config.get("alias_map", {})
self.logger.debug(f"Alias resolver initialized ({len(self.alias_map)} aliases)")
def resolve_aliases(self, entity_name: str, **context) -> Optional[str]:
def resolve_aliases(
self,
entity_name: str,
**context
) -> Optional[str]:
"""
Resolve entity aliases to canonical form.
This method looks up an entity name in the alias map and returns
its canonical form if found.
Args:
entity_name: Entity name
**context: Context information
entity_name: Entity name to resolve
**context: Context information (e.g., entity_type, currently unused)
Returns:
Resolved canonical form or None
Optional[str]: Resolved canonical form if found in alias map,
None otherwise
"""
# Check alias map
entity_lower = entity_name.lower()
@@ -175,30 +258,46 @@ class AliasResolver:
return None
def map_variants(self, entity_name: str, entity_type: str) -> str:
def map_variants(
self,
entity_name: str,
entity_type: str
) -> str:
"""
Map entity name variants.
This method maps entity name variants to a standard form based on
entity type. Currently returns the name as-is; can be extended for
variant mapping.
Args:
entity_name: Entity name
entity_type: Entity type
entity_name: Entity name to map
entity_type: Entity type (e.g., "Person", "Organization")
Returns:
Mapped variant
str: Mapped variant (currently returns entity_name as-is)
"""
# Simple variant mapping
# Simple variant mapping - can be extended
return entity_name
def handle_cultural_variations(self, entity_name: str, culture: Optional[str] = None) -> str:
def handle_cultural_variations(
self,
entity_name: str,
culture: Optional[str] = None
) -> str:
"""
Handle cultural and linguistic variations.
This method handles cultural and linguistic variations in entity names.
Currently returns the name as-is; can be extended for cultural
normalization.
Args:
entity_name: Entity name
culture: Culture identifier
entity_name: Entity name to process
culture: Culture identifier (optional, e.g., "en-US", "zh-CN")
Returns:
Culturally appropriate form
str: Culturally appropriate form (currently returns entity_name as-is)
"""
return entity_name
@@ -207,32 +306,59 @@ class EntityDisambiguator:
"""
Entity disambiguation engine.
• Disambiguates entities with multiple meanings
• Uses contextual information for disambiguation
• Applies machine learning models
• Handles entity type classification
This class provides entity disambiguation capabilities, using context
information to resolve ambiguous entity references.
Features:
- Context-aware entity disambiguation
- Entity type classification
- Confidence score calculation
- Candidate entity generation
Example Usage:
>>> disambiguator = EntityDisambiguator()
>>> result = disambiguator.disambiguate("Apple", context="technology company")
>>> entity_type = disambiguator.classify_entity_type("John Doe")
"""
def __init__(self, **config):
"""
Initialize entity disambiguator.
Sets up the disambiguator with configuration options.
Args:
**config: Configuration options
**config: Configuration options (currently unused)
"""
self.logger = get_logger("entity_disambiguator")
self.config = config
self.logger.debug("Entity disambiguator initialized")
def disambiguate(self, entity_name: str, **context) -> Dict[str, Any]:
def disambiguate(
self,
entity_name: str,
**context
) -> Dict[str, Any]:
"""
Disambiguate entity using context.
This method disambiguates an entity name using context information.
Currently provides a basic implementation; can be extended with
machine learning models for improved disambiguation.
Args:
entity_name: Entity name
**context: Context information
entity_name: Entity name to disambiguate
**context: Context information containing:
- entity_type: Entity type (optional)
- context: Text context (optional)
Returns:
Disambiguation result
dict: Disambiguation result containing:
- entity_name: Original entity name
- entity_type: Detected entity type
- confidence: Confidence score (0.0 to 1.0)
- candidates: List of candidate entity names
"""
entity_type = context.get("entity_type")
text_context = context.get("context", "")
@@ -244,17 +370,30 @@ class EntityDisambiguator:
"candidates": [entity_name]
}
def classify_entity_type(self, entity_name: str, **context) -> str:
def classify_entity_type(
self,
entity_name: str,
**context
) -> str:
"""
Classify entity type for disambiguation.
This method classifies the entity type using simple heuristics based
on name format. Can be extended with more sophisticated classification.
Args:
entity_name: Entity name
**context: Context information
entity_name: Entity name to classify
**context: Context information (currently unused)
Returns:
Entity type
str: Entity type classification:
- "Person": If name starts with uppercase and contains space
- "Organization": If name starts with uppercase
- "Entity": Otherwise
"""
if not entity_name:
return "Entity"
# Simple heuristic-based classification
if entity_name[0].isupper() and ' ' in entity_name:
return "Person"
@@ -263,16 +402,25 @@ class EntityDisambiguator:
else:
return "Entity"
def calculate_confidence(self, candidates: List[str], **context) -> Dict[str, float]:
def calculate_confidence(
self,
candidates: List[str],
**context
) -> Dict[str, float]:
"""
Calculate confidence scores for candidates.
This method calculates confidence scores for candidate entities.
Currently returns a default confidence of 0.8 for all candidates;
can be extended with more sophisticated scoring.
Args:
candidates: List of candidate entities
**context: Context information
candidates: List of candidate entity names
**context: Context information (currently unused)
Returns:
Dictionary of confidence scores
dict: Dictionary mapping candidate names to confidence scores
(0.0 to 1.0)
"""
return {candidate: 0.8 for candidate in candidates}
@@ -281,33 +429,54 @@ class NameVariantHandler:
"""
Name variant processing engine.
• Handles different name formats and variations
• Processes formal and informal names
• Manages name order variations
• Handles title and honorific processing
This class provides name variant handling capabilities, processing different
name formats, titles, and honorifics.
Features:
- Name format normalization (standard, title, lower)
- Title and honorific handling
- Name variant generation
- Format standardization
Example Usage:
>>> handler = NameVariantHandler()
>>> normalized = handler.normalize_name_format("Dr. John Doe", "standard")
>>> title_info = handler.handle_titles_and_honorifics("Mr. John Doe")
"""
def __init__(self, **config):
"""
Initialize name variant handler.
Sets up the handler with titles dictionary and configuration.
Args:
**config: Configuration options
**config: Configuration options:
- titles: Set of title strings (optional, uses default if not provided)
"""
self.logger = get_logger("name_variant_handler")
self.config = config
self.titles = {"Mr.", "Mrs.", "Ms.", "Dr.", "Prof.", "Sir", "Madam"}
self.titles = config.get("titles", {"Mr.", "Mrs.", "Ms.", "Dr.", "Prof.", "Sir", "Madam"})
self.logger.debug(f"Name variant handler initialized ({len(self.titles)} titles)")
def process_variants(self, entity_name: str, **options) -> List[str]:
def process_variants(
self,
entity_name: str,
**options
) -> List[str]:
"""
Process entity name variants.
This method generates a list of name variants for an entity, including
the original name and normalized forms.
Args:
entity_name: Entity name
**options: Processing options
entity_name: Entity name to process
**options: Processing options (unused)
Returns:
List of variants
list: List of name variant strings
"""
variants = [entity_name]
@@ -318,16 +487,26 @@ class NameVariantHandler:
return variants
def normalize_name_format(self, entity_name: str, format_type: str = "standard") -> str:
def normalize_name_format(
self,
entity_name: str,
format_type: str = "standard"
) -> str:
"""
Normalize name format.
This method normalizes the format of an entity name, removing titles
and applying the specified format type.
Args:
entity_name: Entity name
format_type: Format type ('standard', 'title', 'lower')
entity_name: Entity name to normalize
format_type: Format type (default: "standard"):
- "standard": Title case for each word part
- "title": Title case for entire name
- "lower": Lowercase
Returns:
Formatted name
str: Formatted name with titles removed
"""
# Remove titles
name = entity_name
@@ -347,15 +526,23 @@ class NameVariantHandler:
return name
def handle_titles_and_honorifics(self, entity_name: str) -> Dict[str, Any]:
def handle_titles_and_honorifics(
self,
entity_name: str
) -> Dict[str, Any]:
"""
Handle titles and honorifics in names.
This method extracts titles and honorifics from entity names, returning
the name without title and the extracted title.
Args:
entity_name: Entity name
entity_name: Entity name with potential title
Returns:
Dictionary with name and title
dict: Dictionary containing:
- name: Name without title
- title: Extracted title (None if no title found)
"""
title = None
name = entity_name
semantica/normalize/language_detector.py
+158 -41
View File
@@ -1,48 +1,111 @@
"""
Language detection utilities for Semantica framework.
Language Detection Module
This module provides multi-language detection capabilities
using langdetect and other language identification libraries.
This module provides comprehensive language detection capabilities for the
Semantica framework, enabling identification of text language using the
langdetect library.
Key Features:
- Multi-language detection (50+ languages)
- Confidence scoring
- Batch processing
- Top N language detection
- Language code to name mapping
Main Classes:
- LanguageDetector: Language detection coordinator
Example Usage:
>>> from semantica.normalize import LanguageDetector
>>> detector = LanguageDetector()
>>> language = detector.detect("Hello world")
>>> lang, confidence = detector.detect_with_confidence("Bonjour le monde")
>>> languages = detector.detect_multiple(text, top_n=3)
Author: Semantica Contributors
License: MIT
"""
from typing import Any, Dict, List, Optional, Tuple, Union
from langdetect import detect, detect_langs
from langdetect.lang_detect_exception import LangDetectException
try:
from langdetect import detect, detect_langs
from langdetect.lang_detect_exception import LangDetectException
LANGDETECT_AVAILABLE = True
except ImportError:
LANGDETECT_AVAILABLE = False
LangDetectException = Exception
from ..utils.exceptions import ProcessingError
from ..utils.logging import get_logger
class LanguageDetector:
"""Language detection utilities."""
"""
Language detection coordinator.
This class provides language detection capabilities using the langdetect
library, supporting detection with confidence scores and batch processing.
Features:
- Multi-language detection (50+ languages)
- Confidence scoring
- Batch processing
- Top N language detection
- Language code to name mapping
Example Usage:
>>> detector = LanguageDetector()
>>> language = detector.detect("Hello world")
>>> lang, confidence = detector.detect_with_confidence("Bonjour")
>>> is_english = detector.is_language(text, "en")
"""
def __init__(self, **config):
"""
Initialize language detector.
Sets up the detector with default language and minimum confidence threshold.
Args:
**config: Configuration options
**config: Configuration options:
- default_language: Default language code (default: "en")
- min_confidence: Minimum confidence threshold (default: 0.5)
"""
self.logger = get_logger("language_detector")
self.config = config
self.default_language = config.get("default_language", "en")
self.min_confidence = config.get("min_confidence", 0.5)
if not LANGDETECT_AVAILABLE:
self.logger.warning("langdetect library not available, language detection will be limited")
self.logger.debug(f"Language detector initialized (default={self.default_language})")
def detect(self, text: str, **options) -> str:
"""
Detect language of text.
This method detects the language of input text using langdetect.
Returns the default language if detection fails or text is too short.
Args:
text: Input text
**options: Detection options
text: Input text to analyze
**options: Detection options (unused)
Returns:
str: Detected language code
str: Detected language code (e.g., "en", "fr", "de")
Note:
Requires minimum text length of 10 characters for reliable detection.
Returns default_language if text is too short or detection fails.
"""
if not text or len(text.strip()) < 10:
return self.default_language
if not LANGDETECT_AVAILABLE:
return self.default_language
try:
language = detect(text)
return language
@@ -53,20 +116,33 @@ class LanguageDetector:
self.logger.error(f"Language detection error: {e}")
return self.default_language
def detect_with_confidence(self, text: str, **options) -> Tuple[str, float]:
def detect_with_confidence(
self,
text: str,
**options
) -> Tuple[str, float]:
"""
Detect language with confidence score.
This method detects the language of text and returns both the language
code and confidence score. Only returns detected language if confidence
meets the minimum threshold.
Args:
text: Input text
**options: Detection options
text: Input text to analyze
**options: Detection options (unused)
Returns:
tuple: (language_code, confidence_score)
tuple: (language_code, confidence_score) where:
- language_code: Detected language code
- confidence_score: Confidence score between 0.0 and 1.0
"""
if not text or len(text.strip()) < 10:
return (self.default_language, 0.0)
if not LANGDETECT_AVAILABLE:
return (self.default_language, 0.0)
try:
languages = detect_langs(text)
if languages:
@@ -84,21 +160,33 @@ class LanguageDetector:
self.logger.error(f"Language detection error: {e}")
return (self.default_language, 0.0)
def detect_multiple(self, text: str, top_n: int = 3, **options) -> List[Tuple[str, float]]:
def detect_multiple(
self,
text: str,
top_n: int = 3,
**options
) -> List[Tuple[str, float]]:
"""
Detect top N languages with confidence scores.
This method detects multiple candidate languages for text, returning
the top N languages sorted by confidence.
Args:
text: Input text
top_n: Number of top languages to return
**options: Detection options
text: Input text to analyze
top_n: Number of top languages to return (default: 3)
**options: Detection options (unused)
Returns:
list: List of (language_code, confidence_score) tuples
list: List of (language_code, confidence_score) tuples, sorted by
confidence (highest first)
"""
if not text or len(text.strip()) < 10:
return [(self.default_language, 0.0)]
if not LANGDETECT_AVAILABLE:
return [(self.default_language, 0.0)]
try:
languages = detect_langs(text)
if languages:
@@ -112,57 +200,86 @@ class LanguageDetector:
self.logger.error(f"Language detection error: {e}")
return [(self.default_language, 0.0)]
def detect_batch(self, texts: List[str], **options) -> List[str]:
def detect_batch(
self,
texts: List[str],
**options
) -> List[str]:
"""
Detect languages for multiple texts in batch.
This method processes multiple texts in batch, detecting the language
for each text.
Args:
texts: List of texts to analyze
**options: Detection options
**options: Detection options (passed to detect method)
Returns:
list: List of detected language codes
list: List of detected language codes (one per input text)
"""
return [self.detect(text, **options) for text in texts]
def detect_batch_with_confidence(self, texts: List[str], **options) -> List[Tuple[str, float]]:
def detect_batch_with_confidence(
self,
texts: List[str],
**options
) -> List[Tuple[str, float]]:
"""
Detect languages with confidence for multiple texts.
This method processes multiple texts in batch, detecting the language
and confidence score for each text.
Args:
texts: List of texts to analyze
**options: Detection options
**options: Detection options (passed to detect_with_confidence method)
Returns:
list: List of (language_code, confidence_score) tuples
list: List of (language_code, confidence_score) tuples (one per input text)
"""
return [self.detect_with_confidence(text, **options) for text in texts]
def is_language(self, text: str, target_language: str, **options) -> bool:
def is_language(
self,
text: str,
target_language: str,
min_confidence: Optional[float] = None,
**options
) -> bool:
"""
Check if text is in target language.
This method checks whether the detected language matches the target
language and meets the minimum confidence threshold.
Args:
text: Input text
target_language: Target language code
**options: Detection options
text: Input text to check
target_language: Target language code (e.g., "en", "fr")
min_confidence: Minimum confidence threshold (optional, uses
instance min_confidence if not provided)
**options: Detection options (unused)
Returns:
bool: True if text is in target language
bool: True if text is in target language with sufficient confidence,
False otherwise
"""
detected, confidence = self.detect_with_confidence(text, **options)
min_confidence = options.get("min_confidence", self.min_confidence)
return detected == target_language and confidence >= min_confidence
threshold = min_confidence if min_confidence is not None else self.min_confidence
return detected == target_language and confidence >= threshold
def get_language_name(self, language_code: str) -> str:
"""
Get language name from code.
This method converts a language code to its human-readable name.
Args:
language_code: Language code (e.g., 'en', 'fr')
language_code: Language code (e.g., "en", "fr", "de")
Returns:
str: Language name
str: Language name (e.g., "English", "French", "German").
Returns uppercase code if name not found.
"""
language_names = {
'en': 'English',
semantica/normalize/number_normalizer.py
+285 -91
View File
@@ -1,20 +1,33 @@
"""
Number and Quantity Normalization Module
Handles normalization of numbers, quantities, and numerical expressions.
This module provides comprehensive number and quantity normalization capabilities
for the Semantica framework, enabling standardization of numerical data across
various formats and units.
Key Features:
- Number format standardization
- Unit conversion and normalization
- Currency handling
- Number format standardization (integers, floats, percentages)
- Unit conversion and normalization (length, weight, volume)
- Currency handling (symbols, codes, conversion)
- Percentage processing
- Scientific notation handling
- Quantity parsing and normalization
Main Classes:
- NumberNormalizer: Main number normalization class
- NumberNormalizer: Main number normalization coordinator
- UnitConverter: Unit conversion engine
- CurrencyNormalizer: Currency processing
- CurrencyNormalizer: Currency processing engine
- ScientificNotationHandler: Scientific notation processor
Example Usage:
>>> from semantica.normalize import NumberNormalizer
>>> normalizer = NumberNormalizer()
>>> number = normalizer.normalize_number("1,234.56")
>>> quantity = normalizer.normalize_quantity("5 kg")
>>> currency = normalizer.process_currency("$100")
Author: Semantica Contributors
License: MIT
"""
import re
@@ -26,23 +39,37 @@ from ..utils.logging import get_logger
class NumberNormalizer:
"""
Number and quantity normalization handler.
Number and quantity normalization coordinator.
• Normalizes numbers to standard formats
• Handles various number representations
• Processes quantities and units
• Manages currency and percentage values
• Standardizes numerical expressions
• Supports multiple number systems
This class provides comprehensive number and quantity normalization
capabilities, coordinating unit conversion, currency processing, and
scientific notation handling.
Features:
- Number format standardization
- Quantity parsing and normalization
- Unit conversion
- Currency processing
- Percentage handling
- Scientific notation support
Example Usage:
>>> normalizer = NumberNormalizer()
>>> number = normalizer.normalize_number("1,234.56")
>>> quantity = normalizer.normalize_quantity("5 kg")
>>> currency = normalizer.process_currency("$100")
"""
def __init__(self, config=None, **kwargs):
def __init__(self, config: Optional[Dict[str, Any]] = None, **kwargs):
"""
Initialize number normalizer.
Sets up the normalizer with unit converter, currency normalizer, and
scientific notation handler components.
Args:
config: Configuration dictionary
**kwargs: Additional configuration options
config: Configuration dictionary (optional)
**kwargs: Additional configuration options (merged into config)
"""
self.logger = get_logger("number_normalizer")
self.config = config or {}
@@ -51,17 +78,31 @@ class NumberNormalizer:
self.unit_converter = UnitConverter(**self.config)
self.currency_normalizer = CurrencyNormalizer(**self.config)
self.scientific_handler = ScientificNotationHandler(**self.config)
self.logger.debug("Number normalizer initialized")
def normalize_number(self, number_input: Union[str, int, float], **options) -> Union[int, float]:
def normalize_number(
self,
number_input: Union[str, int, float],
**options
) -> Union[int, float]:
"""
Normalize number to standard format.
This method normalizes number input (string, int, or float) to a
standard numeric format, handling percentages and scientific notation.
Args:
number_input: Number input (string, int, or float)
**options: Normalization options
number_input: Number input - can be:
- String (e.g., "1,234.56", "50%", "1.5e3")
- int or float
**options: Normalization options (unused)
Returns:
Normalized number
Union[int, float]: Normalized number (int if no decimal, float otherwise)
Raises:
ValidationError: If number input type is unsupported or parsing fails
"""
if isinstance(number_input, (int, float)):
return number_input
@@ -92,16 +133,29 @@ class NumberNormalizer:
except ValueError:
raise ValidationError(f"Unable to parse number: {number_input}")
def normalize_quantity(self, quantity_input: str, **options) -> Dict[str, Any]:
def normalize_quantity(
self,
quantity_input: str,
**options
) -> Dict[str, Any]:
"""
Normalize quantity with units.
This method parses and normalizes quantity strings containing values
and units (e.g., "5 kg", "10 meters").
Args:
quantity_input: Quantity string (e.g., "5 kg", "10 meters")
**options: Normalization options
quantity_input: Quantity string (e.g., "5 kg", "10 meters", "3.5 liters")
**options: Normalization options (unused)
Returns:
Normalized quantity dictionary
dict: Normalized quantity dictionary containing:
- value: Numeric value (float)
- unit: Normalized unit name (str)
- original: Original quantity string (str)
Raises:
ValidationError: If quantity format is invalid or parsing fails
"""
# Parse quantity and unit
pattern = r'([\d.,\s]+)\s*([a-zA-Z]+)'
@@ -127,51 +181,87 @@ class NumberNormalizer:
else:
raise ValidationError(f"Invalid quantity format: {quantity_input}")
def convert_units(self, value: float, from_unit: str, to_unit: str) -> float:
def convert_units(
self,
value: float,
from_unit: str,
to_unit: str
) -> float:
"""
Convert value between units.
This method converts a numeric value from one unit to another,
supporting length, weight, and volume conversions.
Args:
value: Value to convert
from_unit: Source unit
to_unit: Target unit
value: Numeric value to convert
from_unit: Source unit (e.g., "kg", "meter", "liter")
to_unit: Target unit (e.g., "pound", "mile", "gallon")
Returns:
Converted value
float: Converted value in target unit
Raises:
ValidationError: If units are incompatible or conversion fails
"""
return self.unit_converter.convert_units(value, from_unit, to_unit)
def process_currency(self, currency_input: str, **options) -> Dict[str, Any]:
def process_currency(
self,
currency_input: str,
default_currency: str = "USD",
**options
) -> Dict[str, Any]:
"""
Process currency values.
This method parses and normalizes currency strings, extracting
amount and currency code.
Args:
currency_input: Currency string (e.g., "$100", "100 USD")
**options: Processing options
currency_input: Currency string (e.g., "$100", "100 USD", "€50")
default_currency: Default currency code if not found (default: "USD")
**options: Additional processing options (unused)
Returns:
Normalized currency dictionary
dict: Normalized currency dictionary containing:
- amount: Numeric amount (float or None)
- currency: Currency code (str, e.g., "USD", "EUR")
- original: Original currency string (str)
"""
return self.currency_normalizer.normalize_currency(currency_input, **options)
return self.currency_normalizer.normalize_currency(
currency_input, default_currency=default_currency, **options
)
class UnitConverter:
"""
Unit conversion engine.
• Converts between different units
• Handles various unit systems
• Manages conversion factors
• Processes compound units
• Handles unit validation
This class provides unit conversion capabilities, supporting length,
weight, and volume conversions with validation.
Features:
- Unit conversion (length, weight, volume)
- Conversion factor management
- Unit validation
- Unit normalization
- Support for multiple unit systems
Example Usage:
>>> converter = UnitConverter()
>>> converted = converter.convert_units(100, "kg", "pound")
>>> normalized = converter.normalize_unit("km")
"""
def __init__(self, **config):
"""
Initialize unit converter.
Sets up the converter with conversion factors and unit categories.
Args:
**config: Configuration options
**config: Configuration options (currently unused)
"""
self.logger = get_logger("unit_converter")
self.config = config
@@ -206,18 +296,31 @@ class UnitConverter:
"weight": ["kilogram", "gram", "pound", "ounce"],
"volume": ["liter", "milliliter", "gallon"],
}
self.logger.debug("Unit converter initialized")
def convert_units(self, value: float, from_unit: str, to_unit: str) -> float:
def convert_units(
self,
value: float,
from_unit: str,
to_unit: str
) -> float:
"""
Convert value between units.
This method converts a numeric value from one unit to another within
the same category (length, weight, or volume).
Args:
value: Value to convert
from_unit: Source unit
to_unit: Target unit
value: Numeric value to convert
from_unit: Source unit name (e.g., "kg", "meter", "liter")
to_unit: Target unit name (e.g., "pound", "mile", "gallon")
Returns:
Converted value
float: Converted value in target unit
Raises:
ValidationError: If units are incompatible or not in same category
"""
from_unit = from_unit.lower()
to_unit = to_unit.lower()
@@ -236,16 +339,23 @@ class UnitConverter:
return converted_value
def validate_units(self, from_unit: str, to_unit: str) -> bool:
def validate_units(
self,
from_unit: str,
to_unit: str
) -> bool:
"""
Validate unit conversion compatibility.
This method validates that two units are compatible for conversion,
checking that they exist and are in the same category.
Args:
from_unit: Source unit
to_unit: Target unit
from_unit: Source unit name
to_unit: Target unit name
Returns:
True if units are compatible
bool: True if units are compatible (same category), False otherwise
"""
from_unit = from_unit.lower()
to_unit = to_unit.lower()
@@ -266,16 +376,23 @@ class UnitConverter:
return from_category == to_category
def get_conversion_factor(self, from_unit: str, to_unit: str) -> float:
def get_conversion_factor(
self,
from_unit: str,
to_unit: str
) -> float:
"""
Get conversion factor between units.
This method calculates the conversion factor to convert from one unit
to another. If to_unit is "base", returns factor to base unit.
Args:
from_unit: Source unit
to_unit: Target unit
from_unit: Source unit name
to_unit: Target unit name or "base" for base unit
Returns:
Conversion factor
float: Conversion factor (multiply source value by this to get target)
"""
from_unit = from_unit.lower()
@@ -292,11 +409,15 @@ class UnitConverter:
"""
Normalize unit name to standard form.
This method normalizes unit abbreviations to full unit names
(e.g., "km" -> "kilometer", "kg" -> "kilogram").
Args:
unit: Unit name
unit: Unit name or abbreviation (e.g., "km", "kg", "m")
Returns:
Normalized unit name
str: Normalized unit name (full name if abbreviation found,
original unit otherwise)
"""
unit_lower = unit.lower()
@@ -321,19 +442,30 @@ class CurrencyNormalizer:
"""
Currency normalization engine.
• Normalizes currency values and codes
• Handles currency conversion
• Manages exchange rates
• Processes currency symbols
• Handles multiple currencies
This class provides currency processing capabilities, including symbol
and code recognition, amount extraction, and currency validation.
Features:
- Currency symbol and code recognition
- Amount extraction from currency strings
- Currency code validation
- Support for multiple currencies
- Currency conversion (placeholder for exchange rate integration)
Example Usage:
>>> normalizer = CurrencyNormalizer()
>>> result = normalizer.normalize_currency("$100")
>>> is_valid = normalizer.validate_currency_code("USD")
"""
def __init__(self, **config):
"""
Initialize currency normalizer.
Sets up the normalizer with currency symbols and codes dictionaries.
Args:
**config: Configuration options
**config: Configuration options (currently unused)
"""
self.logger = get_logger("currency_normalizer")
self.config = config
@@ -353,17 +485,31 @@ class CurrencyNormalizer:
}
self.currency_codes = ["USD", "EUR", "GBP", "JPY", "CNY", "INR", "AUD", "CAD", "CHF", "SEK", "NOK", "DKK"]
self.logger.debug("Currency normalizer initialized")
def normalize_currency(self, currency_input: str, **options) -> Dict[str, Any]:
def normalize_currency(
self,
currency_input: str,
default_currency: str = "USD",
**options
) -> Dict[str, Any]:
"""
Normalize currency value and code.
This method parses currency strings, extracting amount and currency
code from symbols or text.
Args:
currency_input: Currency string
**options: Normalization options
currency_input: Currency string (e.g., "$100", "100 USD", "€50")
default_currency: Default currency code if not found (default: "USD")
**options: Additional normalization options (unused)
Returns:
Normalized currency dictionary
dict: Normalized currency dictionary containing:
- amount: Numeric amount (float or None if not found)
- currency: Currency code (str, e.g., "USD", "EUR")
- original: Original currency string (str)
"""
# Extract currency symbol or code
currency_code = None
@@ -404,9 +550,9 @@ class CurrencyNormalizer:
except ValueError:
amount = None
# Default to USD if no currency found
# Default to specified currency if no currency found
if not currency_code:
currency_code = options.get("default_currency", "USD")
currency_code = default_currency
return {
"amount": amount,
@@ -414,19 +560,31 @@ class CurrencyNormalizer:
"original": currency_input
}
def convert_currency(self, amount: float, from_currency: str, to_currency: str) -> float:
def convert_currency(
self,
amount: float,
from_currency: str,
to_currency: str
) -> float:
"""
Convert currency between different currencies.
This method converts an amount from one currency to another.
Currently a placeholder; requires exchange rate API integration
for production use.
Args:
amount: Amount to convert
from_currency: Source currency
to_currency: Target currency
from_currency: Source currency code (e.g., "USD")
to_currency: Target currency code (e.g., "EUR")
Returns:
Converted amount
float: Converted amount (currently returns original amount)
Note:
This is a placeholder implementation. In production, you would
integrate with an exchange rate API to fetch current rates.
"""
# Note: This is a placeholder. In production, you'd fetch exchange rates
self.logger.warning("Currency conversion requires exchange rate API")
return amount
@@ -434,11 +592,14 @@ class CurrencyNormalizer:
"""
Validate currency code.
This method validates that a currency code is in the supported
list of currency codes.
Args:
currency_code: Currency code to validate
currency_code: Currency code to validate (e.g., "USD", "EUR")
Returns:
True if valid
bool: True if currency code is valid, False otherwise
"""
return currency_code.upper() in self.currency_codes
@@ -447,63 +608,96 @@ class ScientificNotationHandler:
"""
Scientific notation processing engine.
• Handles scientific notation numbers
• Processes exponential formats
• Manages precision and significant digits
• Converts between formats
This class provides scientific notation handling capabilities, including
parsing, conversion, and precision normalization.
Features:
- Scientific notation parsing
- Format conversion
- Precision normalization
- Significant digit handling
Example Usage:
>>> handler = ScientificNotationHandler()
>>> number = handler.parse_scientific_notation("1.5e3")
>>> notation = handler.convert_to_scientific(1500, precision=2)
"""
def __init__(self, **config):
"""
Initialize scientific notation handler.
Sets up the handler with configuration options.
Args:
**config: Configuration options
**config: Configuration options (currently unused)
"""
self.logger = get_logger("scientific_notation_handler")
self.config = config
self.logger.debug("Scientific notation handler initialized")
def parse_scientific_notation(self, number_string: str) -> float:
"""
Parse scientific notation number.
This method parses a scientific notation string (e.g., "1.5e3", "2E-4")
and returns the numeric value.
Args:
number_string: Scientific notation string
number_string: Scientific notation string (e.g., "1.5e3", "2E-4")
Returns:
Parsed number as float
float: Parsed numeric value
Raises:
ValidationError: If string is not valid scientific notation
"""
try:
return float(number_string)
except ValueError:
raise ValidationError(f"Invalid scientific notation: {number_string}")
except ValueError as e:
raise ValidationError(f"Invalid scientific notation: {number_string}") from e
def convert_to_scientific(self, number: float, precision: Optional[int] = None) -> str:
def convert_to_scientific(
self,
number: float,
precision: Optional[int] = None
) -> str:
"""
Convert number to scientific notation.
This method converts a numeric value to scientific notation string
format (e.g., "1.5e+03").
Args:
number: Number to convert
precision: Precision (number of decimal places)
number: Numeric value to convert
precision: Number of decimal places (optional, uses default if None)
Returns:
Scientific notation string
str: Scientific notation string (e.g., "1.5e+03", "2.0e-04")
"""
if precision is not None:
return f"{number:.{precision}e}"
else:
return f"{number:e}"
def normalize_precision(self, number: float, significant_digits: int) -> float:
def normalize_precision(
self,
number: float,
significant_digits: int
) -> float:
"""
Normalize number precision.
This method normalizes a number to a specified number of significant
digits, preserving the order of magnitude.
Args:
number: Number to normalize
significant_digits: Number of significant digits
number: Numeric value to normalize
significant_digits: Number of significant digits to preserve
Returns:
Normalized number
float: Normalized number with specified significant digits
"""
if number == 0:
return 0.0
semantica/normalize/text_cleaner.py
+169 -51
View File
@@ -1,32 +1,85 @@
"""
Text cleaning utilities for Semantica framework.
Text Cleaning Module
This module provides text cleaning and preprocessing functions
for HTML removal, whitespace normalization, and text sanitization.
This module provides comprehensive text cleaning and preprocessing capabilities
for the Semantica framework, enabling removal of HTML, normalization of
whitespace and Unicode, and text sanitization.
Key Features:
- HTML tag removal (with BeautifulSoup support)
- Whitespace normalization
- Unicode normalization (NFC, NFD, NFKC, NFKD)
- Special character removal
- Text sanitization (security-focused)
- Batch processing
Main Classes:
- TextCleaner: Text cleaning coordinator
Example Usage:
>>> from semantica.normalize import TextCleaner
>>> cleaner = TextCleaner()
>>> cleaned = cleaner.clean(text, remove_html=True, normalize_whitespace=True)
>>> sanitized = cleaner.sanitize(text)
Author: Semantica Contributors
License: MIT
"""
import re
import unicodedata
from typing import Any, Dict, List, Optional, Union
from bs4 import BeautifulSoup
try:
from bs4 import BeautifulSoup
BEAUTIFULSOUP_AVAILABLE = True
except ImportError:
BEAUTIFULSOUP_AVAILABLE = False
BeautifulSoup = None
from ..utils.exceptions import ProcessingError
from ..utils.logging import get_logger
class TextCleaner:
"""Text cleaning utilities."""
"""
Text cleaning coordinator.
This class provides comprehensive text cleaning capabilities, including
HTML removal, whitespace normalization, Unicode normalization, and
text sanitization.
Features:
- HTML tag removal (with BeautifulSoup support)
- Whitespace normalization
- Unicode normalization
- Special character removal
- Text sanitization
- Batch processing
Example Usage:
>>> cleaner = TextCleaner()
>>> cleaned = cleaner.clean(text, remove_html=True)
>>> sanitized = cleaner.sanitize(text)
>>> batch_cleaned = cleaner.clean_batch(texts)
"""
def __init__(self, **config):
"""
Initialize text cleaner.
Sets up the cleaner with configuration options.
Args:
**config: Configuration options
**config: Configuration options (currently unused)
"""
self.logger = get_logger("text_cleaner")
self.config = config
if not BEAUTIFULSOUP_AVAILABLE:
self.logger.warning("BeautifulSoup not available, HTML removal will use regex fallback")
self.logger.debug("Text cleaner initialized")
def clean(
self,
@@ -35,19 +88,31 @@ class TextCleaner:
normalize_whitespace: bool = True,
normalize_unicode: bool = True,
remove_special_chars: bool = False,
unicode_form: str = "NFC",
allow_spaces: bool = True,
**options
) -> str:
"""
Clean text with various options.
This method performs comprehensive text cleaning by applying multiple
cleaning operations in sequence: HTML removal, Unicode normalization,
whitespace normalization, and special character removal.
Args:
text: Input text to clean
remove_html: Whether to remove HTML tags
normalize_whitespace: Whether to normalize whitespace
normalize_unicode: Whether to normalize unicode
remove_special_chars: Whether to remove special characters
**options: Additional cleaning options
remove_html: Whether to remove HTML tags (default: True)
normalize_whitespace: Whether to normalize whitespace (default: True)
normalize_unicode: Whether to normalize Unicode (default: True)
remove_special_chars: Whether to remove special characters (default: False)
unicode_form: Unicode normalization form (default: "NFC"):
- "NFC": Canonical composition
- "NFD": Canonical decomposition
- "NFKC": Compatibility composition
- "NFKD": Compatibility decomposition
allow_spaces: Whether to allow spaces when removing special chars (default: True)
**options: Additional cleaning options (unused)
Returns:
str: Cleaned text
"""
@@ -62,7 +127,7 @@ class TextCleaner:
# Normalize unicode
if normalize_unicode:
cleaned = self.normalize_unicode(cleaned, form=options.get("unicode_form", "NFC"))
cleaned = self.normalize_unicode(cleaned, form=unicode_form)
# Normalize whitespace
if normalize_whitespace:
@@ -70,7 +135,7 @@ class TextCleaner:
# Remove special characters
if remove_special_chars:
cleaned = self.remove_special_chars(cleaned, allow_spaces=options.get("allow_spaces", True))
cleaned = self.remove_special_chars(cleaned, allow_spaces=allow_spaces)
return cleaned.strip()
@@ -78,22 +143,28 @@ class TextCleaner:
"""
Remove HTML tags from text.
This method removes HTML tags from text, optionally preserving structure
using BeautifulSoup for better parsing.
Args:
text: Input text with HTML
preserve_structure: Whether to preserve structure (use BeautifulSoup)
text: Input text containing HTML tags
preserve_structure: Whether to preserve structure using BeautifulSoup
(default: False, uses regex if False or BeautifulSoup
unavailable)
Returns:
str: Text without HTML tags
str: Text without HTML tags and with HTML entities decoded
"""
if not text:
return ""
if preserve_structure:
if preserve_structure and BEAUTIFULSOUP_AVAILABLE and BeautifulSoup:
try:
soup = BeautifulSoup(text, 'html.parser')
return soup.get_text(separator='\n', strip=True)
except Exception:
# Fallback to regex if BeautifulSoup fails
except Exception as e:
self.logger.warning(f"BeautifulSoup parsing failed, using regex fallback: {e}")
# Fallback to regex
pass
# Remove HTML tags using regex
@@ -116,11 +187,15 @@ class TextCleaner:
"""
Normalize whitespace in text.
This method normalizes whitespace by replacing tabs and newlines with
spaces and collapsing multiple spaces into single spaces.
Args:
text: Input text
text: Input text with potentially irregular whitespace
Returns:
str: Text with normalized whitespace
str: Text with normalized whitespace (tabs/newlines -> spaces,
multiple spaces -> single space)
"""
if not text:
return ""
@@ -135,14 +210,21 @@ class TextCleaner:
def normalize_unicode(self, text: str, form: str = "NFC") -> str:
"""
Normalize unicode characters.
Normalize Unicode characters.
This method normalizes Unicode characters using the specified
normalization form.
Args:
text: Input text
form: Normalization form (NFC, NFD, NFKC, NFKD)
text: Input text to normalize
form: Unicode normalization form (default: "NFC"):
- "NFC": Canonical composition
- "NFD": Canonical decomposition
- "NFKC": Compatibility composition
- "NFKD": Compatibility decomposition
Returns:
str: Normalized text
str: Unicode-normalized text (returns original text if normalization fails)
"""
if not text:
return ""
@@ -153,17 +235,28 @@ class TextCleaner:
self.logger.warning(f"Failed to normalize unicode: {e}")
return text
def remove_special_chars(self, text: str, allow_spaces: bool = True, allow_newlines: bool = False) -> str:
def remove_special_chars(
self,
text: str,
allow_spaces: bool = True,
allow_newlines: bool = False
) -> str:
"""
Remove special characters from text.
This method removes special characters from text, keeping only
alphanumeric characters and optionally spaces and newlines.
Args:
text: Input text
allow_spaces: Whether to allow spaces
allow_newlines: Whether to allow newlines
text: Input text to process
allow_spaces: Whether to allow spaces (default: True)
allow_newlines: Whether to allow newlines (default: False)
Returns:
str: Text without special characters
str: Text with special characters removed, keeping only:
- Alphanumeric characters
- Spaces (if allow_spaces=True)
- Newlines (if allow_newlines=True)
"""
if not text:
return ""
@@ -180,16 +273,25 @@ class TextCleaner:
return re.sub(pattern, '', text)
def sanitize(self, text: str, **options) -> str:
def sanitize(
self,
text: str,
remove_data_urls: bool = False,
**options
) -> str:
"""
Sanitize text for security.
This method sanitizes text by removing potentially dangerous content
like script tags, iframe tags, and javascript: URLs.
Args:
text: Input text
**options: Sanitization options
text: Input text to sanitize
remove_data_urls: Whether to remove data: URLs (default: False)
**options: Additional sanitization options (unused)
Returns:
str: Sanitized text
str: Sanitized text with dangerous content removed
"""
if not text:
return ""
@@ -202,21 +304,30 @@ class TextCleaner:
text = re.sub(r'javascript:', '', text, flags=re.IGNORECASE)
# Remove data: URLs if requested
if options.get("remove_data_urls", False):
if remove_data_urls:
text = re.sub(r'data:[^;]*;base64,', '', text, flags=re.IGNORECASE)
return text
def trim(self, text: str, **options) -> str:
def trim(
self,
text: str,
remove_empty_lines: bool = False,
**options
) -> str:
"""
Trim text.
This method trims leading and trailing whitespace from text, optionally
removing empty lines.
Args:
text: Input text
**options: Trimming options
text: Input text to trim
remove_empty_lines: Whether to remove empty lines (default: False)
**options: Additional trimming options (unused)
Returns:
str: Trimmed text
str: Trimmed text with leading/trailing whitespace removed
"""
if not text:
return ""
@@ -225,21 +336,28 @@ class TextCleaner:
text = text.strip()
# Remove empty lines if requested
if options.get("remove_empty_lines", False):
if remove_empty_lines:
lines = [line for line in text.split('\n') if line.strip()]
text = '\n'.join(lines)
return text
def clean_batch(self, texts: List[str], **options) -> List[str]:
def clean_batch(
self,
texts: List[str],
**options
) -> List[str]:
"""
Clean multiple texts in batch.
This method processes multiple texts in batch, applying the same
cleaning operations to each text.
Args:
texts: List of texts to clean
**options: Cleaning options
**options: Cleaning options (passed to clean method)
Returns:
list: List of cleaned texts
list: List of cleaned texts (one per input text)
"""
return [self.clean(text, **options) for text in texts]
semantica/normalize/text_normalizer.py
+276 -93
View File
@@ -1,20 +1,32 @@
"""
Text Normalization Module
Handles text cleaning, normalization, and standardization.
This module provides comprehensive text normalization capabilities for the
Semantica framework, enabling standardization of text content across various
formats and encodings.
Key Features:
- Text cleaning and sanitization
- Unicode normalization
- Case normalization
- Whitespace handling
- Special character processing
- Unicode normalization (NFC, NFD, NFKC, NFKD)
- Case normalization (lower, upper, title, preserve)
- Whitespace handling (normalization, line breaks, indentation)
- Special character processing (punctuation, diacritics)
- Format standardization
Main Classes:
- TextNormalizer: Main text normalization class
- UnicodeNormalizer: Unicode processing
- WhitespaceNormalizer: Whitespace handling
- SpecialCharacterProcessor: Special character handling
- TextNormalizer: Main text normalization coordinator
- UnicodeNormalizer: Unicode processing engine
- WhitespaceNormalizer: Whitespace handling engine
- SpecialCharacterProcessor: Special character processing engine
Example Usage:
>>> from semantica.normalize import TextNormalizer
>>> normalizer = TextNormalizer()
>>> normalized = normalizer.normalize_text("Hello World", case="lower")
>>> cleaned = normalizer.clean_text(text, remove_html=True)
Author: Semantica Contributors
License: MIT
"""
import re
@@ -28,23 +40,38 @@ from .text_cleaner import TextCleaner
class TextNormalizer:
"""
Text normalization and cleaning handler.
Text normalization and cleaning coordinator.
• Cleans and normalizes text content
• Handles various text encodings
• Processes special characters and symbols
• Standardizes text formatting
• Removes unwanted content and noise
• Supports multiple languages and scripts
This class provides comprehensive text normalization capabilities, coordinating
Unicode normalization, whitespace handling, special character processing,
and text cleaning.
Features:
- Text cleaning and sanitization
- Unicode normalization
- Case normalization
- Whitespace handling
- Special character processing
- Format standardization
- Batch processing
Example Usage:
>>> normalizer = TextNormalizer()
>>> normalized = normalizer.normalize_text("Hello World", case="lower")
>>> cleaned = normalizer.clean_text(text, remove_html=True)
>>> batch = normalizer.process_batch(texts)
"""
def __init__(self, config=None, **kwargs):
def __init__(self, config: Optional[Dict[str, Any]] = None, **kwargs):
"""
Initialize text normalizer.
Sets up the normalizer with text cleaner, Unicode normalizer, whitespace
normalizer, and special character processor components.
Args:
config: Configuration dictionary
**kwargs: Additional configuration options
config: Configuration dictionary (optional)
**kwargs: Additional configuration options (merged into config)
"""
self.logger = get_logger("text_normalizer")
self.config = config or {}
@@ -54,17 +81,44 @@ class TextNormalizer:
self.unicode_normalizer = UnicodeNormalizer(**self.config)
self.whitespace_normalizer = WhitespaceNormalizer(**self.config)
self.special_char_processor = SpecialCharacterProcessor(**self.config)
self.logger.debug("Text normalizer initialized")
def normalize_text(self, text: str, **options) -> str:
def normalize_text(
self,
text: str,
unicode_form: str = "NFC",
case: str = "preserve",
normalize_diacritics: bool = False,
line_break_type: str = "unix",
**options
) -> str:
"""
Normalize text content.
This method performs comprehensive text normalization by applying
Unicode normalization, whitespace normalization, special character
processing, and case normalization in sequence.
Args:
text: Input text
**options: Normalization options
text: Input text to normalize
unicode_form: Unicode normalization form (default: "NFC"):
- "NFC": Canonical composition
- "NFD": Canonical decomposition
- "NFKC": Compatibility composition
- "NFKD": Compatibility decomposition
case: Case normalization type (default: "preserve"):
- "preserve": Keep original case
- "lower": Convert to lowercase
- "upper": Convert to uppercase
- "title": Convert to title case
normalize_diacritics: Whether to normalize diacritics (default: False)
line_break_type: Line break type for whitespace normalization
(default: "unix")
**options: Additional normalization options (passed to sub-processors)
Returns:
Normalized text
str: Normalized text
"""
if not text:
return ""
@@ -72,22 +126,24 @@ class TextNormalizer:
normalized = text
# Unicode normalization
unicode_form = options.get("unicode_form", "NFC")
normalized = self.unicode_normalizer.normalize_unicode(normalized, form=unicode_form)
# Whitespace normalization
normalized = self.whitespace_normalizer.normalize_whitespace(normalized, **options)
normalized = self.whitespace_normalizer.normalize_whitespace(
normalized, line_break_type=line_break_type, **options
)
# Special character processing
normalized = self.special_char_processor.process_special_chars(normalized, **options)
normalized = self.special_char_processor.process_special_chars(
normalized, normalize_diacritics=normalize_diacritics, **options
)
# Case normalization
case_type = options.get("case", "preserve")
if case_type == "lower":
if case == "lower":
normalized = normalized.lower()
elif case_type == "upper":
elif case == "upper":
normalized = normalized.upper()
elif case_type == "title":
elif case == "title":
normalized = normalized.title()
return normalized.strip()
@@ -96,25 +152,37 @@ class TextNormalizer:
"""
Clean and sanitize text content.
This method delegates to the text cleaner for comprehensive text cleaning.
Args:
text: Input text
**options: Cleaning options
text: Input text to clean
**options: Cleaning options (passed to TextCleaner.clean)
Returns:
Cleaned text
str: Cleaned text
"""
return self.text_cleaner.clean(text, **options)
def standardize_format(self, text: str, format_type: str = "standard") -> str:
def standardize_format(
self,
text: str,
format_type: str = "standard"
) -> str:
"""
Standardize text format.
This method standardizes text format by applying format-specific
transformations (compact, preserve, or standard).
Args:
text: Input text
format_type: Format type ('standard', 'compact', 'preserve')
text: Input text to standardize
format_type: Format type (default: "standard"):
- "standard": Apply standard formatting
- "compact": Remove extra whitespace
- "preserve": Preserve original formatting
Returns:
Formatted text
str: Formatted text
"""
if format_type == "compact":
# Remove extra whitespace
@@ -125,16 +193,23 @@ class TextNormalizer:
return text.strip()
def process_batch(self, texts: List[str], **options) -> List[str]:
def process_batch(
self,
texts: List[str],
**options
) -> List[str]:
"""
Process multiple texts in batch.
This method processes multiple texts in batch, applying the same
normalization operations to each text.
Args:
texts: List of texts
**options: Processing options
texts: List of texts to process
**options: Processing options (passed to normalize_text method)
Returns:
List of processed texts
list: List of normalized texts (one per input text)
"""
return [self.normalize_text(text, **options) for text in texts]
@@ -143,33 +218,52 @@ class UnicodeNormalizer:
"""
Unicode normalization engine.
• Handles Unicode normalization
• Processes different Unicode forms
• Manages character encoding
• Handles special Unicode characters
• Supports various scripts and languages
This class provides Unicode normalization capabilities, handling different
Unicode forms and special character processing.
Features:
- Unicode normalization (NFC, NFD, NFKC, NFKD)
- Character encoding conversion
- Special Unicode character processing
- Support for various scripts and languages
Example Usage:
>>> normalizer = UnicodeNormalizer()
>>> normalized = normalizer.normalize_unicode(text, form="NFC")
>>> processed = normalizer.process_special_chars(text)
"""
def __init__(self, **config):
"""
Initialize Unicode normalizer.
Sets up the normalizer with configuration options.
Args:
**config: Configuration options
**config: Configuration options (currently unused)
"""
self.logger = get_logger("unicode_normalizer")
self.config = config
self.logger.debug("Unicode normalizer initialized")
def normalize_unicode(self, text: str, form: str = "NFC") -> str:
"""
Normalize Unicode text.
This method normalizes Unicode characters using the specified
normalization form.
Args:
text: Input text
form: Unicode normalization form (NFC, NFD, NFKC, NFKD)
text: Input text to normalize
form: Unicode normalization form (default: "NFC"):
- "NFC": Canonical composition
- "NFD": Canonical decomposition
- "NFKC": Compatibility composition
- "NFKD": Compatibility decomposition
Returns:
Normalized text
str: Unicode-normalized text (returns original text if normalization fails)
"""
if not text:
return ""
@@ -180,17 +274,26 @@ class UnicodeNormalizer:
self.logger.warning(f"Unicode normalization failed: {e}")
return text
def handle_encoding(self, text: str, source_encoding: str, target_encoding: str = "utf-8") -> str:
def handle_encoding(
self,
text: str,
source_encoding: str,
target_encoding: str = "utf-8"
) -> str:
"""
Handle text encoding conversion.
This method converts text from one encoding to another, handling
both string and bytes input.
Args:
text: Input text
source_encoding: Source encoding
target_encoding: Target encoding
text: Input text (string or bytes)
source_encoding: Source encoding name (e.g., "latin-1", "cp1252")
target_encoding: Target encoding name (default: "utf-8")
Returns:
Converted text
str: Converted text in target encoding (falls back to UTF-8 with
error replacement if conversion fails)
"""
if isinstance(text, bytes):
try:
@@ -204,11 +307,14 @@ class UnicodeNormalizer:
"""
Process special Unicode characters.
This method replaces common special Unicode characters (smart quotes,
dashes, ellipsis) with their ASCII equivalents.
Args:
text: Input text
text: Input text containing special Unicode characters
Returns:
Processed text
str: Text with special Unicode characters replaced with ASCII equivalents
"""
# Replace common special characters
replacements = {
@@ -231,32 +337,56 @@ class WhitespaceNormalizer:
"""
Whitespace normalization engine.
• Normalizes whitespace characters
• Handles different whitespace types
• Manages line breaks and spacing
• Processes indentation and formatting
This class provides whitespace normalization capabilities, handling
different whitespace types, line breaks, and indentation.
Features:
- Whitespace character normalization
- Line break handling (Unix, Windows, Mac)
- Indentation processing
- Multiple whitespace collapse
Example Usage:
>>> normalizer = WhitespaceNormalizer()
>>> normalized = normalizer.normalize_whitespace(text, line_break_type="unix")
>>> processed = normalizer.process_indentation(text, indent_type="spaces")
"""
def __init__(self, **config):
"""
Initialize whitespace normalizer.
Sets up the normalizer with configuration options.
Args:
**config: Configuration options
**config: Configuration options (currently unused)
"""
self.logger = get_logger("whitespace_normalizer")
self.config = config
self.logger.debug("Whitespace normalizer initialized")
def normalize_whitespace(self, text: str, **options) -> str:
def normalize_whitespace(
self,
text: str,
line_break_type: str = "unix",
**options
) -> str:
"""
Normalize whitespace in text.
This method normalizes whitespace by replacing tabs with spaces,
normalizing line breaks, and collapsing multiple spaces.
Args:
text: Input text
**options: Normalization options
text: Input text with potentially irregular whitespace
line_break_type: Line break type (default: "unix"):
- "unix": Unix-style line breaks (\n)
- "windows": Windows-style line breaks (\r\n)
**options: Additional normalization options (unused)
Returns:
Normalized text
str: Text with normalized whitespace
"""
if not text:
return ""
@@ -265,7 +395,6 @@ class WhitespaceNormalizer:
text = text.replace('\t', ' ')
# Normalize line breaks
line_break_type = options.get("line_break_type", "unix")
text = self.handle_line_breaks(text, line_break_type)
# Remove excessive whitespace
@@ -274,16 +403,26 @@ class WhitespaceNormalizer:
return text.strip()
def handle_line_breaks(self, text: str, line_break_type: str = "unix") -> str:
def handle_line_breaks(
self,
text: str,
line_break_type: str = "unix"
) -> str:
"""
Normalize line breaks.
This method normalizes line breaks to the specified type (Unix, Windows,
or Mac).
Args:
text: Input text
line_break_type: Line break type ('unix', 'windows', 'mac')
text: Input text with potentially mixed line breaks
line_break_type: Line break type (default: "unix"):
- "unix": Unix-style (\n)
- "windows": Windows-style (\r\n)
- "mac": Mac-style (\r)
Returns:
Normalized text
str: Text with normalized line breaks
"""
if line_break_type == "unix":
text = text.replace('\r\n', '\n')
@@ -295,16 +434,25 @@ class WhitespaceNormalizer:
return text
def process_indentation(self, text: str, indent_type: str = "spaces") -> str:
def process_indentation(
self,
text: str,
indent_type: str = "spaces"
) -> str:
"""
Normalize text indentation.
This method normalizes text indentation by converting between tabs
and spaces.
Args:
text: Input text
indent_type: Indentation type ('spaces', 'tabs')
text: Input text with potentially mixed indentation
indent_type: Indentation type (default: "spaces"):
- "spaces": Convert tabs to 4 spaces
- "tabs": Convert 4 spaces to tabs
Returns:
Normalized text
str: Text with normalized indentation
"""
if indent_type == "spaces":
text = text.replace('\t', ' ') # Convert tabs to 4 spaces
@@ -318,39 +466,60 @@ class SpecialCharacterProcessor:
"""
Special character processing engine.
• Processes special characters and symbols
• Handles punctuation and diacritics
• Manages mathematical symbols
• Processes currency and unit symbols
This class provides special character processing capabilities, including
punctuation normalization and diacritic handling.
Features:
- Punctuation normalization (quotes, dashes, ellipsis)
- Diacritic processing (normalization or removal)
- Special character replacement
Example Usage:
>>> processor = SpecialCharacterProcessor()
>>> processed = processor.process_special_chars(text, normalize_diacritics=True)
>>> normalized = processor.normalize_punctuation(text)
"""
def __init__(self, **config):
"""
Initialize special character processor.
Sets up the processor with configuration options.
Args:
**config: Configuration options
**config: Configuration options (currently unused)
"""
self.logger = get_logger("special_char_processor")
self.config = config
self.logger.debug("Special character processor initialized")
def process_special_chars(self, text: str, **options) -> str:
def process_special_chars(
self,
text: str,
normalize_diacritics: bool = False,
**options
) -> str:
"""
Process special characters in text.
This method processes special characters by normalizing punctuation
and optionally processing diacritics.
Args:
text: Input text
**options: Processing options
text: Input text to process
normalize_diacritics: Whether to normalize diacritics (default: False)
**options: Additional processing options (passed to process_diacritics)
Returns:
Processed text
str: Text with special characters processed
"""
# Normalize punctuation
text = self.normalize_punctuation(text)
# Process diacritics if requested
if options.get("normalize_diacritics", False):
text = self.process_diacritics(text)
if normalize_diacritics:
text = self.process_diacritics(text, **options)
return text
@@ -358,11 +527,14 @@ class SpecialCharacterProcessor:
"""
Normalize punctuation marks.
This method normalizes various Unicode punctuation marks to their
ASCII equivalents (quotes, dashes, ellipsis).
Args:
text: Input text
text: Input text with potentially mixed punctuation
Returns:
Normalized text
str: Text with normalized punctuation marks
"""
# Normalize quotes
text = re.sub(r'["""]', '"', text)
@@ -376,18 +548,29 @@ class SpecialCharacterProcessor:
return text
def process_diacritics(self, text: str, **options) -> str:
def process_diacritics(
self,
text: str,
remove_diacritics: bool = False,
**options
) -> str:
"""
Process diacritical marks.
This method processes diacritical marks by either normalizing them
(NFC) or removing them entirely.
Args:
text: Input text
**options: Processing options
text: Input text with diacritical marks
remove_diacritics: Whether to remove diacritics (default: False):
- True: Remove all diacritical marks
- False: Normalize diacritics using NFC
**options: Additional processing options (unused)
Returns:
Processed text
str: Text with diacritics processed (normalized or removed)
"""
if options.get("remove_diacritics", False):
if remove_diacritics:
# Remove diacritics
nfd = unicodedata.normalize('NFD', text)
return ''.join(c for c in nfd if unicodedata.category(c) != 'Mn')