[crawler]

crawler_platform/app/config/loader.py

@@ -0,0 +1,154 @@
+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("\"'")