185 lines
7.7 KiB
Python
185 lines
7.7 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from .config import settings
|
|
from .models import StandardAnswer, StandardKind
|
|
from .skillflow import make_standard_id, tags_for_text, load_standard_from_file
|
|
|
|
|
|
class KnowledgeRepository:
|
|
def __init__(self) -> None:
|
|
self._standards: dict[str, StandardAnswer] = {}
|
|
self.reload()
|
|
|
|
def reload(self) -> None:
|
|
standards: dict[str, StandardAnswer] = {}
|
|
for item in self._load_manifest_standards():
|
|
standards[item.standard_id] = item
|
|
for item in self._load_framework_refs():
|
|
standards.setdefault(item.standard_id, item)
|
|
for item in self._load_local_json_standards():
|
|
standards.setdefault(item.standard_id, item)
|
|
self._standards = standards
|
|
|
|
def list_standards(self, query: str = "", kind: StandardKind | None = None) -> list[StandardAnswer]:
|
|
items = list(self._standards.values())
|
|
if kind and kind != StandardKind.unknown:
|
|
items = [item for item in items if item.kind == kind]
|
|
if query.strip():
|
|
tokens = _tokens(query)
|
|
scored = [(score_standard(tokens, item), item) for item in items]
|
|
items = [item for score, item in sorted(scored, key=lambda pair: pair[0], reverse=True) if score > 0]
|
|
return items
|
|
|
|
def require(self, standard_id: str) -> StandardAnswer:
|
|
if standard_id not in self._standards:
|
|
raise KeyError(f"standard answer not found: {standard_id}")
|
|
return self._standards[standard_id]
|
|
|
|
def _load_local_json_standards(self) -> list[StandardAnswer]:
|
|
result: list[StandardAnswer] = []
|
|
for root in settings.standard_skillflow_dirs:
|
|
if not root.exists():
|
|
continue
|
|
for path in root.rglob("*.json"):
|
|
if not _looks_like_skillflow_name(path.name):
|
|
continue
|
|
standard = load_standard_from_file(path)
|
|
if standard:
|
|
result.append(standard)
|
|
return result
|
|
|
|
def _load_manifest_standards(self) -> list[StandardAnswer]:
|
|
manifest_path = settings.standard_skillflow_dirs[0] / "manifest.json"
|
|
if not manifest_path.exists():
|
|
return []
|
|
try:
|
|
manifest = json.loads(manifest_path.read_text(encoding="utf-8-sig"))
|
|
except Exception:
|
|
return []
|
|
entries = manifest.get("entries")
|
|
if not isinstance(entries, list):
|
|
return []
|
|
|
|
result: list[StandardAnswer] = []
|
|
root = manifest_path.parent
|
|
for entry in entries:
|
|
if not isinstance(entry, dict):
|
|
continue
|
|
standard = self._standard_from_manifest_entry(root, entry)
|
|
if standard:
|
|
result.append(standard)
|
|
return result
|
|
|
|
def _standard_from_manifest_entry(self, root: Path, entry: dict[str, Any]) -> StandardAnswer | None:
|
|
preferred = (
|
|
entry.get("modeling_plan_path")
|
|
or entry.get("knowledge_skillflow_path")
|
|
or entry.get("skill_flow_path")
|
|
or ""
|
|
)
|
|
payload_path = root / str(preferred)
|
|
loaded = load_standard_from_file(payload_path) if preferred else None
|
|
if loaded is None:
|
|
return StandardAnswer(
|
|
standard_id=str(entry.get("standard_id") or ""),
|
|
title=str(entry.get("title") or entry.get("source_base_name") or ""),
|
|
kind=_standard_kind(str(entry.get("kind") or "unknown")),
|
|
source_path=str(payload_path) if preferred else "",
|
|
source_type="manifest_ref",
|
|
tags=tags_for_text(f"{entry.get('title', '')} {entry.get('category', '')}"),
|
|
summary="Manifest standard answer reference without a readable local payload.",
|
|
steps=[],
|
|
raw={"manifest_entry": entry},
|
|
)
|
|
|
|
loaded.standard_id = str(entry.get("standard_id") or loaded.standard_id)
|
|
loaded.title = str(entry.get("title") or loaded.title)
|
|
loaded.kind = _standard_kind(str(entry.get("kind") or loaded.kind.value))
|
|
loaded.source_type = "manifest"
|
|
loaded.tags = sorted(set(loaded.tags + tags_for_text(f"{loaded.title} {entry.get('category', '')}")))
|
|
loaded.summary = loaded.summary or f"Manifest standard answer: {loaded.title}"
|
|
loaded.raw = {
|
|
"manifest_entry": entry,
|
|
"payload": loaded.raw,
|
|
}
|
|
return loaded
|
|
|
|
def _load_framework_refs(self) -> list[StandardAnswer]:
|
|
path = settings.knowledge_framework_path if settings.knowledge_framework_path.exists() else settings.legacy_knowledge_framework_path
|
|
if not path.exists():
|
|
return []
|
|
try:
|
|
framework = json.loads(path.read_text(encoding="utf-8-sig"))
|
|
except Exception:
|
|
return []
|
|
|
|
result: list[StandardAnswer] = []
|
|
impl = framework.get("function_implementation_library", {}) if isinstance(framework, dict) else {}
|
|
for section in impl.get("sections", []) if isinstance(impl.get("sections"), list) else []:
|
|
for flow in section.get("skillflows", []) if isinstance(section, dict) and isinstance(section.get("skillflows"), list) else []:
|
|
result.append(_standard_ref(flow, "assembly", path))
|
|
|
|
for part in framework.get("part_library", []) if isinstance(framework, dict) and isinstance(framework.get("part_library"), list) else []:
|
|
category = str(part.get("category_name") or part.get("category_code") or "")
|
|
for flow in part.get("modeling_skillflows", []) if isinstance(part, dict) and isinstance(part.get("modeling_skillflows"), list) else []:
|
|
result.append(_standard_ref(flow, "part", path, category))
|
|
return [item for item in result if item.standard_id]
|
|
|
|
|
|
def _standard_ref(flow: dict[str, Any], kind: str, framework_path: Path, category: str = "") -> StandardAnswer:
|
|
title = str(flow.get("skillflow_name") or flow.get("name") or "")
|
|
source_path = str(flow.get("path") or "")
|
|
text = f"{title} {source_path} {category}"
|
|
return StandardAnswer(
|
|
standard_id=make_standard_id(f"{kind}-{title or source_path}"),
|
|
title=title or source_path,
|
|
kind=_standard_kind(kind),
|
|
source_path=source_path,
|
|
source_type="framework_ref",
|
|
tags=tags_for_text(text),
|
|
summary=f"Knowledge-base standard answer reference from {framework_path.name}.",
|
|
steps=[],
|
|
raw={"framework_path": str(framework_path), "category": category, "ref": flow},
|
|
)
|
|
|
|
|
|
def score_standard(tokens: list[str], item: StandardAnswer) -> int:
|
|
text = " ".join(
|
|
[
|
|
item.standard_id,
|
|
item.title,
|
|
item.summary,
|
|
item.source_path,
|
|
" ".join(item.tags),
|
|
" ".join(step.skill + " " + step.name + " " + step.source_feature for step in item.steps[:80]),
|
|
]
|
|
).lower()
|
|
return sum(text.count(token) for token in tokens)
|
|
|
|
|
|
def _tokens(query: str) -> list[str]:
|
|
lowered = query.lower()
|
|
raw = [item for item in lowered.replace("_", " ").replace("-", " ").split() if len(item) >= 2]
|
|
for index in range(len(lowered) - 1):
|
|
pair = lowered[index : index + 2]
|
|
if any("\u4e00" <= ch <= "\u9fff" for ch in pair):
|
|
raw.append(pair)
|
|
seen: set[str] = set()
|
|
return [item for item in raw if item not in seen and not seen.add(item)]
|
|
|
|
|
|
def _looks_like_skillflow_name(name: str) -> bool:
|
|
lowered = name.lower()
|
|
return "skillflow" in lowered or "skill_flow" in lowered or "modeling_plan" in lowered or "knowledge" in lowered
|
|
|
|
|
|
def _standard_kind(value: str) -> StandardKind:
|
|
try:
|
|
return StandardKind(value)
|
|
except ValueError:
|
|
return StandardKind.unknown
|