222 lines
8.1 KiB
Python
222 lines
8.1 KiB
Python
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import shutil
|
||
from collections import defaultdict
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
|
||
PART_CATEGORY_RULES: list[tuple[str, str]] = [
|
||
("齿轮", "gear"),
|
||
("轴", "shaft"),
|
||
("箱体", "housing"),
|
||
("端盖", "cover"),
|
||
("垫片", "shim"),
|
||
("挡油环", "oil_ring"),
|
||
("油标", "oil_gauge"),
|
||
("通气器", "vent"),
|
||
("窥视孔盖", "inspection_cover"),
|
||
("套筒", "sleeve"),
|
||
("密封油圈", "seal"),
|
||
("毛毡圈", "seal"),
|
||
]
|
||
|
||
ASCII_NAME_ALIASES: list[tuple[str, str]] = [
|
||
("二级减速器", "two-stage-reducer"),
|
||
("上箱体", "upper-housing"),
|
||
("下箱体", "lower-housing"),
|
||
("齿轮1+高速轴", "gear-1-high-speed-shaft"),
|
||
("齿轮120×1.5", "gear-120x1-5"),
|
||
("齿轮82×2.5", "gear-82x2-5"),
|
||
("齿轮24×2.5", "gear-24x2-5"),
|
||
("轴2中间", "shaft-2-intermediate"),
|
||
("轴3低速", "shaft-3-low-speed"),
|
||
("套筒φ30×12", "sleeve-phi30x12"),
|
||
("油标", "oil-gauge"),
|
||
("通气器", "vent"),
|
||
("窥视孔盖", "inspection-cover"),
|
||
("密封油圈", "seal-ring"),
|
||
("毛毡圈35", "felt-ring-35"),
|
||
("毛毡圈50", "felt-ring-50"),
|
||
("挡油环φ30", "oil-slinger-phi30"),
|
||
("挡油环φ40", "oil-slinger-phi40"),
|
||
("挡油环φ55", "oil-slinger-phi55"),
|
||
("调整垫片100", "adjusting-shim-100"),
|
||
("调整垫片62", "adjusting-shim-62"),
|
||
("调整垫片80", "adjusting-shim-80"),
|
||
("端盖中间轴", "cover-intermediate-shaft"),
|
||
("端盖低速轴开口", "cover-low-speed-open"),
|
||
("端盖低速轴闭口", "cover-low-speed-closed"),
|
||
("端盖高速轴开口", "cover-high-speed-open"),
|
||
("端盖高速轴闭口", "cover-high-speed-closed"),
|
||
]
|
||
|
||
|
||
def main() -> None:
|
||
parser = argparse.ArgumentParser(description="Import extracted SolidWorks standard-answer records into agent4.")
|
||
parser.add_argument("--source", default=r"D:\Desktop\提取记录")
|
||
parser.add_argument("--target", default=str(Path(__file__).resolve().parents[1] / "data" / "standards"))
|
||
args = parser.parse_args()
|
||
|
||
source = Path(args.source)
|
||
target = Path(args.target)
|
||
if not source.exists():
|
||
raise FileNotFoundError(source)
|
||
|
||
target.mkdir(parents=True, exist_ok=True)
|
||
groups = group_records(source)
|
||
manifest = {
|
||
"schema_version": "agent4.standard_manifest.v1",
|
||
"source_dir": str(source),
|
||
"entries": [],
|
||
}
|
||
|
||
for base_name, files in sorted(groups.items(), key=lambda item: item[0]):
|
||
kind = "assembly" if "knowledge_skillflow" in base_name else "part"
|
||
category = "reducer" if kind == "assembly" else categorize(base_name)
|
||
entry_dir = target / ("assemblies" if kind == "assembly" else "parts") / category / safe_name(base_name)
|
||
entry_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
copied: dict[str, str] = {}
|
||
for role, path in files.items():
|
||
destination = entry_dir / path.name
|
||
shutil.copy2(path, destination)
|
||
copied[role] = relative(destination, target)
|
||
|
||
validation = read_validation(entry_dir / Path(copied["validation"]).name) if "validation" in copied else {}
|
||
plan_payload = read_json(entry_dir / Path(copied.get("modeling_plan") or copied.get("knowledge_skillflow") or "").name)
|
||
steps = extract_steps(plan_payload)
|
||
skills_used = sorted({skill for skill in (read_skill(step) for step in steps) if skill})
|
||
title = read_title(base_name, plan_payload)
|
||
|
||
manifest["entries"].append(
|
||
{
|
||
"standard_id": standard_id(kind, category, base_name),
|
||
"title": title,
|
||
"kind": kind,
|
||
"category": category,
|
||
"source_base_name": base_name,
|
||
"modeling_plan_path": copied.get("modeling_plan", ""),
|
||
"skill_flow_path": copied.get("skill_flow", ""),
|
||
"validation_path": copied.get("validation", ""),
|
||
"knowledge_skillflow_path": copied.get("knowledge_skillflow", ""),
|
||
"validated": bool(validation.get("Ok") is True or validation.get("ok") is True),
|
||
"total_steps": int(validation.get("TotalSteps") or validation.get("total_steps") or len(steps)),
|
||
"skills_used": validation.get("SkillsUsed") or validation.get("skills_used") or skills_used,
|
||
"errors": validation.get("Errors") or validation.get("errors") or [],
|
||
"warnings": validation.get("Warnings") or validation.get("warnings") or [],
|
||
"teaching_enabled": kind == "part" and bool(copied.get("modeling_plan")) and not validation.get("Errors"),
|
||
"testing_enabled": bool(copied.get("modeling_plan") or copied.get("knowledge_skillflow")),
|
||
}
|
||
)
|
||
|
||
manifest_path = target / "manifest.json"
|
||
manifest_path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
print(f"Imported {len(manifest['entries'])} standard records into {target}")
|
||
print(f"Manifest: {manifest_path}")
|
||
|
||
|
||
def group_records(source: Path) -> dict[str, dict[str, Path]]:
|
||
groups: dict[str, dict[str, Path]] = defaultdict(dict)
|
||
for path in source.glob("*.json"):
|
||
name = path.name
|
||
if name.endswith("_modeling_plan.json"):
|
||
groups[name[: -len("_modeling_plan.json")]]["modeling_plan"] = path
|
||
elif name.endswith("_skill_flow_validation.json"):
|
||
groups[name[: -len("_skill_flow_validation.json")]]["validation"] = path
|
||
elif name.endswith("_skill_flow.json"):
|
||
groups[name[: -len("_skill_flow.json")]]["skill_flow"] = path
|
||
elif name.endswith("_knowledge_skillflow.json"):
|
||
groups[name[: -len("_knowledge_skillflow.json")]]["knowledge_skillflow"] = path
|
||
return groups
|
||
|
||
|
||
def categorize(name: str) -> str:
|
||
for token, category in PART_CATEGORY_RULES:
|
||
if token in name:
|
||
return category
|
||
return "misc"
|
||
|
||
|
||
def safe_name(name: str) -> str:
|
||
result = []
|
||
for char in name:
|
||
if char.isalnum() or char in {"×", "φ", "+", "-", "_"}:
|
||
result.append(char)
|
||
else:
|
||
result.append("_")
|
||
return "".join(result).strip("_") or "standard"
|
||
|
||
|
||
def standard_id(kind: str, category: str, base_name: str) -> str:
|
||
return f"{kind}-{category}-{ascii_slug(base_name)}"
|
||
|
||
|
||
def ascii_slug(name: str) -> str:
|
||
for token, alias in ASCII_NAME_ALIASES:
|
||
if token == name:
|
||
return alias
|
||
chars: list[str] = []
|
||
for char in name.lower():
|
||
if char.isascii() and char.isalnum():
|
||
chars.append(char)
|
||
elif char in {"×", "x"}:
|
||
chars.append("x")
|
||
elif char in {"φ"}:
|
||
chars.extend(["p", "h", "i"])
|
||
else:
|
||
chars.append("-")
|
||
slug = "-".join(part for part in "".join(chars).split("-") if part)
|
||
return slug or safe_name(name)
|
||
|
||
|
||
def relative(path: Path, root: Path) -> str:
|
||
return str(path.relative_to(root)).replace("\\", "/")
|
||
|
||
|
||
def read_json(path: Path) -> Any:
|
||
if not path.exists() or path.is_dir():
|
||
return {}
|
||
try:
|
||
return json.loads(path.read_text(encoding="utf-8-sig"))
|
||
except Exception:
|
||
return {}
|
||
|
||
|
||
def read_validation(path: Path) -> dict[str, Any]:
|
||
payload = read_json(path)
|
||
return payload if isinstance(payload, dict) else {}
|
||
|
||
|
||
def extract_steps(payload: Any) -> list[Any]:
|
||
if isinstance(payload, list):
|
||
return payload
|
||
if not isinstance(payload, dict):
|
||
return []
|
||
if isinstance(payload.get("steps"), list):
|
||
return payload["steps"]
|
||
modeling_plan = payload.get("modelingPlan") or payload.get("modeling_plan")
|
||
if isinstance(modeling_plan, dict) and isinstance(modeling_plan.get("steps"), list):
|
||
return modeling_plan["steps"]
|
||
return []
|
||
|
||
|
||
def read_skill(step: Any) -> str:
|
||
if not isinstance(step, dict):
|
||
return ""
|
||
return str(step.get("skill") or step.get("skillName") or step.get("skill_name") or "")
|
||
|
||
|
||
def read_title(base_name: str, payload: Any) -> str:
|
||
if isinstance(payload, dict):
|
||
for key in ("partName", "title", "workflowId"):
|
||
if payload.get(key):
|
||
return str(payload[key])
|
||
return base_name
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|