Blame for feature/substring.py

from __future__ import annotations
import re
from dataclasses import dataclass, field
from typing import Any, Optional

import numpy as np
import pandas as pd

from .categorical import normalize_value, prepare_pair_frame
from .doc_alignment import DocAlignmentModel
from schema import NodeSchema

SUBSTRING_WEIGHT_PROFILES = {
    "human_relevance": {
        "support": 0.10,
        "prefix_match": 0.02,
        "suffix_match": 0.02,
        "substring_match": 0.05,
        "doc_alignment": 0.05,
    },
    "sample": {
        "support": 0.10,
        "prefix_match": 0.0,
        "suffix_match": 0.0,
        "substring_match": 0.0,
        "doc_alignment": 0.05,
    },
    "file": {
        "support": 0.15,
        "prefix_match": 0.25,
        "suffix_match": 0.30,
        "substring_match": 0.25,
        "doc_alignment": 0.05,
    },
     "case": {
        "support": 0.15,
        "prefix_match": 0.25,
        "suffix_match": 0.30,
        "substring_match": 0.25,
        "doc_alignment": 0.05,
    },
    "study": {
        "support": 0.10,
        "prefix_match": 0.30,
        "suffix_match": 0.30,
        "substring_match": 0.25,
        "doc_alignment": 0.05,
    },
}

def smart_contains(av: str, bv: str) -> bool:
    tokens = re.split(r'[^a-z0-9]+', av.lower())
    bv = bv.lower()
    return any(t and t in bv for t in tokens)

def prefix_match_score(df: pd.DataFrame, a: str, b: str) -> float:
    total = 0
    match = 0
    for _, row in df.iterrows():
        av = normalize_value(row.get(a, "")).lower()
        bv = normalize_value(row.get(b, "")).lower()
        if not av or not bv:
            continue
        total += 1
        if bv.startswith(av):
            match += 1
    return match / total if total else 0.0


def suffix_match_score(df: pd.DataFrame, a: str, b: str) -> float:
    total = 0
    match = 0
    for _, row in df.iterrows():
        av = normalize_value(row.get(a, "")).lower()
        bv = normalize_value(row.get(b, "")).lower()
        if not av or not bv:
            continue
        total += 1
        if bv.endswith(av):
            match += 1
    return match / total if total else 0.0

def substring_match_score(df: pd.DataFrame, a: str, b: str) -> float:
    total = 0
    match = 0
    for _, row in df.iterrows():
        av = normalize_value(row.get(a, "")).lower()
        bv = normalize_value(row.get(b, "")).lower()
        if not av or not bv:
            continue
        total += 1
        if smart_contains(av, bv):
            match += 1
    return match / total if total else 0.0

def get_substring_weights(node_name: str) -> dict[str, float] | None:
    key = (node_name or "").lower()
    return SUBSTRING_WEIGHT_PROFILES.get(key)

@dataclass
class SubstringFeatureAnalyzer:
    node_schema: NodeSchema
    doc_model: DocAlignmentModel
    all_weights: Optional[dict[str, dict[str, float]]] = None

    def __post_init__(self) -> None:
        if self.node_schema is not None:
            key = self.node_schema.name
        else:
            key = "cross_node_match"
        # object.__setattr__(self, "weights", get_substring_weights(key))

        # default_weights = get_substring_weights(key) or {}
        # incoming_weights = self.weights or {}

        # # merge, with incoming weights taking priority
        # merged = {**default_weights, **incoming_weights}
        # object.__setattr__(self, "weights", merged)

        profiles = self.all_weights or SUBSTRING_WEIGHT_PROFILES
        # key = (self.node_schema.name or "").lower()
        object.__setattr__(self, "weights", profiles.get(key, {}))


    @staticmethod
    def classify_strength(score: float) -> str:
        if score >= 0.9:
            return "functional"
        if score >= 0.7:
            return "strong"
        if score >= 0.45:
            return "conditional"
        if score >= 0.2:
            return "weak"
        return "independent"

    def analyze(self, df: pd.DataFrame, a: str, b: str) -> dict[str, Any]:
        if self.weights is None:
            return None
        pair = prepare_pair_frame(df, a, b)
        total_rows = len(df)
        row_count = len(pair)
        support = float(row_count / total_rows) if total_rows else 0.0

        prefix = prefix_match_score(pair, a, b)
        suffix = suffix_match_score(pair, a, b)
        substring = substring_match_score(pair, a, b)
        doc_alignment = float(self.doc_model.score(a, b)) or 0.0

        strength = (
            self.weights["support"] * support
            + self.weights["prefix_match"] * prefix
            + self.weights["suffix_match"] * suffix
            + self.weights["substring_match"] * substring
            + self.weights["doc_alignment"] * doc_alignment
        )

        return {
            "A": a,
            "B": b,
            "feature_type": "substring",
            "support": support,
            "prefix_match": prefix,
            "suffix_match": suffix,
            "substring_match": substring,
            "doc_alignment": doc_alignment,
            "strength": float(np.clip(strength, 0.0, 1.0)),
            "classification": self.classify_strength(strength),
            "row_count": row_count,
            "total_rows": total_rows,
            "evidence": self._build_evidence(pair, a, b),
        }

    def _build_evidence(self, pair: pd.DataFrame, a: str, b: str, limit: int = 20) -> list[dict[str, Any]]:
        if pair.empty:
            return []

        evidence: list[dict[str, Any]] = []
        for _, row in pair.head(limit).iterrows():
            av = str(row[a])
            bv = str(row[b])
            relation = "contains" if av.lower() in bv.lower() else "no_match"
            if bv.lower().startswith(av.lower()):
                relation = "prefix"
            elif bv.lower().endswith(av.lower()):
                relation = "suffix"

            evidence.append({"A_value": av, "B_value": bv, "match_type": relation})

        return evidence


    
    def _build_mapping(self, pair: pd.DataFrame, a: str, b: str) -> dict[str, Any]:
        mapping: dict[str, Any] = {}
        for a_val, subset in pair.groupby(a):
            counts = subset[b].value_counts(normalize=True)
            if counts.empty:
                continue
            top_val = counts.index[0]
            top_prob = float(counts.iloc[0])
            mapping[str(a_val)] = str(top_val) if top_prob >= 0.95 else [str(v) for v in counts.index]
        return mapping