Spaces:
Running
Running
| # -*- 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, | |
| } | |