TurkceOCR / turkceocr /postprocess /entity_engine.py

Release TurkceOCR v2.0 (Apache-2.0)

c8b6043 about 9 hours ago

17.1 kB

"""
UnifiedTurkishEntityEngine — `ictihat_core` × `kvkk-pii` × `TürkçeOCR` Birleşik Varlık Tanıma,
Hiyerarşik Span Kapsama (Hierarchical Span Subsumption) ve KVKK Karartma Motoru.
"""

from __future__ import annotations

import re
import sys
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Dict, List, Optional, Set, Tuple

from turkceocr.postprocess.legal_recognizers import (
    DetectedEntitySpan,
    LegalAndCommercialRecognizerSuite,
)

try:
    from kvkk_pii.engine import PIIAnalyzer as _KvkkPIIAnalyzer  # type: ignore
except Exception:
    _KvkkPIIAnalyzer = None

# Hiyerarşik kapsayıcı varlık tipleri (içlerindeki kısa tarih/plaka/sayı spanları kapsayıcıyı sildirtmez!)
CONTAINER_ENTITY_TYPES: Set[str] = {
    "ADDRESS",
    "HEALTH_DATA",
    "KVKK_SPECIAL",
    "ORG_COMPANY",
    "TAX_OFFICE",
    "TR_COURT_AUTHORITY",
    "TR_LAW_CITATION",
}

# KVKK kapsamında karartılması (redaction) zorunlu olan hassas PII + SUSPECT tipleri
DEFAULT_REDACTABLE_ENTITY_TYPES: Set[str] = {
    "TR_TCKN",
    "TR_TCKN_SUSPECT",
    "TR_YABANCI_KIMLIK",
    "TR_VKN",
    "TR_VKN_SUSPECT",
    "TR_MERSIS",
    "TR_IBAN",
    "TR_IBAN_SUSPECT",
    "TR_PHONE",
    "EMAIL",
    "TR_EMAIL",
    "TR_KEP",
    "TR_PLATE",
    "CREDIT_CARD",
    "TR_CREDIT_CARD",
    "TR_PASSPORT",
    "IP_ADDRESS",
    "TR_IP",
    "TR_IPV6",
    "TR_MAC",
    "SECRET",
    "TR_SECRET",
    "PERSON",
    "ADDRESS",
    "HEALTH_DATA",
    "KVKK_SPECIAL",
    "TR_SGK_SICIL",
    "TR_TAPU_ADA_PARSEL",
}

# Bağlamsal hibrit tanıyıcı kalıpları (NER modeli yüklü olmadığında veya NER'i tamamlayıcı olarak)
_RX_TAX_OFFICE = re.compile(
    r"\b([A-ZÇĞİÖŞÜ][a-zçğıöşüA-ZÇĞİÖŞÜ]{1,20}(?:[^\S\r\n]+[A-ZÇĞİÖŞÜa-zçğıöşü]{1,20}){0,3}[^\S\r\n]+Vergi[^\S\r\n]+Dairesi(?:[^\S\r\n]+Müdürlüğü)?)\b"
)
_RX_ORG_COMPANY = re.compile(
    r"\b([A-ZÇĞİÖŞÜ0-9][A-Za-zÇĞİÖŞÜçğıöşü0-9\.\-&]{1,25}(?:[^\S\r\n]+[A-Za-zÇĞİÖŞÜçğıöşü0-9\.\-&]{1,25}){0,7}[^\S\r\n]+"
    r"(?:T\.?[^\S\r\n]*A\.?[^\S\r\n]*Ş\.?|T\.?[^\S\r\n]*A\.?[^\S\r\n]*O\.?|A\.?[^\S\r\n]*Ş\.?|A\.?[^\S\r\n]*O\.?|LTD\.?[^\S\r\n]*ŞTİ\.?|LİMİTED[^\S\r\n]+ŞİRKETİ|ANONİM[^\S\r\n]+ŞİRKETİ|SAN\.?[^\S\r\n]*(?:VE[^\S\r\n]*)?TİC\.?[^\S\r\n]*(?:T\.?[^\S\r\n]*A\.?[^\S\r\n]*Ş\.?|A\.?[^\S\r\n]*Ş\.?|LTD\.?[^\S\r\n]*ŞTİ\.?)|HOLDİNG[^\S\r\n]+A\.?[^\S\r\n]*Ş\.?))"
)
_RX_LABELED_PERSON = re.compile(
    r"(?:(?:Ad[ıi][^\S\r\n]*Soyad[ıi]|Müşteri(?:[^\S\r\n]*Ad[ıi])?|Hasta(?:[^\S\r\n]*Ad[ıi])?|Alıcı|Borçlu|Alacaklı|Davacı|Davalı|Başvurucu|Sanık|Müşteki|Vekili|Av\.|Avukat|Sayın|Sn\.)"
    r"(?:[^\S\r\n]+(?:Gerçek[^\S\r\n]+Kişi|Ad[ıi][^\S\r\n]*Soyad[ıi]))?[^\S\r\n]*[:\-]?[^\S\r\n]*)"
    r"([A-ZÇĞİÖŞÜ][a-zçğıöşüA-ZÇĞİÖŞÜ]{1,20}(?:[^\S\r\n]+[A-ZÇĞİÖŞÜ][a-zçğıöşüA-ZÇĞİÖŞÜ]{1,22}){1,3})"
)
_RX_LABELED_ADDRESS = re.compile(
    r"(?:(?:İletişim[^\S\r\n]*&[^\S\r\n]*Adres|Adres[i]?|İkametgah|Teslimat[^\S\r\n]*Adresi|Sevk[^\S\r\n]*Adresi)[^\S\r\n]*[:\-][^\S\r\n]*)"
    r"([^\n\r\|]{10,120}?)(?=\s*\||(?<!Mah)(?<!Cad)(?<!Sok)(?<!Sk)(?<!Blv)(?<!Apt)(?<!No)(?<!Kat)\.\s+[A-ZÇĞİÖŞÜ]|[\n\r]|$)"
)
_RX_UNLABELED_ADDRESS = re.compile(
    r"\b([A-ZÇĞİÖŞÜ][a-zçğıöşüA-ZÇĞİÖŞÜ]{1,22}(?:[^\S\r\n]+[A-ZÇĞİÖŞÜa-zçğıöşü]{1,22}){0,2}[^\S\r\n]+(?:Mahallesi|Mah\.|Bulvarı|Blv\.|Caddesi|Cad\.|Sokağı|Sok\.|Sk\.)[^\S\r\n]+[^\n\r\|]{0,95}?(?:(?:\d{1,3}[^\S\r\n]*[/\/][^\S\r\n]*[A-Za-zÇĞİÖŞÜçğıöşü]+)|(?:Kat[^\S\r\n]*[:\.]?[^\S\r\n]*\d+)|(?:Daire[^\S\r\n]*[:\.]?[^\S\r\n]*\d+)|(?:No[^\S\r\n]*[:\.]?[^\S\r\n]*\d+[A-Za-z\/0-9\s]*[A-Za-zÇĞİÖŞÜçğıöşü]+\s*/\s*[A-Za-zÇĞİÖŞÜçğıöşü]+)|(?:No[^\S\r\n]*[:\.]?[^\S\r\n]*\d+[A-Za-z\/0-9]*)))"
)
_RX_HEALTH_DATA = re.compile(
    r"(?:(?:Sağlık[^\S\r\n]*/[^\S\r\n]*Epikriz|Tanı|Teşhis|ICD(?:\-10)?(?:\s*Kodu)?|Reçete|İlaç|Kan\s*Grubu|Epikriz|Poliklinik\s*Notu)\s*[:\-]\s*)"
    r"([^\n\r\|]{5,110}?)(?=\s*\||[\n\r]|$)"
    r"|(\b[A-Z]\d{2}\.\d\s*[-–]?\s*[A-Za-zÇĞİÖŞÜçğıöşü0-9\s]{4,65})"
    r"|(\b(?:0|A|B|AB)\s*Rh\s*[\+\-]\s*\((?:Pozitif|Negatif)\))"
)
_RX_KVKK_SPECIAL = re.compile(
    r"(?:(?:KVKK[^\S\r\n]*Md\.?\s*6[^\S\r\n]*Özel[^\S\r\n]*Veri\s*[:\-]\s*)?"
    r"(?:Dini(?:\s*İnancı)?|Mezhebi|Etnik\s*Kökeni?|Siyasi\s*Düşüncesi|Sendika(?:\s*Üyeliği)?|Dernek(?:\s*Üyeliği)?|Sabıka\s*Kaydı|Adli\s*Sicil|Ceza\s*Mahk[uû]miyeti|Biyometrik\s*Veri)\s*[:\-]\s*)"
    r"([^\n\r\|]{4,100}?)(?=\s*\||[\n\r]|$)"
    r"|(\b[A-ZÇĞİÖŞÜa-zçğıöşü\s]{3,35}Sendikası(?:\s+üyeliği)?(?:\s+ve\s+adli\s+sicil\s+sabıka\s+kaydı)?)"
)


@dataclass
class UnifiedAnalysisReport:
    """Birleşik Varlık Tanıma ve KVKK Analiz Raporu."""

    text: str
    entities: List[DetectedEntitySpan] = field(default_factory=list)
    redacted_text: str = ""
    entity_counts: Dict[str, int] = field(default_factory=dict)

    def to_dict(self) -> Dict[str, Any]:
        return {
            "text_length": len(self.text),
            "entity_count": len(self.entities),
            "entity_counts": self.entity_counts,
            "entities": [e.to_dict() for e in self.entities],
            "redacted_text": self.redacted_text,
        }


class UnifiedTurkishEntityEngine:
    """
    `ictihat_core` (Hukuki Tier-1) + `kvkk-pii` (Kişisel Veri Tier-1 + BERTurkNER v3) +
    `TürkçeOCR` (Ticari Tier-1 + İki Katmanlı OCR-Toleranslı Fail-Closed Dedektör) birleşik motoru.
    """

    def __init__(
        self,
        use_ner: bool = False,
        ner_model_path: Optional[str] = None,
        score_threshold: float = 0.50,
    ) -> None:
        self.use_ner = use_ner
        self.ner_model_path = ner_model_path
        self.score_threshold = score_threshold
        self.legal_suite = LegalAndCommercialRecognizerSuite()
        self.kvkk_analyzer = None
        if _KvkkPIIAnalyzer is not None:
            try:
                self.kvkk_analyzer = _KvkkPIIAnalyzer(ner_model_path=ner_model_path)
            except Exception:
                self.kvkk_analyzer = _KvkkPIIAnalyzer()

    def _extract_contextual_entities(self, text: str) -> List[DetectedEntitySpan]:
        spans: List[DetectedEntitySpan] = []

        for m in _RX_TAX_OFFICE.finditer(text):
            s, e = m.start(1), m.end(1)
            spans.append(
                DetectedEntitySpan(
                    entity_type="TAX_OFFICE",
                    start=s,
                    end=e,
                    score=0.95,
                    text=text[s:e],
                    source="contextual_grammar",
                    tier="VERIFIED",
                )
            )

        for m in _RX_ORG_COMPANY.finditer(text):
            s, e = m.start(1), m.end(1)
            cand_text = text[s:e].strip()
            # Mahkeme/Daire başlığı değilse şirket olarak ekle
            if "Mahkemesi" not in cand_text and "Dairesi" not in cand_text:
                spans.append(
                    DetectedEntitySpan(
                        entity_type="ORG_COMPANY",
                        start=s,
                        end=e,
                        score=0.93,
                        text=text[s:e],
                        source="contextual_grammar",
                        tier="VERIFIED",
                    )
                )

        for m in _RX_LABELED_PERSON.finditer(text):
            s, e = m.start(1), m.end(1)
            cand_text = text[s:e].strip()
            if not any(
                bad in cand_text
                for bad in ("Vergi Dairesi", "Mahkemesi", "Dairesi", "Kanunu", "Müdürlüğü", "Şirketi")
            ):
                spans.append(
                    DetectedEntitySpan(
                        entity_type="PERSON",
                        start=s,
                        end=e,
                        score=0.92,
                        text=text[s:e],
                        source="contextual_grammar",
                        tier="VERIFIED_PII",
                    )
                )

        for m in _RX_LABELED_ADDRESS.finditer(text):
            s, e = m.start(1), m.end(1)
            spans.append(
                DetectedEntitySpan(
                    entity_type="ADDRESS",
                    start=s,
                    end=e,
                    score=0.92,
                    text=text[s:e],
                    source="contextual_grammar",
                    tier="VERIFIED_PII",
                )
            )

        for m in _RX_UNLABELED_ADDRESS.finditer(text):
            s, e = m.start(1), m.end(1)
            spans.append(
                DetectedEntitySpan(
                    entity_type="ADDRESS",
                    start=s,
                    end=e,
                    score=0.90,
                    text=text[s:e],
                    source="contextual_grammar",
                    tier="VERIFIED_PII",
                )
            )

        for m in _RX_HEALTH_DATA.finditer(text):
            s, e = m.start(), m.end()
            for g_idx in (1, 2, 3):
                if m.group(g_idx) is not None:
                    s, e = m.start(g_idx), m.end(g_idx)
                    break
            spans.append(
                DetectedEntitySpan(
                    entity_type="HEALTH_DATA",
                    start=s,
                    end=e,
                    score=0.93,
                    text=text[s:e],
                    source="contextual_grammar",
                    tier="VERIFIED_PII",
                )
            )

        for m in _RX_KVKK_SPECIAL.finditer(text):
            s, e = m.start(), m.end()
            for g_idx in (1, 2):
                if m.group(g_idx) is not None:
                    s, e = m.start(g_idx), m.end(g_idx)
                    break
            spans.append(
                DetectedEntitySpan(
                    entity_type="KVKK_SPECIAL",
                    start=s,
                    end=e,
                    score=0.93,
                    text=text[s:e],
                    source="contextual_grammar",
                    tier="VERIFIED_PII",
                )
            )

        return spans

    @staticmethod
    def deduplicate_hierarchical(entities: List[DetectedEntitySpan]) -> List[DetectedEntitySpan]:
        """
        Hiyerarşik Span Kapsama (Hierarchical Span Subsumption):
        1. `TR_COURT_AUTHORITY` (`score=1.0`) kendisiyle örtüşen yanlış pozitif `ORG_COMPANY`,
           `ADDRESS` veya `PERSON` spanlarını ezer.
        2. Kapsayıcı bir varlık (`ADDRESS`, `HEALTH_DATA`, `KVKK_SPECIAL`, `ORG_COMPANY`,
           `TAX_OFFICE`, `TR_COURT_AUTHORITY`, `TR_LAW_CITATION`) kendisinden daha kısa bir
           alt-span (`TR_DATE`, `TR_PLATE`, `TR_ESAS_KARAR_NO` vb.) ile örtüştüğünde kapsayıcı
           varlık ASLA silinmez; kısa alt-span `container.sub_entities` listesine eklenir.
        """
        if not entities:
            return []

        sorted_ents = sorted(
            entities,
            key=lambda e: (e.start, -(e.end - e.start), -e.score),
        )

        kept: List[DetectedEntitySpan] = []
        for candidate in sorted_ents:
            if not kept:
                kept.append(candidate)
                continue

            prev = kept[-1]
            if candidate.start < prev.end:
                prev_len = prev.end - prev.start
                cand_len = candidate.end - candidate.start

                # Özel kural: TR_COURT_AUTHORITY her zaman ORG_COMPANY / ADDRESS / PERSON yanlış pozitifini ezer
                if prev.entity_type == "TR_COURT_AUTHORITY" and candidate.entity_type in {
                    "ORG_COMPANY",
                    "ADDRESS",
                    "PERSON",
                }:
                    continue
                if candidate.entity_type == "TR_COURT_AUTHORITY" and prev.entity_type in {
                    "ORG_COMPANY",
                    "ADDRESS",
                    "PERSON",
                }:
                    kept[-1] = candidate
                    continue

                # Özel kural: TR_MERSIS (16 hane) içindeki 10 haneli TR_VKN veya TR_PHONE'u alt-varlık olarak yutar
                if prev.entity_type == "TR_MERSIS" and candidate.entity_type in {"TR_VKN", "TR_PHONE"}:
                    prev.sub_entities.append(candidate)
                    continue
                if candidate.entity_type == "TR_MERSIS" and prev.entity_type in {"TR_VKN", "TR_PHONE"}:
                    candidate.sub_entities.append(prev)
                    kept[-1] = candidate
                    continue

                prev_is_container = prev.entity_type in CONTAINER_ENTITY_TYPES
                cand_is_container = candidate.entity_type in CONTAINER_ENTITY_TYPES

                if (
                    prev_is_container
                    and not cand_is_container
                    and candidate.start >= prev.start
                    and candidate.end <= prev.end
                    and prev_len > cand_len
                ):
                    prev.sub_entities.append(candidate)
                    continue

                if (
                    cand_is_container
                    and not prev_is_container
                    and prev.start >= candidate.start
                    and prev.end <= candidate.end
                    and cand_len > prev_len
                ):
                    candidate.sub_entities.append(prev)
                    kept[-1] = candidate
                    continue

                if candidate.score > prev.score or (
                    candidate.score == prev.score and cand_len > prev_len
                ):
                    kept[-1] = candidate
            else:
                kept.append(candidate)

        return kept

    def analyze(
        self,
        text: str,
        redact_types: Optional[Set[str]] = None,
    ) -> UnifiedAnalysisReport:
        if not text:
            return UnifiedAnalysisReport(text="", entities=[], redacted_text="")

        raw_entities: List[DetectedEntitySpan] = []

        # 1. kvkk-pii çekirdek tanıyıcıları
        if self.kvkk_analyzer is not None:
            for e in self.kvkk_analyzer.analyze(text):
                if isinstance(e, dict):
                    raw_entities.append(
                        DetectedEntitySpan(
                            entity_type=str(e["entity_type"]),
                            start=int(e["start"]),
                            end=int(e["end"]),
                            score=float(e["score"]),
                            text=str(e["text"]),
                            source=str(e.get("source", "kvkk_pii")),
                            tier="VERIFIED_PII",
                            redaction_token=f"[{e['entity_type']}]",
                        )
                    )
                else:
                    raw_entities.append(
                        DetectedEntitySpan(
                            entity_type=e.entity_type,
                            start=e.start,
                            end=e.end,
                            score=e.score,
                            text=e.text,
                            source=e.source,
                            tier="VERIFIED_PII",
                            redaction_token=f"[{e.entity_type}]",
                        )
                    )

        # 2. Hukuki (`ictihat_core`), Ticari ve Fail-Closed OCR-Toleranslı (`VERIFIED_PII` + `SUSPECT_PII`) tanıyıcılar
        raw_entities.extend(self.legal_suite.analyze(text))

        # 3. Bağlamsal hibrit tanıyıcılar (`PERSON`, `ADDRESS`, `HEALTH_DATA`, `KVKK_SPECIAL`, `ORG_COMPANY`, `TAX_OFFICE`)
        raw_entities.extend(self._extract_contextual_entities(text))

        # Eşik filtreleme ve Hiyerarşik Span Kapsama
        filtered = [e for e in raw_entities if e.score >= self.score_threshold]
        deduped = self.deduplicate_hierarchical(filtered)

        counts: Dict[str, int] = {}
        for ent in deduped:
            counts[ent.entity_type] = counts.get(ent.entity_type, 0) + 1

        redacted = self.redact_text(text, deduped, redact_types=redact_types)
        return UnifiedAnalysisReport(
            text=text,
            entities=deduped,
            redacted_text=redacted,
            entity_counts=counts,
        )

    @staticmethod
    def redact_text(
        text: str,
        entities: List[DetectedEntitySpan],
        redact_types: Optional[Set[str]] = None,
    ) -> str:
        target_types = redact_types if redact_types is not None else DEFAULT_REDACTABLE_ENTITY_TYPES
        to_redact = [e for e in entities if e.entity_type in target_types]
        if not to_redact:
            return text

        result = text
        for ent in sorted(to_redact, key=lambda x: x.start, reverse=True):
            token = ent.redaction_token or f"[{ent.entity_type}]"
            result = result[: ent.start] + token + result[ent.end :]
        return result

	"""
	UnifiedTurkishEntityEngine — `ictihat_core` × `kvkk-pii` × `TürkçeOCR` Birleşik Varlık Tanıma,
	Hiyerarşik Span Kapsama (Hierarchical Span Subsumption) ve KVKK Karartma Motoru.
	"""

	from __future__ import annotations

	import re
	import sys
	from dataclasses import dataclass, field
	from pathlib import Path
	from typing import Any, Dict, List, Optional, Set, Tuple

	from turkceocr.postprocess.legal_recognizers import (
	DetectedEntitySpan,
	LegalAndCommercialRecognizerSuite,
	)

	try:
	from kvkk_pii.engine import PIIAnalyzer as _KvkkPIIAnalyzer # type: ignore
	except Exception:
	_KvkkPIIAnalyzer = None

	# Hiyerarşik kapsayıcı varlık tipleri (içlerindeki kısa tarih/plaka/sayı spanları kapsayıcıyı sildirtmez!)
	CONTAINER_ENTITY_TYPES: Set[str] = {
	"ADDRESS",
	"HEALTH_DATA",
	"KVKK_SPECIAL",
	"ORG_COMPANY",
	"TAX_OFFICE",
	"TR_COURT_AUTHORITY",
	"TR_LAW_CITATION",
	}

	# KVKK kapsamında karartılması (redaction) zorunlu olan hassas PII + SUSPECT tipleri
	DEFAULT_REDACTABLE_ENTITY_TYPES: Set[str] = {
	"TR_TCKN",
	"TR_TCKN_SUSPECT",
	"TR_YABANCI_KIMLIK",
	"TR_VKN",
	"TR_VKN_SUSPECT",
	"TR_MERSIS",
	"TR_IBAN",
	"TR_IBAN_SUSPECT",
	"TR_PHONE",
	"EMAIL",
	"TR_EMAIL",
	"TR_KEP",
	"TR_PLATE",
	"CREDIT_CARD",
	"TR_CREDIT_CARD",
	"TR_PASSPORT",
	"IP_ADDRESS",
	"TR_IP",
	"TR_IPV6",
	"TR_MAC",
	"SECRET",
	"TR_SECRET",
	"PERSON",
	"ADDRESS",
	"HEALTH_DATA",
	"KVKK_SPECIAL",
	"TR_SGK_SICIL",
	"TR_TAPU_ADA_PARSEL",
	}

	# Bağlamsal hibrit tanıyıcı kalıpları (NER modeli yüklü olmadığında veya NER'i tamamlayıcı olarak)
	_RX_TAX_OFFICE = re.compile(
	r"\b([A-ZÇĞİÖŞÜ][a-zçğıöşüA-ZÇĞİÖŞÜ]{1,20}(?:[^\S\r\n]+[A-ZÇĞİÖŞÜa-zçğıöşü]{1,20}){0,3}[^\S\r\n]+Vergi[^\S\r\n]+Dairesi(?:[^\S\r\n]+Müdürlüğü)?)\b"
	)
	_RX_ORG_COMPANY = re.compile(
	r"\b([A-ZÇĞİÖŞÜ0-9][A-Za-zÇĞİÖŞÜçğıöşü0-9\.\-&]{1,25}(?:[^\S\r\n]+[A-Za-zÇĞİÖŞÜçğıöşü0-9\.\-&]{1,25}){0,7}[^\S\r\n]+"
	r"(?:T\.?[^\S\r\n]A\.?[^\S\r\n]Ş\.?\|T\.?[^\S\r\n]A\.?[^\S\r\n]O\.?\|A\.?[^\S\r\n]Ş\.?\|A\.?[^\S\r\n]O\.?\|LTD\.?[^\S\r\n]ŞTİ\.?\|LİMİTED[^\S\r\n]+ŞİRKETİ\|ANONİM[^\S\r\n]+ŞİRKETİ\|SAN\.?[^\S\r\n](?:VE[^\S\r\n])?TİC\.?[^\S\r\n](?:T\.?[^\S\r\n]A\.?[^\S\r\n]Ş\.?\|A\.?[^\S\r\n]Ş\.?\|LTD\.?[^\S\r\n]ŞTİ\.?)\|HOLDİNG[^\S\r\n]+A\.?[^\S\r\n]*Ş\.?))"
	)
	_RX_LABELED_PERSON = re.compile(
	r"(?:(?:Ad[ıi][^\S\r\n]Soyad[ıi]\|Müşteri(?:[^\S\r\n]Ad[ıi])?\|Hasta(?:[^\S\r\n]*Ad[ıi])?\|Alıcı\|Borçlu\|Alacaklı\|Davacı\|Davalı\|Başvurucu\|Sanık\|Müşteki\|Vekili\|Av\.\|Avukat\|Sayın\|Sn\.)"
	r"(?:[^\S\r\n]+(?:Gerçek[^\S\r\n]+Kişi\|Ad[ıi][^\S\r\n]Soyad[ıi]))?[^\S\r\n][:\-]?[^\S\r\n]*)"
	r"([A-ZÇĞİÖŞÜ][a-zçğıöşüA-ZÇĞİÖŞÜ]{1,20}(?:[^\S\r\n]+[A-ZÇĞİÖŞÜ][a-zçğıöşüA-ZÇĞİÖŞÜ]{1,22}){1,3})"
	)
	_RX_LABELED_ADDRESS = re.compile(
	r"(?:(?:İletişim[^\S\r\n]&[^\S\r\n]Adres\|Adres[i]?\|İkametgah\|Teslimat[^\S\r\n]Adresi\|Sevk[^\S\r\n]Adresi)[^\S\r\n][:\-][^\S\r\n])"
	r"([^\n\r\\|]{10,120}?)(?=\s*\\|\|(?<!Mah)(?<!Cad)(?<!Sok)(?<!Sk)(?<!Blv)(?<!Apt)(?<!No)(?<!Kat)\.\s+[A-ZÇĞİÖŞÜ]\|[\n\r]\|$)"
	)
	_RX_UNLABELED_ADDRESS = re.compile(
	r"\b([A-ZÇĞİÖŞÜ][a-zçğıöşüA-ZÇĞİÖŞÜ]{1,22}(?:[^\S\r\n]+[A-ZÇĞİÖŞÜa-zçğıöşü]{1,22}){0,2}[^\S\r\n]+(?:Mahallesi\|Mah\.\|Bulvarı\|Blv\.\|Caddesi\|Cad\.\|Sokağı\|Sok\.\|Sk\.)[^\S\r\n]+[^\n\r\\|]{0,95}?(?:(?:\d{1,3}[^\S\r\n][/\/][^\S\r\n][A-Za-zÇĞİÖŞÜçğıöşü]+)\|(?:Kat[^\S\r\n][:\.]?[^\S\r\n]\d+)\|(?:Daire[^\S\r\n][:\.]?[^\S\r\n]\d+)\|(?:No[^\S\r\n][:\.]?[^\S\r\n]\d+[A-Za-z\/0-9\s][A-Za-zÇĞİÖŞÜçğıöşü]+\s/\s[A-Za-zÇĞİÖŞÜçğıöşü]+)\|(?:No[^\S\r\n][:\.]?[^\S\r\n]\d+[A-Za-z\/0-9])))"
	)
	_RX_HEALTH_DATA = re.compile(
	r"(?:(?:Sağlık[^\S\r\n]/[^\S\r\n]Epikriz\|Tanı\|Teşhis\|ICD(?:\-10)?(?:\sKodu)?\|Reçete\|İlaç\|Kan\sGrubu\|Epikriz\|Poliklinik\sNotu)\s[:\-]\s*)"
	r"([^\n\r\\|]{5,110}?)(?=\s*\\|\|[\n\r]\|$)"
	r"\|(\b[A-Z]\d{2}\.\d\s[-–]?\s[A-Za-zÇĞİÖŞÜçğıöşü0-9\s]{4,65})"
	r"\|(\b(?:0\|A\|B\|AB)\sRh\s[\+\-]\s*\((?:Pozitif\|Negatif)\))"
	)
	_RX_KVKK_SPECIAL = re.compile(
	r"(?:(?:KVKK[^\S\r\n]Md\.?\s6[^\S\r\n]Özel[^\S\r\n]Veri\s[:\-]\s)?"
	r"(?:Dini(?:\sİnancı)?\|Mezhebi\|Etnik\sKökeni?\|Siyasi\sDüşüncesi\|Sendika(?:\sÜyeliği)?\|Dernek(?:\sÜyeliği)?\|Sabıka\sKaydı\|Adli\sSicil\|Ceza\sMahk[uû]miyeti\|Biyometrik\sVeri)\s[:\-]\s*)"
	r"([^\n\r\\|]{4,100}?)(?=\s*\\|\|[\n\r]\|$)"
	r"\|(\b[A-ZÇĞİÖŞÜa-zçğıöşü\s]{3,35}Sendikası(?:\s+üyeliği)?(?:\s+ve\s+adli\s+sicil\s+sabıka\s+kaydı)?)"
	)


	@dataclass
	class UnifiedAnalysisReport:
	"""Birleşik Varlık Tanıma ve KVKK Analiz Raporu."""

	text: str
	entities: List[DetectedEntitySpan] = field(default_factory=list)
	redacted_text: str = ""
	entity_counts: Dict[str, int] = field(default_factory=dict)

	def to_dict(self) -> Dict[str, Any]:
	return {
	"text_length": len(self.text),
	"entity_count": len(self.entities),
	"entity_counts": self.entity_counts,
	"entities": [e.to_dict() for e in self.entities],
	"redacted_text": self.redacted_text,
	}


	class UnifiedTurkishEntityEngine:
	"""
	`ictihat_core` (Hukuki Tier-1) + `kvkk-pii` (Kişisel Veri Tier-1 + BERTurkNER v3) +
	`TürkçeOCR` (Ticari Tier-1 + İki Katmanlı OCR-Toleranslı Fail-Closed Dedektör) birleşik motoru.
	"""

	def __init__(
	self,
	use_ner: bool = False,
	ner_model_path: Optional[str] = None,
	score_threshold: float = 0.50,
	) -> None:
	self.use_ner = use_ner
	self.ner_model_path = ner_model_path
	self.score_threshold = score_threshold
	self.legal_suite = LegalAndCommercialRecognizerSuite()
	self.kvkk_analyzer = None
	if _KvkkPIIAnalyzer is not None:
	try:
	self.kvkk_analyzer = _KvkkPIIAnalyzer(ner_model_path=ner_model_path)
	except Exception:
	self.kvkk_analyzer = _KvkkPIIAnalyzer()

	def _extract_contextual_entities(self, text: str) -> List[DetectedEntitySpan]:
	spans: List[DetectedEntitySpan] = []

	for m in _RX_TAX_OFFICE.finditer(text):
	s, e = m.start(1), m.end(1)
	spans.append(
	DetectedEntitySpan(
	entity_type="TAX_OFFICE",
	start=s,
	end=e,
	score=0.95,
	text=text[s:e],
	source="contextual_grammar",
	tier="VERIFIED",
	)
	)

	for m in _RX_ORG_COMPANY.finditer(text):
	s, e = m.start(1), m.end(1)
	cand_text = text[s:e].strip()
	# Mahkeme/Daire başlığı değilse şirket olarak ekle
	if "Mahkemesi" not in cand_text and "Dairesi" not in cand_text:
	spans.append(
	DetectedEntitySpan(
	entity_type="ORG_COMPANY",
	start=s,
	end=e,
	score=0.93,
	text=text[s:e],
	source="contextual_grammar",
	tier="VERIFIED",
	)
	)

	for m in _RX_LABELED_PERSON.finditer(text):
	s, e = m.start(1), m.end(1)
	cand_text = text[s:e].strip()
	if not any(
	bad in cand_text
	for bad in ("Vergi Dairesi", "Mahkemesi", "Dairesi", "Kanunu", "Müdürlüğü", "Şirketi")
	):
	spans.append(
	DetectedEntitySpan(
	entity_type="PERSON",
	start=s,
	end=e,
	score=0.92,
	text=text[s:e],
	source="contextual_grammar",
	tier="VERIFIED_PII",
	)
	)

	for m in _RX_LABELED_ADDRESS.finditer(text):
	s, e = m.start(1), m.end(1)
	spans.append(
	DetectedEntitySpan(
	entity_type="ADDRESS",
	start=s,
	end=e,
	score=0.92,
	text=text[s:e],
	source="contextual_grammar",
	tier="VERIFIED_PII",
	)
	)

	for m in _RX_UNLABELED_ADDRESS.finditer(text):
	s, e = m.start(1), m.end(1)
	spans.append(
	DetectedEntitySpan(
	entity_type="ADDRESS",
	start=s,
	end=e,
	score=0.90,
	text=text[s:e],
	source="contextual_grammar",
	tier="VERIFIED_PII",
	)
	)

	for m in _RX_HEALTH_DATA.finditer(text):
	s, e = m.start(), m.end()
	for g_idx in (1, 2, 3):
	if m.group(g_idx) is not None:
	s, e = m.start(g_idx), m.end(g_idx)
	break
	spans.append(
	DetectedEntitySpan(
	entity_type="HEALTH_DATA",
	start=s,
	end=e,
	score=0.93,
	text=text[s:e],
	source="contextual_grammar",
	tier="VERIFIED_PII",
	)
	)

	for m in _RX_KVKK_SPECIAL.finditer(text):
	s, e = m.start(), m.end()
	for g_idx in (1, 2):
	if m.group(g_idx) is not None:
	s, e = m.start(g_idx), m.end(g_idx)
	break
	spans.append(
	DetectedEntitySpan(
	entity_type="KVKK_SPECIAL",
	start=s,
	end=e,
	score=0.93,
	text=text[s:e],
	source="contextual_grammar",
	tier="VERIFIED_PII",
	)
	)

	return spans

	@staticmethod
	def deduplicate_hierarchical(entities: List[DetectedEntitySpan]) -> List[DetectedEntitySpan]:
	"""
	Hiyerarşik Span Kapsama (Hierarchical Span Subsumption):
	1. `TR_COURT_AUTHORITY` (`score=1.0`) kendisiyle örtüşen yanlış pozitif `ORG_COMPANY`,
	`ADDRESS` veya `PERSON` spanlarını ezer.
	2. Kapsayıcı bir varlık (`ADDRESS`, `HEALTH_DATA`, `KVKK_SPECIAL`, `ORG_COMPANY`,
	`TAX_OFFICE`, `TR_COURT_AUTHORITY`, `TR_LAW_CITATION`) kendisinden daha kısa bir
	alt-span (`TR_DATE`, `TR_PLATE`, `TR_ESAS_KARAR_NO` vb.) ile örtüştüğünde kapsayıcı
	varlık ASLA silinmez; kısa alt-span `container.sub_entities` listesine eklenir.
	"""
	if not entities:
	return []

	sorted_ents = sorted(
	entities,
	key=lambda e: (e.start, -(e.end - e.start), -e.score),
	)

	kept: List[DetectedEntitySpan] = []
	for candidate in sorted_ents:
	if not kept:
	kept.append(candidate)
	continue

	prev = kept[-1]
	if candidate.start < prev.end:
	prev_len = prev.end - prev.start
	cand_len = candidate.end - candidate.start

	# Özel kural: TR_COURT_AUTHORITY her zaman ORG_COMPANY / ADDRESS / PERSON yanlış pozitifini ezer
	if prev.entity_type == "TR_COURT_AUTHORITY" and candidate.entity_type in {
	"ORG_COMPANY",
	"ADDRESS",
	"PERSON",
	}:
	continue
	if candidate.entity_type == "TR_COURT_AUTHORITY" and prev.entity_type in {
	"ORG_COMPANY",
	"ADDRESS",
	"PERSON",
	}:
	kept[-1] = candidate
	continue

	# Özel kural: TR_MERSIS (16 hane) içindeki 10 haneli TR_VKN veya TR_PHONE'u alt-varlık olarak yutar
	if prev.entity_type == "TR_MERSIS" and candidate.entity_type in {"TR_VKN", "TR_PHONE"}:
	prev.sub_entities.append(candidate)
	continue
	if candidate.entity_type == "TR_MERSIS" and prev.entity_type in {"TR_VKN", "TR_PHONE"}:
	candidate.sub_entities.append(prev)
	kept[-1] = candidate
	continue

	prev_is_container = prev.entity_type in CONTAINER_ENTITY_TYPES
	cand_is_container = candidate.entity_type in CONTAINER_ENTITY_TYPES

	if (
	prev_is_container
	and not cand_is_container
	and candidate.start >= prev.start
	and candidate.end <= prev.end
	and prev_len > cand_len
	):
	prev.sub_entities.append(candidate)
	continue

	if (
	cand_is_container
	and not prev_is_container
	and prev.start >= candidate.start
	and prev.end <= candidate.end
	and cand_len > prev_len
	):
	candidate.sub_entities.append(prev)
	kept[-1] = candidate
	continue

	if candidate.score > prev.score or (
	candidate.score == prev.score and cand_len > prev_len
	):
	kept[-1] = candidate
	else:
	kept.append(candidate)

	return kept

	def analyze(
	self,
	text: str,
	redact_types: Optional[Set[str]] = None,
	) -> UnifiedAnalysisReport:
	if not text:
	return UnifiedAnalysisReport(text="", entities=[], redacted_text="")

	raw_entities: List[DetectedEntitySpan] = []

	# 1. kvkk-pii çekirdek tanıyıcıları
	if self.kvkk_analyzer is not None:
	for e in self.kvkk_analyzer.analyze(text):
	if isinstance(e, dict):
	raw_entities.append(
	DetectedEntitySpan(
	entity_type=str(e["entity_type"]),
	start=int(e["start"]),
	end=int(e["end"]),
	score=float(e["score"]),
	text=str(e["text"]),
	source=str(e.get("source", "kvkk_pii")),
	tier="VERIFIED_PII",
	redaction_token=f"[{e['entity_type']}]",
	)
	)
	else:
	raw_entities.append(
	DetectedEntitySpan(
	entity_type=e.entity_type,
	start=e.start,
	end=e.end,
	score=e.score,
	text=e.text,
	source=e.source,
	tier="VERIFIED_PII",
	redaction_token=f"[{e.entity_type}]",
	)
	)

	# 2. Hukuki (`ictihat_core`), Ticari ve Fail-Closed OCR-Toleranslı (`VERIFIED_PII` + `SUSPECT_PII`) tanıyıcılar
	raw_entities.extend(self.legal_suite.analyze(text))

	# 3. Bağlamsal hibrit tanıyıcılar (`PERSON`, `ADDRESS`, `HEALTH_DATA`, `KVKK_SPECIAL`, `ORG_COMPANY`, `TAX_OFFICE`)
	raw_entities.extend(self._extract_contextual_entities(text))

	# Eşik filtreleme ve Hiyerarşik Span Kapsama
	filtered = [e for e in raw_entities if e.score >= self.score_threshold]
	deduped = self.deduplicate_hierarchical(filtered)

	counts: Dict[str, int] = {}
	for ent in deduped:
	counts[ent.entity_type] = counts.get(ent.entity_type, 0) + 1

	redacted = self.redact_text(text, deduped, redact_types=redact_types)
	return UnifiedAnalysisReport(
	text=text,
	entities=deduped,
	redacted_text=redacted,
	entity_counts=counts,
	)

	@staticmethod
	def redact_text(
	text: str,
	entities: List[DetectedEntitySpan],
	redact_types: Optional[Set[str]] = None,
	) -> str:
	target_types = redact_types if redact_types is not None else DEFAULT_REDACTABLE_ENTITY_TYPES
	to_redact = [e for e in entities if e.entity_type in target_types]
	if not to_redact:
	return text

	result = text
	for ent in sorted(to_redact, key=lambda x: x.start, reverse=True):
	token = ent.redaction_token or f"[{ent.entity_type}]"
	result = result[: ent.start] + token + result[ent.end :]
	return result