Spaces:
Sleeping
Sleeping
| """Shared data model + helpers for the brain compiler. | |
| The graph is two flat lists: entities and edges. Each extractor returns Entity/Edge | |
| objects; build_graph merges them, dedupes, and computes backlinks. | |
| """ | |
| from __future__ import annotations | |
| import re | |
| from dataclasses import dataclass, field, asdict | |
| from typing import Any | |
| # ---- id helpers ----------------------------------------------------------- | |
| def slug(text: str) -> str: | |
| """Filesystem-safe slug, preserving case-insensitive uniqueness via lowercasing.""" | |
| s = re.sub(r"[^A-Za-z0-9_.-]+", "_", str(text).strip()) | |
| return s.strip("_") or "_" | |
| def eid(etype: str, key: str) -> str: | |
| """Canonical entity id, e.g. 'sp:pocrmn_sp_podbyr'. Keys are lowercased for join safety.""" | |
| return f"{etype}:{str(key).strip().lower()}" | |
| # ---- model ---------------------------------------------------------------- | |
| class Entity: | |
| id: str | |
| type: str # activity | screen | sp | api | table | error | |
| name: str | |
| source: str = "" # raw path relative to ARTIFACTS_ROOT | |
| attrs: dict[str, Any] = field(default_factory=dict) | |
| def to_dict(self) -> dict: | |
| return asdict(self) | |
| class Edge: | |
| src: str # entity id | |
| rel: str # shown_on | runs | invokes | writes | reads | calls | raises | backed_by | |
| dst: str # entity id | |
| source: str = "" | |
| inferred: bool = False | |
| attrs: dict[str, Any] = field(default_factory=dict) | |
| def key(self) -> tuple: | |
| return (self.src, self.rel, self.dst) | |
| def to_dict(self) -> dict: | |
| return asdict(self) | |
| # ---- merge helpers -------------------------------------------------------- | |
| def merge_entity(store: dict[str, Entity], ent: Entity) -> Entity: | |
| """Insert or merge an entity by id. Later attrs update earlier; lists are unioned.""" | |
| existing = store.get(ent.id) | |
| if existing is None: | |
| store[ent.id] = ent | |
| return ent | |
| if not existing.source and ent.source: | |
| existing.source = ent.source | |
| for k, v in ent.attrs.items(): | |
| if isinstance(v, list): | |
| cur = existing.attrs.setdefault(k, []) | |
| for item in v: | |
| if item not in cur: | |
| cur.append(item) | |
| else: | |
| existing.attrs.setdefault(k, v) | |
| return existing | |
| def dedupe_edges(edges: list[Edge]) -> list[Edge]: | |
| seen: dict[tuple, Edge] = {} | |
| for e in edges: | |
| k = e.key() | |
| if k in seen: | |
| # merge attrs; a parsed (non-inferred) edge wins over inferred | |
| cur = seen[k] | |
| if cur.inferred and not e.inferred: | |
| cur.inferred = False | |
| for ak, av in e.attrs.items(): | |
| if isinstance(av, list): | |
| lst = cur.attrs.setdefault(ak, []) | |
| for it in av: | |
| if it not in lst: | |
| lst.append(it) | |
| else: | |
| cur.attrs.setdefault(ak, av) | |
| else: | |
| seen[k] = e | |
| return list(seen.values()) | |