"""Enhanced taxonomy loader with profile merging and validation. Supports: - JSON and YAML taxonomy files - Domain profile merging (Base + Fintech/Healthcare/Automotive) - Dynamic namespace validation - Canonical English policy enforcement """ import copy import json from pathlib import Path from typing import Any, Optional import structlog import yaml logger = structlog.get_logger(__name__) class TaxonomyLoader: """Loads and merges tag taxonomy from JSON/YAML configuration files. Phase 2 Enhancement: Supports ISO-compliant taxonomy + domain profiles. """ def __init__( self, standard_path: str | Path = "config/taxonomy/standard_tags.json", namespaces_path: str | Path = "config/taxonomy/namespaces.json", profile: Optional[str] = None, ): """Initialize taxonomy loader. Args: standard_path: Path to standard taxonomy JSON file namespaces_path: Path to namespace rules JSON file profile: Optional domain profile name (e.g., "fintech", "healthcare", "automotive") """ self.standard_path = Path(standard_path) self.namespaces_path = Path(namespaces_path) self.profile_name = profile self.taxonomy: dict[str, Any] = {} self.namespace_rules: dict[str, Any] = {} self._load() def _load(self) -> None: """Load taxonomy from JSON files with optional profile merging.""" # Load standard taxonomy (base) self.taxonomy = self._load_file(self.standard_path) # Load namespace rules self.namespace_rules = self._load_file(self.namespaces_path) # Merge domain profile if specified if self.profile_name: profile_path = Path(f"config/profiles/{self.profile_name}.json") if profile_path.exists(): profile_data = self._load_file(profile_path) self._merge_profile(profile_data) logger.info( "profile_merged", profile=self.profile_name, additional_tags=self._count_profile_tags(profile_data), ) else: logger.warning("profile_not_found", profile=self.profile_name, path=str(profile_path)) logger.info( "taxonomy_loaded", version=self.taxonomy.get("version", "unknown"), namespaces=list(self.taxonomy.get("namespaces", {}).keys()), profile=self.profile_name or "standard", ) def _load_file(self, file_path: Path) -> dict[str, Any]: """Load JSON or YAML file. Args: file_path: Path to config file Returns: Parsed config dict Raises: FileNotFoundError: If file doesn't exist """ if not file_path.exists(): raise FileNotFoundError(f"Config file not found: {file_path}") with open(file_path, encoding="utf-8") as f: if file_path.suffix == ".json": return json.load(f) elif file_path.suffix in [".yaml", ".yml"]: return yaml.safe_load(f) else: raise ValueError(f"Unsupported file format: {file_path.suffix}") def _merge_profile(self, profile_data: dict[str, Any]) -> None: """Merge domain profile tags into base taxonomy. Args: profile_data: Profile configuration dict """ additional_namespaces = profile_data.get("additional_namespaces", {}) for ns_name, ns_data in additional_namespaces.items(): # If namespace exists, append tags if ns_name in self.taxonomy["namespaces"]: existing_tags = self.taxonomy["namespaces"][ns_name].get("tags", []) new_tags = ns_data.get("tags", []) self.taxonomy["namespaces"][ns_name]["tags"] = existing_tags + new_tags else: # Create new namespace self.taxonomy["namespaces"][ns_name] = ns_data def _count_profile_tags(self, profile_data: dict[str, Any]) -> int: """Count total tags in profile. Args: profile_data: Profile configuration dict Returns: Total tag count """ count = 0 for ns_data in profile_data.get("additional_namespaces", {}).values(): count += len(ns_data.get("tags", [])) return count def get_all_tags(self) -> list[str]: """Get flattened list of all pre-defined tags (excludes dynamic namespaces). Returns: List of tags in format "namespace:tag_name" """ tags = [] namespaces = self.taxonomy.get("namespaces", {}) for ns_name, ns_data in namespaces.items(): # Skip dynamic namespaces (defined in namespaces.json) if self._is_dynamic_namespace(ns_name): continue for tag_def in ns_data.get("tags", []): tag_name = tag_def.get("name") if tag_name: tags.append(f"{ns_name}:{tag_name}") return tags def get_namespace_info(self, namespace: str) -> dict[str, Any] | None: """Get information about a specific namespace. Args: namespace: Namespace name (e.g., "req", "risk") Returns: Namespace data dict or None if not found """ return self.taxonomy.get("namespaces", {}).get(namespace) def get_dynamic_namespaces(self) -> list[str]: """Get list of allowed dynamic namespaces from namespace rules. Returns: List of dynamic namespace names (e.g., ["topic", "entity", "tech", "domain"]) """ allowed_dynamic = [] for ns_name, ns_data in self.namespace_rules.get("namespaces", {}).items(): if ns_data.get("dynamic", False) and ns_data.get("allowed", False): allowed_dynamic.append(ns_name) return allowed_dynamic def _is_dynamic_namespace(self, namespace: str) -> bool: """Check if namespace is dynamic (per namespaces.json). Args: namespace: Namespace to check Returns: True if dynamic namespace """ ns_data = self.namespace_rules.get("namespaces", {}).get(namespace, {}) return ns_data.get("dynamic", False) def is_valid_tag(self, tag: str) -> tuple[bool, Optional[str]]: """Validate a tag against taxonomy and namespace rules. Args: tag: Tag string in format "namespace:value" Returns: Tuple of (is_valid, reason) - (True, None) if valid - (False, "reason") if invalid """ if ":" not in tag: return False, "missing_namespace_separator" namespace, value = tag.split(":", 1) # Check if forbidden namespace forbidden = self.namespace_rules.get("forbidden_namespaces", []) if namespace in forbidden: return False, f"forbidden_namespace:{namespace}" # Check if pre-defined tag if tag in self.get_all_tags(): return True, None # Check if valid dynamic namespace dynamic_namespaces = self.get_dynamic_namespaces() if namespace in dynamic_namespaces: # Validate dynamic tag value ns_rules = self.namespace_rules["namespaces"][namespace] validation = ns_rules.get("validation", {}) # Check length min_len = validation.get("min_length", 1) max_len = validation.get("max_length", 100) if not (min_len <= len(value) <= max_len): return False, f"length_violation:{min_len}-{max_len}" # Check canonical English (unless exception) canonical_lang = validation.get("canonical_language", "en") if canonical_lang == "en" and not value.isascii(): return False, "non_ascii_value" return True, None # Unknown namespace in discovery mode if self.namespace_rules.get("strict_mode", False): return False, f"unknown_namespace:{namespace}" else: # Discovery mode: flag but allow return True, f"unknown_namespace_flagged:{namespace}" def get_forbidden_namespaces(self) -> list[str]: """Get list of forbidden namespace prefixes. Returns: List of forbidden namespaces """ return self.namespace_rules.get("forbidden_namespaces", [])