Human-in-the-Loop Entity Resolution

# Common title prefixes that don't change entity identity
_TITLE_PREFIXES = (
    "detective", "det.", "officer", "sergeant", "sgt.", "lieutenant", "lt.",
    "captain", "cpt.", "chief", "deputy", "agent", "special agent",
    "dr.", "dr", "doctor", "prof.", "professor",
    "mr.", "mr", "mrs.", "mrs", "ms.", "ms", "miss",
    "judge", "justice", "hon.", "honorable",
    "senator", "sen.", "representative", "rep.", "governor", "gov.",
    "president", "vice president",
    "attorney", "atty.", "counsel", "esquire", "esq.",
    "reverend", "rev.", "father", "sister", "brother",
    "sir", "dame", "lord", "lady",
)


def _strip_titles(name: str) -> str:
    """Strip common title prefixes from a name."""
    changed = True
    while changed:
        changed = False
        for prefix in _TITLE_PREFIXES:
            if name.startswith(prefix + " "):
                name = name[len(prefix) + 1:].strip()
                changed = True
                break
    return name

def _find_cross_type_duplicates(kg: KnowledgeGraph) -> list[MergeProposal]:
    """Find entities with the same name but different types.

    When the LLM extracts "reading comprehension" as both CONCEPT and
    PHENOMENON, these are the same entity with an inconsistent type.
    The canonical is the one with more connections (more context for
    the type assignment). All relations get combined on merge.
    """
    from collections import defaultdict

    # Group by normalized name
    name_groups: dict[str, list[tuple[str, str, int]]] = defaultdict(list)
    for nid, data in kg.graph.nodes(data=True):
        entity_type = data.get("entity_type", "")
        if entity_type in SKIP_TYPES:
            continue
        name = data.get("name", "").strip().lower()
        if not name:
            continue
        degree = kg.graph.degree(nid)
        name_groups[name].append((nid, entity_type, degree))

class MergeProposal(BaseModel):
    """A proposed merge of multiple entities into one canonical entity.

    The `members` list contains the non-canonical entities that will be merged
    INTO the canonical entity. The canonical entity itself is identified by
    `canonical_id` and is not included in `members`.
    """

    canonical_id: str
    canonical_name: str
    entity_type: str
    status: StatusType = "DRAFT"
    members: list[MergeMember] = Field(min_length=1)
    reason: str = ""  # LLM's explanation for why these should merge


class MergeFile(BaseModel):
    """Top-level model for merge_proposals.yaml."""

    proposals: list[MergeProposal] = Field(default_factory=list)

    @property
    def confirmed(self) -> list[MergeProposal]:
        return [p for p in self.proposals if p.status == "CONFIRMED"]

    @property
    def draft(self) -> list[MergeProposal]:
        return [p for p in self.proposals if p.status == "DRAFT"]

    drafts = merge_file.draft
    if not drafts:
        console.print("[dim]No merge proposals to review.[/dim]")
        return {"auto_approved": 0, "approved": 0, "rejected": 0, "skipped": 0}

    # Auto-approve high-confidence proposals
    auto_approved = []
    manual_review = []
    for proposal in drafts:
        min_conf = min(m.confidence for m in proposal.members)
        if auto_approve_threshold < 1.0 and min_conf >= auto_approve_threshold:
            proposal.status = "CONFIRMED"
            auto_approved.append(proposal)
        else:
            manual_review.append(proposal)

    """Apply confirmed merge proposals to the knowledge graph.

    For each confirmed proposal:
    1. Merge member node data into canonical node
    2. Rewrite all edges pointing to/from members to point to canonical
    3. Remove member nodes
    4. Remove self-loops created by merging

    Args:
        kg: KnowledgeGraph to modify in place
        merge_file: MergeFile with proposals (only CONFIRMED are applied)

    Returns:
        Stats dict with counts
    """
    confirmed = merge_file.confirmed
    if not confirmed:
        logger.info("No confirmed merges to apply")
        return {"merges_applied": 0, "nodes_removed": 0, "self_loops_removed": 0}

    stats = {"merges_applied": 0, "nodes_removed": 0, "self_loops_removed": 0}

    # Build full merge map: member_id → canonical_id
    merge_map: dict[str, str] = {}
    for proposal in confirmed:
        for member in proposal.members:
            if member.id != proposal.canonical_id:
                merge_map[member.id] = proposal.canonical_id