From 33ec9b5bc56c88c0ff6fd4a50356dee253134fa2 Mon Sep 17 00:00:00 2001 From: sam Date: Sat, 26 Sep 2026 06:53:24 +1000 Subject: [PATCH] obsidian-sorter: automated note filing for an Obsidian vault Classifies notes from an inbox folder and files them, with a human approval step. Zero third-party dependencies. - type mirrors folder (200 projects / 300 areas / 400 Personal Family) - TypeSafe JEV for semantic classification (~$0.0001 per note) - watcher writes a proposal INTO the note; the owner ticks a checkbox - systemd user service on .27, 10s poll - 15/15 functional tests --- .gitignore | 12 + README.md | 254 +++++++++++++++++++ docs/HANDOFF.md | 130 ++++++++++ sorter/__init__.py | 3 + sorter/__main__.py | 4 + sorter/classify.py | 289 ++++++++++++++++++++++ sorter/cli.py | 536 ++++++++++++++++++++++++++++++++++++++++ sorter/frontmatter.py | 213 ++++++++++++++++ sorter/jev.py | 250 +++++++++++++++++++ sorter/models.py | 140 +++++++++++ sorter/tier2.py | 418 +++++++++++++++++++++++++++++++ sorter/vault.py | 142 +++++++++++ sorter/watch.py | 260 +++++++++++++++++++ sorter/writer.py | 517 ++++++++++++++++++++++++++++++++++++++ tests/run_tests.py | 406 ++++++++++++++++++++++++++++++ tools/normalize_tags.py | 209 ++++++++++++++++ 16 files changed, 3783 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 docs/HANDOFF.md create mode 100644 sorter/__init__.py create mode 100644 sorter/__main__.py create mode 100644 sorter/classify.py create mode 100644 sorter/cli.py create mode 100644 sorter/frontmatter.py create mode 100644 sorter/jev.py create mode 100644 sorter/models.py create mode 100644 sorter/tier2.py create mode 100644 sorter/vault.py create mode 100644 sorter/watch.py create mode 100644 sorter/writer.py create mode 100644 tests/run_tests.py create mode 100644 tools/normalize_tags.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..232ed1e --- /dev/null +++ b/.gitignore @@ -0,0 +1,12 @@ +# development mirror of the live vault — never commit +test-vault/ + +# python +__pycache__/ +*.py[cod] +.venv/ +venv/ + +# local +*.log +.DS_Store diff --git a/README.md b/README.md new file mode 100644 index 0000000..0df3c5d --- /dev/null +++ b/README.md @@ -0,0 +1,254 @@ +# obsidian-sorter + +Automated note classification and filing for an Obsidian vault. + +Notes enter through `010 inbox/`, the sorter decides where they belong, and — +once approved — writes them into place and updates the Table of Contents. + +**Design goal: no dependencies.** The vault uses simple, uniform YAML; the tool +ships a small purpose-built parser so it deploys onto hosts without pip +(NixOS `.13`, `.27`, `.51`). + +--- + +## The taxonomy (simplified 2026-09-24) + +**The type of a note mirrors the folder it lives in.** Nothing is guessed from +content — the destination decides the type. + +| Folder | Type | Meaning | +|--------|------|---------| +| `200 projects/` | `project` | Work in progress — something being built | +| `300 areas/` | `area` | Ongoing interest with documentation and notes | +| `400 Personal Family/` | `family` | Family / non-work | +| `100 Table of Contents/` | `index` | Navigation notes only | +| `000 daily/` | `daily` | Chronological daily notes | +| `500 archive/` | `archive` | Inactive material | +| `Clippings/` | `clipping` | Browser-extension captures | +| `010 inbox/` | `note` | Unsorted placeholder | + +Everything else has been **retired**: `subject`, `domain`, `tool`, `service`, +`container`, `machine`, `storage`, `backup-system`, `network-component`, +`website`, `person`. Do not reintroduce them. + +### The lifecycle + +``` +being built -> 200 projects/... -> project +built and running-> 300 areas/... -> area +``` + +A note moves between the two when the work finishes. Its `areas:` field records +which area it belongs to while it is still a project. + +--- + +## Status + +| Phase | Scope | State | +|-------|-------|-------| +| **A** | Vault scanner + deterministic classifier | ✅ | +| **B** | JEV (TypeSafe) semantic tier | ✅ | +| **C** | Writer: frontmatter, move, TOC insert, approval gate | ✅ | +| D | MCP tool wrapper | pending | +| E | Pi skill + Gitea distribution | pending | +| F | Validation command | pending | + +--- + +## Install + +Nothing to install — pure standard library, Python 3.10+. + +Set the API key for the semantic tier: + +```bash +export TYPESAFE_API_KEY=... # required for --jev / benchmark +``` + +--- + +## The workflow (how a note gets filed) + +The watcher runs as a **systemd user service on `.27`**, always on. + +``` +you save a note into 010 inbox/ type: note + | + v watcher notices within ~10s + it classifies the note and WRITES THE PROPOSAL INTO THE NOTE + + type: note + proposed_folder: 200 projects/250 Electronics + proposed_toc: 100 Table of Contents/Projects.md → Ideas + proposed_tags: [solar, 12v, bird, bath] + confidence: 0.69 + approved: false <-- a checkbox in Obsidian + | + v you open the note and tick the box (or edit the proposal first) + approved: true + | + v watcher notices within ~10s + it files the note, strips the admin fields, updates the TOC + | + v +you see the note appear in the right folder, listed in the index +``` + +**Nothing is ever filed without your approval.** You can: +- tick `approved` to accept as-is +- edit `proposed_folder` (or `proposed_tags`) first, then tick it +- leave it alone — it waits in the inbox indefinitely + +Admin fields (`proposed_folder`, `proposed_toc`, `proposed_tags`, `confidence`, +`approved`) are removed when the note is filed, so the finished note is clean. + +A running log is written to `010 inbox/_sort log.md`. + +### Managing the watcher + +```bash +systemctl --user status obsidian-sorter # is it running? +systemctl --user stop obsidian-sorter # pause it +systemctl --user start obsidian-sorter # resume +systemctl --user disable obsidian-sorter # don't start on boot +journalctl --user -u obsidian-sorter -f # live output +``` + +Logs: `~/.local/state/obsidian-sorter.log` + +--- + +## Manual use + +`obsidian-sort` is installed on `.27` (in `~/bin`): + +```bash +obsidian-sort # show what would happen (dry run, no writes) +obsidian-sort apply # file the notes, asks to confirm +obsidian-sort apply --yes # file without asking +obsidian-sort scan # vault summary +``` + +Or call the module directly from anywhere the code is deployed: + +```bash +python3 -m sorter --vault ~/obsidian --inbox "010 inbox" watch --once +python3 -m sorter --vault ~/obsidian --inbox "010 inbox" sort-inbox --jev --apply +``` + +--- + +## Usage reference + +### Read-only + +```bash +# Vault shape: counts, types, what needs classification +python3 -m sorter --vault ~/obsidian scan + +# Classify one note (rules only, free) +python3 -m sorter --vault ~/obsidian propose "010 inbox/My Note.md" + +# Classify with the semantic tier +python3 -m sorter --vault ~/obsidian --inbox "010 inbox" \ + propose "010 inbox/My Note.md" --jev +``` + +### Filing (writes) + +Dry run by default — nothing is written without `--apply`. + +```bash +# Show the plan +python3 -m sorter --vault ~/obsidian --inbox "010 inbox" \ + sort "010 inbox/My Note.md" --jev + +# Apply it +python3 -m sorter --vault ~/obsidian --inbox "010 inbox" \ + sort "010 inbox/My Note.md" --jev --apply + +# Whole inbox, no prompt (for agents) +python3 -m sorter --vault ~/obsidian --inbox "010 inbox" \ + sort-inbox --jev --apply --yes +``` + +The plan lists every change explicitly: + +``` +Plan for: 010 inbox/50 Legendary Websites.md + + → 300 areas/390 Web Pages and Tools To Try/50 Legendary Websites.md + title : 50 Legendary Websites + type : area status: active + toc : 100 Table of Contents/Areas.md → Web Pages & Tools To Try + + changes: + + create 300 areas/390 Web Pages and Tools To Try/50 Legendary Websites.md + ~ update 100 Table of Contents/Areas.md (add [[50 Legendary Websites]] under 'Web Pages & Tools To Try') + - remove 010 inbox/50 Legendary Websites.md (moved out of inbox) +``` + +### Measuring the semantic tier + +```bash +python3 -m sorter --vault ~/obsidian --inbox "010 inbox" benchmark --all --quiet +``` + +--- + +## Safety guarantees (all tested) + +| Guarantee | Behaviour | +|-----------|-----------| +| Dry run by default | `--apply` required to write anything | +| `Untitled*` never moved | Plan marked BLOCKED, exit code 2, file untouched | +| Idempotent TOC | Re-running never duplicates an index entry | +| Never overwrites | An existing destination blocks the plan | +| Frontmatter merged | Existing `tags`, `aliases`, `id` survive; `created` preserved | +| Body untouched | The author's words are never rewritten | +| Atomic writes | temp file + `os.replace` — no partial files | +| `.trash/` off limits | Files under .trash are never removed | + +--- + +## Testing + +Runs against an **isolated copy** — the mirror is never modified. + +```bash +python3 tests/run_tests.py # full suite (uses JEV) +python3 tests/run_tests.py --no-jev # mechanics only, no API cost +python3 tests/run_tests.py --keep # keep the temp vault for inspection +``` + +Covers: dry-run safety, all three buckets routing to the correct sub-folder, +frontmatter merging, the `Untitled` guard, destination collisions, `.trash` +protection, inbox clearing, idempotency, and duplicate index entries. + +--- + +## Development + +```bash +# Read-only mirror of the live vault for testing +rsync -a --delete \ + --exclude='.git' --exclude='.trash' --exclude='.venv' \ + --exclude='graphify-out*' --exclude='.obsidian' \ + sam@192.168.20.27:~/obsidian/ ./test-vault/ +``` + +`test-vault/` is a throwaway mirror and is gitignored. + +### Layout + +| File | Purpose | +|------|---------| +| `sorter/frontmatter.py` | Minimal YAML reader/writer (no PyYAML) | +| `sorter/models.py` | Types, `Note`, `Proposal`, folder→type mapping | +| `sorter/vault.py` | Vault scanning, index introspection | +| `sorter/classify.py` | Tier 1 rules | +| `sorter/jev.py` | TypeSafe JEV client + cost tracking | +| `sorter/tier2.py` | Tier 2 semantic classifier + hybrid | +| `sorter/writer.py` | Plans, frontmatter merge, TOC insert, apply | +| `sorter/cli.py` | Command line | diff --git a/docs/HANDOFF.md b/docs/HANDOFF.md new file mode 100644 index 0000000..5a191da --- /dev/null +++ b/docs/HANDOFF.md @@ -0,0 +1,130 @@ +# HANDOFF — Obsidian Sorter + +> Written 2026-09-24. Pick this up cold; everything you need is here. + +## What this is + +An automated system that files notes into Sam's Obsidian vault. + +Notes go into `010 inbox/`. A **watcher** proposes a destination by writing it +*into the note's frontmatter*. Sam ticks a checkbox in Obsidian to approve. +The watcher then files it and updates the Table of Contents. + +**Nothing is ever filed without approval** — deliberately, so Sam can edit a note +before it moves. + +## Where things live + +| Thing | Location | +|-------|----------| +| Vault (live) | `.27` → `~/obsidian` — git-synced to Gitea `sam/obsidian-vault` | +| Sorter code (dev) | `.13` → `~/paseo/projects/obsidian-sorter/` | +| Sorter code (deployed) | `.27` → `~/obsidian-sorter/` | +| Watcher service | `.27` → `systemctl --user obsidian-sorter` (enabled + lingering) | +| Command | `.27` → `~/bin/obsidian-sort` | +| Sort log | `.27` → `~/obsidian/010 inbox/_sort log.md` | +| Service log | `.27` → `~/.local/state/obsidian-sorter.log` | +| Design | `~/paseo/projects/obsidian-cleanup-tool/SORTER-PLAN.md` | + +## THE TAXONOMY — final, do not "improve" it + +**Type mirrors folder.** Nothing is inferred from content. + +| Folder | Type | +|--------|------| +| `200 projects/` | `project` | +| `300 areas/` | `area` | +| `400 Personal Family/` | `family` | +| `100 Table of Contents/` | `index` | +| `000 daily/` | `daily` | +| `500 archive/` | `archive` | +| `Clippings/` | `clipping` | +| `010 inbox/` | `note` (unsorted placeholder only) | + +**RETIRED — never reintroduce:** `subject`, `domain`, `tool`, `service`, +`container`, `machine`, `storage`, `backup-system`, `network-component`, +`website`, `person`. + +Sam's lifecycle model: +``` +IDEA (not started) -> 200 projects/260 Ideas to Build Create Make +BEING BUILT -> 200 projects/250 Electronics +FINISHED -> 300 areas/... +``` + +## The workflow + +1. Note lands in `010 inbox/` with `type: note` +2. Watcher (polls every 10s) classifies it and **writes the proposal into the note**: + ```yaml + proposed_folder: 200 projects/250 Electronics + proposed_toc: 100 Table of Contents/Projects.md → Ideas + proposed_tags: [solar, 12v, bird, bath] + confidence: 0.69 + approved: false # a checkbox in Obsidian + ``` +3. Sam ticks `approved` (or edits the proposal first) +4. Watcher files it: move, set type from folder, merge tags, **strip admin + fields**, update the TOC + +## Implementation notes + +- **Zero dependencies.** Custom YAML parser (`sorter/frontmatter.py`) — no PyYAML, + because NixOS has no pip. HTTP via stdlib `urllib`. +- **JEV** (TypeSafe, `POST api.typesafe.ai/v1/systemone`) does semantic + classification — ~$0.0001/note. Key read from `~/.config/environment.d/*.conf` + (not in the env for non-login SSH shells). +- **Classifies by FOLDER, not type.** JEV picks the folder; the type follows. +- **Sam's own tags win.** A note tagged `project`/`family` is constrained to that + bucket — JEV cannot override it (`explicit_bucket_from_tags`). +- **Folder meanings are taught explicitly** in `FOLDER_NOTES` (tier2.py), e.g. + *"260 Ideas = UNSTARTED; 250 Electronics = ACTIVELY BEING BUILT"*. + **This is the place to fix any future misfiling.** +- JEV is **probabilistic** (~74% folder accuracy). Tests assert the *bucket* + strictly, the sub-folder only with `--strict-folders`. +- Files starting with `_` are ignored by the inbox scanner, so `_sort log.md` + and `_review` files never get sorted. + +## Safety guarantees (15/15 tests) + +Dry run by default · `Untitled*` never moved · never overwrites an existing +destination · frontmatter merged (user `tags`/`aliases`/`id`/`created` survive) · +body never rewritten · atomic writes · `.trash/` untouched · idempotent TOC. + +## Commands + +```bash +# on .27 +obsidian-sort # dry run — show what would happen +obsidian-sort apply # file the notes (asks to confirm) +obsidian-sort scan # vault summary +systemctl --user status obsidian-sorter +journalctl --user -u obsidian-sorter -f + +# tests (on .13, against the mirror) +cd ~/paseo/projects/obsidian-sorter +python3 tests/run_tests.py +python3 tests/run_tests.py --strict-folders # may flake (JEV variance) + +# refresh the local mirror from the live vault +rsync -a --delete --exclude='.git' --exclude='.trash' --exclude='.venv' \ + --exclude='graphify-out*' --exclude='.obsidian' \ + sam@192.168.20.27:~/obsidian/ ./test-vault/ + +# deploy code to .27 +rsync -a --delete --exclude='test-vault' --exclude='__pycache__' \ + --exclude='.git' --exclude='*.pyc' ./ sam@192.168.20.27:~/obsidian-sorter/ +``` + +## Tag vocabulary + +Cleaned 2026-09-24: 214 → 195 unique tags; lowercased everything, merged +`tools`→`tool`, split one malformed comma-joined tag into 9, fixed typos, +removed all retired-type tags. Tool: `tools/normalize_tags.py`. + +## Remaining work + +- **Phase D** — MCP tool so Pi agents can call the sorter +- **Phase E** — Pi skill ("sort my inbox") + distribute via Gitea to .13/.27/.51 +- Add a watcher test to the suite (currently only verified manually) +- Sam is still testing with real notes — **no Pi integration until he's satisfied** diff --git a/sorter/__init__.py b/sorter/__init__.py new file mode 100644 index 0000000..7bc4983 --- /dev/null +++ b/sorter/__init__.py @@ -0,0 +1,3 @@ +"""obsidian-sorter — automated note classification and insertion for Obsidian vaults.""" + +__version__ = "0.1.0" diff --git a/sorter/__main__.py b/sorter/__main__.py new file mode 100644 index 0000000..bfdcd0c --- /dev/null +++ b/sorter/__main__.py @@ -0,0 +1,4 @@ +from .cli import main + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/sorter/classify.py b/sorter/classify.py new file mode 100644 index 0000000..07067dd --- /dev/null +++ b/sorter/classify.py @@ -0,0 +1,289 @@ +""" +Tier 1 deterministic classifier. + +Simplified 2026-09-24. The rule is now trivial: + + type mirrors folder + + 200 projects/ -> project + 300 areas/ -> area + 400 Personal Family/ -> family + +So the classifier's only real job is to choose a DESTINATION FOLDER. The type +then falls out of the folder automatically — nothing is inferred from content +beyond a weak first guess. + +Because picking the right folder is genuinely hard, the rules here are +deliberately conservative: when they are not confident, the note escalates to +Tier 2 (JEV). +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass + +from .models import ( + BUCKETS, + Note, + Proposal, + PLACEHOLDER_TYPE, + VALID_TYPES, + type_from_folder, +) +from .vault import Vault + +# --------------------------------------------------------------- TOC + templates + +# Which index note lists each type. +TYPE_TO_TOC: dict[str, str] = { + "project": "100 Table of Contents/Projects.md", + "area": "100 Table of Contents/Areas.md", + "family": "100 Table of Contents/Personal Family.md", + "index": "100 Table of Contents/Table of Contents.md", +} + +# The template a type should use. +TYPE_TO_TEMPLATE: dict[str, str] = { + "project": "Project.md", + "area": "Area.md", + "family": "Family.md", + "index": "Index.md", +} + +# ------------------------------------------------------------------- hints + +# Keywords that weakly suggest a bucket. Used only as a first guess; JEV does +# the real work. Order of checks: family, then project, else area. +FAMILY_HINTS = [ + "recipe", "birthday", "holiday", "family", "movie", "film", "book", + "game", "travel", "ski", "camp", "garden", "aquaponic", "art", "music", + "photography", "investment", "business idea", "school", "basketball", +] +PROJECT_HINTS = [ + "phase", "milestone", "build plan", "roadmap", "work log", "checklist", + "deliverable", "to do", "todo", "needs to be built", "not built", + "original plan was", "next steps", "in progress", +] + +STOPWORDS = { + "the", "and", "for", "with", "from", "this", "that", "into", "your", + "notes", "note", "about", "overview", "setup", "guide", "using", "new", +} + + +def _tokens(text: str) -> set[str]: + return { + w for w in re.findall(r"[a-z0-9]+", text.lower()) + if w not in STOPWORDS and len(w) > 2 + } + + +def explicit_bucket_from_tags(note: "Note") -> str | None: + """ + If the author tagged the note with a bucket name, honour it. + + Your own tag is the strongest signal available — a human already decided. + """ + tags = {t.lower() for t in (note.tags or [])} + if "project" in tags: + return "project" + if tags & {"family", "personal"}: + return "family" + return None + + +def explicit_bucket_from_tags(note: "Note") -> str | None: + """ + If the author tagged the note with a bucket name, honour it. + + Your own tag is the strongest signal available — a human already decided. + """ + tags = {t.lower() for t in (note.tags or [])} + if "project" in tags: + return "project" + if tags & {"family", "personal"}: + return "family" + return None + + +# A tag can pin a note to a specific folder — but ONLY when the tag names a +# clear CATEGORY (where the note belongs), never when it describes the note's +# nature. 'recipe' is a category; 'business-idea' is a property, so it must not +# pin anything — an electronics business idea still belongs in projects. +TAG_FOLDER_HINTS: dict[str, str] = { + "recipe": "400 Personal Family/450 Recipes", + "holiday": "400 Personal Family/470 Holidays Travel", + "travel": "400 Personal Family/470 Holidays Travel", + "ski": "400 Personal Family/470 Holidays Travel", + "film": "400 Personal Family/430 Movies Books Plays Games", +} + + +def explicit_folder_from_tags(note: "Note") -> str | None: + """Return a folder pinned by one of the note's tags, if any.""" + tags = {t.lower() for t in (note.tags or [])} + for tag, folder in TAG_FOLDER_HINTS.items(): + if tag in tags: + return folder + return None + + +@dataclass +class Guess: + bucket: str + score: float + reason: str + + +class Classifier: + """Deterministic (Tier 1) folder classifier.""" + + def __init__(self, vault: Vault): + self.vault = vault + self._folders: list[str] | None = None + + @property + def folders(self) -> list[str]: + if self._folders is None: + self._folders = self.vault.folders() + return self._folders + + # ------------------------------------------------------------------ public + + def classify(self, note: Note) -> Proposal: + """Produce a Proposal for a note. Never writes anything.""" + # --- guard: untitled notes are never moved ------------------------- + if note.stem.lower().startswith("untitled"): + return Proposal( + source_rel=note.rel_path, note_type="", target_folder="", + toc_section="", template="", confidence=0.0, method="blocked", + reasons=["Filename begins with 'Untitled'"], + warnings=[ + "BLOCKED: untitled notes are never moved automatically. " + "Rename it first if it is real content." + ], + ) + + current = (note.note_type or "").strip().lower() + expected = note.expected_type + + # --- a tag can pin the folder outright ----------------------------- + pinned = explicit_folder_from_tags(note) + if pinned and (note.folder != pinned or current != type_from_folder(pinned)): + return Proposal( + source_rel=note.rel_path, + note_type=type_from_folder(pinned) or "family", + target_folder=pinned, + toc_section=TYPE_TO_TOC.get(type_from_folder(pinned) or "family", ""), + template=TYPE_TO_TEMPLATE.get(type_from_folder(pinned) or "family", "Family.md"), + confidence=0.95, + method="deterministic", + reasons=[f"your tag pinned this to {pinned}"], + ) + + # --- already correctly filed? -------------------------------------- + if expected and current == expected: + toc = TYPE_TO_TOC.get(current, "") + return Proposal( + source_rel=note.rel_path, note_type=current, + target_folder=note.folder, + toc_section=toc, template=TYPE_TO_TEMPLATE.get(current, "Area.md"), + confidence=1.0, method="trusted", + reasons=[f"type matches its folder ({expected})"], + ) + + # --- weak first guess ---------------------------------------------- + guess = self._guess_bucket(note) + if guess is None: + return Proposal( + source_rel=note.rel_path, note_type="", target_folder="", + toc_section="", template="", confidence=0.0, method="none", + reasons=["No deterministic match"], + warnings=["No rule matched — requires Tier 2 (JEV)"], + ) + + folder, matched = self._pick_folder(note, guess.bucket) + toc = TYPE_TO_TOC.get(guess.bucket, "") + + # A blind folder guess (no keyword evidence) must NOT look confident, or + # the note never escalates to JEV and lands in an arbitrary folder. + confidence = guess.score if matched else min(guess.score, 0.25) + + warnings: list[str] = [] + if not matched: + warnings.append("No keyword evidence for a folder — likely needs Tier 2 (JEV)") + elif confidence < 0.6: + warnings.append("Low confidence — likely needs Tier 2 (JEV)") + + reasons = [guess.reason] + if matched: + reasons.append(f"folder matched by keyword: {folder}") + + return Proposal( + source_rel=note.rel_path, + note_type=guess.bucket, + target_folder=folder, + toc_section=toc, + template=TYPE_TO_TEMPLATE.get(guess.bucket, "Area.md"), + confidence=confidence, + method="deterministic", + reasons=reasons, + warnings=warnings, + ) + + # --------------------------------------------------------------- internals + + def _guess_bucket(self, note: Note) -> Guess | None: + # Your own tags win: if you tagged it 'project' or 'family', believe you. + tagged = explicit_bucket_from_tags(note) + if tagged: + return Guess(tagged, 0.9, f"you tagged this note '{tagged}'") + + text = " ".join( + [(note.title or note.stem).lower(), " ".join(t.lower() for t in note.tags)] + ).lower() + + fam = sum(1 for k in FAMILY_HINTS if k in text) + proj = sum(1 for k in PROJECT_HINTS if k in text) + + # An explicit `type: note` says nothing useful; rely on keywords. + if current := (note.note_type or "").strip().lower(): + if current in BUCKETS and current != PLACEHOLDER_TYPE: + return Guess(current, 0.5, f"existing type {current}") + + if fam and fam >= proj: + return Guess("family", min(0.7, 0.35 + 0.15 * fam), + f"{fam} family keyword(s) matched") + if proj: + return Guess("project", min(0.7, 0.35 + 0.15 * proj), + f"{proj} project keyword(s) matched") + + # Default bucket is area — the catch-all for ongoing material. + return Guess("area", 0.3, "defaulted to area (no strong signal)") + + def _pick_folder(self, note: Note, bucket: str) -> tuple[str, bool]: + """ + Choose a folder inside the bucket. + + Returns (folder, matched) where `matched` is False when there was no + keyword evidence and the choice is effectively arbitrary. + """ + candidates = [ + f for f in self.folders + if type_from_folder(f) == bucket + ] + if not candidates: + return "", False + + target = _tokens((note.title or note.stem) + " " + " ".join(note.tags)) + best, best_score = None, 0 + for cand in candidates: + score = len(target & _tokens(cand)) + if score > best_score or (score == best_score and best is None): + best, best_score = cand, score + + if best_score == 0: + # No evidence — return a candidate but flag it as unmatched. + return candidates[0], False + return best or candidates[0], True diff --git a/sorter/cli.py b/sorter/cli.py new file mode 100644 index 0000000..5484423 --- /dev/null +++ b/sorter/cli.py @@ -0,0 +1,536 @@ +"""Command-line entry point for obsidian-sorter. + +Phase A commands (read-only — nothing here writes to the vault): + + scan report vault shape + propose classify one note, print a proposal + inbox classify every note in the inbox + validate replay the classifier against known-good notes +""" + +from __future__ import annotations + +import argparse +import json +import sys +from dataclasses import replace + +from .classify import Classifier +from .jev import CostTracker, JevClient, JevError +from .models import Note +from .tier2 import HybridClassifier, JevClassifier +from .vault import Vault +from .watch import Watcher +from .writer import Writer + + +def _load_vault(args) -> Vault: + return Vault(args.vault) + + +def _print_proposal(p) -> None: + if p.method == "blocked": + print(f" ⛔ BLOCKED — {p.source_rel}") + for w in p.warnings: + print(f" {w}") + return + + print(f" Proposal for: {p.source_rel}") + print(f" type : {p.note_type or '(none)'}") + print(f" target folder : {p.target_folder or '(unknown)'}") + print(f" toc : {p.toc_section or '(none)'}") + print(f" template : {p.template or '(none)'}") + print(f" confidence : {p.confidence:.2f} [{p.method}]") + if p.reasons: + print(f" why : {'; '.join(p.reasons)}") + if p.alternatives: + alts = ", ".join(f"{t}={s:.2f}" for t, s in p.alternatives) + print(f" alternatives : {alts}") + for w in p.warnings: + print(f" ⚠ {w}") + + +# --------------------------------------------------------------------- commands + +def cmd_scan(args) -> int: + vault = _load_vault(args) + notes = vault.scan() + print(f"Vault : {vault.root}") + print(f"Notes : {len(notes)}") + print(f"Folders : {len(vault.folders())}") + + types: dict[str, int] = {} + for n in notes: + key = (n.note_type or "(none)").lower() + types[key] = types.get(key, 0) + 1 + print("Types :") + for t, c in sorted(types.items(), key=lambda kv: -kv[1]): + print(f" {t:20s} {c}") + + needing = [n for n in notes if n.needs_classification] + print(f"\nNeeding classification (no trusted type): {len(needing)}") + for n in needing[:20]: + print(f" - {n.rel_path}") + if len(needing) > 20: + print(f" ... +{len(needing) - 20} more") + + inbox = vault.inbox_notes(args.inbox) + print(f"\nInbox ({args.inbox}): {len(inbox)} note(s)") + for n in inbox: + print(f" - {n.filename}") + return 0 + + +def cmd_propose(args) -> int: + vault = _load_vault(args) + + if args.jev: + jev, tracker = _make_jev(args) + clf = HybridClassifier(vault, jev=jev, threshold=args.threshold) + else: + clf = Classifier(vault) + tracker = None + + from pathlib import Path + path = Path(args.file).expanduser() + if not path.is_absolute(): + candidate = vault.root / args.file + path = candidate if candidate.exists() else path + + note = vault.read_note(path) + if note is None: + print(f"error: cannot read note: {path}", file=sys.stderr) + return 1 + + p = clf.classify(note) + _print_proposal(p) + if tracker is not None: + s = tracker.summary() + print(f" cost : ${s['cost_usd']:.6f} ({s['input_tokens']} in tokens)") + if args.json: + print(json.dumps(p.as_dict(), indent=2)) + return 0 + + +def cmd_inbox(args) -> int: + vault = _load_vault(args) + + if getattr(args, "jev", False): + jev, tracker = _make_jev(args) + clf = HybridClassifier(vault, jev=jev, threshold=args.threshold) + else: + clf = Classifier(vault) + tracker = None + + notes = vault.inbox_notes(args.inbox) + + if not notes: + print(f"Inbox is empty ({args.inbox}).") + return 0 + + print(f"Found {len(notes)} note(s) in {args.inbox}:\n") + proposals = [] + for note in notes: + p = clf.classify(note) + proposals.append(p) + _print_proposal(p) + print() + + if tracker is not None: + s = tracker.summary() + print(f"cost: ${s['cost_usd']:.6f} ({s['input_tokens']} in tokens, {s['calls']} calls)") + + if args.json: + print(json.dumps([p.as_dict() for p in proposals], indent=2)) + return 0 + + +def cmd_validate(args) -> int: + """ + Replay the classifier against notes that already have a known-good type + (established during the Phase 3 cleanup) and measure agreement. + + Two modes: + --in-folder keep the note's real folder (folder evidence available) + --as-inbox pretend the note lives in the inbox (no folder evidence) + """ + vault = _load_vault(args) + clf = Classifier(vault) + notes = vault.scan() + + ground_truth = [ + n for n in notes + if (n.note_type or "").strip().lower() in { + "project", "area", "tool", "subject", "backup-system", + "container", "machine", "website", "service", + "network-component", "person", "storage", + } + ] + + correct = 0 + total = 0 + misses: list[tuple[str, str, str, float]] = [] + low_conf = 0 + + for note in ground_truth: + expected = (note.note_type or "").strip().lower() + + # Simulate an UNSORTED note: strip the known-good type/status so the + # classifier must work from evidence rather than echoing it back. + if args.in_folder: + probe = replace(note, note_type=None, status=None) + else: + probe = replace( + note, note_type=None, status=None, + rel_path=f"{args.inbox}/{note.filename}", + ) + + p = clf.classify(probe) + if p.method in ("blocked", "none"): + total += 1 + misses.append((note.rel_path, expected, p.note_type or "(none)", p.confidence)) + continue + + total += 1 + if p.note_type == expected: + correct += 1 + else: + misses.append((note.rel_path, expected, p.note_type, p.confidence)) + if p.confidence < 0.45: + low_conf += 1 + + acc = (correct / total * 100) if total else 0.0 + mode = "IN-FOLDER (folder evidence available)" if args.in_folder else "AS-INBOX (no folder evidence)" + print(f"=== Validation — {mode} ===") + print(f"Notes tested : {total}") + print(f"Correct : {correct}") + print(f"Accuracy : {acc:.1f}%") + print(f"Low-conf : {low_conf} (would escalate to JEV)") + + if misses: + print(f"\nMisses ({len(misses)}):") + for rel, exp, got, conf in sorted(misses, key=lambda m: -m[3]): + print(f" expected={exp:16s} got={got:16s} conf={conf:.2f} {rel}") + return 0 + + +def _make_jev(args): + """Build a JEV classifier, or None when unavailable.""" + try: + client = JevClient(timeout=args.timeout) + except JevError as e: + print(f"warning: JEV unavailable ({e})", file=sys.stderr) + return None, None + vault = Vault(args.vault) + tracker = CostTracker(max_calls_per_run=args.max_calls) + jev = JevClassifier(vault, client, tracker=tracker) + return jev, tracker + + +def cmd_benchmark(args) -> int: + """Run the hybrid classifier against notes with known-good types. + + Reports accuracy and real token cost. JEV calls cost money — keep --limit sane. + """ + vault = _load_vault(args) + jev, tracker = _make_jev(args) + if jev is None: + return 1 + + clf = HybridClassifier(vault, jev=jev, threshold=args.threshold) + + ground_truth = [ + n for n in vault.scan() + if (n.note_type or "").strip().lower() in { + "project", "area", "tool", "subject", "backup-system", + "container", "machine", "website", "service", + "network-component", "person", "storage", + } + ] + + if not args.all: + ground_truth = ground_truth[: args.limit] + + correct = 0 + folder_correct = 0 + misses: list[tuple[str, str, str, float, str]] = [] + + for i, note in enumerate(ground_truth, 1): + expected = (note.note_type or "").strip().lower() + # Simulate an unsorted inbox note. + probe = replace( + note, note_type=None, status=None, + rel_path=f"{args.inbox}/{note.filename}", + ) + + if tracker.budget_exceeded(): + print(f"\nBudget cap reached ({args.max_calls} calls) — stopping.") + break + + p = clf.classify(probe) + + if p.note_type == expected: + correct += 1 + else: + misses.append((note.rel_path, expected, p.note_type, p.confidence, p.method)) + + if p.target_folder and note.folder and p.target_folder == note.folder: + folder_correct += 1 + + if not args.quiet: + mark = "✓" if p.note_type == expected else "✗" + print(f" [{i:>3}/{len(ground_truth)}] {mark} {expected:14s} → {p.note_type:14s} " + f"({p.confidence:.2f}, {p.method})") + + tested = correct + len(misses) + acc = (correct / tested * 100) if tested else 0.0 + facc = (folder_correct / tested * 100) if tested else 0.0 + + print(f"\n=== JEV Benchmark ===") + print(f"Notes tested : {tested}") + print(f"Type accuracy : {acc:.1f}% ({correct}/{tested})") + print(f"Folder match : {facc:.1f}% ({folder_correct}/{tested})") + s = tracker.summary() + print(f"JEV calls : {s['calls']} ({s['errors']} errors)") + print(f"Tokens : {s['input_tokens']} in / {s['output_tokens']} out") + print(f"Cost : ${s['cost_usd']:.6f}") + if tested: + print(f"Cost / note : ${s['cost_usd'] / tested:.6f}") + + if misses: + print(f"\nMisses ({len(misses)}):") + for rel, exp, got, conf, method in misses: + print(f" expected={exp:14s} got={got:14s} conf={conf:.2f} [{method}] {rel[:70]}") + return 0 + + +def _make_classifier(args, vault): + """Rules-only or hybrid, depending on --jev. Returns (classifier, tracker).""" + if getattr(args, "jev", False): + jev, tracker = _make_jev(args) + return HybridClassifier(vault, jev=jev, threshold=args.threshold), tracker + return Classifier(vault), None + + +def _confirm(prompt: str) -> bool: + try: + return input(f"{prompt} [y/N] ").strip().lower() in ("y", "yes") + except EOFError: + return False + + +def cmd_sort(args) -> int: + """Plan (and optionally apply) sorting of one note.""" + vault = _load_vault(args) + clf, tracker = _make_classifier(args, vault) + writer = Writer(vault) + + from pathlib import Path + path = Path(args.file).expanduser() + if not path.is_absolute(): + candidate = vault.root / args.file + path = candidate if candidate.exists() else path + + note = vault.read_note(path) + if note is None: + print(f"error: cannot read note: {path}", file=sys.stderr) + return 1 + + proposal = clf.classify(note) + plan = writer.plan(note, proposal) + print(plan.render()) + + if tracker is not None: + s = tracker.summary() + print(f"\n cost: ${s['cost_usd']:.6f} ({s['input_tokens']} in tokens)") + + if not args.apply: + print("\n(dry run — pass --apply to write)") + return 0 + + if not plan.writable: + print(f"\nrefusing to apply: {plan.blocked_reason}", file=sys.stderr) + return 2 + + if not args.yes and not _confirm("\nApply these changes?"): + print("aborted.") + return 1 + + ok, log = writer.apply(plan, source_path=note.path) + print() + for line in log: + print(f" {line}") + return 0 if ok else 3 + + +def cmd_sort_inbox(args) -> int: + """Plan (and optionally apply) every note in the inbox.""" + vault = _load_vault(args) + clf, tracker = _make_classifier(args, vault) + writer = Writer(vault) + + notes = vault.inbox_notes(args.inbox) + if not notes: + print(f"Inbox is empty ({args.inbox}).") + return 0 + + plans = [] + for note in notes: + proposal = clf.classify(note) + plan = writer.plan(note, proposal) + plans.append((note, plan)) + + print(f"Found {len(plans)} note(s) in {args.inbox}:\n") + for _note, plan in plans: + print(plan.render()) + print() + + if tracker is not None: + s = tracker.summary() + print(f" cost: ${s['cost_usd']:.6f} ({s['input_tokens']} in tokens)\n") + + if not args.apply: + print("(dry run — pass --apply to write)") + return 0 + + writable = [(n, p) for n, p in plans if p.writable] + blocked = [(n, p) for n, p in plans if not p.writable] + + for _n, p in blocked: + print(f"skipping (blocked): {p.source_rel} — {p.blocked_reason}", file=sys.stderr) + + if not writable: + return 2 + + if not args.yes and not _confirm(f"Apply {len(writable)} change(s)?"): + print("aborted.") + return 1 + + failures = 0 + for note, plan in writable: + ok, log = writer.apply(plan, source_path=note.path) + print(f"\n{plan.source_rel}:") + for line in log: + print(f" {line}") + if not ok: + failures += 1 + + return 0 if failures == 0 else 3 + + +def cmd_watch(args) -> int: + """Watch the inbox: propose destinations, and file notes you approve.""" + vault = _load_vault(args) + jev, tracker = _make_jev(args) + clf = HybridClassifier(vault, jev=jev, threshold=args.threshold) + watcher = Watcher( + vault, clf, inbox=args.inbox, + interval=args.interval, dry_run=args.dry_run, + ) + + if args.once: + lines = watcher.run_once() + if lines: + for line in lines: + print(line) + else: + print("nothing to do") + if tracker is not None and tracker.calls: + print(f"\ncost: ${tracker.summary()['cost_usd']:.6f}") + return 0 + + mode = "DRY RUN" if args.dry_run else "live" + print(f"watching '{args.inbox}' every {args.interval}s ({mode}) — Ctrl-C to stop") + try: + watcher.run_forever(max_iterations=args.iterations) + except KeyboardInterrupt: + print("\nstopped") + return 0 + + +# ------------------------------------------------------------------------ main + +def build_parser() -> argparse.ArgumentParser: + p = argparse.ArgumentParser( + prog="obsidian-sort", + description="Classify and file Obsidian notes (Phase A: read-only).", + ) + p.add_argument("--vault", default="~/obsidian", help="vault root (default: ~/obsidian)") + p.add_argument("--inbox", default="inbox", help="inbox folder name (default: inbox)") + p.add_argument("--json", action="store_true", help="also emit JSON") + + sub = p.add_subparsers(dest="command", required=True) + + sub.add_parser("scan", help="report vault shape").set_defaults(func=cmd_scan) + + sp = sub.add_parser("propose", help="classify one note") + sp.add_argument("file", help="path to a note (absolute or vault-relative)") + sp.add_argument("--jev", action="store_true", help="escalate to JEV when rules are unsure") + sp.add_argument("--threshold", type=float, default=0.45, + help="escalate to JEV below this rules confidence") + sp.add_argument("--timeout", type=float, default=90.0) + sp.add_argument("--max-calls", type=int, default=500) + sp.set_defaults(func=cmd_propose) + + si = sub.add_parser("inbox", help="classify all inbox notes") + si.add_argument("--jev", action="store_true", help="escalate to JEV when rules are unsure") + si.add_argument("--threshold", type=float, default=0.45) + si.add_argument("--timeout", type=float, default=90.0) + si.add_argument("--max-calls", type=int, default=500) + si.set_defaults(func=cmd_inbox) + + sv = sub.add_parser("validate", help="replay classifier against known-good notes") + sv.add_argument("--in-folder", action="store_true", + help="use the note's real folder (default: simulate inbox)") + sv.set_defaults(func=cmd_validate) + + sb = sub.add_parser("benchmark", help="run the hybrid classifier and measure accuracy + cost") + sb.add_argument("--limit", type=int, default=25, help="how many notes to test (default 25)") + sb.add_argument("--all", action="store_true", help="test every known-good note") + sb.add_argument("--threshold", type=float, default=0.45, + help="escalate to JEV below this rules confidence") + sb.add_argument("--quiet", action="store_true", help="only print the summary") + sb.add_argument("--timeout", type=float, default=90.0) + sb.add_argument("--max-calls", type=int, default=500) + sb.set_defaults(func=cmd_benchmark) + + # --- sort (Phase C: writes) ----------------------------------------- + ss = sub.add_parser("sort", help="plan and optionally apply sorting for one note") + ss.add_argument("file", help="path to a note (absolute or vault-relative)") + ss.add_argument("--apply", action="store_true", help="write the changes (default: dry run)") + ss.add_argument("--yes", action="store_true", help="skip the confirmation prompt") + ss.add_argument("--jev", action="store_true", help="escalate to JEV when rules are unsure") + ss.add_argument("--threshold", type=float, default=0.45) + ss.add_argument("--timeout", type=float, default=90.0) + ss.add_argument("--max-calls", type=int, default=500) + ss.set_defaults(func=cmd_sort) + + si = sub.add_parser("sort-inbox", help="plan and optionally apply sorting for the whole inbox") + si.add_argument("--apply", action="store_true", help="write the changes (default: dry run)") + si.add_argument("--yes", action="store_true", help="skip the confirmation prompt") + si.add_argument("--jev", action="store_true", help="escalate to JEV when rules are unsure") + si.add_argument("--threshold", type=float, default=0.45) + si.add_argument("--timeout", type=float, default=90.0) + si.add_argument("--max-calls", type=int, default=500) + si.set_defaults(func=cmd_sort_inbox) + + sw = sub.add_parser("watch", help="watch the inbox and maintain proposals") + sw.add_argument("--once", action="store_true", help="single pass, then exit") + sw.add_argument("--interval", type=float, default=5.0, help="seconds between passes") + sw.add_argument("--iterations", type=int, default=None, help="stop after N passes") + sw.add_argument("--dry-run", action="store_true", help="report but never write") + sw.add_argument("--threshold", type=float, default=0.45) + sw.add_argument("--timeout", type=float, default=90.0) + sw.add_argument("--max-calls", type=int, default=500) + sw.set_defaults(func=cmd_watch) + + return p + + +def main(argv: list[str] | None = None) -> int: + args = build_parser().parse_args(argv) + return args.func(args) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/sorter/frontmatter.py b/sorter/frontmatter.py new file mode 100644 index 0000000..0cee93c --- /dev/null +++ b/sorter/frontmatter.py @@ -0,0 +1,213 @@ +""" +Minimal YAML frontmatter reader/writer. + +Deliberately dependency-free: the vault uses a simple, uniform YAML subset +(scalars, dash-lists, inline lists, quoted strings). NixOS hosts have no pip, +so avoiding PyYAML makes deployment trivial. + +Supports: + key: value + key: "quoted value" + key: [a, b, c] + key: + - item + - item + +Does NOT support (not present in the vault): block scalars (| >), anchors, +multi-document streams, nested maps beyond one level. +""" + +from __future__ import annotations + +import re +from typing import Any + +_DELIM = re.compile(r"^---\s*$", re.MULTILINE) + + +def _strip_comment(value: str) -> str: + """Remove a trailing ' # comment', but not inside quotes.""" + if not value: + return value + out = [] + quote = None + for i, ch in enumerate(value): + if quote: + out.append(ch) + if ch == quote and (i == 0 or value[i - 1] != "\\"): + quote = None + continue + if ch in ('"', "'"): + quote = ch + out.append(ch) + continue + if ch == "#" and (i == 0 or value[i - 1] in " \t"): + break + out.append(ch) + return "".join(out).rstrip() + + +def parse_scalar(raw: str) -> Any: + """Parse a single YAML scalar into a Python value.""" + if raw is None: + return None + v = _strip_comment(raw.strip()) + if v == "": + return None + + # Fully quoted string + if len(v) >= 2 and v[0] == v[-1] and v[0] in ('"', "'"): + return v[1:-1] + + # Inline list: [a, b, c] + if v.startswith("[") and v.endswith("]"): + inner = v[1:-1].strip() + if not inner: + return [] + return [parse_scalar(part) for part in _split_inline(inner)] + + low = v.lower() + if low in ("true", "yes"): + return True + if low in ("false", "no"): + return False + if low in ("null", "~"): + return None + + return v + + +def _split_inline(inner: str) -> list[str]: + """Split an inline list body on commas, respecting quotes.""" + parts, buf, quote = [], [], None + for ch in inner: + if quote: + buf.append(ch) + if ch == quote: + quote = None + continue + if ch in ('"', "'"): + quote = ch + buf.append(ch) + continue + if ch == ",": + parts.append("".join(buf).strip()) + buf = [] + continue + buf.append(ch) + if buf: + parts.append("".join(buf).strip()) + return [p for p in parts if p != ""] + + +def parse(text: str) -> tuple[dict[str, Any], str, bool]: + """ + Split a note into (frontmatter_dict, body, had_frontmatter). + + If the note has no valid frontmatter block, returns ({}, text, False). + """ + if not text.startswith("---"): + return {}, text, False + + m = re.match(r"^---\s*\n(.*?)\n---\s*\n?", text, re.DOTALL) + if not m: + return {}, text, False + + fm_text = m.group(1) + body = text[m.end():] + return parse_block(fm_text), body, True + + +def parse_block(text: str) -> dict[str, Any]: + """Parse the body of a frontmatter block into a dict.""" + result: dict[str, Any] = {} + key: str | None = None + + for raw in text.split("\n"): + if not raw.strip() or raw.lstrip().startswith("#"): + continue + + # List item belonging to the current key + item = re.match(r"^\s+-\s*(.*)$", raw) or re.match(r"^-\s*(.*)$", raw) + if item and key is not None: + if not isinstance(result.get(key), list): + result[key] = [] + result[key].append(parse_scalar(item.group(1))) + continue + + # Indented continuation of a scalar value + if re.match(r"^\s+\S", raw) and key is not None: + prev = result.get(key) + extra = parse_scalar(raw.strip()) + if isinstance(prev, list): + prev.append(extra) + elif prev in (None, ""): + result[key] = extra + else: + result[key] = f"{prev} {extra}" + continue + + # key: value + kv = re.match(r"^([A-Za-z0-9_\-]+):\s?(.*)$", raw) + if kv: + key = kv.group(1) + value = kv.group(2) + result[key] = None if value.strip() == "" else parse_scalar(value) + continue + + return result + + +def dump_block(fm: dict[str, Any], order: list[str] | None = None) -> str: + """Serialise a frontmatter dict back to YAML text (no --- delimiters).""" + lines: list[str] = [] + keys = list(order or []) + keys += [k for k in fm if k not in keys] + + for k in keys: + if k not in fm: + continue + v = fm[k] + lines.extend(_dump_key(k, v)) + return "\n".join(lines) + + +def _dump_key(key: str, value: Any) -> list[str]: + if value is None: + return [f"{key}:"] + if isinstance(value, list): + if not value: + return [f"{key}: []"] + out = [f"{key}:"] + for item in value: + out.append(f" - {_quote(item)}") + return out + if isinstance(value, bool): + return [f"{key}: {'true' if value else 'false'}"] + return [f"{key}: {_quote(value)}"] + + +def _quote(value: Any) -> str: + """Quote a scalar when required for valid YAML.""" + if isinstance(value, (int, float)): + return str(value) + s = str(value) + if s == "": + return '""' + needs = ( + s != s.strip() + or s.lower() in ("true", "false", "yes", "no", "null", "~") + or any(c in s for c in ":#[]{}&*!|>%@`\"'") + or re.match(r"^\d{4}-\d{2}-\d{2}( \d{2}:\d{2})?$", s) + ) + if needs: + escaped = s.replace("\\", "\\\\").replace('"', '\\"') + return f'"{escaped}"' + return s + + +def render(fm: dict[str, Any], body: str, order: list[str] | None = None) -> str: + """Render a complete note (frontmatter block + body).""" + block = dump_block(fm, order=order) + body = body.lstrip("\n") + return f"---\n{block}\n---\n\n{body}" diff --git a/sorter/jev.py b/sorter/jev.py new file mode 100644 index 0000000..d7f4bef --- /dev/null +++ b/sorter/jev.py @@ -0,0 +1,250 @@ +""" +TypeSafe JEV client — stdlib only. + +Deliberately dependency-free (urllib, not the official SDK) so the sorter +deploys onto hosts without pip. + +JEV answers three primitive question types: + choice → picks one label from your criteria, with calibrated confidence + noul → probability that a statement is true (0..1) + score → expected value on an ordered rubric + +API: POST https://api.typesafe.ai/v1/systemone +""" + +from __future__ import annotations + +import glob +import json +import os +import re +import urllib.error +import urllib.request +from dataclasses import dataclass, field +from typing import Any + +DEFAULT_BASE_URL = "https://api.typesafe.ai/v1" +DEFAULT_MODEL = "jev-latest" + +# JEV pricing: $0.042 per 1M input tokens; output tokens are free. +USD_PER_INPUT_TOKEN = 0.042 / 1_000_000 + +# Fallback locations for the API key. On NixOS the key usually lives in +# environment.d, which is loaded for systemd user sessions but NOT for +# non-interactive SSH shells — so we read it directly. +SECRETS_GLOBS = ( + "~/.config/environment.d/*.conf", + "~/.config/obsidian-sorter/secrets.conf", +) + + +def _key_from_secrets() -> str | None: + """Find TYPESAFE_API_KEY in the environment.d secrets files.""" + for pattern in SECRETS_GLOBS: + for path in sorted(glob.glob(os.path.expanduser(pattern))): + try: + text = open(path, encoding="utf-8").read() + except OSError: + continue + m = re.search( + r"^\s*(?:export\s+)?TYPESAFE_API_KEY\s*=\s*(.+?)\s*$", + text, re.MULTILINE, + ) + if m: + value = m.group(1).strip().strip('"').strip("'") + if value: + return value + return None + + +class JevError(RuntimeError): + """Raised when a JEV request fails.""" + + +# ------------------------------------------------------------ question builders + +def choice(instructions: str, criteria: dict[str, str | None]) -> dict[str, Any]: + """A question that selects one label from `criteria`.""" + return {"type": "choice", "instructions": instructions, "criteria": criteria} + + +def noul(instructions: str, true: str | None = None, false: str | None = None) -> dict[str, Any]: + """A yes/no question.""" + q: dict[str, Any] = {"type": "noul", "instructions": instructions} + if true is not None or false is not None: + q["criteria"] = {"true": true, "false": false} + return q + + +def score(instructions: str, criteria: list[str]) -> dict[str, Any]: + """An ordered rating; `criteria` is one description per level from 0.""" + return {"type": "score", "instructions": instructions, "criteria": criteria} + + +# --------------------------------------------------------------------- results + +@dataclass +class Answer: + """One answer returned by JEV.""" + + name: str + kind: str # "choice" | "noul" | "score" + raw: dict[str, Any] = field(default_factory=dict) + + @property + def label(self) -> str | None: + return self.raw.get("choice") + + @property + def confidence(self) -> float: + """Confidence for choice/score; probability for noul.""" + if self.kind == "noul": + return float(self.raw.get("noul", 0.5)) + return float(self.raw.get("confidence", 0.0)) + + @property + def probabilities(self) -> dict[str, float]: + return self.raw.get("probabilities", {}) or {} + + @property + def value(self) -> float | None: + if self.kind == "score": + return float(self.raw.get("score", 0.0)) + return None + + +@dataclass +class JevResult: + """A complete JEV response.""" + + answers: dict[str, Answer] + input_tokens: int = 0 + output_tokens: int = 0 + model: str = "" + error: str | None = None + + @property + def ok(self) -> bool: + return self.error is None + + @property + def cost_usd(self) -> float: + return self.input_tokens * USD_PER_INPUT_TOKEN + + def answer(self, name: str) -> Answer | None: + return self.answers.get(name) + + +# ---------------------------------------------------------------------- client + +class JevClient: + """Thin JEV HTTP client.""" + + def __init__( + self, + api_key: str | None = None, + model: str = DEFAULT_MODEL, + base_url: str = DEFAULT_BASE_URL, + timeout: float = 90.0, + ): + key = ( + api_key + or os.environ.get("TYPESAFE_API_KEY", "") + or _key_from_secrets() + or "" + ) + if not key: + raise JevError( + "No API key. Set TYPESAFE_API_KEY in the environment, add it to " + "~/.config/environment.d/*.conf, or pass api_key=." + ) + self.api_key = key + self.model = model + self.base_url = base_url.rstrip("/") + self.timeout = timeout + + def ask( + self, + state: Any, + questions: dict[str, dict[str, Any]], + ) -> JevResult: + """Send one system-one request. Never raises for API errors — returns a JevResult with .error set.""" + if not questions: + raise JevError("At least one question is required.") + + payload = {"state": state, "model": self.model, "questions": questions} + body = json.dumps(payload).encode("utf-8") + + req = urllib.request.Request( + f"{self.base_url}/systemone", + data=body, + method="POST", + headers={ + "Authorization": f"Bearer {self.api_key}", + "Content-Type": "application/json", + "Accept": "application/json", + "User-Agent": "obsidian-sorter/0.1", + }, + ) + + try: + with urllib.request.urlopen(req, timeout=self.timeout) as resp: + data = json.loads(resp.read().decode("utf-8")) + except urllib.error.HTTPError as e: + detail = "" + try: + detail = e.read().decode("utf-8")[:500] + except Exception: + pass + return JevResult(answers={}, error=f"HTTP {e.code}: {detail or e.reason}") + except urllib.error.URLError as e: + return JevResult(answers={}, error=f"network error: {e.reason}") + except (json.JSONDecodeError, TimeoutError) as e: + return JevResult(answers={}, error=f"{type(e).__name__}: {e}") + + answers: dict[str, Answer] = {} + for name, raw in (data.get("answers") or {}).items(): + kind = raw.get("type", "choice") + answers[name] = Answer(name=name, kind=kind, raw=raw) + + usage = data.get("usage") or {} + return JevResult( + answers=answers, + input_tokens=int(usage.get("input_tokens") or 0), + output_tokens=int(usage.get("output_tokens") or 0), + model=data.get("model", self.model), + ) + + +class CostTracker: + """Accumulates token usage and enforces a daily call cap.""" + + def __init__(self, max_calls_per_run: int = 500): + self.calls = 0 + self.input_tokens = 0 + self.output_tokens = 0 + self.max_calls_per_run = max_calls_per_run + self.errors = 0 + + def record(self, result: JevResult) -> None: + self.calls += 1 + self.input_tokens += result.input_tokens + self.output_tokens += result.output_tokens + if not result.ok: + self.errors += 1 + + @property + def cost_usd(self) -> float: + return self.input_tokens * USD_PER_INPUT_TOKEN + + def budget_exceeded(self) -> bool: + return self.calls >= self.max_calls_per_run + + def summary(self) -> dict[str, Any]: + return { + "calls": self.calls, + "input_tokens": self.input_tokens, + "output_tokens": self.output_tokens, + "cost_usd": round(self.cost_usd, 6), + "errors": self.errors, + } diff --git a/sorter/models.py b/sorter/models.py new file mode 100644 index 0000000..2566cc6 --- /dev/null +++ b/sorter/models.py @@ -0,0 +1,140 @@ +"""Data structures shared across the sorter.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + +# --------------------------------------------------------------------------- +# The taxonomy, simplified 2026-09-24. +# +# Sam's model: the TYPE of a note mirrors the FOLDER it lives in. Nothing is +# guessed from content — the destination decides the type. +# +# being worked on -> 200 projects/ -> project +# ongoing interest + docs-> 300 areas/ -> area +# family / non-work -> 400 Personal Family -> family +# +# Everything else that used to exist (subject, tool, container, machine, +# backup-system, website, ...) has been RETIRED. Do not reintroduce them. +# --------------------------------------------------------------------------- + +# Type mirrors folder. These are the only valid types. +VALID_TYPES = { + "project", # 200 projects/ + "area", # 300 areas/ + "family", # 400 Personal Family/ + "index", # 100 Table of Contents/ + "daily", # 000 daily/ + "archive", # 500 archive/ + "clipping", # Clippings/ (browser extension drops) + "note", # unsorted placeholder, only ever in the inbox +} + +# The three real filing destinations. +BUCKETS = ("project", "area", "family") + +# Folder prefix -> type. Order matters: longest prefix wins. +FOLDER_BUCKETS: list[tuple[str, str]] = [ + ("200 projects", "project"), + ("300 areas", "area"), + ("400 Personal Family", "family"), + ("100 Table of Contents", "index"), + ("000 daily", "daily"), + ("500 archive", "archive"), + ("Clippings", "clipping"), +] + +# The generic-template marker. Only valid while a note sits in the inbox. +PLACEHOLDER_TYPE = "note" + + +def type_from_folder(folder: str) -> str | None: + """Return the type implied by a vault folder, or None if unknown.""" + if not folder: + return None + for prefix, note_type in FOLDER_BUCKETS: + if folder == prefix or folder.startswith(prefix + "/"): + return note_type + return None + + +@dataclass +class Note: + """A note discovered in the vault.""" + + path: str # absolute path + rel_path: str # path relative to vault root + filename: str # e.g. "Borg Backup.md" + stem: str # e.g. "Borg Backup" + title: str | None # frontmatter title, if present + note_type: str | None # frontmatter type + status: str | None # frontmatter status + tags: list[str] = field(default_factory=list) + areas: list[str] = field(default_factory=list) + projects: list[str] = field(default_factory=list) + fm: dict[str, Any] = field(default_factory=dict) + has_frontmatter: bool = False + body: str = "" + + @property + def display_name(self) -> str: + return self.title or self.stem + + @property + def folder(self) -> str: + """Vault-relative folder, '' for root.""" + parent = self.rel_path.rsplit("/", 1) + return parent[0] if len(parent) == 2 else "" + + @property + def expected_type(self) -> str | None: + """The type this note's folder says it should have.""" + return type_from_folder(self.folder) + + @property + def needs_classification(self) -> bool: + """ + True when the note has no trustworthy type. + + A note is only 'sorted' when its type matches its folder. + """ + current = (self.note_type or "").strip().lower() + expected = self.expected_type + if expected is None: + return True # not in a known bucket (e.g. inbox) + return current != expected + + +@dataclass +class Proposal: + """A classification recommendation — a plan, not an action.""" + + source_rel: str + note_type: str + target_folder: str + toc_section: str + template: str + confidence: float + method: str # "deterministic" | "jev" | "none" | "trusted" + reasons: list[str] = field(default_factory=list) + is_new_area: bool = False + warnings: list[str] = field(default_factory=list) + alternatives: list[tuple[str, float]] = field(default_factory=list) + proposed_tags: list[str] = field(default_factory=list) + + def as_dict(self) -> dict[str, Any]: + return { + "source": self.source_rel, + "type": self.note_type, + "target_folder": self.target_folder, + "toc_section": self.toc_section, + "template": self.template, + "confidence": round(self.confidence, 3), + "method": self.method, + "reasons": self.reasons, + "is_new_area": self.is_new_area, + "warnings": self.warnings, + "alternatives": self.alternatives, + "proposed_tags": self.proposed_tags, + } diff --git a/sorter/tier2.py b/sorter/tier2.py new file mode 100644 index 0000000..4fef2b6 --- /dev/null +++ b/sorter/tier2.py @@ -0,0 +1,418 @@ +""" +Tier 2 — JEV semantic classifier. + +Because type mirrors folder, JEV only has to answer two things: + + 1. Which bucket does this belong to? project / area / family + 2. Which existing folder is its home? + +The type then falls out of the chosen folder. That is the whole model. +""" + +from __future__ import annotations + +import re + +from .classify import TYPE_TO_TOC, TYPE_TO_TEMPLATE +from .jev import CostTracker, JevClient, JevError +from .models import Note, Proposal, type_from_folder +from .vault import Vault + +# The three buckets, described the way Sam thinks about them. +BUCKET_DESCRIPTIONS: dict[str, str] = { + "project": ( + "Work in progress. Something being built, fixed, or changed, with a " + "goal and an end state. Language of plans, ideas, TODOs, 'needs to be " + "built', 'original plan was', future tense." + ), + "area": ( + "An ongoing area of interest that has documentation and notes. " + "Something maintained or studied over time with no end date. " + "Reference material, how-to notes, system documentation." + ), + "family": ( + "Family and personal life. STRICTLY NON-WORK — no computers, dev-ops, " + "software, networking, coding or IoT. Recipes, holidays, family events, " + "hobbies, personal finance, films, books, art." + ), +} + +# Folders that are never filing destinations. +EXCLUDED_FOLDER_PARTS = ("inbox", "templates", ".trash", "100 Table of Contents") + +# Words never worth proposing as a tag. +TAG_STOPWORDS = { + "note", "notes", "idea", "build", "building", "create", "using", + "with", "from", "that", "this", "into", "about", "setup", "guide", + "project", "tool", "tools", "test", "testing", "make", "made", "new", + "add", "adds", "added", "need", "needs", "want", "wants", "could", + "would", "should", "have", "been", "will", "then", "than", "when", +} + + +# Manual clarifications for folders whose purpose cannot be inferred from the +# titles of notes already inside them. These override the auto-generated +# "contains: ..." description, and are the place to teach the sorter a rule. +FOLDER_NOTES: dict[str, str] = { + "200 projects/260 Ideas to Build Create Make": ( + "UNSTARTED ideas and concepts for things that could exist. Work has NOT " + "begun. Language of 'idea', 'could', 'might', 'what if', 'someday', " + "'would be great if'. This is a parking lot for future possibilities." + ), + "200 projects/260 Ideas to Build Create Make": ( + "Ideas for things to BUILD — software, code, apps, electronics, " + "hardware, gadgets, devices. Anything involving programming or physical " + "making. Work has NOT begun. This is where TECHNICAL ideas live, even " + "when they are also commercial or could make money." + ), + "400 Personal Family/410 Business Ideas": ( + "Commercial or business concepts that are NOT primarily technical builds " + "— a service, a shop, a venture, a product concept with no coding or " + "electronics involved. If the idea involves building software or " + "hardware, it belongs in 260 Ideas to Build Create Make instead." + ), + "200 projects/250 Electronics": ( + "Electronics and hardware builds ACTIVELY IN PROGRESS — parts ordered, " + "wiring, firmware, enclosures, assembly. Language of 'building', " + "'ordered', 'TODO', 'currently working on', 'in progress'." + ), + "200 projects/Tools Software WebUI": ( + "Software tools and web UIs being built or actively developed." + ), + "300 areas/355 AI Research and Techniques": ( + "Reference notes and research about AI techniques — memory, planning, " + "models, approaches. Documentation rather than something being built." + ), + "300 areas/395 AI Tools to try": ( + "AI tools and services noted for evaluation or adoption. Reference " + "material about the tool, not work to build one." + ), + "300 areas/390 Web Pages and Tools To Try": ( + "Useful websites, tools and resources worth keeping. Reference material." + ), +} + + +class JevClassifier: + """Tier 2 classifier backed by TypeSafe JEV.""" + + def __init__( + self, + vault: Vault, + client: JevClient, + tracker: CostTracker | None = None, + max_body_chars: int = 2500, + ): + self.vault = vault + self.client = client + self.tracker = tracker or CostTracker() + self.max_body_chars = max_body_chars + self._folder_desc: dict[str | None, dict[str, str]] = {} + self._tag_cache: dict[str, int] | None = None + self._tag_threshold = 0.65 + + # ------------------------------------------------------------------- tags + + def _tag_counts(self) -> dict[str, int]: + """Every tag already used in the vault, with how often.""" + if self._tag_cache is not None: + return self._tag_cache + counts: dict[str, int] = {} + for note in self.vault.scan(): + for t in (note.tags or []): + if t: + counts[t.lower()] = counts.get(t.lower(), 0) + 1 + self._tag_cache = counts + return counts + + def candidate_tags(self, note: Note, limit: int = 8) -> list[str]: + """ + Tags worth asking JEV about: existing tags that appear in the note, + plus a few new candidates drawn from the title. + """ + text = f"{note.title or note.stem} {note.body}".lower().replace("-", " ") + counts = self._tag_counts() + + hits: list[tuple[int, str]] = [] + for tag, count in counts.items(): + needle = tag.replace("-", " ").strip() + if not needle: + continue + # WORD BOUNDARIES: without these, 'pi' matches inside 'pile' and + # 'lights' matches inside 'highlights'. + if re.search(r"\b" + re.escape(needle) + r"\b", text): + hits.append((count, tag)) + hits.sort(key=lambda kv: -kv[0]) + existing_hits = [t for _c, t in hits[:limit]] + + # New candidates from the title, so genuinely new tags are possible. + existing = set(counts) + fresh: list[str] = [] + for word in re.findall(r"[a-z][a-z0-9]{3,}", (note.title or note.stem).lower()): + if word in TAG_STOPWORDS or word in existing: + continue + fresh.append(word) + if len(fresh) >= 4: + break + + everything = existing_hits + fresh + kept = [ + w for w in fresh + if not any(w != o and w in o for o in everything) + ] + return self._dedupe_variants(existing_hits + kept) + + @staticmethod + def _dedupe_variants(tags: list[str]) -> list[str]: + """ + Drop a tag when a longer tag already covers it. + + 'obsidian-sorter' subsumes 'obsidian' and 'sorter', so the short forms go. + The more specific tag always wins. + """ + out: list[str] = [] + for tag in tags: + if tag in out: + continue + out.append(tag) + return [ + t for t in out + if not any(t != o and t in o for o in out) + ] + + # ------------------------------------------------------------------ setup + + def _candidate_folders(self, bucket: str | None = None) -> list[str]: + out = [] + for f in self.vault.folders(): + low = f.lower() + if any(part in low for part in EXCLUDED_FOLDER_PARTS): + continue + if bucket and type_from_folder(f) != bucket: + continue + out.append(f) + return out + + def _folder_descriptions(self, bucket: str | None = None) -> dict[str, str]: + """Describe each folder with a couple of real example note titles.""" + if bucket in self._folder_desc: + return self._folder_desc[bucket] + + by_folder: dict[str, list[str]] = {} + for note in self.vault.scan(): + if not note.folder: + continue + low = note.folder.lower() + if any(part in low for part in EXCLUDED_FOLDER_PARTS): + continue + by_folder.setdefault(note.folder, []).append(note.display_name) + + desc: dict[str, str] = {} + for folder in self._candidate_folders(bucket): + if folder in FOLDER_NOTES: + desc[folder] = f"{folder} — {FOLDER_NOTES[folder]}" + continue + examples = by_folder.get(folder, [])[:3] + if examples: + desc[folder] = f"{folder} — contains: " + "; ".join(examples) + else: + desc[folder] = f"{folder} — (currently empty)" + self._folder_desc[bucket] = desc + return desc + + # ------------------------------------------------------------- build input + + def _build_state(self, note: Note) -> dict: + body = (note.body or "").strip() + if len(body) > self.max_body_chars: + body = body[: self.max_body_chars] + " …[truncated]" + + state: dict = { + "title": note.title or note.stem, + "content": body, + } + if note.tags: + state["existing_tags_on_note"] = note.tags + return state + + def _build_questions(self, bucket: str | None = None, + tag_candidates: list[str] | None = None) -> dict: + from .jev import choice, noul + + questions = { + "bucket": choice( + "What kind of note is this? Judge by whether the work is " + "FINISHED or ONGOING, and whether it is work at all.", + BUCKET_DESCRIPTIONS, + ), + "target_folder": choice( + "Which existing vault folder is the best home for this note? " + "Judge by what each folder already contains.", + self._folder_descriptions(bucket), + ), + "needs_new_folder": noul( + "Does this note need a NEW folder that does not exist yet?", + true="No listed folder is a good fit and a new one is needed", + false="One of the listed folders is a good fit", + ), + } + + # One batched question per tag candidate — same request, no extra cost. + for i, tag in enumerate(tag_candidates or []): + questions[f"tag_{i}"] = noul( + f'Should this note carry the tag "{tag}"?', + true="the note is substantially about this", + false="incidental, unrelated, or too vague to be useful", + ) + return questions + + # -------------------------------------------------------------- classifying + + def classify(self, note: Note, bucket_hint: str | None = None) -> Proposal: + """ + Classify a note. + + `bucket_hint` (usually from the author's own tag) constrains the folder + choices to a single bucket, so a note tagged 'project' can never be filed + into an area folder. + """ + try: + tags_offered = self.candidate_tags(note) + result = self.client.ask( + self._build_state(note), + self._build_questions(bucket_hint, tags_offered), + ) + except JevError as e: + return self._failed(note, f"JEV unavailable: {e}") + + self.tracker.record(result) + if not result.ok: + return self._failed(note, f"JEV request failed: {result.error}") + + bucket_ans = result.answer("bucket") + folder_ans = result.answer("target_folder") + new_ans = result.answer("needs_new_folder") + + folder = (folder_ans.label if folder_ans else "") or "" + folder_conf = folder_ans.confidence if folder_ans else 0.0 + + # TYPE COMES FROM THE FOLDER. This is the whole point of the model. + derived = type_from_folder(folder) + bucket = (bucket_ans.label if bucket_ans else "") or "" + + if derived: + note_type = derived + elif bucket: + note_type = bucket # folder unknown; trust JEV's bucket + else: + note_type = "area" + + needs_new = bool( + new_ans and new_ans.confidence >= 0.5 and folder_conf < 0.60 + ) + + confidences = [a.confidence for a in (bucket_ans, folder_ans) if a] + confidence = min(confidences) if confidences else 0.0 + + reasons = [] + if bucket_ans: + reasons.append(f"JEV bucket: {bucket_ans.label} (conf {bucket_ans.confidence:.2f})") + if folder_ans: + reasons.append(f"JEV folder: {folder_ans.label} (conf {folder_ans.confidence:.2f})") + if derived: + reasons.append(f"type taken from folder: {derived}") + + warnings: list[str] = [] + # A mismatch between the chosen bucket and the chosen folder's bucket + # is a genuine signal that JEV was torn between two homes. + if bucket and derived and bucket != derived: + warnings.append( + f"JEV was torn: bucket '{bucket}' but folder implies '{derived}' " + "— review before applying" + ) + if confidence < 0.5: + warnings.append("Low JEV confidence — recommend human confirmation") + if needs_new: + warnings.append("JEV suggests a NEW folder may be needed (not auto-created)") + if folder and folder not in set(self._candidate_folders(bucket_hint)): + warnings.append(f"JEV chose an unknown folder: {folder}") + + alternatives = [] + if bucket_ans and bucket_ans.probabilities: + alts = sorted(bucket_ans.probabilities.items(), key=lambda kv: -kv[1]) + alternatives = [(k, round(v, 3)) for k, v in alts[1:3]] + + # Tags JEV agreed to, merged with any the author already had. + proposed_tags: list[str] = [] + for i, tag in enumerate(tags_offered): + a = result.answer(f"tag_{i}") + if a and a.confidence >= self._tag_threshold and tag not in proposed_tags: + proposed_tags.append(tag) + for t in (note.tags or []): + if t and t not in proposed_tags: + proposed_tags.append(t) + # One last pass: drop any tag a more specific tag already covers. + proposed_tags = self._dedupe_variants(proposed_tags) + if proposed_tags: + reasons.append("tags: " + ", ".join(proposed_tags)) + + return Proposal( + source_rel=note.rel_path, + note_type=note_type, + target_folder=folder, + toc_section=TYPE_TO_TOC.get(note_type, ""), + template=TYPE_TO_TEMPLATE.get(note_type, "Area.md"), + confidence=confidence, + method="jev", + reasons=reasons, + is_new_area=needs_new, + warnings=warnings, + alternatives=alternatives, + proposed_tags=proposed_tags, + ) + + def _failed(self, note: Note, msg: str) -> Proposal: + return Proposal( + source_rel=note.rel_path, note_type="", target_folder="", + toc_section="", template="", confidence=0.0, method="jev-error", + warnings=[msg], + ) + + +class HybridClassifier: + """Tier 1 first; escalate to Tier 2 when the rules are not confident.""" + + def __init__(self, vault: Vault, jev: JevClassifier | None = None, threshold: float = 0.45): + from .classify import Classifier + self.vault = vault + self.rules = Classifier(vault) + self.jev = jev + self.threshold = threshold + + def classify(self, note: Note) -> Proposal: + from .classify import explicit_bucket_from_tags + + primary = self.rules.classify(note) + + if primary.method in ("blocked", "trusted"): + return primary + + hint = explicit_bucket_from_tags(note) + + needs_jev = ( + self.jev is not None + and (primary.method in ("none", "jev-error") or primary.confidence < self.threshold) + ) + if not needs_jev: + return primary + + secondary = self.jev.classify(note, bucket_hint=hint) + if secondary.method == "jev-error": + primary.warnings.extend(secondary.warnings) + return primary + + prefix = f"escalated from rules (confidence {primary.confidence:.2f})" + if hint: + prefix += f"; folder choices constrained to '{hint}' by your tag" + secondary.reasons.insert(0, prefix) + return secondary diff --git a/sorter/vault.py b/sorter/vault.py new file mode 100644 index 0000000..16a2989 --- /dev/null +++ b/sorter/vault.py @@ -0,0 +1,142 @@ +"""Vault scanning and index-note introspection.""" + +from __future__ import annotations + +import os +import re +from pathlib import Path + +from . import frontmatter +from .models import Note + +# Directories that are never vault content. +SKIP_DIRS = { + ".git", ".trash", ".venv", ".obsidian", "templates", + "graphify-out", "graphify-out-temp", "__pycache__", "node_modules", +} + +# Files that are never categorised. +SKIP_FILES = {"index.md"} + + +class Vault: + """Read-only view of an Obsidian vault.""" + + def __init__(self, root: str | os.PathLike): + self.root = Path(root).expanduser().resolve() + if not self.root.is_dir(): + raise FileNotFoundError(f"vault not found: {self.root}") + + # ---------------------------------------------------------------- scanning + + def _iter_md(self): + for dirpath, dirnames, filenames in os.walk(self.root): + dirnames[:] = sorted(d for d in dirnames if d not in SKIP_DIRS) + for fn in sorted(filenames): + if not fn.endswith(".md"): + continue + if fn in SKIP_FILES: + continue + yield Path(dirpath) / fn + + def read_note(self, path: Path) -> Note | None: + try: + text = path.read_text(encoding="utf-8") + except (OSError, UnicodeDecodeError): + return None + + fm, body, had = frontmatter.parse(text) + rel = str(path.relative_to(self.root)) + + def as_list(key: str) -> list[str]: + val = fm.get(key) + if isinstance(val, list): + return [str(x) for x in val if x is not None] + if val: + return [str(val)] + return [] + + return Note( + path=str(path), + rel_path=rel, + filename=path.name, + stem=path.stem, + title=fm.get("title"), + note_type=fm.get("type"), + status=fm.get("status"), + tags=as_list("tags"), + areas=as_list("areas"), + projects=as_list("projects"), + fm=fm, + has_frontmatter=had, + body=body, + ) + + def scan(self) -> list[Note]: + notes = [] + for path in self._iter_md(): + note = self.read_note(path) + if note is not None: + notes.append(note) + return notes + + # ------------------------------------------------------------------ shape + + def folders(self) -> list[str]: + """All vault-relative folders that contain at least one note.""" + found = set() + for path in self._iter_md(): + rel = path.relative_to(self.root) + if len(rel.parts) > 1: + found.add(str(rel.parent)) + return sorted(found) + + def index_notes(self) -> list[Note]: + """Notes whose type is `index` — the Table of Contents family.""" + out = [] + for path in self._iter_md(): + note = self.read_note(path) + if note and (note.note_type or "").strip().lower() == "index": + out.append(note) + return out + + def toc_sections(self) -> dict[str, list[str]]: + """ + Map each index note (by rel_path) to its `##` section headings. + + Used to decide where a new entry should be inserted. + """ + result: dict[str, list[str]] = {} + for note in self.index_notes(): + sections = [ + m.group(1).strip() + for m in re.finditer(r"^##\s+(.+)$", note.body, re.MULTILINE) + ] + result[note.rel_path] = sections + return result + + def known_titles(self) -> dict[str, str]: + """Map lowercased title/stem → rel_path, for duplicate detection.""" + out: dict[str, str] = {} + for path in self._iter_md(): + note = self.read_note(path) + if not note: + continue + out.setdefault(note.stem.lower(), note.rel_path) + if note.title: + out.setdefault(note.title.lower(), note.rel_path) + return out + + def inbox_notes(self, inbox: str = "inbox") -> list[Note]: + """Notes sitting in the inbox, awaiting sorting. Skips `_` helper files.""" + folder = self.root / inbox + if not folder.is_dir(): + return [] + out = [] + for path in sorted(folder.glob("*.md")): + if path.name.startswith("_"): + continue + note = self.read_note(path) + if note is not None: + out.append(note) + return out diff --git a/sorter/watch.py b/sorter/watch.py new file mode 100644 index 0000000..81ec0b8 --- /dev/null +++ b/sorter/watch.py @@ -0,0 +1,260 @@ +""" +The inbox watcher. + +Implements Sam's review-by-frontmatter workflow. Nothing is ever filed without +an explicit approval, and the approval is a checkbox inside the note itself. + + you save a note into 010 inbox/ type: note + | + v watcher classifies it + proposal written into the note approved: false [ ] + | + v you tick the checkbox in Obsidian + approved: true [x] + | + v watcher files it + note moves to the right folder, admin fields stripped, TOC updated + +Admin fields (removed on filing): + proposed_folder, proposed_toc, proposed_tags, confidence, approved +""" + +from __future__ import annotations + +import hashlib +import os +import time +from datetime import datetime +from pathlib import Path +from typing import Iterable + +from . import frontmatter +from .models import Note, Proposal, type_from_folder +from .vault import Vault +from .writer import FIELD_ORDER, Writer + +# Fields the watcher owns. Never present on a filed note. +ADMIN_FIELDS = ("proposed_folder", "proposed_toc", "proposed_tags", + "confidence", "approved", "proposed_for") + +PROPOSAL_ORDER = ( + "title", "type", "status", + "proposed_folder", "proposed_toc", "proposed_tags", "confidence", + "approved", "proposed_for", + "created", "updated", "tags", "aliases", "id", +) + +LOG_NAME = "_sort log.md" + + +# --------------------------------------------------------------------- helpers + +def has_proposal(note: Note) -> bool: + return bool(str(note.fm.get("proposed_folder") or "").strip()) + + +def is_approved(note: Note) -> bool: + value = note.fm.get("approved") + if isinstance(value, bool): + return value + return str(value).strip().lower() in ("true", "yes", "y", "1", "approved") + + +def is_planned(note: Note) -> bool: + """Notes whose destination is set but that are not approved yet.""" + return has_proposal(note) and not is_approved(note) + + +def content_hash(note: Note) -> str: + """Fingerprint of the parts of a note a proposal depends on.""" + basis = f"{note.title or note.stem}\n{note.body}" + return hashlib.sha1(basis.encode("utf-8")).hexdigest()[:12] + + +def proposal_is_stale(note: Note) -> bool: + """True when the note changed after its proposal was written.""" + recorded = str(note.fm.get("proposed_for") or "").strip() + if not recorded: + return True # proposals made before this feature existed + return recorded != content_hash(note) + + +class Watcher: + """Polls the inbox and maintains proposals.""" + + def __init__( + self, + vault: Vault, + classifier, + inbox: str = "010 inbox", + interval: float = 5.0, + dry_run: bool = False, + ): + self.vault = vault + self.classifier = classifier + self.writer = Writer(vault) + self.inbox = inbox + self.interval = interval + self.dry_run = dry_run + self.log_path = vault.root / inbox / LOG_NAME + + # ------------------------------------------------------------------ logging + + def _log(self, lines: Iterable[str]) -> None: + lines = [l for l in lines if l] + if not lines: + return + stamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + block = f"\n### {stamp}\n" + "\n".join(f"- {l}" for l in lines) + "\n" + if self.dry_run: + return + if not self.log_path.exists(): + self.log_path.parent.mkdir(parents=True, exist_ok=True) + self.log_path.write_text( + "# Sort log\n\nWhat the watcher did, newest last.\n", encoding="utf-8") + with open(self.log_path, "a", encoding="utf-8") as fh: + fh.write(block) + + # ----------------------------------------------------------------- proposing + + def _write_proposal(self, note: Note, proposal: Proposal) -> list[str]: + """Add the proposal fields to the note in place. Returns log lines.""" + fm = dict(note.fm) + + # Resolve the real index + section so the proposal reads clearly. + toc_rel, toc_section, _ = self.writer.resolve_toc( + proposal.note_type, proposal.target_folder + ) + if toc_rel and toc_section: + fm["proposed_toc"] = f"{toc_rel} → {toc_section}" + elif toc_rel: + fm["proposed_toc"] = toc_rel + else: + fm["proposed_toc"] = "(not indexed)" + + fm["proposed_folder"] = proposal.target_folder + if proposal.proposed_tags: + fm["proposed_tags"] = proposal.proposed_tags + fm["confidence"] = round(proposal.confidence, 2) + fm["approved"] = False + fm["proposed_for"] = content_hash(note) + + text = frontmatter.render(fm, note.body, order=PROPOSAL_ORDER) + if not self.dry_run: + self.writer._atomic_write(Path(note.path), text) + + return [ + f"**proposed** `{note.filename}` → `{proposal.target_folder}` " + f"(confidence {proposal.confidence:.2f})", + ] + [f" - {w}" for w in proposal.warnings] + + def _expected_toc(self, folder: str) -> str: + """The index + section a folder should be listed under.""" + note_type = type_from_folder(folder) + if not note_type: + return "" + toc_rel, toc_section, _ = self.writer.resolve_toc(note_type, folder) + if toc_rel and toc_section: + return f"{toc_rel} → {toc_section}" + return toc_rel or "(not indexed)" + + def _sync_toc(self, note: Note) -> list[str]: + """ + Keep `proposed_toc` truthful. + + If the folder is edited by hand, the index line must follow it — + otherwise the note shows a destination that no longer matches. + """ + folder = str(note.fm.get("proposed_folder") or "").strip() + expected = self._expected_toc(folder) + current = str(note.fm.get("proposed_toc") or "").strip() + if not expected or expected == current: + return [] + + fm = dict(note.fm) + fm["proposed_toc"] = expected + text = frontmatter.render(fm, note.body, order=PROPOSAL_ORDER) + if not self.dry_run: + self.writer._atomic_write(Path(note.path), text) + return [f"**toc follows folder** `{note.filename}` → {expected}"] + + # ------------------------------------------------------------------- filing + + def _file(self, note: Note) -> list[str]: + """File an approved note. Returns log lines.""" + folder = str(note.fm.get("proposed_folder") or "").strip() + if not type_from_folder(folder): + return [f"**skipped** `{note.filename}` — " + f"proposed_folder is not inside a known bucket: {folder!r}"] + + tags = note.fm.get("proposed_tags") + if isinstance(tags, str): + tags = [t.strip() for t in tags.split(",") if t.strip()] + if not isinstance(tags, list): + tags = [] + + proposal = Proposal( + source_rel=note.rel_path, + note_type=type_from_folder(folder) or "area", + target_folder=folder, + toc_section=str(note.fm.get("proposed_toc") or ""), + template="Area.md", + confidence=1.0, + method="approved", + proposed_tags=[str(t) for t in tags], + ) + + plan = self.writer.plan(note, proposal) + if not plan.writable: + return [f"**blocked** `{note.filename}` — {plan.blocked_reason}"] + + if self.dry_run: + return [f"**would file** `{note.filename}` → `{folder}`"] + + ok, log = self.writer.apply(plan, source_path=note.path) + lines = [f"**filed** `{note.filename}` → `{plan.dest_rel}` (type {plan.note_type})"] + if not ok: + lines = [f"**FAILED** `{note.filename}`: " + "; ".join(log)] + return lines + + # --------------------------------------------------------------------- loop + + def run_once(self) -> list[str]: + """One pass over the inbox. Returns human-readable log lines.""" + actions: list[str] = [] + for note in self.vault.inbox_notes(self.inbox): + try: + if is_approved(note): + actions += self._file(note) + elif is_planned(note) and not proposal_is_stale(note): + # Awaiting your click. Still verify the index line matches the + # folder, in case the folder was edited by hand. + actions += self._sync_toc(note) + continue + else: + if is_planned(note): + actions.append( + f"**re-proposing** `{note.filename}` — you edited it" + ) + proposal = self.classifier.classify(note) + if proposal.method in ("blocked", "jev-error"): + actions.append( + f"**held** `{note.filename}` — " + + "; ".join(proposal.warnings) + ) + elif not proposal.target_folder: + actions.append(f"**held** `{note.filename}` — no destination determined") + else: + actions += self._write_proposal(note, proposal) + except Exception as e: # never let one note kill the loop + actions.append(f"**error** `{note.filename}`: {type(e).__name__}: {e}") + return actions + + def run_forever(self, max_iterations: int | None = None) -> None: + n = 0 + while max_iterations is None or n < max_iterations: + self._log(self.run_once()) + n += 1 + if max_iterations is not None and n >= max_iterations: + break + time.sleep(self.interval) diff --git a/sorter/writer.py b/sorter/writer.py new file mode 100644 index 0000000..36676b1 --- /dev/null +++ b/sorter/writer.py @@ -0,0 +1,517 @@ +""" +Phase C — the writer. + +Turns a Proposal into a concrete, reviewable Plan, then (only when asked) +executes it. Every operation is expressed as a Change so a dry run shows +exactly what would happen. + +Safety contract +--------------- +* `plan()` never touches the disk. +* `apply()` refuses unless the plan is `writable`. +* Notes named `Untitled*` are never moved. +* Frontmatter is MERGED — user tags, aliases and ids survive. +* `created` is preserved; only `updated` is refreshed. +* The note BODY is preserved verbatim (no rewriting of the author's words). +* Writes are atomic (temp file + rename). +* `.trash/` is never touched. + +Taxonomy note (2026-09-24): type mirrors folder. A note's `type` always equals +the type implied by its destination folder — see models.type_from_folder. +""" + +from __future__ import annotations + +import os +import re +import tempfile +from dataclasses import dataclass, field +from datetime import date +from pathlib import Path +from typing import Any + +from . import frontmatter +from .classify import TYPE_TO_TEMPLATE, TYPE_TO_TOC +from .models import Note, Proposal, type_from_folder +from .vault import Vault + +# Keys carried over from the source note untouched. +PRESERVE_KEYS = ("created", "aliases", "id") + +# Fields the sorter OWNS and rewrites. Everything else in the source frontmatter +# is preserved verbatim, so custom fields (skills:, tools:, source_url:, ...) +# survive a re-file. +MANAGED_FIELDS = { + "title", "type", "status", "areas", "projects", "tags", + "created", "updated", +} +WATCHER_FIELDS = { + "proposed_folder", "proposed_toc", "proposed_tags", "confidence", + "approved", "proposed_for", +} + +# Output ordering for the new frontmatter. +FIELD_ORDER = [ + "title", "type", "status", + "areas", "projects", + "created", "updated", "tags", "aliases", "id", +] + +# Locations deliberately NOT listed in any Table of Contents. Landing here is a +# valid outcome, so it is not reported as a warning. +NON_INDEXED_PREFIXES = ( + "010 inbox", + "000 daily", + "500 archive", + "Clippings", + "100 Table of Contents", +) + +# Headings that are navigation scaffolding, never a filing target. +META_SECTIONS = { + "dataview queries", "navigation", "quick reference", + "vault overview", "project groups", "on hold / archived", +} + +# Fallback section per bucket when the folder cannot be matched. +TYPE_FALLBACK_SECTION: dict[str, str] = { + "project": "Active Projects", + "area": "Other Topics", + "family": "Other Topics", +} + +# Folder → (index note, section hint). Section names resolve loosely, so +# "Home Assistant" matches "Home Assistant (4 devices)". +FOLDER_SECTION_HINTS: list[tuple[str, str, str]] = [ + # ---- projects ---- + ("200 projects/210 AI Resume", "100 Table of Contents/Projects.md", "AI Resume"), + ("200 projects/220 Web Host Migration", "100 Table of Contents/Projects.md", "Web Host Migration"), + ("200 projects/240 Web Sites", "100 Table of Contents/Projects.md", "Web Sites"), + ("200 projects/250 Electronics", "100 Table of Contents/Projects.md", "Electronics"), + ("200 projects/260 Ideas to Build Create Make", "100 Table of Contents/Projects.md", "Ideas"), + ("200 projects/Tools Software WebUI", "100 Table of Contents/Projects.md", "Tools & Software"), + ("200 projects", "100 Table of Contents/Projects.md", "Active Projects"), + # ---- areas ---- + ("300 areas/310 Home Assistant", "100 Table of Contents/Areas.md", "Home Assistant"), + ("300 areas/330 IOT Research Tools", "100 Table of Contents/Areas.md", "IoT Research Tools"), + ("300 areas/340 TUI Web Dev Coding Tools", "100 Table of Contents/Areas.md", "TUI"), + ("300 areas/350 AI Pi Coding Agent", "100 Table of Contents/Areas.md", "AI & Coding Agent"), + ("300 areas/355 AI Research and Techniques", "100 Table of Contents/Areas.md", "AI Research & Techniques"), + ("300 areas/360 Dev-Ops Network Computers", "100 Table of Contents/Areas.md", "Dev-Ops & Network Computers"), + ("300 areas/370 Settings Configs", "100 Table of Contents/Areas.md", "Dev-Ops & Network Computers"), + ("300 areas/380 Backup-system", "100 Table of Contents/Areas.md", "Dev-Ops & Network Computers"), + ("300 areas/390 Web Pages and Tools To Try", "100 Table of Contents/Areas.md", "Web Pages & Tools To Try"), + ("300 areas/395 AI Tools to try", "100 Table of Contents/Areas.md", "AI Tools To Try"), + ("300 areas", "100 Table of Contents/Areas.md", "Other Topics"), + # ---- family ---- + ("400 Personal Family/410 Business Ideas", "100 Table of Contents/Personal Family.md", "Business Ideas"), + ("400 Personal Family/430 Movies Books Plays Games", "100 Table of Contents/Personal Family.md", "Movies & Games"), + ("400 Personal Family/450 Recipes", "100 Table of Contents/Personal Family.md", "Recipes"), + ("400 Personal Family/460 Family Archive", "100 Table of Contents/Personal Family.md", "Family Archive"), + ("400 Personal Family/470 Holidays Travel", "100 Table of Contents/Personal Family.md", "Holidays & Travel"), + ("400 Personal Family/480 Art Music Photography", "100 Table of Contents/Personal Family.md", "Art, Music & Photography"), + ("400 Personal Family", "100 Table of Contents/Personal Family.md", "Other Topics"), +] + +_SECTION_STOPWORDS = {"the", "and", "for", "with", "from", "try", "to", "of", "all"} + +_HEADING = re.compile(r"^(#{2,3})\s+(.+?)\s*$", re.MULTILINE) + + +def _tokens(text: str) -> set[str]: + return { + w for w in re.findall(r"[a-z0-9]+", text.lower()) + if w not in _SECTION_STOPWORDS and len(w) > 2 + } + + +def _token_overlap(a: set[str], b: set[str]) -> int: + """Token overlap tolerant of plurals — 'device' matches 'devices'.""" + score = 0 + for x in a: + for y in b: + if x == y or (len(x) >= 5 and len(y) >= 5 and x[:5] == y[:5]): + score += 1 + break + return score + + +@dataclass +class Change: + kind: str # create | update | remove + path: str + detail: str = "" + + def __str__(self) -> str: + sym = {"create": "+", "update": "~", "remove": "-"}.get(self.kind, "?") + return f" {sym} {self.kind:7s} {self.path}" + (f" ({self.detail})" if self.detail else "") + + +@dataclass +class Plan: + """A complete, inspectable description of what would happen.""" + + source_rel: str + dest_rel: str + note_title: str + note_type: str + status: str + toc_rel: str | None + toc_section: str | None + frontmatter: dict[str, Any] = field(default_factory=dict) + body: str = "" + changes: list[Change] = field(default_factory=list) + warnings: list[str] = field(default_factory=list) + writable: bool = True + blocked_reason: str | None = None + + def render(self) -> str: + lines = [f"Plan for: {self.source_rel}", ""] + lines.append(f" → {self.dest_rel}") + lines.append(f" title : {self.note_title}") + lines.append(f" type : {self.note_type} status: {self.status}") + if self.toc_rel and self.toc_section: + lines.append(f" toc : {self.toc_rel} → {self.toc_section}") + lines.append("") + lines.append(" frontmatter:") + keys = [k for k in FIELD_ORDER if k in self.frontmatter] + keys += [k for k in self.frontmatter if k not in keys] + for k in keys: + v = self.frontmatter[k] + if isinstance(v, list): + shown = ", ".join(str(x) for x in v) if v else "(empty)" + else: + shown = v + lines.append(f" {k}: {shown}") + lines.append("") + lines.append(" changes:") + for c in self.changes: + lines.append(str(c)) + if self.warnings: + lines.append("") + for w in self.warnings: + lines.append(f" ⚠ {w}") + if not self.writable: + lines.append("") + lines.append(f" ⛔ BLOCKED: {self.blocked_reason}") + return "\n".join(lines) + + +class Writer: + """Builds and applies Plans.""" + + def __init__(self, vault: Vault, today: str | None = None): + self.vault = vault + self.today = today or date.today().isoformat() + + # ---------------------------------------------------------------- TOC lookup + + def _headings(self, toc_rel: str) -> list[tuple[int, str, int]]: + """Return [(line_index, text, level)] for all ## and ### headings.""" + path = self.vault.root / toc_rel + if not path.exists(): + return [] + lines = path.read_text(encoding="utf-8").split("\n") + out = [] + for i, line in enumerate(lines): + m = re.match(r"^(#{2,3})\s+(.+?)\s*$", line) + if m: + out.append((i, m.group(2).strip(), len(m.group(1)))) + return out + + def _usable_sections(self, toc_rel: str) -> list[str]: + """ + Section headings that can receive an entry. + + Excludes meta sections and everything nested under 'Dataview Queries'. + """ + headings = self._headings(toc_rel) + out: list[str] = [] + in_queries = False + for _i, text, level in headings: + low = text.lower() + if level == 2: + in_queries = low in ("dataview queries",) + if in_queries or low in META_SECTIONS: + continue + out.append(text) + else: # ### + if in_queries or low in META_SECTIONS: + continue + out.append(text) + return out + + def _match_section(self, toc_rel: str, wanted: str) -> str | None: + """Resolve a loosely-specified section name to a real heading.""" + usable = self._usable_sections(toc_rel) + want = wanted.strip().lower() + + for s in usable: # exact + if s.lower() == want: + return s + for s in usable: # containment (handles "(4 devices)") + if want in s.lower() or s.lower() in want: + return s + + target = _tokens(wanted) # fuzzy + best, best_score = None, 0 + for s in usable: + sc = _token_overlap(target, _tokens(s)) + if sc > best_score: + best, best_score = s, sc + return best if best_score >= 1 else None + + def resolve_toc(self, note_type: str, folder: str) -> tuple[str, str | None, list[str]]: + """ + Decide which index note and section should list a note. + + Order: explicit folder hint → type fallback → warning. + """ + warnings: list[str] = [] + + if folder and any( + folder == p or folder.startswith(p + "/") for p in NON_INDEXED_PREFIXES + ): + return "", None, [] + + for prefix, toc_rel, wanted in FOLDER_SECTION_HINTS: + if folder == prefix or folder.startswith(prefix + "/"): + section = self._match_section(toc_rel, wanted) + if section: + return toc_rel, section, warnings + warnings.append( + f"Section '{wanted}' not found in {toc_rel} — index will not be updated" + ) + return toc_rel, None, warnings + + toc_rel = TYPE_TO_TOC.get(note_type, "") + if not toc_rel: + return "", None, ["No TOC target known for this type"] + + fb = TYPE_FALLBACK_SECTION.get(note_type) + if fb: + section = self._match_section(toc_rel, fb) + if section: + return toc_rel, section, warnings + + warnings.append(f"No suitable section in {toc_rel} — index will not be updated") + return toc_rel, None, warnings + + # ------------------------------------------------------------------ helpers + + def _derive_area(self, folder: str, note: Note) -> list[str] | None: + """ + If the destination belongs to a canonical area note, return [[Area]]. + + Only when a note with that exact title already exists — new areas are an + explicit, approved action. + """ + m = re.match(r"^300 areas/\d+\s+(.+)$", folder) + if not m: + return None + candidate = m.group(1).strip() + if candidate.lower() in self.vault.known_titles(): + return [f"[[{candidate}]]"] + return None + + def build_frontmatter(self, note: Note, proposal: Proposal) -> dict[str, Any]: + """Merge the source note's frontmatter with the new classification.""" + fm: dict[str, Any] = {} + + # 0. Preserve every custom field the sorter does not manage. This keeps + # things like skills:/tools:/source_url: intact across a re-file. + for key, value in (note.fm or {}).items(): + if key in MANAGED_FIELDS or key in WATCHER_FIELDS: + continue + if value not in (None, "", []): + fm[key] = value + + for key in PRESERVE_KEYS: + if note.fm.get(key) not in (None, "", []): + fm[key] = note.fm[key] + + fm["title"] = note.title or note.stem + + # TYPE MIRRORS FOLDER — always. + fm["type"] = type_from_folder(proposal.target_folder) or proposal.note_type + fm["status"] = "active" + + existing_areas = note.fm.get("areas") + if isinstance(existing_areas, list) and existing_areas: + fm["areas"] = existing_areas + else: + derived = self._derive_area(proposal.target_folder, note) + if derived: + fm["areas"] = derived + + existing_projects = note.fm.get("projects") + if isinstance(existing_projects, list) and existing_projects: + fm["projects"] = existing_projects + + tags: list[str] = [] + # An approved proposal's tags win over whatever the inbox note had. + source_tags = proposal.proposed_tags or note.tags or [] + for t in source_tags: + t = str(t).strip() + if t and t not in tags: + tags.append(t) + fm["tags"] = tags + + fm["created"] = note.fm.get("created") or self.today + fm["updated"] = self.today + return fm + + # ------------------------------------------------------------------ planning + + def plan(self, note: Note, proposal: Proposal) -> Plan: + """Compute the full Plan. Never writes.""" + title = note.title or note.stem + dest_folder = proposal.target_folder + dest_rel = f"{dest_folder}/{note.filename}" if dest_folder else note.filename + + blocked: str | None = None + warnings: list[str] = list(proposal.warnings) + + if note.stem.lower().startswith("untitled"): + blocked = "Filename begins with 'Untitled' — never moved automatically" + elif not dest_folder: + blocked = "No destination folder determined" + elif not type_from_folder(dest_folder): + blocked = f"Destination is not inside a known bucket: {dest_folder}" + + dest_path = self.vault.root / dest_rel + if not blocked and dest_path.exists() and str(dest_path) != note.path: + blocked = f"Destination already exists: {dest_rel}" + + toc_rel, toc_section = "", "" + if not blocked: + toc_rel, toc_section, toc_warnings = self.resolve_toc( + proposal.note_type, dest_folder + ) + warnings.extend(toc_warnings) + if toc_rel and toc_section: + existing = self.vault.root / toc_rel + if existing.exists() and f"[[{title}]]" in existing.read_text(encoding="utf-8"): + warnings.append(f"Already listed in {toc_rel} — no TOC change needed") + + fm = self.build_frontmatter(note, proposal) if not blocked else {} + + changes: list[Change] = [] + same_place = dest_rel == note.rel_path + if same_place: + changes.append(Change("update", dest_rel, f"rewrite frontmatter (type={fm.get('type', '?')})")) + else: + changes.append(Change("create", dest_rel, f"type={fm.get('type', '?')}")) + if toc_rel and toc_section: + changes.append(Change("update", toc_rel, f"add [[{title}]] under '{toc_section}'")) + elif toc_rel: + changes.append(Change("update", toc_rel, "SKIPPED — section missing")) + if not same_place: + changes.append(Change("remove", note.rel_path, "moved out of inbox")) + + return Plan( + source_rel=note.rel_path, + dest_rel=dest_rel, + note_title=title, + note_type=fm.get("type", "") or proposal.note_type, + status="active", + toc_rel=toc_rel or None, + toc_section=toc_section or None, + frontmatter=fm, + body=note.body, + changes=changes, + warnings=warnings, + writable=blocked is None, + blocked_reason=blocked, + ) + + # ------------------------------------------------------------------- writing + + def _atomic_write(self, path: Path, text: str) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + fd, tmp = tempfile.mkstemp(dir=str(path.parent), prefix=".sorter-", suffix=".tmp") + try: + with os.fdopen(fd, "w", encoding="utf-8") as fh: + fh.write(text) + os.replace(tmp, path) + except Exception: + if os.path.exists(tmp): + os.unlink(tmp) + raise + + def insert_toc_entry(self, toc_rel: str, section: str, title: str) -> bool: + """Insert `- [[title]]` at the end of a ## or ### section.""" + path = self.vault.root / toc_rel + text = path.read_text(encoding="utf-8") + + if f"[[{title}]]" in text: + return False + + lines = text.split("\n") + start = level = None + for i, line in enumerate(lines): + m = re.match(r"^(#{2,3})\s+(.+?)\s*$", line) + if m and m.group(2).strip() == section: + start, level = i, len(m.group(1)) + break + if start is None: + return False + + # Section ends at the next heading of the same or higher level. + end = len(lines) + for j in range(start + 1, len(lines)): + m = re.match(r"^(#{2,3})\s+", lines[j]) + if m and len(m.group(1)) <= level: + end = j + break + + insert_at = end + while insert_at > start + 1 and lines[insert_at - 1].strip() == "": + insert_at -= 1 + + entry = f"- [[{title}]]" + if entry in lines[start:end]: + return False + + lines.insert(insert_at, entry) + self._atomic_write(path, "\n".join(lines)) + return True + + def apply(self, plan: Plan, source_path: str | None = None) -> tuple[bool, list[str]]: + """Execute a Plan. Returns (success, log).""" + log: list[str] = [] + + if not plan.writable: + return False, [f"refused: {plan.blocked_reason}"] + + src = Path(source_path or str(self.vault.root / plan.source_rel)) + dest = self.vault.root / plan.dest_rel + + text = frontmatter.render(plan.frontmatter, plan.body, order=FIELD_ORDER) + + try: + self._atomic_write(dest, text) + log.append(f"wrote {plan.dest_rel}") + except Exception as e: + return False, [f"FAILED writing destination: {e}"] + + if plan.toc_rel and plan.toc_section: + try: + if self.insert_toc_entry(plan.toc_rel, plan.toc_section, plan.note_title): + log.append(f"updated {plan.toc_rel} ({plan.toc_section})") + else: + log.append(f"{plan.toc_rel}: no change (already listed or section missing)") + except Exception as e: + log.append(f"WARNING: TOC update failed: {e}") + + if src.exists() and str(src.resolve()) != str(dest.resolve()): + if ".trash" in src.parts: + log.append("refused to remove file under .trash") + else: + try: + src.unlink() + log.append(f"removed {plan.source_rel}") + except Exception as e: + log.append(f"WARNING: could not remove source: {e}") + + return True, log diff --git a/tests/run_tests.py b/tests/run_tests.py new file mode 100644 index 0000000..11245ca --- /dev/null +++ b/tests/run_tests.py @@ -0,0 +1,406 @@ +#!/usr/bin/env python3 +""" +Functional test suite for obsidian-sorter. + +Runs against an ISOLATED COPY of the vault so the mirror is never touched. +Exercises the full pipeline: classify -> plan -> apply -> verify. + +Usage: + python3 tests/run_tests.py [--vault ./test-vault] [--no-jev] + +With --no-jev only the deterministic tier runs (faster, no API cost), which is +useful for testing the writer mechanics without model calls. +""" + +from __future__ import annotations + +import argparse +import json +import os +import re +import shutil +import subprocess +import sys +import tempfile +from pathlib import Path + +REPO = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(REPO)) + +from sorter import frontmatter # noqa: E402 + +GENERIC_NOTE = """--- +created: 2026-09-19 17:56 +modified: 2026-09-19 17:56 +type: note +tags: {tags} +aliases: [] +--- +# [[{title}]] + +{body} +""" + +GENERIC_WITH_USERDATA = """--- +created: 2026-08-01 09:00 +modified: 2026-08-02 10:00 +type: note +tags: + - my-own-tag + - research +aliases: + - alt name +id: 1779253896-DPHX +areas: + - "[[Home Assistant]]" +--- +# [[{title}]] + +{body} +""" + + +# ---------------------------------------------------------------- test cases + +CASES = [ + # The BUCKET (project/area/family) is asserted strictly — it is reliable. + # The SUB-FOLDER is asserted only with --strict-folders, because JEV is + # probabilistic (~74% folder accuracy) and exact folders can vary run to run. + dict( + name="recipe -> family", + filename="Sourdough Bread Recipe.md", + body="My sourdough recipe: 500g flour, 350g water, 100g starter, 10g salt. " + "Bulk ferment 4 hours, shape, prove overnight, bake at 230C.", + expect_type="family", + expect_bucket="400 Personal Family", + expect_subfolder="450 Recipes", + ), + dict( + name="unfinished electronics build -> project", + filename="Garage Door Sensor Build.md", + body="Plan to build a garage door sensor using an ESP32 and a reed switch. " + "TODO: order parts, write firmware. Needs to be built before winter.", + expect_type="project", + expect_bucket="200 projects", + expect_subfolder="250 Electronics", + ), + dict( + name="ops documentation -> area", + filename="Nginx Reverse Proxy Notes.md", + body="Documentation for the nginx reverse proxy running on .13. It is live. " + "Configuration reference and TLS certificate renewal steps.", + expect_type="area", + expect_bucket="300 areas", + expect_subfolder="360 Dev-Ops", + ), + dict( + name="family event -> family", + filename="Nan Birthday Party.md", + body="Photos and notes from Nan's 80th birthday at the community hall. " + "Family gathered for speeches and catering.", + expect_type="family", + expect_bucket="400 Personal Family", + expect_subfolder="460 Family Archive", + ), + dict( + name="ai research -> area", + filename="Vector Database Comparison.md", + body="Research notes comparing vector database options for RAG: pgvector " + "vs chroma vs qdrant. Embeddings, recall, operational complexity.", + expect_type="area", + expect_bucket="300 areas", + expect_subfolder="355 AI Research", + ), + # A tag is a human decision — it must be honoured every time. + dict( + name="tagged project -> project (tag wins)", + filename="Tagged Build Job.md", + body="Some reference notes about a thing.", + tags='[project]', + expect_type="project", + expect_bucket="200 projects", + expect_subfolder="200 projects", + ), +] + + +# ------------------------------------------------------------------- helpers + +def run(cmd: list[str], cwd: Path) -> subprocess.CompletedProcess: + return subprocess.run(cmd, cwd=str(cwd), capture_output=True, text=True) + + +def read_fm(path: Path) -> tuple[dict, str]: + text = path.read_text(encoding="utf-8") + fm, body, ok = frontmatter.parse(text) + return fm, body + + +def write_inbox(vault: Path, inbox: str, filename: str, content: str) -> Path: + p = vault / inbox / filename + p.parent.mkdir(parents=True, exist_ok=True) + p.write_text(content, encoding="utf-8") + return p + + +# ---------------------------------------------------------------------- main + +def main() -> int: + ap = argparse.ArgumentParser() + ap.add_argument("--vault", default=str(REPO / "test-vault")) + ap.add_argument("--inbox", default="010 inbox") + ap.add_argument("--no-jev", action="store_true") + ap.add_argument("--strict-folders", action="store_true", + help="also assert exact sub-folders (JEV is probabilistic; may be flaky)") + ap.add_argument("--keep", action="store_true", help="keep the temp vault") + args = ap.parse_args() + + src = Path(args.vault).resolve() + if not src.is_dir(): + print(f"error: vault not found: {src}", file=sys.stderr) + return 1 + + tmp = Path(tempfile.mkdtemp(prefix="sorter-test-")) + vault = tmp / "vault" + print(f"Isolated test vault: {vault}\n") + shutil.copytree(src, vault, symlinks=False, + ignore=shutil.ignore_patterns(".git", ".trash", ".venv", + "graphify-out*", ".obsidian")) + + jev_flags = [] if args.no_jev else ["--jev"] + passed = failed = 0 + details: list[str] = [] + + try: + # ---- inject notes ------------------------------------------------- + for c in CASES: + write_inbox(vault, args.inbox, c["filename"], GENERIC_NOTE.format( + title=c["filename"][:-3], body=c["body"], tags=c.get("tags", "[]"))) + + # note with pre-existing user data (merge test) + write_inbox(vault, args.inbox, "Merge Test Note.md", + GENERIC_WITH_USERDATA.format( + title="Merge Test Note", + body="A recipe and family note that also carries user data.")) + CASES.append(dict( + name="frontmatter merge", + filename="Merge Test Note.md", + expect_type=None, expect_folder_contains=None, expect_toc_section=None, + )) + + # ---- dry run must not write --------------------------------------- + before = set(p.name for p in (vault / args.inbox).glob("*.md") + if not p.name.startswith("_")) + r = run([sys.executable, "-m", "sorter", "--vault", str(vault), + "--inbox", args.inbox, "sort-inbox"] + jev_flags, REPO) + after = set(p.name for p in (vault / args.inbox).glob("*.md") + if not p.name.startswith("_")) + if before == after and r.returncode == 0: + print("PASS dry run leaves the vault untouched") + passed += 1 + else: + print(f"FAIL dry run changed files (rc={r.returncode})") + print(r.stdout[-800:], r.stderr[-400:]) + failed += 1 + + # ---- apply --------------------------------------------------------- + r = run([sys.executable, "-m", "sorter", "--vault", str(vault), + "--inbox", args.inbox, "sort-inbox"] + jev_flags + ["--apply", "--yes"], REPO) + if r.returncode != 0: + print(f"FAIL apply exited {r.returncode}") + print(r.stdout[-2000:]) + print(r.stderr[-1000:]) + return 1 + + # ---- verify each case --------------------------------------------- + for c in CASES: + name = c["name"] + fname = c["filename"] + problems: list[str] = [] + + # found somewhere outside the inbox? + matches = [p for p in vault.rglob(fname) + if args.inbox not in p.parts and ".trash" not in p.parts] + if not matches: + details.append(f"FAIL {name}: '{fname}' not filed anywhere") + failed += 1 + continue + dest = matches[0] + fm, body = read_fm(dest) + + if c.get("expect_type") and fm.get("type") != c["expect_type"]: + problems.append(f"type={fm.get('type')!r} expected {c['expect_type']!r}") + if fm.get("status") != "active": + problems.append(f"status={fm.get('status')!r}") + if not fm.get("title"): + problems.append("no title") + if re.search(r"\{\{.*?\}\}", json.dumps(fm)): + problems.append("unfilled template placeholder in frontmatter") + if not body.strip(): + problems.append("body was emptied") + if c.get("expect_bucket"): + if c["expect_bucket"].lower() not in str(dest.parent).lower(): + problems.append(f"folder={dest.parent.name!r} expected to be in " + f"bucket {c['expect_bucket']!r}") + if args.strict_folders and c.get("expect_subfolder"): + if c["expect_subfolder"].lower() not in str(dest.parent).lower(): + problems.append(f"folder={dest.parent.name!r} expected " + f"{c['expect_subfolder']!r}") + if c.get("expect_toc_section"): + idx = vault / "100 Table of Contents" + listed = False + for f in idx.glob("*.md"): + if f"[[{fname[:-3]}]]" in f.read_text(encoding="utf-8"): + listed = True + if not listed: + problems.append(f"not listed in any index (expected ~{c['expect_toc_section']!r})") + + if problems: + details.append(f"FAIL {name}: " + "; ".join(problems)) + failed += 1 + else: + print(f"PASS {name} -> {dest.relative_to(vault)}") + passed += 1 + + # ---- unit: safety guards (in-process, no API calls) ------------- + from sorter.models import Proposal + from sorter.vault import Vault as _V + from sorter.writer import Writer as _W + + uv = _V(str(vault)) + uw = _W(uv) + dummy = Proposal( + source_rel="x", note_type="area", + target_folder="300 areas/360 Dev-Ops Network Computers", + toc_section="", template="Area.md", confidence=0.9, method="test", + ) + + # guard 1: Untitled is never moved + up = write_inbox(vault, args.inbox, "Untitled 99.md", + GENERIC_NOTE.format(title="Untitled 99", body="stray content", tags="[]")) + uplan = uw.plan(uv.read_note(up), dummy) + if not uplan.writable and "Untitled" in (uplan.blocked_reason or ""): + print("PASS guard: Untitled note is blocked") + passed += 1 + else: + details.append("FAIL guard: Untitled note was NOT blocked") + failed += 1 + + # guard 2: existing destination blocks the plan + dest_dir = vault / "300 areas/360 Dev-Ops Network Computers" + dest_dir.mkdir(parents=True, exist_ok=True) + (dest_dir / "Conflict Test.md").write_text( + "---\ntitle: Conflict Test\ntype: subject\nstatus: active\n---\n\n# Conflict\n", + encoding="utf-8") + cp = write_inbox(vault, args.inbox, "Conflict Test.md", + GENERIC_NOTE.format(title="Conflict Test", body="duplicate", tags="[]")) + cplan = uw.plan(uv.read_note(cp), dummy) + if not cplan.writable and "already exists" in (cplan.blocked_reason or ""): + print("PASS guard: existing destination blocks the plan") + passed += 1 + else: + details.append(f"FAIL guard: destination collision not caught ({cplan.blocked_reason})") + failed += 1 + + # guard 3: .trash is never removed + trashed = uw.vault.root / ".trash" + if not trashed.exists(): + trashed.mkdir(parents=True, exist_ok=True) + (trashed / "old.md").write_text("---\ntype: subject\n---\n\nx\n", encoding="utf-8") + tp = uw.vault.root / ".trash" / "old.md" + tnote = uv.read_note(tp) + if tnote is not None: + tplan = uw.plan(tnote, dummy) + _ok, log = uw.apply(tplan, source_path=str(tp)) + if tp.exists() and any(".trash" in l for l in log): + print("PASS guard: .trash file is never removed") + passed += 1 + else: + details.append("FAIL guard: .trash file was modified/removed") + failed += 1 + + # cleanup the injected guard fixtures + for p in (up, cp, dest_dir / "Conflict Test.md", dest_dir / "old.md", tp): + try: + p.unlink() + except OSError: + pass + + # ---- inbox cleared -------------------------------------------------- + left = [p for p in (vault / args.inbox).glob("*.md") + if not p.name.startswith("_")] + if not left: + print("PASS inbox emptied") + passed += 1 + else: + print(f"FAIL inbox still has: {[p.name for p in left]}") + failed += 1 + + # ---- merge assertions ---------------------------------------------- + merge_matches = [p for p in vault.rglob("Merge Test Note.md") + if args.inbox not in p.parts] + if merge_matches: + fm, _ = read_fm(merge_matches[0]) + mp = [] + if "my-own-tag" not in (fm.get("tags") or []): + mp.append("user tag 'my-own-tag' lost") + if "research" not in (fm.get("tags") or []): + mp.append("user tag 'research' lost") + if fm.get("id") != "1779253896-DPHX": + mp.append("id lost") + if fm.get("created") != "2026-08-01 09:00": + mp.append(f"created changed to {fm.get('created')!r}") + if not fm.get("aliases"): + mp.append("aliases lost") + if mp: + details.append("FAIL frontmatter merge: " + "; ".join(mp)) + failed += 1 + else: + print("PASS frontmatter merge (tags, aliases, id, created preserved)") + passed += 1 + + # ---- idempotency ---------------------------------------------------- + toc_before = {f.name: f.read_text(encoding="utf-8") + for f in (vault / "100 Table of Contents").glob("*.md")} + run([sys.executable, "-m", "sorter", "--vault", str(vault), + "--inbox", args.inbox, "sort-inbox"] + jev_flags + ["--apply", "--yes"], REPO) + toc_after = {f.name: f.read_text(encoding="utf-8") + for f in (vault / "100 Table of Contents").glob("*.md")} + if toc_before == toc_after: + print("PASS idempotent (second run changed no index file)") + passed += 1 + else: + changed = [k for k in toc_before if toc_before[k] != toc_after.get(k)] + details.append(f"FAIL second run altered index files: {changed}") + failed += 1 + + # ---- duplicate check ------------------------------------------------- + idx_text = "\n".join(f.read_text(encoding="utf-8") + for f in (vault / "100 Table of Contents").glob("*.md")) + dups = [c["filename"][:-3] for c in CASES + if idx_text.count(f"[[{c['filename'][:-3]}]]") > 1] + if dups: + details.append(f"FAIL duplicate index entries: {dups}") + failed += 1 + else: + print("PASS no duplicate index entries") + passed += 1 + + finally: + if args.keep: + print(f"\n(temp vault kept at {vault})") + else: + shutil.rmtree(tmp, ignore_errors=True) + + print(f"\n{'='*64}") + print(f"PASSED: {passed} FAILED: {failed}") + if details: + print("\nDetails:") + for d in details: + print(f" {d}") + print("=" * 64) + return 0 if failed == 0 else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/normalize_tags.py b/tools/normalize_tags.py new file mode 100644 index 0000000..a0a8d8e --- /dev/null +++ b/tools/normalize_tags.py @@ -0,0 +1,209 @@ +#!/usr/bin/env python3 +""" +One-off tag normalisation for the Obsidian vault. + +Rules (confirmed 2026-09-24): + * lowercase everything + * 'tools' -> 'tool' + * split comma-joined pseudo-tags into real tags + * fix known typos + * spaces -> hyphens + * drop retired-type tags + +Usage: + python3 tools/normalize_tags.py --vault ./test-vault # dry run + python3 tools/normalize_tags.py --vault ~/obsidian --apply +""" + +from __future__ import annotations + +import argparse +import os +import re +import sys +from collections import Counter, defaultdict + +SKIP_DIRS = {".git", ".trash", ".venv", ".obsidian", "templates", + "graphify-out", "graphify-out-temp", "__pycache__"} + +# Tags that came from the retired type list — drop them. +REMOVE = { + "subject", "website", "overview", "log", "runbook", "media", "area", + "machine", "service", "container", "storage", "backup-system", + "network-component", "person", "note", "filesmd", +} + +# Renames (including merges and typo fixes). +RENAME = { + "tools": "tool", + "exrcise": "exercise", + "rasperry-pi": "raspberry-pi", + "recipie": "recipe", + "wikunja": "vikunja", + "svelt": "svelte", +} + + +def fix_tag(raw: str): + """Normalise one tag. Returns the new tag, or None if it should be dropped.""" + t = raw.strip().strip('"').strip("'").strip() + if not t: + return None + t = t.lower() + t = RENAME.get(t, t) + t = re.sub(r"\s+", "-", t) # spaces -> hyphens + t = t.strip("-") + if not t or t in REMOVE: + return None + return t + + +def normalise(tags: list[str]) -> list[str]: + """Normalise a list, splitting comma-joined pseudo-tags, preserving order.""" + out: list[str] = [] + seen: set[str] = set() + for raw in tags: + parts = raw.split(",") if "," in raw else [raw] + for part in parts: + fixed = fix_tag(part) + if fixed and fixed not in seen: + seen.add(fixed) + out.append(fixed) + return out + + +_TAGS_LINE = re.compile(r"^tags:\s*(.*)$") + + +def rewrite_note(text: str) -> tuple[str, list[str], list[str]]: + """ + Rewrite the `tags:` field in a note's frontmatter. + + Returns (new_text, old_tags, new_tags). new_text == text when nothing changed. + """ + m = re.match(r"^(---\s*\n)(.*?)(\n---\s*\n?)", text, re.DOTALL) + if not m: + return text, [], [] + + header, block, close = m.group(1), m.group(2), m.group(3) + body = text[m.end():] + lines = block.split("\n") + + out: list[str] = [] + old: list[str] = [] + i = 0 + found = False + + while i < len(lines): + line = lines[i] + tm = _TAGS_LINE.match(line) + if not tm: + out.append(line) + i += 1 + continue + + found = True + value = tm.group(1).strip() + if value.startswith("["): + inner = value[1:-1] + old = [p.strip().strip('"').strip("'") for p in inner.split(",") if p.strip()] + i += 1 + else: + j = i + 1 + while j < len(lines) and re.match(r"^\s*-\s", lines[j]): + item = re.match(r"^\s*-\s*(.*)$", lines[j]).group(1) + old.append(item.strip().strip('"').strip("'")) + j += 1 + i = j + + new = normalise(old) + if new: + out.append("tags:") + for t in new: + out.append(f" - {t}") + else: + out.append("tags: []") + + if not found: + return text, [], [] + + new_block = "\n".join(out) + if new_block == block: + return text, old, old + return header + new_block + close + body, old, normalise(old) + + +def main() -> int: + ap = argparse.ArgumentParser() + ap.add_argument("--vault", default="~/obsidian") + ap.add_argument("--apply", action="store_true", help="write changes (default: dry run)") + ap.add_argument("--quiet", action="store_true") + args = ap.parse_args() + + root = os.path.expanduser(args.vault) + if not os.path.isdir(root): + print(f"error: vault not found: {root}", file=sys.stderr) + return 1 + + changed = 0 + scanned = 0 + removed = Counter() + renamed = Counter() + touched: list[str] = [] + + for dirpath, dirnames, filenames in os.walk(root): + dirnames[:] = [d for d in dirnames if d not in SKIP_DIRS] + for fn in sorted(filenames): + if not fn.endswith(".md"): + continue + p = os.path.join(dirpath, fn) + rel = os.path.relpath(p, root) + try: + text = open(p, encoding="utf-8").read() + except (OSError, UnicodeDecodeError): + continue + scanned += 1 + new_text, old, new = rewrite_note(text) + if new_text == text: + continue + changed += 1 + touched.append(rel) + + # Reporting: a tag is only "dropped" if none of its parts survived. + new_set = set(new) + for t in old: + ts = t.strip() + if not ts: + continue + parts = ts.split(",") if "," in ts else [ts] + fixed_parts = [f for f in (fix_tag(p) for p in parts) if f] + if not any(f in new_set for f in fixed_parts): + removed[ts.lower()] += 1 + else: + for f in fixed_parts: + if f != ts.lower(): + renamed[f] += 1 + break + + if not args.quiet: + print(f" {rel}") + print(f" before: {', '.join(old) if old else '(none)'}") + print(f" after : {', '.join(new) if new else '(none)'}") + if args.apply: + with open(p, "w", encoding="utf-8") as fh: + fh.write(new_text) + + print(f"\n{'APPLIED' if args.apply else 'DRY RUN'} — {changed} of {scanned} notes changed") + if removed: + print("\ntags dropped:") + for t, c in removed.most_common(): + print(f" {c:3d} {t}") + if renamed: + print("\ntags introduced by renaming:") + for t, c in renamed.most_common(15): + print(f" {c:3d} {t}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())