scripts.spec_word_cloud

Reference › scripts › spec_word_cloud

scripts/spec_word_cloud.py is a term-prominence tool, an MVP word cloud, for the specs and requirements of orchestrata-ultimate: it shows which terms dominate the specification corpus. A human runs it. With --apply-to-graph it also matches the top terms against entity names in the semantic graph, using orchestrata_ultimate.semantic_graph.adapters.memgraph, orchestrata_ultimate.semantic_graph.config and orchestrata_ultimate.semantic_graph.repository.

discover_documents concatenates the text of each spec directory under the specs directory. tokenize counts terms per document, splitting CamelCase and snake_case identifiers into words (is_identifier_token, split_identifier), and compute_tfidf produces a global ranking and a per-spec ranking. write_json_report writes the JSON, and build_svg renders a self-contained SVG word cloud with greedy row packing, sizing fonts by weight. For the graph step, entity_name_candidates and _canonical_form normalise names to lowercase alphanumeric strings so terms can be matched. Only the top-N terms are matched, because matching the full ranking against thousands of entities runs for hours. Options include the specs directory, --output-json, --output-svg, --top and --apply-to-graph. It prints the document count, the output paths and the top ten global terms.

At a glance

Class diagram

Dependency diagram

API

Classes

TermStat (dataclass)

Decorators: dataclass · Line 141

A single (term, weight, count) ranking entry.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/spec_word_cloud.py

"""Spec/requirements term-prominence tool (MVP word cloud) for orchestrata-ultimate.

Scans ``specs/*/spec.md`` + ``specs/*/requirements.md``, tokenizes the text,
splits identifiers (CamelCase / snake_case) into constituent words while ALSO
keeping the intact identifier as a 2x-weighted "object/node" term, computes a
classic TF-IDF ranking per spec-document and globally, and emits:

- a JSON term-prominence report (``plans/spec_term_prominence.json``)
- a hand-laid-out, self-contained SVG word cloud (``docs/diagrams/spec-word-cloud.svg``)

Optionally (``--apply-to-graph``), it writes a ``prominence`` property onto
matching entities in the semantic graph (Memgraph), fail-soft if the graph is
unreachable.

Weighting choice (documented per the brief): tf is the raw per-spec term count
(with the 2x boost for intact identifiers baked directly into the count, not
applied twice), idf is computed GLOBALLY as ``log(N_docs / (1 + df))`` where
df is the number of spec-documents containing the term at least once. The
per-spec weight is ``tf * idf``. The GLOBAL weight for a term is the sum of
its per-spec weights across all documents that contain it — this rewards
terms that are both frequent and either concentrated or broadly used, while
still following the classic single global idf (simplest option consistent
between per-spec and global views).
"""

from __future__ import annotations

import argparse
import json
import math
import re
from collections import Counter
from dataclasses import dataclass
from pathlib import Path
from typing import Any

PROJECT_ROOT = Path(__file__).resolve().parent.parent

# --- stopwords -------------------------------------------------------------

_ENGLISH_STOPWORDS = frozenset("""
    a an the is are of to in on for and or that this be as by with from at
    it its into if when then which will shall must should may can not no
    than such other these those we you your our
    """.split())

_SPEC_BOILERPLATE_STOPWORDS = frozenset("""
    spec specs requirement requirements section table contents overview
    summary story stories us req fr nfr id ids status draft approved todo
    tbd note notes example examples figure diagram
    """.split())

STOPWORDS = _ENGLISH_STOPWORDS | _SPEC_BOILERPLATE_STOPWORDS

# raw word-ish tokens: letters/digits/underscore runs, must start with a letter
_TOKEN_RE = re.compile(r"[A-Za-z][A-Za-z0-9_]*")

# camelCase boundary: lowercase/digit followed by uppercase, OR a run of
# uppercase letters followed by a lowercase (e.g. "CXPPacket" -> "CXP","Packet")
_CAMEL_BOUNDARY_RE = re.compile(r"(?<=[a-z0-9])(?=[A-Z])|(?<=[A-Z])(?=[A-Z][a-z])")


def is_identifier_token(token: str) -> bool:
    """True if a token reads as a CamelCase or snake_case identifier."""
    if "_" in token:
        return True
    # internal uppercase after a lowercase somewhere in the token
    return bool(re.search(r"[a-z][A-Z]", token)) or bool(
        re.search(r"[A-Z]{2,}[a-z]", token)
    )


def split_identifier(token: str) -> list[str]:
    """Split a CamelCase/snake_case identifier into lowercase constituent words."""
    parts: list[str] = []
    for chunk in token.split("_"):
        if not chunk:
            continue
        parts.extend(p for p in _CAMEL_BOUNDARY_RE.split(chunk) if p)
    return [p.lower() for p in parts if p]


def tokenize(text: str) -> Counter[str]:
    """Tokenize ``text`` into a term->count Counter for one document.

    Intact identifiers (CamelCase/snake_case) are counted under their
    ORIGINAL casing with a 2x weight (baked into the count). Their
    constituent words are ALSO counted (lowercased, normal 1x weight),
    unless they are stopwords or shorter than 2 chars.
    """
    counts: Counter[str] = Counter()
    for match in _TOKEN_RE.finditer(text):
        token = match.group(0)
        if len(token) < 2:
            continue
        if is_identifier_token(token):
            counts[token] += 2
            for word in split_identifier(token):
                if len(word) < 2 or word in STOPWORDS:
                    continue
                counts[word] += 1
        else:
            word = token.lower()
            if len(word) < 2 or word in STOPWORDS:
                continue
            counts[word] += 1
    return counts


# --- document discovery -----------------------------------------------------

_SPEC_ID_RE = re.compile(r"SPEC-\d{3}\b")


def discover_documents(specs_dir: Path) -> dict[str, str]:
    """Return ``{spec_key: concatenated_text}`` for each spec directory.

    ``spec_key`` prefers an embedded ``SPEC-XXX`` id found in the document
    text; falls back to the directory name otherwise.
    """
    documents: dict[str, str] = {}
    for spec_dir in sorted(p for p in specs_dir.iterdir() if p.is_dir()):
        pieces: list[str] = []
        for filename in ("spec.md", "requirements.md"):
            file_path = spec_dir / filename
            if file_path.exists():
                pieces.append(file_path.read_text(encoding="utf-8", errors="replace"))
        if not pieces:
            continue
        text = "\n".join(pieces)
        match = _SPEC_ID_RE.search(text)
        key = match.group(0) if match else spec_dir.name
        documents[key] = text
    return documents


# --- TF-IDF ------------------------------------------------------------------


@dataclass
class TermStat:
    """A single (term, weight, count) ranking entry."""

    term: str
    weight: float
    count: int

    def to_dict(self) -> dict[str, Any]:
        return {"term": self.term, "weight": self.weight, "count": self.count}


def compute_tfidf(
    documents: dict[str, str],
) -> tuple[list[TermStat], dict[str, list[TermStat]]]:
    """Compute global and per-spec TF-IDF rankings.

    Returns ``(global_ranking, per_spec_rankings)``, both already sorted
    descending by weight (ties broken by term for determinism).
    """
    per_doc_counts: dict[str, Counter[str]] = {
        key: tokenize(text) for key, text in documents.items()
    }
    n_docs = len(per_doc_counts)

    df: Counter[str] = Counter()
    for counts in per_doc_counts.values():
        for term in counts:
            df[term] += 1

    def idf(term: str) -> float:
        # Smoothed IDF (as in scikit-learn's TfidfVectorizer default): always
        # non-negative, so a term's weight/prominence is never negative even
        # when it appears in every document.
        return math.log((1 + n_docs) / (1 + df[term])) + 1

    per_spec: dict[str, list[TermStat]] = {}
    global_weight: dict[str, float] = {}
    global_count: Counter[str] = Counter()

    for key, counts in per_doc_counts.items():
        stats = []
        for term, tf in counts.items():
            weight = tf * idf(term)
            stats.append(TermStat(term=term, weight=weight, count=tf))
            global_weight[term] = global_weight.get(term, 0.0) + weight
            global_count[term] += tf
        stats.sort(key=lambda s: (-s.weight, s.term))
        per_spec[key] = stats

    global_ranking = [
        TermStat(term=term, weight=global_weight[term], count=global_count[term])
        for term in global_weight
    ]
    global_ranking.sort(key=lambda s: (-s.weight, s.term))

    return global_ranking, per_spec


# --- SVG word cloud ----------------------------------------------------------

_PALETTE = ["#3b6ea5", "#a5573b", "#4f8f6d", "#8a5ca8", "#b58a2e", "#5c7a8a"]

_SVG_WIDTH = 1000
_SVG_HEIGHT = 700
_MIN_FONT = 12.0
_MAX_FONT = 64.0


def _font_size(weight: float, min_weight: float, max_weight: float) -> float:
    if max_weight <= min_weight:
        return (_MIN_FONT + _MAX_FONT) / 2
    frac = (weight - min_weight) / (max_weight - min_weight)
    frac = math.sqrt(max(0.0, min(1.0, frac)))
    return _MIN_FONT + frac * (_MAX_FONT - _MIN_FONT)


def _estimate_text_width(text: str, font_size: float) -> float:
    # Rough average glyph width for a generic sans-serif font.
    return len(text) * font_size * 0.6


def build_svg(terms: list[TermStat], max_terms: int = 80) -> str:
    """Render a self-contained SVG word cloud (greedy row-packing layout)."""
    chosen = terms[:max_terms]
    if not chosen:
        return (
            '<svg xmlns="http://www.w3.org/2000/svg" '
            f'width="{_SVG_WIDTH}" height="{_SVG_HEIGHT}" '
            f'viewBox="0 0 {_SVG_WIDTH} {_SVG_HEIGHT}">'
            f'<rect width="{_SVG_WIDTH}" height="{_SVG_HEIGHT}" fill="#1b1f24"/>'
            "</svg>"
        )

    weights = [t.weight for t in chosen]
    min_weight, max_weight = min(weights), max(weights)

    sized = [(t, _font_size(t.weight, min_weight, max_weight)) for t in chosen]

    padding = 12.0
    margin = 20.0
    x = margin
    y = margin
    row_height = 0.0
    texts: list[str] = []

    for index, (term, font_size) in enumerate(sized):
        width = _estimate_text_width(term.term, font_size)
        if x + width + padding > _SVG_WIDTH - margin and x > margin:
            x = margin
            y += row_height + padding
            row_height = 0.0
        if y + font_size > _SVG_HEIGHT - margin:
            # Ran out of vertical space; stop placing further (lower-weight) terms.
            break
        color = _PALETTE[index % len(_PALETTE)]
        safe_term = (
            term.term.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
        )
        texts.append(
            f'<text x="{x:.1f}" y="{(y + font_size):.1f}" '
            f'font-family="sans-serif" font-size="{font_size:.1f}" '
            f'fill="{color}">{safe_term}</text>'
        )
        x += width + padding
        row_height = max(row_height, font_size)

    body = "\n  ".join(texts)
    return (
        '<svg xmlns="http://www.w3.org/2000/svg" '
        f'width="{_SVG_WIDTH}" height="{_SVG_HEIGHT}" '
        f'viewBox="0 0 {_SVG_WIDTH} {_SVG_HEIGHT}">\n'
        f'  <rect width="{_SVG_WIDTH}" height="{_SVG_HEIGHT}" fill="#1b1f24"/>\n'
        f"  {body}\n"
        "</svg>"
    )


# --- entity/graph matching ----------------------------------------------------


def _canonical_form(text: str) -> str:
    """Normalize an identifier/name to a canonical lowercase-alnum string."""
    words: list[str] = []
    for token in _TOKEN_RE.findall(text):
        if is_identifier_token(token):
            words.extend(split_identifier(token))
        else:
            words.append(token.lower())
    return "".join(words)


def entity_name_candidates(entity_id: str, name_prop: str | None = None) -> set[str]:
    """Derive matchable name fragments from a structured entity id.

    Real ids look like ``orchestrata-ultimate:spec:006-session-management`` or
    ``orchestrata-ultimate:code:pkg.module:ClassName:method`` — the whole id
    never equals a term, so matching works on the meaningful fragments: each
    colon segment past the project prefix, plus its hyphen/dot-separated words
    and intact identifiers.
    """
    candidates: set[str] = set()
    if name_prop:
        candidates.add(name_prop)
    segments = entity_id.split(":")[2:] or [entity_id]
    for seg in segments:
        candidates.add(seg)
        for part in re.split(r"[-./]", seg):
            part = re.sub(r"^\d+$", "", part)
            if len(part) >= 3:
                candidates.add(part)
    return candidates


def term_matches_entity_name(term: str, entity_name: str) -> bool:
    """True if ``term`` and ``entity_name`` refer to the same identifier,
    modulo case and camelCase/snake_case formatting differences."""
    if not term or not entity_name:
        return False
    if term.lower() == entity_name.lower():
        return True
    return _canonical_form(term) == _canonical_form(entity_name)


# --- output writers ------------------------------------------------------------


def write_json_report(
    output_path: Path,
    global_ranking: list[TermStat],
    per_spec: dict[str, list[TermStat]],
    top: int,
) -> None:
    payload = {
        "global": [t.to_dict() for t in global_ranking[:top]],
        "per_spec": {
            key: [t.to_dict() for t in stats[:30]] for key, stats in per_spec.items()
        },
    }
    output_path.parent.mkdir(parents=True, exist_ok=True)
    output_path.write_text(json.dumps(payload, indent=2), encoding="utf-8")


def write_svg_report(output_path: Path, global_ranking: list[TermStat]) -> None:
    svg = build_svg(global_ranking, max_terms=min(80, len(global_ranking)))
    output_path.parent.mkdir(parents=True, exist_ok=True)
    output_path.write_text(svg, encoding="utf-8")


# --- optional semantic-graph enrichment ---------------------------------------


async def apply_to_graph(global_ranking: list[TermStat]) -> None:
    """Best-effort: write a ``prominence`` property onto matching graph nodes."""
    from orchestrata_ultimate.semantic_graph.adapters.memgraph import (
        MemgraphSemanticGraph,
    )
    from orchestrata_ultimate.semantic_graph.config import SemanticGraphConfig
    from orchestrata_ultimate.semantic_graph.repository import create_semantic_graph

    config = SemanticGraphConfig.from_env()
    graph = create_semantic_graph(config)
    try:
        connected = await graph.connect()
        if not connected:
            print(
                "Semantic graph unreachable — skipping graph enrichment: "
                "connect() returned False"
            )
            return

        if not isinstance(graph, MemgraphSemanticGraph):
            print(
                "Semantic graph backend does not support enumeration — "
                "skipping graph enrichment"
            )
            return

        entities = await graph._all_entities()  # noqa: SLF001
        updates: list[dict[str, Any]] = []
        for entity in entities:
            best_weight: float | None = None
            name_prop = entity.properties.get("name")
            candidate_names = entity_name_candidates(
                entity.id, name_prop if isinstance(name_prop, str) else None
            )
            for term_stat in global_ranking:
                if any(
                    term_matches_entity_name(term_stat.term, name)
                    for name in candidate_names
                ):
                    if best_weight is None or term_stat.weight > best_weight:
                        best_weight = term_stat.weight
            if best_weight is not None:
                updates.append({"id": entity.id, "val": best_weight})
        # One UNWIND write per batch — a per-entity round trip takes >12min
        # at 10k entities; batching finishes in seconds.
        batch_size = 1000
        for start in range(0, len(updates), batch_size):
            await graph._run_write(  # noqa: SLF001
                "UNWIND $rows AS row "
                "MATCH (e:Entity {id: row.id}) SET e.prominence = row.val",
                {"rows": updates[start : start + batch_size]},
            )
        print(
            f"Matched {len(updates)} / {len(entities)} nodes; "
            f"updated {len(updates)} nodes with prominence scores."
        )
    finally:
        await graph.close()


# --- CLI -----------------------------------------------------------------------


def _build_arg_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="Compute spec/requirements term prominence (TF-IDF word cloud)."
    )
    parser.add_argument("--top", type=int, default=200)
    parser.add_argument(
        "--output-json",
        type=Path,
        default=PROJECT_ROOT / "plans" / "spec_term_prominence.json",
    )
    parser.add_argument(
        "--output-svg",
        type=Path,
        default=PROJECT_ROOT / "docs" / "diagrams" / "spec-word-cloud.svg",
    )
    parser.add_argument("--specs-dir", type=Path, default=PROJECT_ROOT / "specs")
    parser.add_argument("--apply-to-graph", action="store_true", default=False)
    return parser


def main(argv: list[str] | None = None) -> int:
    args = _build_arg_parser().parse_args(argv)

    documents = discover_documents(args.specs_dir)
    global_ranking, per_spec = compute_tfidf(documents)

    write_json_report(args.output_json, global_ranking, per_spec, args.top)
    write_svg_report(args.output_svg, global_ranking)

    print(f"Documents: {len(documents)}")
    print(f"Wrote {args.output_json}")
    print(f"Wrote {args.output_svg}")
    print("Top 10 global terms:")
    for stat in global_ranking[:10]:
        print(f"  {stat.term}: weight={stat.weight:.3f} count={stat.count}")

    if args.apply_to_graph:
        import asyncio

        # Only the top-N terms define prominence; matching the FULL ranking
        # (thousands of terms x 10k entities) runs for hours.
        asyncio.run(apply_to_graph(global_ranking[: args.top]))

    return 0


if __name__ == "__main__":
    raise SystemExit(main())