from __future__ import annotations from dataclasses import dataclass, field import json from pathlib import Path from typing import Any @dataclass(slots=True) class SourceConfig: name: str type: str = "unknown" trust_level: float = 0.5 base_url: str | None = None allowed_paths: list[str] = field(default_factory=list) parser: str = "generic" fetcher: str = "requests" rate_limit_per_minute: int = 30 respect_robots_txt: bool = True @dataclass(slots=True) class ProjectConfig: project_name: str domain: str target_entities: list[str] fields: list[str] sources: list[SourceConfig] ontology: dict[str, Any] = field(default_factory=dict) recommendation: dict[str, Any] = field(default_factory=dict) update_policy: dict[str, Any] = field(default_factory=dict) def source_by_name(self, name: str) -> SourceConfig: for source in self.sources: if source.name == name: return source raise KeyError(f"Unknown source in project config: {name}") def load_project_config(path: str | Path) -> ProjectConfig: config_path = Path(path) data = _load_mapping(config_path) sources = [SourceConfig(**item) for item in data.get("sources", [])] return ProjectConfig( project_name=data["project_name"], domain=data["domain"], target_entities=list(data.get("target_entities", [])), fields=list(data.get("fields", [])), sources=sources, ontology=dict(data.get("ontology", {})), recommendation=dict(data.get("recommendation", {})), update_policy=dict(data.get("update_policy", {})), ) def _load_mapping(path: Path) -> dict[str, Any]: text = path.read_text(encoding="utf-8") if path.suffix.lower() == ".json": return json.loads(text) try: import yaml except ImportError as exc: return _load_simple_yaml(text) return yaml.safe_load(text) def _load_simple_yaml(text: str) -> dict[str, Any]: """Small YAML fallback for project configs when PyYAML is unavailable. It intentionally supports only the subset used by the sample project files: nested mappings, lists, booleans, ints, floats, and strings. """ lines: list[tuple[int, str]] = [] for raw_line in text.splitlines(): if not raw_line.strip() or raw_line.lstrip().startswith("#"): continue indent = len(raw_line) - len(raw_line.lstrip(" ")) lines.append((indent, raw_line.strip())) value, index = _parse_yaml_block(lines, 0, 0) if index != len(lines): raise RuntimeError("Could not parse full YAML config. Install PyYAML for full YAML support.") if not isinstance(value, dict): raise RuntimeError("Project config root must be a mapping.") return value def _parse_yaml_block(lines: list[tuple[int, str]], index: int, indent: int) -> tuple[Any, int]: if index >= len(lines): return {}, index current_indent, content = lines[index] if current_indent < indent: return {}, index if content.startswith("- "): result: list[Any] = [] while index < len(lines): item_indent, item_content = lines[index] if item_indent != indent or not item_content.startswith("- "): break item_raw = item_content[2:].strip() index += 1 if not item_raw: child, index = _parse_yaml_block(lines, index, indent + 2) result.append(child) elif ":" in item_raw: key, raw_value = _split_key_value(item_raw) item: dict[str, Any] = {key: _parse_scalar(raw_value)} if raw_value else {key: None} if index < len(lines) and lines[index][0] > indent: child, index = _parse_yaml_block(lines, index, lines[index][0]) if isinstance(child, dict): item.update(child) result.append(item) else: result.append(_parse_scalar(item_raw)) return result, index result: dict[str, Any] = {} while index < len(lines): line_indent, line_content = lines[index] if line_indent != indent or line_content.startswith("- "): break key, raw_value = _split_key_value(line_content) index += 1 if raw_value: result[key] = _parse_scalar(raw_value) elif index < len(lines) and lines[index][0] > indent: child, index = _parse_yaml_block(lines, index, lines[index][0]) result[key] = child else: result[key] = None return result, index def _split_key_value(content: str) -> tuple[str, str]: key, _, raw_value = content.partition(":") return key.strip(), raw_value.strip() def _parse_scalar(value: str) -> Any: if value == "": return None lowered = value.lower() if lowered == "true": return True if lowered == "false": return False if lowered in {"null", "none"}: return None try: if "." in value: return float(value) return int(value) except ValueError: return value.strip("\"'")