317 lines
14 KiB
Python
317 lines
14 KiB
Python
# Copyright (c) 2025 Patrick Motsch
|
|
# All rights reserved.
|
|
"""
|
|
Data Neutralization Service
|
|
Handles file processing for data neutralization including SharePoint integration
|
|
DSGVO-konformer Daten-Neutralisierer für KI-Agentensysteme
|
|
Unterstützt TXT, JSON, CSV, Excel und Word-Dateien
|
|
Mehrsprachig: DE, EN, FR, IT
|
|
"""
|
|
|
|
import logging
|
|
import re
|
|
import json
|
|
from typing import Dict, List, Any, Optional
|
|
|
|
from modules.features.neutralization.datamodelFeatureNeutralizer import DataNeutraliserConfig, DataNeutralizerAttributes
|
|
from modules.features.neutralization.interfaceFeatureNeutralizer import InterfaceFeatureNeutralizer, getInterface as getNeutralizerInterface
|
|
|
|
# Import all necessary classes and functions for neutralization
|
|
from .subProcessCommon import CommonUtils, NeutralizationResult, NeutralizationAttribute
|
|
from .subProcessText import TextProcessor, PlainText
|
|
from .subProcessList import ListProcessor, TableData
|
|
from .subProcessBinary import BinaryProcessor
|
|
from .subPatterns import HeaderPatterns, DataPatterns, TextTablePatterns
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class NeutralizationService:
|
|
"""Service for handling data neutralization operations"""
|
|
|
|
def __init__(self, serviceCenter=None, NamesToParse: List[str] = None):
|
|
"""Initialize the service with user context and anonymization processors
|
|
|
|
Args:
|
|
serviceCenter: Service center instance for accessing other services
|
|
NamesToParse: List of names to parse and replace (case-insensitive)
|
|
"""
|
|
self.services = serviceCenter
|
|
self.interfaceDbComponent = serviceCenter.interfaceDbComponent
|
|
|
|
# Create feature-specific interface for neutralizer DB operations
|
|
self.interfaceNeutralizer: InterfaceFeatureNeutralizer = None
|
|
if serviceCenter and serviceCenter.interfaceDbApp:
|
|
dbApp = serviceCenter.interfaceDbApp
|
|
self.interfaceNeutralizer = getNeutralizerInterface(
|
|
currentUser=dbApp.currentUser,
|
|
mandateId=dbApp.mandateId,
|
|
featureInstanceId=getattr(dbApp, 'featureInstanceId', None)
|
|
)
|
|
|
|
# Initialize anonymization processors
|
|
self.NamesToParse = NamesToParse or []
|
|
self.textProcessor = TextProcessor(NamesToParse)
|
|
self.listProcessor = ListProcessor(NamesToParse)
|
|
self.binaryProcessor = BinaryProcessor()
|
|
self.commonUtils = CommonUtils()
|
|
|
|
def getConfig(self) -> Optional[DataNeutraliserConfig]:
|
|
"""Get the neutralization configuration for the current user's mandate"""
|
|
if not self.interfaceNeutralizer:
|
|
return None
|
|
return self.interfaceNeutralizer.getNeutralizationConfig()
|
|
|
|
def saveConfig(self, configData: Dict[str, Any]) -> DataNeutraliserConfig:
|
|
"""Save or update the neutralization configuration"""
|
|
if not self.interfaceNeutralizer:
|
|
raise ValueError("User context required for saving configuration")
|
|
return self.interfaceNeutralizer.createOrUpdateNeutralizationConfig(configData)
|
|
|
|
# Public API: process text or file
|
|
|
|
def processText(self, text: str) -> Dict[str, Any]:
|
|
"""Neutralize a raw text string and return a standard result dict."""
|
|
return self._neutralizeText(text, 'text')
|
|
|
|
def processFile(self, fileId: str) -> Dict[str, Any]:
|
|
"""Neutralize a file referenced by its fileId using component interface.
|
|
Binary files are not neutralized but will be indicated in the result."""
|
|
if not self.interfaceDbComponent:
|
|
raise ValueError("Component interface is required to process a file by fileId")
|
|
# Fetch file data and metadata
|
|
fileInfo = None
|
|
try:
|
|
# getFile returns an object; fallback to dict-like
|
|
fileInfo = self.interfaceDbComponent.getFile(fileId)
|
|
except Exception:
|
|
fileInfo = None
|
|
fileName = getattr(fileInfo, 'fileName', None) if fileInfo else None
|
|
mimeType = getattr(fileInfo, 'mimeType', None) if fileInfo else None
|
|
|
|
# Check if file is binary and cannot be neutralized
|
|
if self._isBinaryMimeType(mimeType or ''):
|
|
# Return a result indicating binary file (not neutralized)
|
|
return {
|
|
'file_id': fileId,
|
|
'is_binary': True,
|
|
'mime_type': mimeType or 'unknown',
|
|
'file_name': fileName or 'unknown',
|
|
'neutralized_text': None,
|
|
'processed_info': {
|
|
'type': 'binary',
|
|
'status': 'skipped',
|
|
'message': 'Binary file neutralization will be implemented in the future'
|
|
}
|
|
}
|
|
|
|
fileData = self.interfaceDbComponent.getFileData(fileId)
|
|
if not fileData:
|
|
raise ValueError(f"No file data found for fileId: {fileId}")
|
|
|
|
# Determine textType from mime
|
|
textType = self._getContentTypeFromMime(mimeType or '')
|
|
|
|
# Decode to text
|
|
try:
|
|
textContent = fileData.decode('utf-8')
|
|
except UnicodeDecodeError:
|
|
decoded = None
|
|
for enc in ['latin-1', 'cp1252', 'iso-8859-1']:
|
|
try:
|
|
decoded = fileData.decode(enc)
|
|
break
|
|
except UnicodeDecodeError:
|
|
continue
|
|
if decoded is None:
|
|
raise ValueError("Unable to decode file content as text. This may indicate a binary file that cannot be neutralized.")
|
|
textContent = decoded
|
|
|
|
result = self._neutralizeText(textContent, textType)
|
|
# Add a reasonable output filename if original known
|
|
if fileName:
|
|
result['neutralized_file_name'] = f"neutralized_{fileName}"
|
|
result['file_id'] = fileId
|
|
result['is_binary'] = False
|
|
return result
|
|
|
|
def resolveText(self, text: str) -> str:
|
|
if not self.interfaceNeutralizer:
|
|
return text
|
|
try:
|
|
placeholderPattern = r'\[([a-z]+)\.([a-f0-9-]{36})\]'
|
|
matches = re.findall(placeholderPattern, text)
|
|
resolvedText = text
|
|
for placeholderType, uid in matches:
|
|
attribute = self.interfaceNeutralizer.getAttributeById(uid)
|
|
if attribute:
|
|
placeholder = f"[{placeholderType}.{uid}]"
|
|
resolvedText = resolvedText.replace(placeholder, attribute["originalText"])
|
|
return resolvedText
|
|
except Exception:
|
|
return text
|
|
|
|
def getAttributes(self) -> List[DataNeutralizerAttributes]:
|
|
"""Get all neutralization attributes for the current user's mandate"""
|
|
if not self.interfaceNeutralizer:
|
|
return []
|
|
try:
|
|
# Use the interface method which properly converts dicts to objects
|
|
return self.interfaceNeutralizer.getNeutralizationAttributes()
|
|
except Exception as e:
|
|
logger.error(f"Error getting neutralization attributes: {str(e)}")
|
|
return []
|
|
|
|
def deleteNeutralizationAttributes(self, fileId: str) -> bool:
|
|
"""Delete neutralization attributes for a specific file"""
|
|
if not self.interfaceNeutralizer:
|
|
return False
|
|
return self.interfaceNeutralizer.deleteNeutralizationAttributes(fileId)
|
|
|
|
def _reloadNamesFromConfig(self) -> None:
|
|
"""Reload names from config and update processors"""
|
|
try:
|
|
config = self.getConfig()
|
|
if not config:
|
|
return
|
|
|
|
# Parse namesToParse string into list
|
|
names_list = []
|
|
if config.namesToParse:
|
|
names_list = [name.strip() for name in config.namesToParse.split('\n') if name.strip()]
|
|
|
|
# Update internal list
|
|
self.NamesToParse = names_list
|
|
|
|
# Recreate processors with updated names
|
|
self.textProcessor = TextProcessor(names_list)
|
|
self.listProcessor = ListProcessor(names_list)
|
|
|
|
logger.debug(f"Reloaded {len(names_list)} names from config")
|
|
except Exception as e:
|
|
logger.error(f"Error reloading names from config: {str(e)}")
|
|
# Continue with existing names if reload fails
|
|
|
|
# Helper functions
|
|
|
|
def _neutralizeText(self, text: str, textType: str = None) -> Dict[str, Any]:
|
|
"""Process text and return unified dict for API consumption."""
|
|
try:
|
|
# Reload names from config before processing to ensure we have the latest names
|
|
self._reloadNamesFromConfig()
|
|
|
|
# Auto-detect content type if not provided
|
|
if textType is None:
|
|
textType = self.commonUtils.detectContentType(text)
|
|
|
|
# Check if content is binary data
|
|
if self.binaryProcessor.isBinaryContent(text):
|
|
data, mapping, replaced_fields, processed_info = self.binaryProcessor.processBinaryContent(text)
|
|
neutralized_text = text if isinstance(data, str) else str(data)
|
|
attributes = [NeutralizationAttribute(original=k, placeholder=v) for k, v in mapping.items()]
|
|
return NeutralizationResult(
|
|
neutralized_text=neutralized_text,
|
|
mapping=mapping,
|
|
attributes=attributes,
|
|
processed_info=processed_info
|
|
).model_dump()
|
|
|
|
# Inline former _processData routing
|
|
if textType in ['csv', 'json', 'xml']:
|
|
if textType == 'csv':
|
|
data, mapping, replaced_fields, processed_info = self.listProcessor.processCsvContent(text)
|
|
elif textType == 'json':
|
|
data, mapping, replaced_fields, processed_info = self.listProcessor.processJsonContent(text)
|
|
else: # xml
|
|
data, mapping, replaced_fields, processed_info = self.listProcessor.processXmlContent(text)
|
|
else:
|
|
data, mapping, replaced_fields, processed_info = self.textProcessor.processTextContent(text)
|
|
# Stringify data consistently
|
|
if textType == 'csv':
|
|
try:
|
|
neutralized_text = data.to_csv(index=False)
|
|
except Exception:
|
|
neutralized_text = str(data)
|
|
elif textType == 'json':
|
|
neutralized_text = json.dumps(data, ensure_ascii=False)
|
|
elif textType == 'xml':
|
|
neutralized_text = str(data)
|
|
else:
|
|
neutralized_text = str(data)
|
|
|
|
attributes = [NeutralizationAttribute(original=k, placeholder=v) for k, v in mapping.items()]
|
|
return NeutralizationResult(
|
|
neutralized_text=neutralized_text,
|
|
mapping=mapping,
|
|
attributes=attributes,
|
|
processed_info=processed_info
|
|
).model_dump()
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error processing content: {str(e)}")
|
|
return NeutralizationResult(
|
|
neutralized_text='',
|
|
mapping={},
|
|
attributes=[],
|
|
processed_info={'type': 'error', 'error': str(e)}
|
|
).model_dump()
|
|
|
|
def _isBinaryMimeType(self, mime_type: str) -> bool:
|
|
"""Check if a MIME type represents binary content that cannot be neutralized as text"""
|
|
if not mime_type:
|
|
return False
|
|
|
|
mime_type_lower = mime_type.lower()
|
|
|
|
# Text-based MIME types that CAN be neutralized (explicit list)
|
|
text_mime_types = [
|
|
'text/plain', 'text/html', 'text/css', 'text/markdown', 'text/csv',
|
|
'text/javascript', 'text/xml', 'text/json',
|
|
'application/json', 'application/xml', 'application/javascript',
|
|
'application/csv'
|
|
]
|
|
|
|
# Check explicit text types first
|
|
if mime_type_lower in text_mime_types:
|
|
return False
|
|
|
|
# Text-based prefixes that can be neutralized
|
|
if mime_type_lower.startswith('text/'):
|
|
return False
|
|
|
|
# Binary MIME types that CANNOT be neutralized
|
|
binary_mime_prefixes = [
|
|
'image/', 'audio/', 'video/',
|
|
'application/pdf', 'application/zip',
|
|
'application/octet-stream', 'application/x-',
|
|
'application/vnd.', 'application/msword',
|
|
'application/vnd.ms-', 'application/vnd.openxmlformats-'
|
|
]
|
|
|
|
# Check if it's a binary type by prefix
|
|
if any(mime_type_lower.startswith(prefix) for prefix in binary_mime_prefixes):
|
|
return True
|
|
|
|
# Additional specific binary document types
|
|
binary_mime_types = [
|
|
'application/pdf', 'application/msword', 'application/vnd.ms-excel',
|
|
'application/vnd.ms-powerpoint',
|
|
'application/vnd.openxmlformats-officedocument.wordprocessingml.document',
|
|
'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
|
|
'application/vnd.openxmlformats-officedocument.presentationml.presentation',
|
|
'application/zip', 'application/x-rar-compressed', 'application/x-7z-compressed',
|
|
'application/x-tar', 'application/gzip'
|
|
]
|
|
|
|
return mime_type_lower in binary_mime_types
|
|
|
|
def _getContentTypeFromMime(self, mime_type: str) -> str:
|
|
"""Determine content type from MIME type for neutralization processing"""
|
|
if mime_type.startswith('text/'):
|
|
return 'text'
|
|
elif mime_type in ['application/json', 'application/xml', 'text/xml']:
|
|
return 'json' if 'json' in mime_type else 'xml'
|
|
elif mime_type in ['text/csv', 'application/csv']:
|
|
return 'csv'
|
|
else:
|
|
return 'text' # Default to text processing
|
|
|