""" 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