Files
mech-ai/backend/agent4/skillflow.py
T
2026-07-17 17:45:56 +08:00

107 lines
3.7 KiB
Python

from __future__ import annotations
import json
from pathlib import Path
from typing import Any
from .models import SkillStep, StandardAnswer, StandardKind
def load_standard_from_file(path: Path) -> StandardAnswer | None:
try:
raw = json.loads(path.read_text(encoding="utf-8-sig"))
except Exception:
return None
steps = normalize_steps(raw)
if not steps:
return None
schema_version = raw.get("schemaVersion", "") if isinstance(raw, dict) else ""
kind = StandardKind.assembly if "assembly" in str(schema_version).lower() else StandardKind.part
title = path.stem
modeling_plan = raw.get("modelingPlan") if isinstance(raw, dict) and isinstance(raw.get("modelingPlan"), dict) else {}
approved_text = raw.get("approvedPlanText") if isinstance(raw, dict) else ""
summary = str(modeling_plan.get("summary") or approved_text or "")
return StandardAnswer(
standard_id=make_standard_id(path.stem),
title=title,
kind=kind,
source_path=str(path),
source_type="local_json",
tags=tags_for_text(f"{title} {summary}"),
summary=summary,
steps=steps,
raw=raw if isinstance(raw, dict) else {"steps": raw},
)
def normalize_steps(payload: Any) -> list[SkillStep]:
raw_steps: list[Any] = []
if isinstance(payload, list):
raw_steps = payload
elif isinstance(payload, dict):
if isinstance(payload.get("steps"), list):
raw_steps = payload["steps"]
elif isinstance(payload.get("modelingPlan"), dict) and isinstance(payload["modelingPlan"].get("steps"), list):
raw_steps = payload["modelingPlan"]["steps"]
elif isinstance(payload.get("modeling_plan"), dict) and isinstance(payload["modeling_plan"].get("steps"), list):
raw_steps = payload["modeling_plan"]["steps"]
result: list[SkillStep] = []
for index, item in enumerate(raw_steps, 1):
if not isinstance(item, dict):
continue
skill = str(item.get("skill") or item.get("skillName") or item.get("skill_name") or "")
if not skill:
continue
args = item.get("args") or item.get("arguments") or {}
if not isinstance(args, dict):
args = {}
result.append(
SkillStep(
id=str(item.get("id") or f"step-{index:03d}"),
order=int(item.get("step") or index),
skill=skill,
name=str(item.get("name") or item.get("source_feature") or skill),
description=str(item.get("description") or item.get("note") or ""),
arguments=args,
source_feature=str(item.get("source_feature") or item.get("sourceFeature") or item.get("name") or ""),
note=str(item.get("note") or item.get("description") or ""),
)
)
return result
def make_standard_id(text: str) -> str:
normalized = []
for ch in text.lower():
if ch.isalnum() or "\u4e00" <= ch <= "\u9fff":
normalized.append(ch)
else:
normalized.append("-")
return "-".join(part for part in "".join(normalized).split("-") if part)[:120] or "standard"
def tags_for_text(text: str) -> list[str]:
tags: list[str] = []
mapping = {
"齿轮": "gear",
"榻": "gear",
"轴": "shaft",
"箱": "housing",
"端盖": "cover",
"减速器": "reducer",
"装配": "assembly",
"assembly": "assembly",
"shell": "shell",
"抽壳": "shell",
"loft": "loft",
"放样": "loft",
}
lowered = text.lower()
for key, tag in mapping.items():
if key.lower() in lowered and tag not in tags:
tags.append(tag)
return tags