from __future__ import annotations import json import re import sqlite3 from pathlib import Path from typing import Any ROOT = Path(__file__).resolve().parent.parent ACADEMY_ROOT = ROOT / "data" / "academy" / "design_mastery" MANIFEST_PATH = ACADEMY_ROOT / "manifest.json" CATALOG_PATH = ACADEMY_ROOT / "curriculum" / "catalog.json" INDEX_PATH = ACADEMY_ROOT / "runtime" / "academy_index.sqlite3" _TOKEN_RE = re.compile(r"[a-z0-9]+") _STOP_WORDS = { "a", "about", "all", "also", "an", "and", "any", "are", "as", "at", "be", "build", "can", "could", "create", "design", "do", "for", "from", "give", "help", "how", "i", "in", "into", "is", "it", "make", "me", "my", "need", "of", "on", "or", "please", "proper", "put", "should", "that", "the", "this", "to", "want", "we", "what", "when", "which", "with", "would", "you" } class DesignAcademyService: """Read-only retrieval and status access for the installed Design Academy.""" def __init__(self) -> None: self.root = ACADEMY_ROOT self.manifest_path = MANIFEST_PATH self.catalog_path = CATALOG_PATH self.index_path = INDEX_PATH def _load_json(self, path: Path, default: Any) -> Any: try: return json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return default def _connect(self) -> sqlite3.Connection: connection = sqlite3.connect( f"file:{self.index_path}?mode=ro", uri=True, timeout=5, ) connection.row_factory = sqlite3.Row return connection def status(self) -> dict[str, Any]: manifest = self._load_json(self.manifest_path, {}) result: dict[str, Any] = { "available": False, "package_id": manifest.get("package_id"), "name": manifest.get("name"), "version": manifest.get("version"), "domains": 0, "skills": 0, "knowledge_cards": 0, "certified_skills": 0, "forge_locked": True, "index_path": str(self.index_path), "error": None, } if not self.index_path.is_file(): result["error"] = "Academy index is missing." return result try: with self._connect() as connection: result["domains"] = connection.execute( "SELECT COUNT(DISTINCT domain_id) FROM skills" ).fetchone()[0] result["skills"] = connection.execute( "SELECT COUNT(*) FROM skills" ).fetchone()[0] result["knowledge_cards"] = connection.execute( "SELECT COUNT(*) FROM cards" ).fetchone()[0] result["certified_skills"] = connection.execute( "SELECT COUNT(*) FROM skills WHERE forge_allowed = 1" ).fetchone()[0] except (sqlite3.Error, OSError) as exc: result["error"] = str(exc) return result expected_domains = int(manifest.get("domain_count") or 0) expected_skills = int(manifest.get("skill_count") or 0) expected_cards = int(manifest.get("knowledge_card_count") or 0) result["available"] = bool( result["domains"] and result["skills"] and result["knowledge_cards"] and (not expected_domains or result["domains"] == expected_domains) and (not expected_skills or result["skills"] == expected_skills) and (not expected_cards or result["knowledge_cards"] == expected_cards) ) result["forge_locked"] = result["certified_skills"] < result["skills"] if not result["available"] and result["error"] is None: result["error"] = "Academy index counts do not match the installed manifest." return result def _query_tokens(self, prompt: str, maximum: int = 12) -> list[str]: tokens: list[str] = [] seen: set[str] = set() for token in _TOKEN_RE.findall((prompt or "").lower()): if len(token) < 3 or token in _STOP_WORDS or token in seen: continue seen.add(token) tokens.append(token) if len(tokens) >= maximum: break return tokens def _fts_query(self, prompt: str) -> str: return " OR ".join(f"{token}*" for token in self._query_tokens(prompt)) def search(self, prompt: str, limit: int = 6) -> list[dict[str, Any]]: if not self.index_path.is_file(): return [] query = self._fts_query(prompt) if not query: return [] limit = max(1, min(int(limit), 12)) fetch_limit = min(limit * 5, 60) try: with self._connect() as connection: rows = connection.execute( """ SELECT skill_id, domain_id, skill_name, card_type, question, answer, bm25(cards) AS rank FROM cards WHERE cards MATCH ? ORDER BY rank LIMIT ? """, (query, fetch_limit), ).fetchall() except (sqlite3.Error, OSError): return [] results = [] seen_skills = set() for row in rows: skill_id = str(row["skill_id"]) if skill_id in seen_skills: continue seen_skills.add(skill_id) results.append( { "skill_id": skill_id, "domain_id": row["domain_id"], "skill_name": row["skill_name"], "card_type": row["card_type"], "question": row["question"], "answer": row["answer"], "rank": float(row["rank"]), "forge_allowed": False, } ) if len(results) >= limit: break return results def context_for_prompt(self, prompt: str, limit: int = 6) -> str: status = self.status() if not status.get("available"): return "" rows = self.search(prompt, limit=limit) if not rows: return "" lines = [ f"SHiRE Design Academy {status.get('version') or ''} — REFERENCE ONLY", ( f"Installed library: {status['domains']} domains, {status['skills']} skills, " f"{status['knowledge_cards']} knowledge cards." ), ( "All starter skills remain Forge-disabled until their mandatory Academy " "practice, validation, provenance and safety evidence passes." ), "Relevant Academy cards:", ] for row in rows: lines.extend( [ f"- SKILL: {row['skill_id']}", f" DOMAIN: {row['domain_id']}", f" TYPE: {row['card_type']}", f" QUESTION: {row['question']}", f" GUIDANCE: {row['answer']}", ] ) return "\n".join(lines)[:8000] def dashboard_data(self) -> dict[str, Any]: status = self.status() output = { **status, "library_integrity_percent": 100 if status.get("available") else 0, "certification_percent": 0, "domain_rows": [], } if not status.get("available"): return output catalog = self._load_json(self.catalog_path, []) names = { str(item.get("domain_id")): str(item.get("domain_name")) for item in catalog if item.get("domain_id") and item.get("domain_name") } try: with self._connect() as connection: rows = connection.execute( """ SELECT domain_id, COUNT(*) AS skill_count, SUM(CASE WHEN forge_allowed = 1 THEN 1 ELSE 0 END) AS certified_count FROM skills GROUP BY domain_id ORDER BY domain_id """ ).fetchall() except (sqlite3.Error, OSError): output["available"] = False output["library_integrity_percent"] = 0 output["error"] = "Could not read Academy domain summary." return output domain_rows = [] for row in rows: skill_count = int(row["skill_count"] or 0) certified_count = int(row["certified_count"] or 0) progress = round((certified_count / skill_count) * 100) if skill_count else 0 domain_id = str(row["domain_id"]) domain_rows.append( { "domain_id": domain_id, "domain_name": names.get( domain_id, domain_id.replace("_", " ").title(), ), "skill_count": skill_count, "certified_count": certified_count, "progress": progress, } ) output["domain_rows"] = domain_rows if status["skills"]: output["certification_percent"] = round( (status["certified_skills"] / status["skills"]) * 100 ) return output design_academy_service = DesignAcademyService()