# semantic_search/processor.py
import logging
import re
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
import hashlib
logger = logging.getLogger(__name__)
@dataclass
class DocumentChunk:
"""Represents a chunk of a document."""
chunk_id: str
document_id: str
text: str
metadata: Dict[str, Any]
chunk_index: int
total_chunks: int
class DocumentProcessor:
"""
Processes legal documents for semantic search.
Handles chunking, cleaning, and metadata extraction.
"""
def __init__(self,
chunk_size: int = 1000,
chunk_overlap: int = 200,
min_chunk_size: int = 100):
"""
Initialize document processor.
Args:
chunk_size: Target size for each chunk in characters
chunk_overlap: Number of overlapping characters between chunks
min_chunk_size: Minimum chunk size to keep
"""
self.chunk_size = chunk_size
self.chunk_overlap = chunk_overlap
self.min_chunk_size = min_chunk_size
logger.info(f"Initialized DocumentProcessor (chunk_size={chunk_size}, overlap={chunk_overlap})")
def process_document(self,
document_id: str,
text: str,
metadata: Optional[Dict[str, Any]] = None) -> List[DocumentChunk]:
"""
Process a single document into chunks.
Args:
document_id: Unique document identifier
text: Document text content
metadata: Optional document metadata
Returns:
List of document chunks
"""
if not text or len(text.strip()) < self.min_chunk_size:
logger.warning(f"Document {document_id} too short to process")
return []
# Clean text
cleaned_text = self._clean_text(text)
# Extract metadata from text if not provided
if metadata is None:
metadata = {}
# Add extracted metadata
extracted_metadata = self._extract_metadata(cleaned_text)
metadata.update(extracted_metadata)
# Create chunks
chunks = self._create_chunks(cleaned_text)
# Create DocumentChunk objects
document_chunks = []
for i, chunk_text in enumerate(chunks):
chunk_id = self._generate_chunk_id(document_id, i)
chunk = DocumentChunk(
chunk_id=chunk_id,
document_id=document_id,
text=chunk_text,
metadata={
**metadata,
'chunk_index': i,
'total_chunks': len(chunks)
},
chunk_index=i,
total_chunks=len(chunks)
)
document_chunks.append(chunk)
logger.info(f"Processed document {document_id} into {len(chunks)} chunks")
return document_chunks
def _clean_text(self, text: str) -> str:
"""
Clean and normalize text for processing.
Args:
text: Raw text
Returns:
Cleaned text
"""
# Remove excessive whitespace
text = re.sub(r'\s+', ' ', text)
# Remove special characters but keep Turkish characters
# Keep: letters, numbers, spaces, and common punctuation
text = re.sub(r'[^\w\s\.\,\;\:\!\?\-\(\)\"\'ÇĞIİÖŞÜçğıiöşü]', ' ', text)
# Remove multiple spaces
text = re.sub(r' +', ' ', text)
# Trim
text = text.strip()
return text
def _extract_metadata(self, text: str) -> Dict[str, Any]:
"""
Extract metadata from legal document text.
Args:
text: Document text
Returns:
Extracted metadata
"""
metadata = {}
# Extract case numbers (Esas/Karar)
esas_pattern = r'E(?:sas)?[\s\.\:]*(\d{4})[\/\-](\d+)'
karar_pattern = r'K(?:arar)?[\s\.\:]*(\d{4})[\/\-](\d+)'
esas_match = re.search(esas_pattern, text[:500]) # Look in first 500 chars
if esas_match:
metadata['esas_no'] = f"E.{esas_match.group(1)}/{esas_match.group(2)}"
karar_match = re.search(karar_pattern, text[:500])
if karar_match:
metadata['karar_no'] = f"K.{karar_match.group(1)}/{karar_match.group(2)}"
# Extract dates (DD.MM.YYYY or DD/MM/YYYY format)
date_pattern = r'(\d{1,2})[\.\/](\d{1,2})[\.\/](\d{4})'
dates = re.findall(date_pattern, text[:1000]) # Look in first 1000 chars
if dates:
# Take the first date as decision date
day, month, year = dates[0]
metadata['karar_tarihi'] = f"{year}-{month.zfill(2)}-{day.zfill(2)}"
# Extract court/chamber name
chamber_patterns = [
r'(\d+)\.\s*Hukuk\s+Dairesi',
r'(\d+)\.\s*Ceza\s+Dairesi',
r'Hukuk\s+Genel\s+Kurulu',
r'Ceza\s+Genel\s+Kurulu',
r'(\d+)\.\s*Daire'
]
for pattern in chamber_patterns:
match = re.search(pattern, text[:500], re.IGNORECASE)
if match:
metadata['chamber'] = match.group(0)
break
return metadata
def _create_chunks(self, text: str) -> List[str]:
"""
Create overlapping chunks from text.
Args:
text: Cleaned document text
Returns:
List of text chunks
"""
chunks = []
# Split by sentences for better semantic coherence
sentences = self._split_sentences(text)
current_chunk = []
current_size = 0
for sentence in sentences:
sentence_size = len(sentence)
# If adding this sentence exceeds chunk size
if current_size + sentence_size > self.chunk_size and current_chunk:
# Save current chunk
chunk_text = ' '.join(current_chunk)
chunks.append(chunk_text)
# Create overlap for next chunk
overlap_size = 0
overlap_sentences = []
# Add sentences from the end until we reach overlap size
for sent in reversed(current_chunk):
overlap_size += len(sent)
overlap_sentences.insert(0, sent)
if overlap_size >= self.chunk_overlap:
break
# Start new chunk with overlap
current_chunk = overlap_sentences
current_size = sum(len(s) for s in current_chunk)
# Add sentence to current chunk
current_chunk.append(sentence)
current_size += sentence_size
# Add final chunk if not empty
if current_chunk:
chunk_text = ' '.join(current_chunk)
if len(chunk_text) >= self.min_chunk_size:
chunks.append(chunk_text)
return chunks
def _split_sentences(self, text: str) -> List[str]:
"""
Split text into sentences.
Args:
text: Text to split
Returns:
List of sentences
"""
# Simple sentence splitting for Turkish text
# Split on period, question mark, exclamation, but not on abbreviations
# Common Turkish abbreviations to preserve
abbreviations = ['Dr', 'Prof', 'Av', 'Md', 'Yrd', 'Doç', 'No', 'S', 'vs', 'vb', 'bkz']
# Replace abbreviations temporarily
temp_text = text
replacements = {}
for i, abbr in enumerate(abbreviations):
placeholder = f"__ABBR{i}__"
temp_text = temp_text.replace(f"{abbr}.", placeholder)
replacements[placeholder] = f"{abbr}."
# Split sentences
sentence_endings = re.compile(r'[.!?]+')
sentences = sentence_endings.split(temp_text)
# Restore abbreviations and clean
cleaned_sentences = []
for sentence in sentences:
# Restore abbreviations
for placeholder, original in replacements.items():
sentence = sentence.replace(placeholder, original)
# Clean and add if not empty
sentence = sentence.strip()
if sentence and len(sentence) > 10: # Minimum sentence length
cleaned_sentences.append(sentence)
return cleaned_sentences
def _generate_chunk_id(self, document_id: str, chunk_index: int) -> str:
"""
Generate unique chunk ID.
Args:
document_id: Parent document ID
chunk_index: Index of chunk in document
Returns:
Unique chunk ID
"""
chunk_string = f"{document_id}_chunk_{chunk_index}"
chunk_hash = hashlib.md5(chunk_string.encode()).hexdigest()[:8]
return f"{document_id}_c{chunk_index}_{chunk_hash}"
def combine_chunks(self, chunks: List[DocumentChunk]) -> str:
"""
Combine chunks back into full document text.
Args:
chunks: List of document chunks
Returns:
Combined text
"""
if not chunks:
return ""
# Sort by chunk index
sorted_chunks = sorted(chunks, key=lambda x: x.chunk_index)
# For overlapping chunks, we need to be careful about duplication
# Simple approach: just concatenate with space
combined = " ".join([chunk.text for chunk in sorted_chunks])
return combined
