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()