aaronrene / gabriel-muse public
stats.py python
161 lines 5.2 KB
Raw
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8 docs(KD-STAGING): sync governance after KD-6b DONE Human 12 days ago
1 """Vault statistics collector for the Knowtation domain.
2
3 Computes lightweight, filesystem-only statistics by walking the vault and
4 parsing frontmatter — no vector store, no network, no Node.js required.
5
6 Designed to be called by ``muse plumbing domain-info`` when the active domain
7 is ``knowtation``, and by any agent that needs a quick vault health summary
8 without doing a full ``snapshot()``.
9
10 Output shape (:class:`VaultStats`)::
11
12 {
13 "note_count": 76,
14 "source_types": ["chatgpt-export", "slack", "telegram"],
15 "projects": ["born-free", "dreambolt-network"],
16 "entity_count": 14,
17 "tag_count": 31,
18 "index_freshness": "2026-05-12T17:30:00+00:00" // or null
19 }
20
21 ``index_freshness`` is the ISO 8601 UTC mtime of the ``data/`` directory at
22 the vault root (where Knowtation stores its vector/sqlite index). Returns
23 ``null`` when the directory does not exist, indicating the vault has never
24 been indexed.
25 """
26
27 from __future__ import annotations
28
29 import os
30 import pathlib
31 from datetime import datetime, timezone
32 from typing import TypedDict
33
34 from muse.plugins.knowtation.parser import parse_frontmatter
35 from muse.plugins.knowtation.plugin import _ALWAYS_IGNORE_DIRS
36
37 # ---------------------------------------------------------------------------
38 # Constants
39 # ---------------------------------------------------------------------------
40
41 _MARKDOWN_EXTS: frozenset[str] = frozenset({".md", ".markdown", ".mdx"})
42
43 # Structural directories that hold scaffold files, not authored vault notes
44 # (mirrors the exclusion applied by the data-integrity tests).
45 _STRUCTURAL_DIRS: frozenset[str] = frozenset({"templates", "meta"})
46
47 _IGNORE_DIRS: frozenset[str] = _ALWAYS_IGNORE_DIRS | _STRUCTURAL_DIRS
48
49
50 # ---------------------------------------------------------------------------
51 # VaultStats TypedDict
52 # ---------------------------------------------------------------------------
53
54
55 class VaultStats(TypedDict):
56 """Statistics snapshot for one Knowtation vault.
57
58 ``note_count``
59 Total number of Markdown files found (excluding structural dirs).
60 ``source_types``
61 Sorted list of unique ``source`` / ``source_type`` values seen across
62 all notes. Identifies which capture/import pipelines have contributed.
63 ``projects``
64 Sorted list of unique ``project`` slugs seen across all notes.
65 ``entity_count``
66 Number of unique entity labels seen across all ``entity:`` fields.
67 ``tag_count``
68 Number of unique tag strings seen across all ``tags:`` fields.
69 ``index_freshness``
70 ISO 8601 UTC mtime of the ``data/`` index directory, or ``None`` when
71 the vault has never been indexed.
72 """
73
74 note_count: int
75 source_types: list[str]
76 projects: list[str]
77 entity_count: int
78 tag_count: int
79 index_freshness: str | None
80
81
82 # ---------------------------------------------------------------------------
83 # Collection
84 # ---------------------------------------------------------------------------
85
86
87 def collect_vault_stats(root: pathlib.Path) -> VaultStats:
88 """Walk *root* and return a :class:`VaultStats` snapshot.
89
90 Parses only YAML frontmatter (via :func:`~parser.parse_frontmatter`);
91 does not read note bodies or call any external service. Skips files and
92 directories that are unreadable without raising.
93
94 Args:
95 root: Vault root directory (same as the Muse repo root for a
96 ``knowtation`` repo).
97
98 Returns:
99 A :class:`VaultStats` mapping with counts and sorted lists.
100 """
101 note_count: int = 0
102 source_types: set[str] = set()
103 projects: set[str] = set()
104 entities: set[str] = set()
105 tags: set[str] = set()
106
107 root_str = str(root)
108
109 for dirpath, dirnames, filenames in os.walk(root_str, followlinks=False):
110 # Prune ignored and hidden directories in-place so os.walk skips them.
111 dirnames[:] = sorted(
112 d for d in dirnames
113 if d not in _IGNORE_DIRS and not d.startswith(".")
114 )
115
116 for fname in sorted(filenames):
117 suffix = pathlib.PurePosixPath(fname).suffix.lower()
118 if suffix not in _MARKDOWN_EXTS:
119 continue
120
121 note_count += 1
122 fpath = pathlib.Path(dirpath) / fname
123
124 try:
125 content = fpath.read_bytes()
126 except OSError:
127 continue
128
129 fm = parse_frontmatter(content)
130 if fm is None:
131 continue
132
133 src = fm.effective_source
134 if src:
135 source_types.add(src)
136
137 if fm.project:
138 projects.add(fm.project)
139
140 entities.update(fm.entity)
141 tags.update(fm.tags)
142
143 # Index freshness — mtime of data/ directory (Knowtation's index store).
144 index_freshness: str | None = None
145 data_dir = root / "data"
146 if data_dir.exists():
147 try:
148 mtime = data_dir.stat().st_mtime
149 dt = datetime.fromtimestamp(mtime, tz=timezone.utc)
150 index_freshness = dt.isoformat()
151 except OSError:
152 pass
153
154 return VaultStats(
155 note_count=note_count,
156 source_types=sorted(source_types),
157 projects=sorted(projects),
158 entity_count=len(entities),
159 tag_count=len(tags),
160 index_freshness=index_freshness,
161 )
File History 1 commit
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8 docs(KD-STAGING): sync governance after KD-6b DONE Human 12 days ago