YG_FT/backend/app/modules/data_process/algorithms/quality.py

"""数据处理算法 - 质量评分和去重。"""

from __future__ import annotations

import hashlib
import json
import math
import re
import unicodedata
from collections import Counter
from collections.abc import Iterable, Mapping, Sequence
from copy import deepcopy
from typing import Any

from .text_utils import normalize_text
from .types import (
    _MAX_ANOMALY_TEXT_CHARS,
    _MOJIBAKE_MARKERS,
    _TOKEN_PATTERN,
    ProcessedStructuredRecord,
    QualityScore,
)


def estimate_token_count(text: str) -> int:
    """粗略估计文本的 token 数量。"""
    return len(_TOKEN_PATTERN.findall(text))


def content_quality_flags(
    text: str,
    *,
    min_chars: int = 20,
    min_tokens: int = 5,
    max_chars: int = _MAX_ANOMALY_TEXT_CHARS,
) -> tuple[str, ...]:
    """返回非结构化内容的确定性低质量原因。"""

    if min_chars < 0 or min_tokens < 0 or max_chars <= 0:
        raise ValueError("content quality limits must be non-negative")
    normalized = normalize_text(text)
    if not normalized:
        return ("empty_content",)
    flags: list[str] = []
    if len(normalized) < min_chars or estimate_token_count(normalized) < min_tokens:
        flags.append("content_too_short")
    if len(normalized) > max_chars:
        flags.append("content_too_long")
    if any(marker in normalized for marker in _MOJIBAKE_MARKERS):
        flags.append("mojibake")
    nonspace = [char for char in normalized if not char.isspace()]
    if nonspace:
        readable_ratio = sum(
            char.isprintable()
            and unicodedata.category(char) not in {"Co", "Cs", "Cn"}
            for char in nonspace
        ) / len(nonspace)
        if readable_ratio < 0.85:
            flags.append("low_printable_ratio")
        if len(nonspace) >= 100:
            most_common = Counter(nonspace).most_common(1)[0][1]
            if most_common / len(nonspace) > 0.9:
                flags.append("repetitive_content")
    return tuple(dict.fromkeys(flags))

def is_low_quality_content(
    text: str,
    *,
    min_chars: int = 20,
    min_tokens: int = 5,
    max_chars: int = _MAX_ANOMALY_TEXT_CHARS,
) -> bool:
    """判断内容是否命中任一低质量规则。"""

    return bool(
        content_quality_flags(
            text,
            min_chars=min_chars,
            min_tokens=min_tokens,
            max_chars=max_chars,
        )
    )

def _deduplicate_structured_entries(
    entries: Sequence[ProcessedStructuredRecord],
) -> list[ProcessedStructuredRecord]:
    """仅按整条 canonical JSON 稳定去重，避免误删同 ID 的更新记录。"""

    # canonical_record_json 位于 structured_processing，延迟导入以断开循环依赖。
    from .structured_processing import canonical_record_json

    exact_seen: set[str] = set()
    unique: list[ProcessedStructuredRecord] = []
    for entry in entries:
        record = entry.record
        fingerprint = hashlib.sha256(canonical_record_json(record).encode("utf-8")).hexdigest()
        if fingerprint in exact_seen:
            continue
        exact_seen.add(fingerprint)
        unique.append(
            ProcessedStructuredRecord(
                entry.source_index,
                deepcopy(dict(record)),
            )
        )
    return unique

def deduplicate_structured_records(
    records: Sequence[Mapping[str, Any]],
) -> list[dict[str, Any]]:
    """仅按整条 canonical JSON 稳定去重。"""

    entries = [
        ProcessedStructuredRecord(index, deepcopy(dict(record)))
        for index, record in enumerate(records)
    ]
    return [entry.record for entry in _deduplicate_structured_entries(entries)]

def _near_duplicate_features(text: str, shingle_size: int) -> tuple[str, ...]:
    if isinstance(shingle_size, bool) or not isinstance(shingle_size, int):
        raise TypeError("shingle_size must be an integer")
    if shingle_size <= 0:
        raise ValueError("shingle_size must be greater than 0")
    tokens = re.findall(
        r"[\u3400-\u4dbf\u4e00-\u9fff]|[A-Za-z0-9_]+",
        normalize_text(text).casefold(),
    )
    if not tokens:
        return ()
    if len(tokens) < shingle_size:
        return ("\x1f".join(tokens),)
    return tuple(
        "\x1f".join(tokens[index : index + shingle_size])
        for index in range(len(tokens) - shingle_size + 1)
    )

def near_duplicate_fingerprint(text: str, *, shingle_size: int = 3) -> str:
    """生成 64 位 SimHash 指纹，用于低成本近重复候选筛选。"""

    if isinstance(shingle_size, bool) or not isinstance(shingle_size, int):
        raise TypeError("shingle_size must be an integer")
    if shingle_size <= 0:
        raise ValueError("shingle_size must be greater than 0")
    features = Counter(_near_duplicate_features(text, shingle_size))
    if not features:
        return "0" * 16
    vector = [0] * 64
    for feature, weight in features.items():
        digest = int.from_bytes(hashlib.sha256(feature.encode("utf-8")).digest()[:8], "big")
        for bit in range(64):
            vector[bit] += weight if digest & (1 << bit) else -weight
    fingerprint = sum(1 << bit for bit, value in enumerate(vector) if value >= 0)
    return f"{fingerprint:016x}"

def fingerprints_are_near_duplicate(
    left: str,
    right: str,
    *,
    max_hamming_distance: int = 3,
) -> bool:
    """比较两个 64 位十六进制 SimHash 指纹。"""

    if isinstance(max_hamming_distance, bool) or not isinstance(max_hamming_distance, int):
        raise TypeError("max_hamming_distance must be an integer")
    if not 0 <= max_hamming_distance <= 64:
        raise ValueError("max_hamming_distance must be in [0, 64]")
    if not re.fullmatch(r"[0-9a-fA-F]{16}", left) or not re.fullmatch(
        r"[0-9a-fA-F]{16}", right
    ):
        raise ValueError("fingerprints must be 16-character hexadecimal strings")
    distance = (int(left, 16) ^ int(right, 16)).bit_count()
    return distance <= max_hamming_distance

def is_near_duplicate(
    left: str,
    right: str,
    *,
    shingle_size: int = 3,
    similarity_threshold: float = 0.9,
    max_hamming_distance: int = 3,
) -> bool:
    """结合词片 Jaccard 和 SimHash 判断两段内容是否近重复。"""

    if isinstance(similarity_threshold, bool) or not isinstance(
        similarity_threshold, (int, float)
    ):
        raise TypeError("similarity_threshold must be a number")
    if not 0 <= similarity_threshold <= 1:
        raise ValueError("similarity_threshold must be in [0, 1]")
    if isinstance(max_hamming_distance, bool) or not isinstance(max_hamming_distance, int):
        raise TypeError("max_hamming_distance must be an integer")
    if not 0 <= max_hamming_distance <= 64:
        raise ValueError("max_hamming_distance must be in [0, 64]")
    left_normalized = normalize_text(left)
    right_normalized = normalize_text(right)
    if not left_normalized or not right_normalized:
        return left_normalized == right_normalized
    if left_normalized.casefold() == right_normalized.casefold():
        return True
    left_features = set(_near_duplicate_features(left_normalized, shingle_size))
    right_features = set(_near_duplicate_features(right_normalized, shingle_size))
    union = left_features | right_features
    similarity = len(left_features & right_features) / len(union) if union else 1.0
    if similarity >= similarity_threshold:
        return True
    return fingerprints_are_near_duplicate(
        near_duplicate_fingerprint(left_normalized, shingle_size=shingle_size),
        near_duplicate_fingerprint(right_normalized, shingle_size=shingle_size),
        max_hamming_distance=max_hamming_distance,
    )

def record_fingerprint(record: Mapping[str, Any]) -> str:
    """计算与字典键顺序无关的稳定记录指纹。"""

    canonical = {
        "instruction": normalize_text(str(record.get("instruction") or "")),
        "input": normalize_text(str(record.get("input") or "")),
        "output": normalize_text(str(record.get("output") or "")),
    }
    raw = json.dumps(canonical, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()

def _readability_score(text: str) -> float:
    if not text:
        return 0.0
    nonspace = [char for char in text if not char.isspace()]
    if not nonspace:
        return 0.0
    printable_ratio = sum(char.isprintable() for char in nonspace) / len(nonspace)
    useful_ratio = sum(
        char.isalnum() or "\u3400" <= char <= "\u9fff" or unicodedata.category(char).startswith("P")
        for char in nonspace
    ) / len(nonspace)
    return round(100 * (0.65 * printable_ratio + 0.35 * useful_ratio), 2)


def _internal_duplicate_score(text: str) -> float:
    units = [unit.strip().lower() for unit in re.split(r"[\n。！？!?；;]+", text) if unit.strip()]
    if len(units) <= 1:
        return 100.0
    return round(100 * len(set(units)) / len(units), 2)


def _source_relevance_score(record: Mapping[str, Any], source_content: str) -> float:
    """估算结果与来源文本的词元覆盖率。

    这是无外部模型依赖、可重复的首版评分。没有来源文本（例如人工新增结果）
    时不扣分；存在来源时，以结果中的有效词元被来源覆盖的比例计分。
    """

    source = normalize_text(source_content)
    if not source:
        return 100.0
    candidate = normalize_text(
        "\n".join(
            str(record.get(field) or "") for field in ("instruction", "input", "output")
        )
    )

    def semantic_tokens(text: str) -> set[str]:
        return {
            token.lower()
            for token in _TOKEN_PATTERN.findall(text)
            if token.isalnum() or "\u3400" <= token <= "\u9fff"
        }

    source_tokens = semantic_tokens(source)
    candidate_tokens = semantic_tokens(candidate)
    if not candidate_tokens:
        return 0.0
    if not source_tokens:
        return 0.0
    return round(100 * len(candidate_tokens & source_tokens) / len(candidate_tokens), 2)


def score_quality(
    record: Mapping[str, Any],
    *,
    min_output_length: int = 20,
    source_content: str = "",
    known_fingerprints: Iterable[str] = (),
    threshold: float = 60.0,
) -> QualityScore:
    """按完整性、长度、可读性、来源相关性和重复度计算质量分。"""

    if min_output_length <= 0:
        raise ValueError("min_output_length must be greater than 0")
    if not 0 <= threshold <= 100:
        raise ValueError("threshold must be in [0, 100]")

    instruction = normalize_text(str(record.get("instruction") or ""))
    input_text = normalize_text(str(record.get("input") or ""))
    output = normalize_text(str(record.get("output") or ""))
    flags: list[str] = []

    completeness = 100.0
    if not instruction:
        completeness -= 50
        flags.append("missing_instruction")
    if not output:
        completeness -= 50
        flags.append("missing_output")

    output_length = len(output)
    length_score = round(min(100.0, output_length / min_output_length * 100), 2)
    if output_length < min_output_length:
        flags.append("output_too_short")

    readability = _readability_score("\n".join((instruction, input_text, output)))
    if readability < 70:
        flags.append("low_readability")

    relevance = _source_relevance_score(record, source_content)
    if source_content and relevance < 30:
        flags.append("low_source_relevance")

    fingerprint = record_fingerprint(record)
    known = set(known_fingerprints)
    duplicate = 0.0 if fingerprint in known else _internal_duplicate_score(output)
    if duplicate == 0:
        flags.append("duplicate_record")
    elif duplicate < 70:
        flags.append("repetitive_output")

    overall = round(
        completeness * 0.35
        + length_score * 0.20
        + readability * 0.20
        + relevance * 0.15
        + duplicate * 0.10,
        2,
    )
    hard_valid = bool(instruction and output)
    return QualityScore(
        overall=overall,
        completeness=completeness,
        length=length_score,
        readability=readability,
        relevance=relevance,
        duplicate=duplicate,
        is_valid=hard_valid and overall >= threshold,
        flags=tuple(flags),
        fingerprint=fingerprint,
    )


def _cosine_similarity(left: Sequence[float], right: Sequence[float]) -> float:
    if not left or not right or len(left) != len(right):
        return 0.0
    dot = math.fsum(a * b for a, b in zip(left, right))
    norm_left = math.sqrt(math.fsum(a * a for a in left))
    norm_right = math.sqrt(math.fsum(b * b for b in right))
    if not norm_left or not norm_right:
        return 0.0
    return dot / (norm_left * norm_right)


def semantic_quality_scores(
    record: Mapping[str, Any],
    *,
    source_content: str = "",
    embed_model: Any = None,
) -> dict[str, Any] | None:
    """用本地嵌入向量计算语义相关性（0-100）。

    返回 ``question_answer``（问题↔答案）、``answer_source``（答案↔来源，
    无来源时缺省）与 ``overall``；嵌入模型不可用时返回 None 降级，不阻断流程。
    """

    try:
        if embed_model is None:
            from .embedding import semantic_embedding_model

            embed_model = semantic_embedding_model()
        if embed_model is None:
            return None

        question = normalize_text(
            " ".join(
                str(record.get(field) or "")
                for field in ("instruction", "input")
            )
        )
        answer = normalize_text(
            str(record.get("output") or "") or str(record.get("chosen") or "")
        )
        source = normalize_text(source_content)
        texts = [text for text in {question, answer, source} if text]
        if not texts:
            return None
        vectors = {text: embed_model.get_text_embedding(text) for text in texts}
    except Exception:
        return None

    scores: dict[str, Any] = {}
    if question and answer:
        scores["question_answer"] = round(
            100 * max(0.0, _cosine_similarity(vectors[question], vectors[answer])), 2
        )
    if answer and source:
        scores["answer_source"] = round(
            100 * max(0.0, _cosine_similarity(vectors[answer], vectors[source])), 2
        )
    if not scores:
        return None
    scores["overall"] = round(sum(scores.values()) / len(scores), 2)
    return scores


def composite_overall(
    *,
    rule: float | None,
    semantic: float | None = None,
    judge: float | None = None,
) -> float:
    """三层加权组合：规则 35% + 语义 20% + 评审 45%，缺失层自动重归一。"""

    if rule is None:
        rule = 0.0
    if judge is not None and semantic is not None:
        overall = rule * 0.35 + semantic * 0.20 + judge * 0.45
    elif semantic is not None:
        overall = rule * 0.60 + semantic * 0.40
    elif judge is not None:
        overall = rule * 0.55 + judge * 0.45
    else:
        overall = rule
    return round(max(0.0, min(100.0, overall)), 2)