ada-pc: 2026-06-16 20:18:31 | 9
Affected files: .obsidian/graph.json 99 System/INDEX.md 99 System/Tools/agent-tools/AGENT_TOOLS.md 99 System/Tools/agent-tools/vault_audit.py 99 System/Tools/agent-tools/vault_common.py 99 System/Tools/agent-tools/vault_frontmatter.py 99 System/Tools/agent-tools/vault_index.py 99 System/Tools/agent-tools/vault_ingest.py AGENTS.md
This commit is contained in:
@@ -0,0 +1,279 @@
|
||||
#!/usr/bin/env python3
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
from datetime import date, datetime
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[3]
|
||||
INDEX_PATH = ROOT / "99 System" / "INDEX.md"
|
||||
EXCLUDED_DIRS = {".git", ".trash", ".obsidian", ".smart-env"}
|
||||
REQUIRED_FRONTMATTER = {"title", "status", "type", "tags", "created", "updated"}
|
||||
VALID_STATUS = {"seed", "processing", "stable"}
|
||||
VALID_TYPE = {"concept", "moc", "lab", "knowledge", "guide"}
|
||||
|
||||
|
||||
try:
|
||||
import yaml
|
||||
except Exception: # pragma: no cover - depends on agent environment
|
||||
yaml = None
|
||||
|
||||
|
||||
def rel(path: Path) -> str:
|
||||
return path.relative_to(ROOT).as_posix()
|
||||
|
||||
|
||||
def json_print(data: Any) -> None:
|
||||
print(json.dumps(data, ensure_ascii=False, indent=2, sort_keys=True))
|
||||
|
||||
|
||||
def note_files(include_index: bool = False) -> list[Path]:
|
||||
files: list[Path] = []
|
||||
for path in ROOT.rglob("*.md"):
|
||||
parts = set(path.relative_to(ROOT).parts)
|
||||
if parts & EXCLUDED_DIRS:
|
||||
continue
|
||||
if not include_index and path == INDEX_PATH:
|
||||
continue
|
||||
files.append(path)
|
||||
return sorted(files, key=lambda p: rel(p).lower())
|
||||
|
||||
|
||||
def split_frontmatter(text: str) -> tuple[dict[str, Any], str, str | None]:
|
||||
lines = text.splitlines()
|
||||
if not lines or lines[0].strip() != "---":
|
||||
return {}, text, None
|
||||
|
||||
end = None
|
||||
for index, line in enumerate(lines[1:], start=1):
|
||||
if line.strip() == "---":
|
||||
end = index
|
||||
break
|
||||
|
||||
if end is None:
|
||||
return {}, text, None
|
||||
|
||||
raw = "\n".join(lines[1:end])
|
||||
body = "\n".join(lines[end + 1 :])
|
||||
return parse_frontmatter(raw), body, raw
|
||||
|
||||
|
||||
def parse_frontmatter(raw: str) -> dict[str, Any]:
|
||||
if yaml is not None:
|
||||
try:
|
||||
data = yaml.safe_load(raw) or {}
|
||||
return data if isinstance(data, dict) else {}
|
||||
except Exception:
|
||||
return {}
|
||||
|
||||
data: dict[str, Any] = {}
|
||||
current_key: str | None = None
|
||||
|
||||
for line in raw.splitlines():
|
||||
if not line.strip():
|
||||
continue
|
||||
|
||||
list_item = re.match(r"^\s*-\s+(.*)$", line)
|
||||
if list_item and current_key:
|
||||
data.setdefault(current_key, [])
|
||||
if isinstance(data[current_key], list):
|
||||
data[current_key].append(clean_scalar(list_item.group(1)))
|
||||
continue
|
||||
|
||||
pair = re.match(r"^([A-Za-zА-Яа-я0-9_-]+):\s*(.*)$", line)
|
||||
if not pair:
|
||||
continue
|
||||
|
||||
key, value = pair.group(1), pair.group(2).strip()
|
||||
current_key = key
|
||||
if value == "":
|
||||
data[key] = [] if key in {"aliases", "tags"} else ""
|
||||
elif value.startswith("[") and value.endswith("]"):
|
||||
data[key] = parse_inline_list(value)
|
||||
else:
|
||||
data[key] = clean_scalar(value)
|
||||
|
||||
return data
|
||||
|
||||
|
||||
def parse_inline_list(value: str) -> list[str]:
|
||||
inner = value.strip()[1:-1].strip()
|
||||
if not inner:
|
||||
return []
|
||||
return [clean_scalar(item.strip()) for item in inner.split(",") if item.strip()]
|
||||
|
||||
|
||||
def clean_scalar(value: Any) -> str:
|
||||
text = str(value).strip()
|
||||
if len(text) >= 2 and text[0] == text[-1] and text[0] in {"'", '"'}:
|
||||
return text[1:-1]
|
||||
return text
|
||||
|
||||
|
||||
def as_list(value: Any) -> list[str]:
|
||||
if value is None or value == "":
|
||||
return []
|
||||
if isinstance(value, list):
|
||||
return [clean_scalar(item) for item in value if clean_scalar(item)]
|
||||
return [clean_scalar(value)]
|
||||
|
||||
|
||||
def as_cell(value: Any, default: str = "—") -> str:
|
||||
if value is None or value == "":
|
||||
return default
|
||||
if isinstance(value, (date, datetime)):
|
||||
return value.isoformat()
|
||||
return str(value)
|
||||
|
||||
|
||||
def tags_cell(value: Any) -> str:
|
||||
tags = as_list(value)
|
||||
return "[" + ", ".join(tags) + "]"
|
||||
|
||||
|
||||
def note_record(path: Path) -> dict[str, Any]:
|
||||
text = path.read_text(encoding="utf-8")
|
||||
fm, body, raw_fm = split_frontmatter(text)
|
||||
return {
|
||||
"path": rel(path),
|
||||
"name": path.stem,
|
||||
"title": as_cell(fm.get("title"), ""),
|
||||
"aliases": as_list(fm.get("aliases")),
|
||||
"status": as_cell(fm.get("status")),
|
||||
"type": as_cell(fm.get("type")),
|
||||
"tags": as_list(fm.get("tags")),
|
||||
"created": as_cell(fm.get("created"), ""),
|
||||
"updated": as_cell(fm.get("updated"), ""),
|
||||
"frontmatter": fm,
|
||||
"has_frontmatter": raw_fm is not None,
|
||||
"body": body,
|
||||
}
|
||||
|
||||
|
||||
def all_records() -> list[dict[str, Any]]:
|
||||
return [note_record(path) for path in note_files()]
|
||||
|
||||
|
||||
def wikilinks(text: str) -> list[str]:
|
||||
text = strip_markdown_code(text)
|
||||
links = []
|
||||
for match in re.finditer(r"\[\[([^\]]+)\]\]", text):
|
||||
target = match.group(1).split("|", 1)[0].split("#", 1)[0].strip()
|
||||
if target:
|
||||
links.append(target)
|
||||
return links
|
||||
|
||||
|
||||
def strip_markdown_code(text: str) -> str:
|
||||
text = re.sub(r"```.*?```", "", text, flags=re.DOTALL)
|
||||
text = re.sub(r"`[^`\n]*`", "", text)
|
||||
return text
|
||||
|
||||
|
||||
def known_note_names(records: list[dict[str, Any]]) -> set[str]:
|
||||
names: set[str] = set()
|
||||
for record in records:
|
||||
names.add(record["name"])
|
||||
names.add(record["path"])
|
||||
names.update(record["aliases"])
|
||||
return names
|
||||
|
||||
|
||||
def tokenize(value: str) -> set[str]:
|
||||
return {token.lower() for token in re.findall(r"[A-Za-zА-Яа-я0-9_+-]+", value)}
|
||||
|
||||
|
||||
def search_records(records: list[dict[str, Any]], term: str, limit: int = 10) -> list[dict[str, Any]]:
|
||||
query = tokenize(term)
|
||||
if not query:
|
||||
return []
|
||||
|
||||
scored = []
|
||||
for record in records:
|
||||
haystack = " ".join(
|
||||
[
|
||||
record["name"],
|
||||
record["title"],
|
||||
" ".join(record["aliases"]),
|
||||
" ".join(record["tags"]),
|
||||
record["path"],
|
||||
]
|
||||
)
|
||||
tokens = tokenize(haystack)
|
||||
score = len(query & tokens)
|
||||
if term.lower() in haystack.lower():
|
||||
score += 3
|
||||
if score:
|
||||
scored.append((score, record))
|
||||
|
||||
scored.sort(key=lambda item: (-item[0], item[1]["path"]))
|
||||
return [compact_record(record) | {"score": score} for score, record in scored[:limit]]
|
||||
|
||||
|
||||
def compact_record(record: dict[str, Any]) -> dict[str, Any]:
|
||||
return {
|
||||
"path": record["path"],
|
||||
"name": record["name"],
|
||||
"title": record["title"],
|
||||
"aliases": record["aliases"],
|
||||
"status": record["status"],
|
||||
"type": record["type"],
|
||||
"tags": record["tags"],
|
||||
"created": record["created"],
|
||||
"updated": record["updated"],
|
||||
}
|
||||
|
||||
|
||||
def write_index(records: list[dict[str, Any]]) -> None:
|
||||
by_dir: dict[str, list[dict[str, Any]]] = {}
|
||||
for record in records:
|
||||
directory = str(Path(record["path"]).parent)
|
||||
if directory == ".":
|
||||
directory = "."
|
||||
by_dir.setdefault(directory, []).append(record)
|
||||
|
||||
total = len(records)
|
||||
lines = [
|
||||
"---",
|
||||
f"generated: {date.today().isoformat()}",
|
||||
f"total: {total}",
|
||||
"---",
|
||||
"",
|
||||
"# Vault Index",
|
||||
"",
|
||||
"## Статистика",
|
||||
"",
|
||||
"| Параметр | Значение |",
|
||||
"|----------|----------|",
|
||||
f"| Всего заметок | {total} |",
|
||||
"",
|
||||
]
|
||||
|
||||
for directory in sorted(by_dir, key=str.lower):
|
||||
lines.extend(
|
||||
[
|
||||
f"## {directory}",
|
||||
"",
|
||||
"| Заметка | Aliases | Status | Type | Tags |",
|
||||
"|---------|---------|--------|------|------|",
|
||||
]
|
||||
)
|
||||
for record in sorted(by_dir[directory], key=lambda item: item["name"].lower()):
|
||||
aliases = ", ".join(record["aliases"]) if record["aliases"] else "—"
|
||||
status = record["status"] or "—"
|
||||
note_type = record["type"] or "—"
|
||||
tags = "[" + ", ".join(record["tags"]) + "]"
|
||||
lines.append(f"| [[{record['name']}]] | {aliases} | {status} | {note_type} | {tags} |")
|
||||
lines.append("")
|
||||
|
||||
INDEX_PATH.write_text("\n".join(lines), encoding="utf-8")
|
||||
|
||||
|
||||
def source_values(value: Any) -> list[str]:
|
||||
if isinstance(value, list):
|
||||
return [clean_scalar(item) for item in value if clean_scalar(item)]
|
||||
if value:
|
||||
return [clean_scalar(value)]
|
||||
return []
|
||||
Reference in New Issue
Block a user