Dmitry
22eda53d79
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
280 lines
8.1 KiB
Python
280 lines
8.1 KiB
Python
#!/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 []
|