Blame for core/advanced_method_registry.py

# -*- coding: utf-8 -*-
"""Deterministic two-design manual selector and capability preflight for SmartPaper.

Production research design is intentionally limited to two supported paths:
qualitative research and survey-based quantitative research. New tasks default to
qualitative research, and the user may explicitly adjust the design before task
creation. Title/Keywords edits may reset an unlaunched design to the configured
default through deterministic reconciliation; materials never select a design.

This module never selects providers/models and never rewrites Title/Keywords.
"""
from __future__ import annotations

import os
import re
from typing import Any, Iterable

METHOD_RESOLVER_VERSION = "smartpaper-manual-method-selector-v2-single-authority"
METHOD_CAPABILITY_REGISTRY_VERSION = "smartpaper-method-capability-registry-v4"
DEFAULT_METHOD_FAMILY = "qualitative"

# Reader-visible / task-owning method families.  Order is intentional:
# qualitative is the product default and appears first in the selector.
METHOD_FAMILIES = (
    "qualitative",
    "survey_path",
)

_METHOD_CAPABILITIES: dict[str, dict[str, Any]] = {
    "qualitative": {
        "legacy_policy_lane": "asq",
        "capability_status": "production_supported",
        "evidence_timing": "pre_literature",
        "engine_family": "qualitative_engine",
        "template_profile": "qualitative_materials",
        "material_profile": "qualitative_materials",
        "research_material_extensions": ("txt",),
        "material_routing": {
            "structured": "data",
            "text": "interview",
            "document_reference": "reference",
            "document_research": "interview",
            # P0-R7-B: explicit per-file user roles remain registry-owned.
            "user_reference": "reference",
            "user_research": "interview",
        },
    },
    "survey_path": {
        "legacy_policy_lane": "amj",
        "capability_status": "production_supported",
        "evidence_timing": "pre_literature",
        "engine_family": "quantitative_engine",
        "template_profile": "survey_quantitative_data",
        "material_profile": "survey_quantitative_data",
        "research_material_extensions": ("csv",),
        "material_routing": {
            "structured": "data",
            "text": "reference",
            "document_reference": "reference",
            "document_research": "reference",
            # P0-R7-B: user research material is empirical evidence for survey work.
            "user_reference": "reference",
            "user_research": "data",
        },
    },
}

# Specialized methods are deliberately NOT method families in U1-R2C.  They
# are scope cues only.  A hit triggers a notice and a qualitative fallback.
_SPECIALIZED_METHOD_PATTERNS: tuple[tuple[str, tuple[str, ...]], ...] = (
    (
        "fsqca",
        (
            r"\bfsqca\b", r"\bfuzzy[- ]set\s+qca\b", r"\bfuzzy[- ]set\s+qualitative\s+comparative\s+analysis\b",
            r"\bconfigurational\s+(?:analysis|paths?|approach)\b", r"\btruth\s+table\b",
            r"模糊集定性比较分析", r"定性比较分析", r"组态(?:分析|路径)",
            r"нечетк\w*.{0,18}сравнительн\w*.{0,18}анализ", r"конфигурационн\w*.{0,12}анализ",
            r"бұлыңғыр.{0,16}сапалық.{0,16}салыстырмалы.{0,16}талдау", r"конфигурациялық.{0,12}талдау",
        ),
    ),
    (
        "meta_analysis",
        (
            r"\bmeta[- ]analysis\b", r"\bmeta[- ]analytic\b", r"\bmeta analysis\b",
            r"元分析", r"荟萃分析", r"мета[- ]?анализ", r"метааналитическ", r"мета[- ]?талдау", r"метааналитикалық",
        ),
    ),
    (
        "panel_econometrics",
        (
            r"\bpanel\s+data\b", r"\bfirm[- ]year\b", r"\bentity\s+fixed\s+effects?\b", r"\bfixed\s+effects?\b",
            r"\brandom\s+effects?\b", r"\bpanel\s+econometrics?\b", r"\btwo[- ]way\s+fixed\s+effects?\b",
            r"面板数据", r"固定效应", r"双向固定效应", r"公司[-—– ]?年度",
            r"панельн\w*.{0,8}данн", r"фиксированн\w*.{0,8}эффект", r"панельдік.{0,8}дерек", r"тұрақты.{0,8}әсер",
        ),
    ),
    (
        "bibliometric_review",
        (
            r"\bbibliometric\b", r"\bscience\s+mapping\b", r"\bco[- ]citation\b", r"\bco[- ]word\b",
            r"\bintegrative\s+(?:theory\s+)?review\b", r"\btheory\s+review\b", r"\bsystematic\s+literature\s+review\b",
            r"文献计量", r"科学知识图谱", r"共被引", r"共词", r"整合性(?:理论)?综述", r"系统性文献综述", r"文献综述",
            r"библиометр", r"научн\w*.{0,8}картирован", r"коцитирован", r"интегративн\w*.{0,8}обзор", r"систематическ\w*.{0,8}обзор",
            r"библиометр", r"ғылыми.{0,8}карталау", r"бірлескен.{0,8}дәйексөз", r"интегративті.{0,8}шолу", r"жүйелі.{0,8}әдебиет.{0,8}шолу",
        ),
    ),
)

_SUPPORTED_METHOD_PATTERNS: tuple[tuple[str, tuple[str, ...]], ...] = (
    (
        "qualitative",
        (
            r"\bqualitative\s+(?:study|research|analysis)\b", r"\bgioia\b", r"\bgrounded\s+theory\b",
            r"\bcase\s+study\b", r"\bcomparative\s+case\b", r"\bprocess\s+study\b", r"\bethnograph",
            r"\bsemi[- ]structured\s+interviews?\b", r"\binterviews?\b",
            r"定性研究", r"质性研究", r"扎根理论", r"案例研究", r"个案研究", r"访谈", r"民族志",
            r"качественн\w*.{0,8}(?:исследован|анализ)", r"обоснованн\w*.{0,8}теори", r"интервью", r"кейс[- ]?исследован",
            r"сапалық.{0,8}(?:зерттеу|талдау)", r"негізделген.{0,8}теори", r"сұхбат", r"кейс.{0,8}зерттеу",
        ),
    ),
    (
        "survey_path",
        (
            r"\bsurvey\b", r"\bquestionnaire\b", r"\bstructural\s+equation\s+model", r"\bsem\b", r"\bpls[- ]sem\b",
            r"\bregression\b", r"\bmediating\s+role\b", r"\bmoderating\s+role\b", r"\bmediation\b", r"\bmoderation\b",
            r"问卷", r"结构方程", r"回归", r"中介(?:作用|效应|变量)?", r"调节(?:作用|效应|变量)?",
            r"опрос", r"анкет", r"структурн\w*.{0,10}уравнен", r"регресс", r"медиирующ", r"модерирующ",
            r"сауалнама", r"құрылымдық.{0,10}теңдеу", r"регресс", r"медиаторлық", r"модераторлық",
        ),
    ),
)


def _normalized_text(value: Any) -> str:
    return re.sub(r"\s+", " ", str(value or "")).strip()


def _keyword_items(value: Any) -> list[str]:
    if isinstance(value, (list, tuple)):
        raw = list(value)
    else:
        raw = re.split(r"[;\n|]+", str(value or ""))
        if len(raw) == 1:
            raw = re.split(r"[,]", str(value or ""))
    return [_normalized_text(item) for item in raw if _normalized_text(item)]


def _material_descriptors(materials: Iterable[Any] | None) -> list[dict[str, str]]:
    out: list[dict[str, str]] = []
    for item in materials or []:
        if isinstance(item, dict):
            name = _normalized_text(item.get("name") or item.get("filename"))
            mime = _normalized_text(item.get("mime_type") or item.get("type"))
        else:
            name = _normalized_text(getattr(item, "name", ""))
            mime = _normalized_text(getattr(item, "type", ""))
        if not name and not mime:
            continue
        out.append({
            "name": name,
            "mime_type": mime,
            "extension": os.path.splitext(name)[1].lower().lstrip("."),
        })
    return out


def default_method_family() -> str:
    """Return the single configured default research-design family."""
    if DEFAULT_METHOD_FAMILY not in METHOD_FAMILIES:
        raise RuntimeError("default_method_family_not_registered")
    return DEFAULT_METHOD_FAMILY


def normalize_method_family(value: Any, *, fallback_to_default: bool = False) -> str:
    """Normalize a reader-visible method family without inventing new values."""
    family = str(value or "").strip().casefold()
    if family in METHOD_FAMILIES:
        return family
    return default_method_family() if fallback_to_default else ""


def method_family_for_legacy_lane(selected_lane: Any) -> str:
    """Project a legacy AMJ/ASQ lane back to its registered method family."""
    lane = str(selected_lane or "").strip().casefold()
    matches = [
        family for family in METHOD_FAMILIES
        if str((_METHOD_CAPABILITIES.get(family) or {}).get("legacy_policy_lane") or "").strip().casefold() == lane
    ]
    return matches[0] if len(matches) == 1 else ""


def reconcile_method_authority(
    *,
    current_authority: Any,
    current_ui_family: Any,
    previous_ui_family: Any,
    current_intent_signature: Any,
    previous_intent_signature: Any,
) -> dict[str, Any]:
    """Deterministically reconcile the one mutable new-task design authority.

    Explicit Research Design changes win over a simultaneous Title/Keywords
    change. Otherwise, a genuine intent change resets to the configured default.
    Unrelated reruns preserve the existing authority.
    """
    default_family = default_method_family()
    authority = normalize_method_family(current_authority) or default_family
    current_ui = normalize_method_family(current_ui_family) or authority
    previous_ui = normalize_method_family(previous_ui_family) or current_ui
    current_sig = str(current_intent_signature or "").strip()
    previous_sig = str(previous_intent_signature or "").strip()

    ui_changed = current_ui != previous_ui
    intent_changed = bool(previous_sig) and current_sig != previous_sig

    if ui_changed:
        chosen = current_ui
        reason = "user_selected_method_family"
    elif intent_changed:
        chosen = default_family
        reason = "intent_changed_reset_to_default"
    else:
        chosen = authority
        reason = "preserve_existing_authority"

    return {
        "method_family": chosen,
        "reason": reason,
        "ui_changed": ui_changed,
        "intent_changed": intent_changed,
        "default_method_family": default_family,
    }


def method_capability(method_family: str) -> dict[str, Any]:
    """Return a copy of versioned capability metadata for one supported family."""
    family = str(method_family or "").strip().casefold()
    row = dict(_METHOD_CAPABILITIES.get(family) or {})
    if not row:
        return {
            "registry_version": METHOD_CAPABILITY_REGISTRY_VERSION,
            "method_family": "undetermined",
            "legacy_policy_lane": None,
            "capability_status": "ambiguous",
            "evidence_timing": "undetermined",
            "engine_family": "undetermined",
            "template_profile": None,
            "material_profile": None,
            "material_routing": {},
        }
    row.update({
        "registry_version": METHOD_CAPABILITY_REGISTRY_VERSION,
        "method_family": family,
    })
    return row


def _pattern_hits(text: str, pattern_rows: tuple[tuple[str, tuple[str, ...]], ...]) -> list[dict[str, Any]]:
    hits: list[dict[str, Any]] = []
    for family, patterns in pattern_rows:
        matched = [pattern for pattern in patterns if re.search(pattern, text, flags=re.I | re.UNICODE)]
        if matched:
            hits.append({"method_family": family, "source": "title_keywords", "match_count": len(matched)})
    return hits


def resolve_method_family(
    title: Any,
    keywords: Any,
    *,
    materials: Iterable[Any] | None = None,
) -> dict[str, Any]:
    """Return the fixed default design for a new task.

    Automatic research-design detection is intentionally disabled. Title,
    Keywords, and Research Materials are accepted for API compatibility only and
    never change the default. New tasks start from qualitative research unless the
    user explicitly selects survey-based quantitative research in preflight.
    """
    _ = title, keywords
    descriptors = _material_descriptors(materials)
    family = default_method_family()
    capability = method_capability(family)
    return {
        "resolver_version": METHOD_RESOLVER_VERSION,
        "shadow_only": False,
        "method_family": family,
        "confidence": "DEFAULT_CONFIGURED",
        "reason_codes": ["manual_design_control", "default_method_family"],
        "title_keyword_hits": [],
        "specialized_method_detected": False,
        "specialized_method_families": [],
        "specialized_method_hits": [],
        "material_hint_families": [],
        "material_count": len(descriptors),
        "derived_policy_lane": capability.get("legacy_policy_lane"),
        "capability": capability,
    }

def compare_shadow_to_legacy_lane(resolution: dict[str, Any], selected_lane: Any) -> dict[str, Any]:
    """Compare inferred design with a legacy AMJ/ASQ lane."""
    selected = str(selected_lane or "").strip().casefold()
    derived = str((resolution or {}).get("derived_policy_lane") or "").strip().casefold()
    return {
        "selected_legacy_lane": selected or None,
        "derived_policy_lane": derived or None,
        "lane_match": bool(selected and derived and selected == derived),
        "comparable": bool(selected and derived),
    }


def resolve_generation_preflight(
    title: Any,
    keywords: Any,
    *,
    materials: Iterable[Any] | None = None,
    confirmed_method_family: Any = None,
) -> dict[str, Any]:
    """Resolve the two supported designs before any task/quota creation.

    The configured default is used unless a supported method family is supplied
    by the single UI authority. This function validates/projections capability;
    it does not infer a design from Title, Keywords, or Research Materials.
    """
    resolution = resolve_method_family(title, keywords, materials=materials)
    family = default_method_family()

    confirmed = normalize_method_family(confirmed_method_family)
    if confirmed in METHOD_FAMILIES:
        family = confirmed
        capability = method_capability(family)
        resolution = dict(resolution)
        resolution.update({
            "method_family": family,
            "confidence": "USER_SELECTION",
            "derived_policy_lane": capability.get("legacy_policy_lane"),
            "capability": capability,
            "reason_codes": ["manual_design_control", "user_selected_method_family"],
        })

    capability = method_capability(family)
    lane = str(capability.get("legacy_policy_lane") or "").strip().casefold() or None
    status = str(capability.get("capability_status") or "").strip().casefold()
    engine_family = str(capability.get("engine_family") or "").strip()
    template_profile = str(capability.get("template_profile") or "").strip()
    material_profile = str(capability.get("material_profile") or "").strip()
    research_material_extensions = tuple(
        str(item or "").strip().casefold()
        for item in (capability.get("research_material_extensions") or ())
        if str(item or "").strip()
    )
    material_routing = dict(capability.get("material_routing") or {})
    routing_keys = {
        "structured", "text", "document_reference", "document_research",
        "user_reference", "user_research",
    }
    routing_values = {str(value or "").strip() for value in material_routing.values()}
    if (
        family not in METHOD_FAMILIES
        or not lane
        or status != "production_supported"
        or not engine_family
        or not template_profile
        or not material_profile
        or len(research_material_extensions) != 1
        or research_material_extensions[0] not in {"txt", "csv"}
        or set(material_routing) != routing_keys
        or not routing_values.issubset({"reference", "data", "interview"})
    ):
        return {
            "can_start_generation": False,
            "block_reason": "method_resolution_invalid",
            "resolution": resolution,
            "method_family": family or default_method_family(),
            "legacy_policy_lane": None,
            "capability": capability,
        }
    return {
        "can_start_generation": True,
        "block_reason": "",
        "resolution": resolution,
        "method_family": family,
        "legacy_policy_lane": lane,
        "capability": capability,
    }