325 lines
13 KiB
Python
325 lines
13 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Sync Cyberpunk notes from the Obsidian vault into the Git repo.
|
||
|
||
Conversions (Obsidian -> repo):
|
||
- Removes YAML front matter and back-links like "> [[Home]]"
|
||
- Removes private sections (by heading, or between %% private-start %% / %% private-end %%)
|
||
- [[Note]] / [[Note|Alias]] / [[Note#Heading]] -> [Note](./Repo-File.md#heading)
|
||
- [[#Heading]] -> [Heading](#heading)
|
||
- ![[image.jpg]] / ![[image.jpg|800]] -> <img src="./assets/image.jpg" alt="image" width="...">
|
||
- Copies referenced images from the vault into assets/ (only if new or changed)
|
||
- ==highlight== -> plain text (not supported outside Obsidian)
|
||
- Cleans heading artifacts ("# # Title", "## > Title"), stray BOMs, trailing "---"
|
||
- Formats the result with Prettier (same style as format-on-save in VS Code)
|
||
|
||
Setup:
|
||
Copy scripts/.env.example to scripts/.env and set your vault path.
|
||
scripts/.env is git-ignored; everything else in scripts/ is versioned.
|
||
|
||
Usage:
|
||
python scripts/sync-from-obsidian.py # dry run: shows what would change
|
||
python scripts/sync-from-obsidian.py -d / --diff # dry run + unified diff
|
||
python scripts/sync-from-obsidian.py -w / --write # write the files
|
||
|
||
Nothing is committed. Review with `git diff`, then commit yourself.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import difflib
|
||
import filecmp
|
||
import json
|
||
import os
|
||
import re
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
SCRIPT_DIR = Path(__file__).resolve().parent
|
||
CONFIG_PATH = SCRIPT_DIR / "sync-config.json"
|
||
ENV_PATH = SCRIPT_DIR / ".env"
|
||
|
||
WIKILINK_IMAGE = re.compile(r"!\[\[([^\]|]+?\.(?:png|jpe?g|gif|webp|svg))(?:\|(\d+))?\]\]", re.I)
|
||
WIKILINK = re.compile(r"(?<!!)\[\[([^\]|#]*)(?:#([^\]|]+))?(?:\|([^\]]+))?\]\]")
|
||
FRONT_MATTER = re.compile(r"\A---\r?\n.*?\r?\n---[ \t]*\r?\n", re.S)
|
||
HEADING = re.compile(r"^(#{1,6})\s+(.*?)\s*$")
|
||
HIGHLIGHT = re.compile(r"(?<!`)==(?=\S)([^=\n]+?)(?<=\S)==(?!`)")
|
||
PRIVATE_START = re.compile(r"^\s*%%\s*private-start\s*%%\s*$", re.I)
|
||
PRIVATE_END = re.compile(r"^\s*%%\s*private-end\s*%%\s*$", re.I)
|
||
|
||
|
||
# --------------------------------------------------------------------------- helpers
|
||
|
||
def load_env(path: Path) -> dict[str, str]:
|
||
"""Minimal .env reader (KEY=value, # comments, optional quotes). Real env vars take precedence."""
|
||
values: dict[str, str] = {}
|
||
if path.is_file():
|
||
for raw in path.read_text(encoding="utf-8-sig").splitlines():
|
||
line = raw.strip()
|
||
if not line or line.startswith("#") or "=" not in line:
|
||
continue
|
||
key, value = line.split("=", 1)
|
||
value = value.strip()
|
||
if len(value) >= 2 and value[0] == value[-1] and value[0] in "\"'":
|
||
value = value[1:-1]
|
||
values[key.strip()] = value
|
||
values.update({k: v for k, v in os.environ.items() if k in values or k.startswith("OBSIDIAN_")})
|
||
return values
|
||
|
||
|
||
def slugify(text: str) -> str:
|
||
"""GitHub/Gitea-style heading anchor."""
|
||
text = re.sub(r"<[^>]+>", "", text).strip().lower()
|
||
text = re.sub(r"[^\w\- ]", "", text)
|
||
return text.replace(" ", "-")
|
||
|
||
|
||
def heading_text(raw: str) -> str:
|
||
"""Heading text without HTML tags, for matching excluded sections."""
|
||
return re.sub(r"<[^>]+>", "", raw).strip()
|
||
|
||
|
||
def read_text(path: Path) -> str:
|
||
return path.read_text(encoding="utf-8-sig").replace("\r\n", "\n")
|
||
|
||
|
||
# --------------------------------------------------------------------------- conversion
|
||
|
||
class Converter:
|
||
def __init__(self, cfg: dict, vault: Path):
|
||
self.cfg = cfg
|
||
self.vault = vault
|
||
self.notes: dict[str, str] = {Path(n["source"]).stem: n["target"] for n in cfg["notes"]}
|
||
self.default_width = cfg.get("default_image_width", 400)
|
||
self.excluded = {h.lower() for h in cfg.get("exclude_sections", [])}
|
||
self.images: set[str] = set()
|
||
self.warnings: list[str] = []
|
||
|
||
def convert(self, text: str, source: str) -> str:
|
||
text = FRONT_MATTER.sub("", text, count=1)
|
||
text = text.replace("", "")
|
||
lines = self._drop_private(text.split("\n"))
|
||
out, in_code = [], False
|
||
for l in lines:
|
||
if l.lstrip().startswith("```"):
|
||
in_code = not in_code
|
||
if in_code or l.lstrip().startswith("```"):
|
||
out.append(l)
|
||
elif not self._is_backlink(l):
|
||
out.append(self._line(l, source))
|
||
return self._tidy(out)
|
||
|
||
# -- sections
|
||
|
||
def _drop_private(self, lines: list[str]) -> list[str]:
|
||
out, in_code, skip_marker, skip_level = [], False, False, 0
|
||
for line in lines:
|
||
if line.lstrip().startswith("```"):
|
||
in_code = not in_code
|
||
if not in_code:
|
||
if PRIVATE_START.match(line):
|
||
skip_marker = True
|
||
continue
|
||
if PRIVATE_END.match(line):
|
||
skip_marker = False
|
||
continue
|
||
m = HEADING.match(line)
|
||
if m:
|
||
level = len(m.group(1))
|
||
if skip_level and level <= skip_level:
|
||
skip_level = 0
|
||
if not skip_level and heading_text(m.group(2)).lower() in self.excluded:
|
||
skip_level = level
|
||
if not skip_marker and not skip_level:
|
||
out.append(line)
|
||
return out
|
||
|
||
@staticmethod
|
||
def _is_backlink(line: str) -> bool:
|
||
return bool(re.fullmatch(r">\s*\[\[[^\]]+\]\]\s*", line.strip()))
|
||
|
||
# -- inline
|
||
|
||
def _line(self, line: str, source: str) -> str:
|
||
m = HEADING.match(line)
|
||
if m: # "# # Title" / "## > Title" -> clean heading
|
||
title = re.sub(r"^(?:#+\s+|>\s+)", "", m.group(2))
|
||
line = f"{m.group(1)} {title}"
|
||
line = HIGHLIGHT.sub(r"\1", line) # ==text== -> text (Gitea has no highlight)
|
||
line = WIKILINK_IMAGE.sub(self._image, line)
|
||
return WIKILINK.sub(lambda mm: self._link(mm, source), line)
|
||
|
||
def _image(self, m: re.Match) -> str:
|
||
name, width = m.group(1).strip(), m.group(2) or self.default_width
|
||
self.images.add(name)
|
||
return f'<img src="./assets/{name}" alt="{Path(name).stem}" width="{width}">'
|
||
|
||
def _link(self, m: re.Match, source: str) -> str:
|
||
note, anchor, alias = (m.group(1) or "").strip(), m.group(2), m.group(3)
|
||
label = alias or (note if note else anchor)
|
||
if not note: # [[#Heading]]
|
||
return f"[{label}](#{slugify(anchor)})"
|
||
target = self.notes.get(note)
|
||
if not target:
|
||
self.warnings.append(f"{source}: no repo file for [[{note}]], kept as plain text")
|
||
return label
|
||
return f"[{label}](./{target}{'#' + slugify(anchor) if anchor else ''})"
|
||
|
||
# -- layout
|
||
|
||
@staticmethod
|
||
def _tidy(lines: list[str]) -> str:
|
||
out: list[str] = []
|
||
in_code = False
|
||
for line in lines:
|
||
line = line.rstrip()
|
||
if line.startswith("```"):
|
||
in_code = not in_code
|
||
elif not in_code:
|
||
needs_gap = HEADING.match(line) or line == "---"
|
||
if needs_gap and out and out[-1] != "":
|
||
out.append("")
|
||
if out and out[-1] == "" and line == "":
|
||
continue
|
||
if out and out[-1] == "---" and line and not HEADING.match(line):
|
||
out.append("")
|
||
out.append(line)
|
||
while out and out[0] in ("", "---"):
|
||
out.pop(0)
|
||
while out and out[-1] in ("", "---"):
|
||
out.pop()
|
||
return "\n".join(out) + "\n"
|
||
|
||
|
||
class Formatter:
|
||
"""Runs Prettier on the converted Markdown (via stdin), if configured and available."""
|
||
|
||
def __init__(self, cmd: list[str] | None, warnings: list[str]):
|
||
self.cmd, self.warnings, self.disabled = cmd, warnings, not cmd
|
||
if cmd:
|
||
exe = shutil.which(cmd[0]) # resolves npx.cmd on Windows
|
||
if exe:
|
||
self.cmd = [exe, *cmd[1:]]
|
||
else:
|
||
self.disabled = True
|
||
warnings.append(f"formatter '{cmd[0]}' not found, output is not formatted")
|
||
|
||
def __call__(self, text: str, filename: str) -> str:
|
||
if self.disabled:
|
||
return text
|
||
try:
|
||
res = subprocess.run([*self.cmd, "--stdin-filepath", filename], input=text,
|
||
capture_output=True, text=True, encoding="utf-8", timeout=120)
|
||
except (OSError, subprocess.TimeoutExpired) as e:
|
||
res = None
|
||
err = str(e)
|
||
else:
|
||
err = res.stderr.strip()
|
||
if res is None or res.returncode != 0:
|
||
self.disabled = True
|
||
self.warnings.append(f"formatter failed ({err[:200]}), output is not formatted")
|
||
return text
|
||
return res.stdout.replace("\r\n", "\n")
|
||
|
||
|
||
# --------------------------------------------------------------------------- main
|
||
|
||
def find_image(vault: Path, dirs: list[str], name: str) -> Path | None:
|
||
for d in dirs:
|
||
p = vault / d / name
|
||
if p.is_file():
|
||
return p
|
||
hits = [p for p in vault.rglob(name) if ".obsidian" not in p.parts]
|
||
return hits[0] if hits else None
|
||
|
||
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
||
ap.add_argument("-w", "--write", action="store_true", help="write changes (default: dry run)")
|
||
ap.add_argument("-d", "--diff", action="store_true", help="show a unified diff for changed notes")
|
||
ap.add_argument("--vault", type=Path, help="override vault path from the config")
|
||
ap.add_argument("--repo", type=Path, help="override repo path (default: parent of scripts/)")
|
||
args = ap.parse_args()
|
||
|
||
cfg = json.loads(CONFIG_PATH.read_text(encoding="utf-8"))
|
||
env = load_env(ENV_PATH)
|
||
if "OBSIDIAN_EXCLUDE_SECTIONS" in env:
|
||
cfg["exclude_sections"] = [h.strip() for h in env["OBSIDIAN_EXCLUDE_SECTIONS"].split(";") if h.strip()]
|
||
if env.get("OBSIDIAN_IMAGE_DIRS"):
|
||
cfg["image_dirs"] = [d.strip() for d in env["OBSIDIAN_IMAGE_DIRS"].split(";") if d.strip()]
|
||
if not args.vault and not env.get("OBSIDIAN_VAULT"):
|
||
print(f"OBSIDIAN_VAULT is not set. Copy {ENV_PATH.with_name('.env.example').name} "
|
||
f"to {ENV_PATH.name} in {SCRIPT_DIR} and set your vault path.", file=sys.stderr)
|
||
return 1
|
||
vault = args.vault or Path(env["OBSIDIAN_VAULT"])
|
||
repo = args.repo or (Path(cfg["repo"]) if cfg.get("repo") else SCRIPT_DIR.parent)
|
||
if not vault.is_dir() or not repo.is_dir():
|
||
print(f"Vault or repo not found:\n vault: {vault}\n repo: {repo}", file=sys.stderr)
|
||
return 1
|
||
|
||
conv = Converter(cfg, vault)
|
||
fmt = Formatter(cfg.get("formatter"), conv.warnings)
|
||
changed = 0
|
||
print(f"{'WRITE' if args.write else 'DRY RUN'} vault: {vault}\n{' ' * 9}repo: {repo}\n")
|
||
|
||
for note in cfg["notes"]:
|
||
src, dst = vault / note["source"], repo / note["target"]
|
||
if not src.is_file():
|
||
conv.warnings.append(f"missing in vault: {note['source']}")
|
||
continue
|
||
new = conv.convert(read_text(src), note["source"])
|
||
if note.get("title"): # replace or add the H1
|
||
if new.startswith("# "):
|
||
new = f"# {note['title']}\n" + new.split("\n", 1)[1]
|
||
else:
|
||
new = f"# {note['title']}\n\n" + new
|
||
try:
|
||
old = read_text(dst) if dst.exists() else None
|
||
except PermissionError:
|
||
conv.warnings.append(f"cannot read {note['target']} (permission denied), skipped")
|
||
continue
|
||
new = fmt(new, note["target"])
|
||
|
||
if new == old:
|
||
print(f" unchanged {note['target']}")
|
||
continue
|
||
changed += 1
|
||
print(f" {'new ' if old is None else 'changed '} {note['target']}")
|
||
if args.diff:
|
||
sys.stdout.writelines(difflib.unified_diff(
|
||
(old or "").splitlines(True), new.splitlines(True),
|
||
f"repo/{note['target']}", f"vault/{note['source']}"))
|
||
print()
|
||
if args.write:
|
||
dst.write_text(new, encoding="utf-8", newline="\n")
|
||
|
||
assets = repo / cfg.get("assets_dir", "assets")
|
||
for name in sorted(conv.images):
|
||
src = find_image(vault, cfg.get("image_dirs", []), name)
|
||
if not src:
|
||
conv.warnings.append(f"image not found in vault: {name}")
|
||
continue
|
||
dst = assets / name
|
||
if dst.exists() and filecmp.cmp(src, dst, shallow=False):
|
||
continue
|
||
changed += 1
|
||
print(f" {'changed ' if dst.exists() else 'new '} {cfg.get('assets_dir', 'assets')}/{name}")
|
||
if args.write:
|
||
assets.mkdir(exist_ok=True)
|
||
shutil.copy2(src, dst)
|
||
|
||
if conv.warnings:
|
||
print("\nWarnings:")
|
||
for w in conv.warnings:
|
||
print(f" - {w}")
|
||
print(f"\n{changed} file(s) {'written' if args.write else 'would change'}."
|
||
+ ("" if args.write or not changed else " Run with --write to apply."))
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|