feat: implement NER dataset feature engineering with multiple transformation formats
This commit is contained in:
@@ -0,0 +1,77 @@
|
||||
from abc import ABC, abstractmethod
|
||||
from typing import List, Tuple, Dict
|
||||
|
||||
import pandas as pd
|
||||
|
||||
from processing.steps.feature_extraction_step import NameCategory
|
||||
|
||||
|
||||
class BaseNameFormatter(ABC):
|
||||
"""
|
||||
Base class for name formatting transformations.
|
||||
Contains common logic for NER tagging and attribute computation.
|
||||
"""
|
||||
|
||||
def __init__(self, connectors: List[str] = None, additional_surnames: List[str] = None):
|
||||
self.connectors = connectors or ['wa', 'ya', 'ka', 'ba']
|
||||
self.additional_surnames = additional_surnames or [
|
||||
'jean', 'paul', 'marie', 'joseph', 'pierre', 'claude',
|
||||
'andre', 'michel', 'robert'
|
||||
]
|
||||
|
||||
@classmethod
|
||||
def parse_native_components(cls, native_str: str) -> List[str]:
|
||||
"""Parse native name string into individual components"""
|
||||
if pd.isna(native_str) or not native_str:
|
||||
return []
|
||||
return native_str.strip().split()
|
||||
|
||||
def create_ner_tags(self, text: str, native_parts: List[str], surname: str) -> List[Tuple[int, int, str]]:
|
||||
"""Create NER entity tags for transformed text"""
|
||||
entities = []
|
||||
current_pos = 0
|
||||
words = text.split()
|
||||
|
||||
for word in words:
|
||||
start_pos = current_pos
|
||||
end_pos = current_pos + len(word)
|
||||
|
||||
# Determine tag based on word content
|
||||
if word in native_parts or any(connector in word for connector in self.connectors):
|
||||
tag = 'NATIVE'
|
||||
elif word == surname or word in self.additional_surnames:
|
||||
tag = 'SURNAME'
|
||||
else:
|
||||
# Check if it's a compound native word or new surname
|
||||
if any(part in word for part in native_parts):
|
||||
tag = 'NATIVE'
|
||||
else:
|
||||
tag = 'SURNAME'
|
||||
|
||||
entities.append((start_pos, end_pos, tag))
|
||||
current_pos = end_pos + 1 # +1 for space
|
||||
|
||||
return entities
|
||||
|
||||
@classmethod
|
||||
def compute_derived_attributes(cls, name: str) -> Dict:
|
||||
"""Compute all derived attributes for the transformed name"""
|
||||
words_count = len(name.split()) if name else 0
|
||||
length = len(name) if name else 0
|
||||
|
||||
return {
|
||||
'words': words_count,
|
||||
'length': length,
|
||||
'identified_category': NameCategory.SIMPLE if words_count == 3 else NameCategory.COMPOSE,
|
||||
}
|
||||
|
||||
@abstractmethod
|
||||
def transform(self, row: pd.Series) -> Dict:
|
||||
"""Transform a row according to the specific format rules"""
|
||||
pass
|
||||
|
||||
@property
|
||||
@abstractmethod
|
||||
def transformation_type(self) -> str:
|
||||
"""Return the transformation type identifier"""
|
||||
pass
|
||||
@@ -0,0 +1,35 @@
|
||||
import random
|
||||
from typing import Dict
|
||||
|
||||
import pandas as pd
|
||||
|
||||
from processing.ner.formats import BaseNameFormatter
|
||||
|
||||
|
||||
class ConnectorFormatter(BaseNameFormatter):
|
||||
def transform(self, row: pd.Series) -> Dict:
|
||||
native_parts = self.parse_native_components(row['probable_native'])
|
||||
surname = row['probable_surname'] if pd.notna(row['probable_surname']) else ''
|
||||
connector = random.choice(self.connectors)
|
||||
|
||||
if len(native_parts) > 1:
|
||||
connected_native = f" {connector} ".join(native_parts)
|
||||
full_name = f"{connected_native} {surname}".strip()
|
||||
else:
|
||||
connected_native = f"{row['probable_native']} {connector} {row['probable_native']}".strip()
|
||||
full_name = f"{connected_native} {surname}".strip()
|
||||
|
||||
return {
|
||||
'name': full_name,
|
||||
'probable_native': connected_native,
|
||||
'identify_name': connected_native,
|
||||
'probable_surname': surname,
|
||||
'identify_surname': surname,
|
||||
'ner_entities': str(self.create_ner_tags(full_name, native_parts, surname)),
|
||||
'transformation_type': self.transformation_type,
|
||||
**self.compute_derived_attributes(full_name)
|
||||
}
|
||||
|
||||
@property
|
||||
def transformation_type(self) -> str:
|
||||
return 'connector_added'
|
||||
@@ -0,0 +1,32 @@
|
||||
import random
|
||||
from typing import Dict
|
||||
|
||||
import pandas as pd
|
||||
|
||||
from processing.ner.formats import BaseNameFormatter
|
||||
|
||||
|
||||
class ExtendedSurnameFormatter(BaseNameFormatter):
|
||||
def transform(self, row: pd.Series) -> Dict:
|
||||
native_parts = self.parse_native_components(row['probable_native'])
|
||||
original_surname = row['probable_surname'] if pd.notna(row['probable_surname']) else ''
|
||||
|
||||
# Add random additional surname
|
||||
additional_surname = random.choice(self.additional_surnames)
|
||||
combined_surname = f"{additional_surname} {original_surname}".strip()
|
||||
full_name = f"{row['probable_native']} {combined_surname}".strip()
|
||||
|
||||
return {
|
||||
'name': full_name,
|
||||
'probable_native': row['probable_native'],
|
||||
'identify_name': row['probable_native'],
|
||||
'probable_surname': combined_surname,
|
||||
'identity_surname': combined_surname,
|
||||
'ner_entities': str(self.create_ner_tags(full_name, native_parts, combined_surname)),
|
||||
'transformation_type': self.transformation_type,
|
||||
**self.compute_derived_attributes(full_name)
|
||||
}
|
||||
|
||||
@property
|
||||
def transformation_type(self) -> str:
|
||||
return 'extended_surname'
|
||||
@@ -0,0 +1,28 @@
|
||||
from typing import Dict
|
||||
|
||||
import pandas as pd
|
||||
|
||||
from processing.ner.formats import BaseNameFormatter
|
||||
|
||||
|
||||
class NativeOnlyFormatter(BaseNameFormatter):
|
||||
def transform(self, row: pd.Series) -> Dict:
|
||||
native_parts = self.parse_native_components(row['probable_native'])
|
||||
|
||||
# Only native components
|
||||
full_name = row['probable_native']
|
||||
|
||||
return {
|
||||
'name': full_name,
|
||||
'probable_native': row['probable_native'],
|
||||
'identify_name': row['probable_native'],
|
||||
'probable_surname': '',
|
||||
'identify_surname': '',
|
||||
'ner_entities': str(self.create_ner_tags(full_name, native_parts, '')),
|
||||
'transformation_type': self.transformation_type,
|
||||
**self.compute_derived_attributes(full_name)
|
||||
}
|
||||
|
||||
@property
|
||||
def transformation_type(self) -> str:
|
||||
return 'native_only'
|
||||
@@ -0,0 +1,29 @@
|
||||
from typing import Dict
|
||||
|
||||
import pandas as pd
|
||||
|
||||
from processing.ner.formats import BaseNameFormatter
|
||||
|
||||
|
||||
class OriginalFormatter(BaseNameFormatter):
|
||||
def transform(self, row: pd.Series) -> Dict:
|
||||
native_parts = self.parse_native_components(row['probable_native'])
|
||||
surname = row['probable_surname'] if pd.notna(row['probable_surname']) else ''
|
||||
|
||||
# Keep original order: native components + surname
|
||||
full_name = f"{row['probable_native']} {surname}".strip()
|
||||
|
||||
return {
|
||||
'name': full_name,
|
||||
'probable_native': row['probable_native'],
|
||||
'identify_name': row['probable_native'],
|
||||
'probable_surname': surname,
|
||||
'identify_surname': surname,
|
||||
'ner_entities': str(self.create_ner_tags(full_name, native_parts, surname)),
|
||||
'transformation_type': self.transformation_type,
|
||||
**self.compute_derived_attributes(full_name)
|
||||
}
|
||||
|
||||
@property
|
||||
def transformation_type(self) -> str:
|
||||
return 'original'
|
||||
@@ -0,0 +1,29 @@
|
||||
from typing import Dict
|
||||
|
||||
import pandas as pd
|
||||
|
||||
from processing.ner.formats import BaseNameFormatter
|
||||
|
||||
|
||||
class PositionFlippedFormatter(BaseNameFormatter):
|
||||
def transform(self, row: pd.Series) -> Dict:
|
||||
native_parts = self.parse_native_components(row['probable_native'])
|
||||
surname = row['probable_surname'] if pd.notna(row['probable_surname']) else ''
|
||||
|
||||
# Flip order: surname + native components
|
||||
full_name = f"{surname} {row['probable_native']}".strip()
|
||||
|
||||
return {
|
||||
'name': full_name,
|
||||
'probable_native': row['probable_native'],
|
||||
'identify_name': row['probable_native'],
|
||||
'probable_surname': surname,
|
||||
'identify_surname': surname,
|
||||
'ner_entities': str(self.create_ner_tags(full_name, native_parts, surname)),
|
||||
'transformation_type': self.transformation_type,
|
||||
**self.compute_derived_attributes(full_name)
|
||||
}
|
||||
|
||||
@property
|
||||
def transformation_type(self) -> str:
|
||||
return 'position_flipped'
|
||||
@@ -0,0 +1,30 @@
|
||||
from typing import Dict
|
||||
|
||||
import pandas as pd
|
||||
|
||||
from processing.ner.formats import BaseNameFormatter
|
||||
|
||||
|
||||
class ReducedNativeFormatter(BaseNameFormatter):
|
||||
def transform(self, row: pd.Series) -> Dict:
|
||||
native_parts = self.parse_native_components(row['probable_native'])
|
||||
surname = row['probable_surname'] if pd.notna(row['probable_surname']) else ''
|
||||
|
||||
# Keep only first native component + surname
|
||||
reduced_native = native_parts[0] if len(native_parts) > 1 else row['probable_native']
|
||||
full_name = f"{reduced_native} {surname}".strip()
|
||||
|
||||
return {
|
||||
'name': full_name,
|
||||
'probable_native': reduced_native,
|
||||
'identify_name': reduced_native,
|
||||
'probable_surname': surname,
|
||||
'identify_surname': surname,
|
||||
'ner_entities': str(self.create_ner_tags(full_name, [reduced_native], surname)),
|
||||
'transformation_type': self.transformation_type,
|
||||
**self.compute_derived_attributes(full_name)
|
||||
}
|
||||
|
||||
@property
|
||||
def transformation_type(self) -> str:
|
||||
return 'reduced_native'
|
||||
Reference in New Issue
Block a user