Create-knowledge-find-decode/decode_content/scoping.py

"""Scope candidate generation, normalization, and linking."""
from __future__ import annotations

import json
import re
from typing import Any, Protocol

from core.llm import chat_json as default_chat_json
from core.prompts import load_prompt
from core.text_limits import DIRECTIVE_CONTEXT_MAX_CHARS, clip_text
from decode_content.contracts import REUSE_THRESHOLD, SCOPE_TYPE_CN, SkillContract, load_contract
from decode_content.gates import ChatJsonFn


NORMALIZE = load_prompt("normalize_scope")
SRC2CN = SCOPE_TYPE_CN
STRIP_VERBS = ("寻找", "定位", "推导", "核验", "提取", "挖掘", "捕捉", "识别", "梳理", "归纳", "判断", "验证", "确认", "复盘")
SCOPE_CONN = re.compile(r"\s*[和与、，,/／&]\s*")


class ScopeLinker(Protocol):
    def link(self, value: str, *, source_type: str, top_k: int = 3) -> list[dict[str, Any]]:
        ...


def slim_knowledges(knowledges: list[dict[str, Any]]) -> list[dict[str, Any]]:
    slim: list[dict[str, Any]] = []
    for knowledge in knowledges:
        item = {"id": knowledge.get("id"), "type": knowledge.get("type"), "title": knowledge.get("title")}
        if knowledge.get("type") == "how":
            item["steps"] = [
                {
                    "id": step.get("id"),
                    "input": step.get("input"),
                    "directive": clip_text(step.get("directive") or "", DIRECTIVE_CONTEXT_MAX_CHARS),
                    "output": step.get("output"),
                }
                for step in knowledge.get("steps", [])
            ]
        else:
            item["内容"] = {
                key: knowledge.get(key)
                for key in ("概要", "维度拆分规则", "body", "阐述")
                if knowledge.get(key)
            }
        slim.append(item)
    return slim


def scope_candidates(
    knowledges: list[dict[str, Any]],
    *,
    contract: SkillContract | None = None,
    chat_json_fn: ChatJsonFn = default_chat_json,
) -> list[dict[str, Any]]:
    contract = contract or load_contract()
    user = (
        "给下面每颗知识标作用域候选（how 逐步：每个 step 一组；what/why 颗级：整颗一组）。\n"
        "每类平铺列出所有相关值（同类可多条、全部对等、不选主次）；先脑内摊正交、收敛近义；意图通常 1 个。\n"
        "只输出 JSON：{\"scopes\":[{\"knowledge_id\":\"k1\",\"step_id\":\"s1\",\"items\":["
        "{\"scope_type\":\"substance\",\"value\":\"实质值1\"},{\"scope_type\":\"substance\",\"value\":\"实质值2\"},"
        "{\"scope_type\":\"form\",\"value\":\"形式值1\"},{\"scope_type\":\"intent\",\"value\":\"意图值\"}]},"
        "{\"knowledge_id\":\"k2\",\"step_id\":null,\"items\":[...]}]}\n\n"
        + json.dumps(slim_knowledges(knowledges), ensure_ascii=False)
    )
    return chat_json_fn(contract.phase2, user, timeout=120).get("scopes") or []


def strip_verb_tail(value: str) -> str:
    if not value or len(value) < 3:
        return value
    original = value
    for verb in STRIP_VERBS:
        if value.startswith(verb) and len(value) - len(verb) >= 2:
            value = value[len(verb):]
            break
    for verb in STRIP_VERBS:
        if value.endswith(verb) and len(value) - len(verb) >= 2:
            value = value[: -len(verb)]
            break
    return value or original


def nounify_scopes(
    scopes: list[dict[str, Any]],
    *,
    chat_json_fn: ChatJsonFn = default_chat_json,
) -> list[dict[str, Any]]:
    vals = sorted(
        {
            item["value"]
            for scope in scopes
            for item in (scope.get("items") or [])
            if item.get("value") and item.get("scope_type") != "intent"
        }
    )
    mapping: dict[str, str] = {}
    if vals:
        try:
            mapping = chat_json_fn(NORMALIZE, json.dumps(vals, ensure_ascii=False), timeout=90).get("映射") or {}
        except Exception:
            mapping = {}
    for scope in scopes:
        for item in scope.get("items") or []:
            value = item.get("value")
            if not value or item.get("scope_type") == "intent":
                continue
            item["value"] = strip_verb_tail(mapping.get(value) or value)
    return scopes


def link_scope(linker: ScopeLinker, scope_type: str, value: str) -> dict[str, Any]:
    try:
        hits = linker.link(value, source_type=SRC2CN.get(scope_type, scope_type), top_k=3)
    except Exception:
        hits = []
    top = hits[0] if hits else {}
    score = float(top.get("score", 0.0) or 0.0)
    reuse = score >= REUSE_THRESHOLD and top.get("name")
    return {
        "scope_type": scope_type,
        "value": top["name"] if reuse else value,
        "candidate": value,
        "link": "复用" if reuse else "新建",
        "score": round(score, 4),
        "top": [
            {"name": hit["name"], "score": hit["score"], "path": hit.get("path", "")}
            for hit in hits
            if hit.get("name") is not None
        ],
    }


def split_scope_items(items: list[dict[str, Any]] | None) -> list[dict[str, str]]:
    out: list[dict[str, str]] = []
    for item in items or []:
        scope_type = item.get("scope_type")
        value = item.get("value")
        if not (scope_type and value):
            continue
        parts = [part.strip() for part in SCOPE_CONN.split(value) if part.strip()]
        for part in (parts if len(parts) > 1 else [value]):
            out.append({"scope_type": scope_type, "value": part})
    return out


def apply_scopes(
    knowledges: list[dict[str, Any]],
    scopes: list[dict[str, Any]],
    linker: ScopeLinker | None = None,
) -> None:
    by_k = {knowledge.get("id"): knowledge for knowledge in knowledges}
    for scope in scopes:
        knowledge = by_k.get(scope.get("knowledge_id"))
        if not knowledge:
            continue
        if linker is None:
            linked = [
                {"scope_type": item["scope_type"], "value": item["value"], "candidate": item["value"], "link": "候选"}
                for item in split_scope_items(scope.get("items"))
            ]
        else:
            linked = [
                link_scope(linker, item["scope_type"], item["value"])
                for item in split_scope_items(scope.get("items"))
            ]
        if knowledge.get("type") == "how" and scope.get("step_id"):
            for step in knowledge.get("steps", []):
                if step.get("id") == scope["step_id"]:
                    step["作用域"] = linked
        else:
            knowledge["作用域"] = (knowledge.get("作用域") or []) + linked