path_explorer.py

# YAML-access python

import yaml
from pathlib import Path
from typing import Any, Union


PROJECT_PATH = Path(__file__).resolve().parent
PathLike = Union[str, Path, tuple[str, ...]]


# ============================================================
# 1. Dot-access wrapper
# ============================================================

class YAMLNode:
    def __init__(self, data: Any):
        object.__setattr__(self, "_data", data)

    # ============================================================
    # Core access
    # ============================================================

    def __getitem__(self, key):
        return self._wrap(self._data[key])

    def __getattr__(self, key):
        if isinstance(self._data, dict) and key in self._data:
            return self._wrap(self._data[key])

        raise AttributeError(f"Key not found: {key}")

    def __setitem__(self, key, value):
        if isinstance(self._data, dict):
            self._data[key] = self._unwrap(value)
            return

        if isinstance(self._data, list):
            self._data[key] = self._unwrap(value)
            return

        raise TypeError(f"Cannot set item on {type(self._data).__name__}")

    def __setattr__(self, key, value):
        if key == "_data":
            object.__setattr__(self, key, value)
            return

        if isinstance(self._data, dict):
            self._data[key] = self._unwrap(value)
            return

        raise AttributeError("Cannot set attribute on non-dict YAML node")

    # ============================================================
    # Dict-like methods
    # ============================================================

    def get(self, key, default=None):
        if isinstance(self._data, dict):
            return self._wrap(self._data.get(key, default))

        return default

    def items(self):
        if not isinstance(self._data, dict):
            raise TypeError(f"items() is only available on dict nodes, got {type(self._data).__name__}")

        for key, value in self._data.items():
            yield key, self._wrap(value)

    def keys(self):
        if not isinstance(self._data, dict):
            raise TypeError(f"keys() is only available on dict nodes, got {type(self._data).__name__}")

        return self._data.keys()

    def values(self):
        if not isinstance(self._data, dict):
            raise TypeError(f"values() is only available on dict nodes, got {type(self._data).__name__}")

        for value in self._data.values():
            yield self._wrap(value)

    def update(self, other=None, **kwargs):
        if not isinstance(self._data, dict):
            raise TypeError(f"update() is only available on dict nodes, got {type(self._data).__name__}")

        if other:
            if isinstance(other, YAMLNode):
                other = other.raw()

            for key, value in dict(other).items():
                self._data[key] = self._unwrap(value)

        for key, value in kwargs.items():
            self._data[key] = self._unwrap(value)

        return self

    # ============================================================
    # List-like helpers
    # ============================================================

    def append(self, value):
        if not isinstance(self._data, list):
            raise TypeError(f"append() is only available on list nodes, got {type(self._data).__name__}")

        self._data.append(self._unwrap(value))
        return self

    def extend(self, values):
        if not isinstance(self._data, list):
            raise TypeError(f"extend() is only available on list nodes, got {type(self._data).__name__}")

        if isinstance(values, YAMLNode):
            values = values.raw()

        self._data.extend([self._unwrap(v) for v in values])
        return self

    # ============================================================
    # Python protocol support
    # ============================================================

    def __iter__(self):
        if isinstance(self._data, dict):
            return iter(self._data)

        if isinstance(self._data, list):
            return (self._wrap(value) for value in self._data)

        raise TypeError(f"{type(self._data).__name__} node is not iterable")

    def __len__(self):
        if isinstance(self._data, (dict, list, tuple, str)):
            return len(self._data)

        raise TypeError(f"{type(self._data).__name__} node has no len()")

    def __bool__(self):
        return bool(self._data)

    def __contains__(self, key):
        if isinstance(self._data, dict):
            return key in self._data

        if isinstance(self._data, list):
            return key in self._data

        return False

    # ============================================================
    # Conversion
    # ============================================================

    def raw(self):
        return self._data

    def to_dict(self):
        return self._deep_unwrap(self._data)

    @staticmethod
    def _wrap(value):
        if isinstance(value, (dict, list)):
            return YAMLNode(value)

        return value

    @staticmethod
    def _unwrap(value):
        if isinstance(value, YAMLNode):
            return value.raw()

        return value

    @classmethod
    def _deep_unwrap(cls, value):
        if isinstance(value, YAMLNode):
            return cls._deep_unwrap(value.raw())

        if isinstance(value, dict):
            return {key: cls._deep_unwrap(val) for key, val in value.items()}

        if isinstance(value, list):
            return [cls._deep_unwrap(item) for item in value]

        return value

    def __repr__(self):
        return repr(self._data)


# ============================================================
# 2. Path resolving
# ============================================================

def _project_path(path: PathLike = PROJECT_PATH) -> Path:
    if isinstance(path, tuple):
        return PROJECT_PATH.joinpath(*path)

    path = Path(path)

    # "/seeds/planner" should mean relative to PROJECT_PATH,
    # not system root.
    raw = str(path)
    if raw.startswith("/"):
        raw = raw.lstrip("/")
        return PROJECT_PATH / raw

    if path.is_absolute():
        return path

    return PROJECT_PATH / path


def _possible_yaml_paths(base: Path):
    if base.suffix in [".yaml", ".yml"]:
        yield base
    else:
        yield base.with_suffix(".yaml")
        yield base.with_suffix(".yml")
        yield base


def _find_yaml_file_and_inner(path: PathLike):
    """
    Examples:
      /seeds/planner
        -> .project/seeds/planner.yaml, inner=None

      /seeds/planner/final_export.io.dasd
        -> .project/seeds/planner.yaml, inner="final_export.io.dasd"

      /seeds/planner/final_export/io/dasd
        -> .project/seeds/planner.yaml, inner="final_export/io/dasd"
    """

    full = _project_path(path)

    # Direct file check.
    for candidate in _possible_yaml_paths(full):
        if candidate.exists() and candidate.is_file():
            return candidate, None

    parts = full.parts
    project_parts = PROJECT_PATH.parts

    # Remove project prefix from parts.
    if parts[:len(project_parts)] == project_parts:
        rel_parts = parts[len(project_parts):]
    else:
        rel_parts = parts

    # Try longest possible YAML file path.
    # /seeds/planner/final_export/io
    # tries:
    #   seeds/planner/final_export/io.yaml
    #   seeds/planner/final_export.yaml
    #   seeds/planner.yaml
    #   seeds.yaml
    for i in range(len(rel_parts), 0, -1):
        file_candidate = PROJECT_PATH.joinpath(*rel_parts[:i])

        for yaml_candidate in _possible_yaml_paths(file_candidate):
            if yaml_candidate.exists() and yaml_candidate.is_file():
                inner_parts = rel_parts[i:]
                inner = "/".join(inner_parts) if inner_parts else None
                return yaml_candidate, inner

    raise FileNotFoundError(f"No YAML file found for path: {path}")


# ============================================================
# 3. Inner key parser
# ============================================================

def _parse_inner_path(inner: str | None):
    """
    Supports:
      final_export.io.dasd
      final_export['io']['dasd']
      final_export["io"].dasd
      final_export/io/dasd
      items[0].name
    """

    if not inner:
        return []

    tokens = []
    i = 0
    n = len(inner)

    while i < n:
        ch = inner[i]

        if ch in ". /":
            i += 1
            continue

        if ch == "[":
            i += 1

            # Quoted key: ['abc'] or ["abc"]
            if i < n and inner[i] in ["'", '"']:
                quote = inner[i]
                i += 1
                start = i

                while i < n and inner[i] != quote:
                    i += 1

                if i >= n:
                    raise ValueError(f"Unclosed quoted bracket in path: {inner}")

                tokens.append(inner[start:i])

                i += 1

                if i >= n or inner[i] != "]":
                    raise ValueError(f"Missing closing ] in path: {inner}")

                i += 1
                continue

            # Numeric index or bare key: [0] or [name]
            start = i
            while i < n and inner[i] != "]":
                i += 1

            if i >= n:
                raise ValueError(f"Unclosed bracket in path: {inner}")

            raw = inner[start:i].strip()
            if raw.isdigit():
                tokens.append(int(raw))
            else:
                tokens.append(raw)

            i += 1
            continue

        # Bare token: abc until ., /, or [
        start = i
        while i < n and inner[i] not in "./[":
            i += 1

        raw = inner[start:i].strip()
        if raw:
            tokens.append(raw)

    return tokens


# ============================================================
# 4. Walking YAML data
# ============================================================

def _get_from_tokens(data: Any, tokens: list):
    cur = data

    for token in tokens:
        if isinstance(cur, YAMLNode):
            cur = cur.raw()

        if isinstance(token, int):
            if not isinstance(cur, list):
                raise TypeError(f"Expected list for index [{token}], got {type(cur).__name__}")
            cur = cur[token]
        else:
            if not isinstance(cur, dict):
                raise TypeError(f"Expected dict for key '{token}', got {type(cur).__name__}")
            cur = cur[token]

    return cur


def _set_from_tokens(data: Any, tokens: list, value: Any):
    if not tokens:
        return value

    cur = data

    for idx, token in enumerate(tokens[:-1]):
        next_token = tokens[idx + 1]

        if isinstance(token, int):
            if not isinstance(cur, list):
                raise TypeError(f"Expected list for index [{token}], got {type(cur).__name__}")

            while len(cur) <= token:
                cur.append({} if not isinstance(next_token, int) else [])

            if cur[token] is None:
                cur[token] = {} if not isinstance(next_token, int) else []

            cur = cur[token]

        else:
            if not isinstance(cur, dict):
                raise TypeError(f"Expected dict for key '{token}', got {type(cur).__name__}")

            if token not in cur or cur[token] is None:
                cur[token] = {} if not isinstance(next_token, int) else []

            cur = cur[token]

    last = tokens[-1]

    if isinstance(last, int):
        if not isinstance(cur, list):
            raise TypeError(f"Expected list for final index [{last}], got {type(cur).__name__}")

        while len(cur) <= last:
            cur.append(None)

        cur[last] = YAMLNode._unwrap(value)
    else:
        if not isinstance(cur, dict):
            raise TypeError(f"Expected dict for final key '{last}', got {type(cur).__name__}")

        cur[last] = YAMLNode._unwrap(value)

    return data


# ============================================================
# 5. Public API
# ============================================================

def ExploreFromPath(path: PathLike = PROJECT_PATH):
    yaml_file, inner = _find_yaml_file_and_inner(path)
    tokens = _parse_inner_path(inner)

    return {
        "yaml_file": yaml_file,
        "inner": inner,
        "tokens": tokens,
    }


def Get(path: PathLike = PROJECT_PATH, *, raw: bool = False):
    yaml_file, inner = _find_yaml_file_and_inner(path)
    tokens = _parse_inner_path(inner)

    with open(yaml_file, "r", encoding="utf-8") as f:
        data = yaml.safe_load(f) or {}

    value = _get_from_tokens(data, tokens)

    if raw:
        return value

    return YAMLNode._wrap(value)


def Set(path: PathLike, value: Any):
    yaml_file, inner = _find_yaml_file_and_inner(path)
    tokens = _parse_inner_path(inner)

    with open(yaml_file, "r", encoding="utf-8") as f:
        data = yaml.safe_load(f) or {}

    data = _set_from_tokens(data, tokens, value)

    with open(yaml_file, "w", encoding="utf-8") as f:
        yaml.safe_dump(data, f, sort_keys=False, allow_unicode=True)

    return YAMLNode._wrap(data)