LQStandardsService/service/matcher.py

from __future__ import annotations

import csv
import io
import re
import unicodedata
from dataclasses import dataclass, field
from functools import lru_cache
from pathlib import Path

from rapidfuzz import fuzz


_DASH_TRANSLATION = str.maketrans(
    {
        "‐": "-",
        "‑": "-",
        "‒": "-",
        "–": "-",
        "—": "-",
        "―": "-",
        "−": "-",
        "﹣": "-",
        "－": "-",
        "﹕": ":",
        "：": ":",
    }
)
_CODE_PATTERN = re.compile(
    r"(?<![A-Za-z0-9])"
    r"(?P<code>"
    r"[A-Za-z][A-Za-z/]*"
    r"(?:\s*[A-Za-z])?"
    r"\s*(?::\s*[A-Za-z])?"
    r"\s*\d"
    r"[A-Za-z0-9./:\-‐‑‒–—―−﹣－]*"
    r")",
    re.IGNORECASE,
)
_YEAR_PATTERN = re.compile(r"-(19\d{2}|20\d{2})$")
_SHORT_YEAR_PATTERN = re.compile(r"^(.*)-(\d{2})$")


def _nfkc(value: str) -> str:
    return unicodedata.normalize("NFKC", value or "").translate(_DASH_TRANSLATION)


def normalize_code(value: str) -> str:
    normalized = _nfkc(value).upper()
    return re.sub(r"[^A-Z0-9/.:+\-]", "", normalized)


def normalize_title(value: str) -> str:
    normalized = _nfkc(value).lower()
    return "".join(
        character
        for character in normalized
        if unicodedata.category(character)[0] not in {"P", "S", "Z"}
    )


def extract_query_parts(keyword: str) -> tuple[str, str]:
    normalized = _nfkc(keyword)
    prepared = re.sub(r"\s*([/.:+\-])\s*", r"\1", normalized)
    prepared = re.sub(r"(?<=[A-Za-z])\s+(?=[A-Za-z0-9])", "", prepared)
    prepared = re.sub(r"(?<=[0-9])\s+(?=[0-9])", "", prepared)
    match = _CODE_PATTERN.search(prepared)
    if not match:
        return "", normalize_title(prepared)

    query_code = normalize_code(match.group("code"))
    remaining = prepared[: match.start()] + " " + prepared[match.end() :]
    return query_code, normalize_title(remaining)


@lru_cache(maxsize=32768)
def _year(code: str) -> int | None:
    match = _YEAR_PATTERN.search(code)
    return int(match.group(1)) if match else None


@lru_cache(maxsize=32768)
def _without_year(code: str) -> str:
    return _YEAR_PATTERN.sub("", code)


@lru_cache(maxsize=32768)
def _relax_standard_type(code: str) -> str:
    first_digit = next((index for index, char in enumerate(code) if char.isdigit()), len(code))
    prefix = code[:first_digit].replace("/T", "")
    return prefix + code[first_digit:]


@lru_cache(maxsize=32768)
def _loose_code(code: str) -> str:
    return re.sub(r"[^A-Z0-9]", "", code)


@lru_cache(maxsize=65536)
def _two_digit_year_equivalent(left: str, right: str) -> bool:
    for short_code, full_code in ((left, right), (right, left)):
        short_match = _SHORT_YEAR_PATTERN.match(short_code)
        full_year = _year(full_code)
        if not short_match or full_year is None:
            continue
        if short_match.group(1) != _without_year(full_code):
            continue
        if int(short_match.group(2)) == full_year % 100:
            return True
    return False


@lru_cache(maxsize=131072)
def code_similarity(query_code: str, candidate_code: str) -> float:
    if not query_code or not candidate_code:
        return 0.0
    if query_code == candidate_code:
        return 1.0
    if _two_digit_year_equivalent(query_code, candidate_code):
        return 0.98
    if _loose_code(query_code) == _loose_code(candidate_code):
        return 0.98

    query_relaxed = _relax_standard_type(query_code)
    candidate_relaxed = _relax_standard_type(candidate_code)
    if query_relaxed == candidate_relaxed:
        return 0.93
    if _two_digit_year_equivalent(query_relaxed, candidate_relaxed):
        return 0.93

    query_year = _year(query_code)
    candidate_year = _year(candidate_code)
    if query_year is None and candidate_year is not None:
        query_without_type = _relax_standard_type(query_code)
        candidate_stem = _without_year(_relax_standard_type(candidate_code))
        if query_without_type == candidate_stem:
            return 0.90
        if candidate_stem.startswith(query_without_type):
            suffix = candidate_stem[len(query_without_type) :]
            if suffix and suffix[0] in {".", "/", ":", "-"}:
                return 0.80

    if query_year is not None and candidate_year is not None and query_year != candidate_year:
        if _without_year(query_relaxed) == _without_year(candidate_relaxed):
            return 0.70
        return 0.0

    if query_year is not None and candidate_year == query_year:
        ratio = fuzz.ratio(query_code, candidate_code) / 100
        if ratio >= 0.88:
            return 0.75

    return 0.0


def _han_count(value: str) -> int:
    return sum("\u4e00" <= character <= "\u9fff" for character in value)


@lru_cache(maxsize=131072)
def title_similarity(query_title: str, candidate_title: str) -> float:
    if not query_title or not candidate_title:
        return 0.0
    if query_title == candidate_title:
        return 1.0
    if candidate_title in query_title and len(candidate_title) >= 4:
        coverage = len(candidate_title) / len(query_title)
        return min(0.98, 0.90 + 0.08 * coverage)
    if query_title in candidate_title and len(query_title) >= 4:
        coverage = len(query_title) / len(candidate_title)
        return min(0.95, 0.65 + 0.35 * coverage)
    if _han_count(query_title) < 6:
        return 0.0
    return fuzz.WRatio(query_title, candidate_title) / 100


@dataclass(frozen=True)
class Standard:
    code: str
    normalized_code: str
    title: str
    status: str
    relation_text: str
    implement_date: str
    official_key: str = field(init=False)
    alias_key: str = field(init=False)
    normalized_title: str = field(init=False)
    official_year: int = field(init=False)

    def __post_init__(self) -> None:
        official_key = normalize_code(self.code or self.normalized_code)
        object.__setattr__(self, "official_key", official_key)
        object.__setattr__(self, "alias_key", normalize_code(self.normalized_code))
        object.__setattr__(self, "normalized_title", normalize_title(self.title))
        object.__setattr__(self, "official_year", _year(official_key) or 0)

    def code_score(self, query_code: str) -> float:
        score = code_similarity(query_code, self.official_key)

        if self.alias_key and self.alias_key != self.official_key:
            query_to_alias = code_similarity(query_code, self.alias_key)
            alias_to_official = code_similarity(self.alias_key, self.official_key)
            score = max(score, min(query_to_alias, alias_to_official))
        return score

    def as_candidate(self, score: float) -> dict[str, str | float]:
        return {
            "code": self.code,
            "title": self.title,
            "status": self.status,
            "relation_text": self.relation_text,
            "implement_date": self.implement_date,
            "match_score": round(score, 4),
        }


class StandardMatcher:
    def __init__(self, standards: list[Standard]):
        self._standards = standards

    @classmethod
    def from_csv(cls, path: str | Path) -> "StandardMatcher":
        csv_path = Path(path)
        content = cls._read_csv_text(csv_path)
        reader = csv.DictReader(io.StringIO(content))
        standards = []
        for row in reader:
            code = (row.get("csres_code") or "").strip()
            normalized_code = (row.get("csres_normalized_code") or "").strip()
            title = (row.get("csres_title") or "").strip()
            if not (code or normalized_code):
                continue
            standards.append(
                Standard(
                    code=code or normalized_code,
                    normalized_code=normalized_code or code,
                    title=title,
                    status=(row.get("csres_status") or "").strip(),
                    relation_text=(row.get("csres_relation_text") or "").strip(),
                    implement_date=(row.get("csres_implement_date") or "").strip(),
                )
            )
        return cls(standards)

    @staticmethod
    def _read_csv_text(path: Path) -> str:
        for encoding in ("utf-8-sig", "gb18030"):
            try:
                return path.read_text(encoding=encoding)
            except UnicodeDecodeError:
                continue
        raise UnicodeDecodeError("utf-8", b"", 0, 1, f"无法识别文件编码: {path}")

    def search(self, keyword: str, limit: int = 3) -> list[dict[str, str | float]]:
        query_code, query_title = extract_query_parts(keyword)
        if not query_code and not query_title:
            return []

        scored = []
        for standard in self._standards:
            code_score = standard.code_score(query_code) if query_code else 0.0
            name_score = (
                title_similarity(query_title, standard.normalized_title)
                if query_title
                else 0.0
            )
            code_relevant = bool(query_code) and code_score >= 0.70
            name_relevant = bool(query_title) and name_score >= 0.75
            if not (code_relevant or name_relevant):
                continue

            if query_code and query_title:
                higher = max(code_score, name_score)
                lower = min(code_score, name_score)
                final_score = higher * 0.70 + lower * 0.30
            elif query_code:
                final_score = code_score
            else:
                final_score = name_score
            scored.append((standard, final_score, code_score))

        deduplicated: dict[str, tuple[Standard, float, float]] = {}
        for standard, score, code_score in scored:
            key = standard.official_key
            current = deduplicated.get(key)
            if current is None or (score, code_score) > (current[1], current[2]):
                deduplicated[key] = (standard, score, code_score)

        ordered = sorted(
            deduplicated.values(),
            key=lambda item: (
                -item[1],
                -item[2],
                -item[0].official_year,
                item[0].official_key,
            ),
        )
        return [standard.as_candidate(score) for standard, score, _ in ordered[:limit]]