112 lines
3.8 KiB
Python
112 lines
3.8 KiB
Python
#!/usr/bin/env -S uv run --script
|
|
# /// script
|
|
# requires-python = ">=3.11"
|
|
# dependencies = []
|
|
# ///
|
|
"""note_capture.py — dumb, instant capture for the /note skill.
|
|
|
|
Writes the raw input verbatim into notes/inbox/ (atomic tmp -> os.replace) plus one
|
|
audit line to log/note.log, then prints a one-line confirmation. No reformulation,
|
|
no reading of the knowledge doc, no compile — that is the compile step's job
|
|
(inline in immediate mode, or the cron drain in `cron` mode).
|
|
|
|
Used identically by both modes; the only difference is what the agent does *after*
|
|
calling this (immediate: run the compile workflow inline; cron: stop).
|
|
"""
|
|
|
|
import argparse
|
|
import re
|
|
import sys
|
|
import unicodedata
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
# workspace/skills/note/scripts/note_capture.py -> parents[3] = workspace root.
|
|
WORKSPACE = Path(__file__).resolve().parents[3]
|
|
INBOX = WORKSPACE / "notes" / "inbox"
|
|
LOG = WORKSPACE / "log" / "note.log"
|
|
|
|
_SLUG_STRIP_RE = re.compile(r"[^a-z0-9]+")
|
|
_URL_RE = re.compile(r"https?://([^/\s]+)")
|
|
MAX_SLUG_WORDS = 4
|
|
SLUG_MAX_LEN = 40
|
|
|
|
|
|
def _ascii_fold(text: str) -> str:
|
|
"""Drop diacritics so Czech words survive slugging (mazání -> mazani)."""
|
|
return unicodedata.normalize("NFKD", text).encode("ascii", "ignore").decode("ascii")
|
|
|
|
|
|
def _slugify(text: str) -> str:
|
|
"""Short kebab slug from the first words of the input (domain for a bare URL)."""
|
|
first_line = next((line for line in text.splitlines() if line.strip()), "").strip()
|
|
url_match = _URL_RE.match(first_line)
|
|
if url_match:
|
|
host = url_match.group(1).removeprefix("www.")
|
|
slug = _SLUG_STRIP_RE.sub("-", _ascii_fold(host).lower()).strip("-")
|
|
return slug or "note"
|
|
words = first_line.split()[:MAX_SLUG_WORDS]
|
|
slug = _SLUG_STRIP_RE.sub("-", _ascii_fold(" ".join(words)).lower()).strip("-")
|
|
return slug[:SLUG_MAX_LEN].strip("-") or "note"
|
|
|
|
|
|
def _build_content(
|
|
captured_at: str, channel: str | None, chat_id: str | None, body: str
|
|
) -> str:
|
|
lines = [f"captured_at: {captured_at}"]
|
|
if channel:
|
|
lines.append(f"channel: {channel}")
|
|
if chat_id:
|
|
lines.append(f'chat_id: "{chat_id}"')
|
|
frontmatter = "\n".join(lines)
|
|
return f"---\n{frontmatter}\n---\n\n{body.strip()}\n"
|
|
|
|
|
|
def _append_log(filename: str, body: str) -> None:
|
|
LOG.parent.mkdir(parents=True, exist_ok=True)
|
|
stamp = datetime.now().astimezone().isoformat(timespec="seconds")
|
|
summary = " ".join(body.split())[:80]
|
|
with LOG.open("a", encoding="utf-8") as handle:
|
|
handle.write(f"{stamp} CAPTURE {filename} :: {summary}\n")
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Capture a raw note into notes/inbox/")
|
|
parser.add_argument(
|
|
"--text", default=None, help="Raw input; if omitted, read from stdin"
|
|
)
|
|
parser.add_argument(
|
|
"--channel", default=None, help="Origin channel (telegram/websocket/cli)"
|
|
)
|
|
parser.add_argument(
|
|
"--chat-id", default=None, dest="chat_id", help="Origin chat id"
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
body = args.text if args.text is not None else sys.stdin.read()
|
|
body = body.strip()
|
|
if not body:
|
|
print("Nothing to capture (empty input).", file=sys.stderr)
|
|
return 1
|
|
|
|
now = datetime.now().astimezone()
|
|
timestamp = now.strftime("%Y-%m-%d_%H_%M_%S_%f")
|
|
filename = f"{timestamp}-{_slugify(body)}.md"
|
|
content = _build_content(
|
|
now.isoformat(timespec="seconds"), args.channel, args.chat_id, body
|
|
)
|
|
|
|
INBOX.mkdir(parents=True, exist_ok=True)
|
|
tmp_path = INBOX / f".{filename}.tmp"
|
|
final_path = INBOX / filename
|
|
tmp_path.write_text(content, encoding="utf-8")
|
|
tmp_path.replace(final_path)
|
|
|
|
_append_log(filename, body)
|
|
print(f"captured: {filename}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|