1530 lines
59 KiB
Python
1530 lines
59 KiB
Python
#!/usr/bin/env python3
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import html
|
||
import json
|
||
import mimetypes
|
||
import os
|
||
import re
|
||
import socketserver
|
||
import threading
|
||
import webbrowser
|
||
from dataclasses import dataclass
|
||
from datetime import datetime, timezone
|
||
from http.server import BaseHTTPRequestHandler, HTTPServer
|
||
from pathlib import Path
|
||
from typing import Iterable
|
||
from urllib.parse import parse_qs, quote, unquote, urlencode, urlparse
|
||
from urllib.request import Request, urlopen
|
||
|
||
|
||
@dataclass
|
||
class OrgFile:
|
||
path: Path
|
||
relative_path: str
|
||
title: str
|
||
file_id: str | None
|
||
content: str
|
||
mtime: float = 0.0
|
||
pinned: bool = False
|
||
|
||
|
||
ROOT_DIR = Path(__file__).resolve().parent
|
||
TEMPLATE_PATH = ROOT_DIR / "templates" / "index.html"
|
||
STATIC_DIR = ROOT_DIR / "static"
|
||
TITLE_RE = re.compile(r"^\s*#\+TITLE:\s*(.+?)\s*$", flags=re.IGNORECASE)
|
||
ID_RE = re.compile(r"^\s*:ID:\s*(\S+)\s*$", flags=re.IGNORECASE)
|
||
ORG_LINK_RE = re.compile(r"\[\[([^\]]+)\](?:\[([^\]]*)\])?\]")
|
||
URL_RE = re.compile(r"https?://[^\s<>'\"()\[\]]+")
|
||
CREATED_LINE_RE = re.compile(r"^\s*(?:#\+)?CREATED:\s*(.+?)\s*$", flags=re.IGNORECASE)
|
||
ORG_TIMESTAMP_RE = re.compile(r"[\[<](\d{4})-(\d{2})-(\d{2})(?:\s+\w{3})?(?:\s+(\d{2}):(\d{2}))?[\]>]")
|
||
PROPERTY_LINE_RE = re.compile(r"^\s*:([A-Za-z_][A-Za-z0-9_-]*):\s+(.+?)\s*$")
|
||
INLINE_TIMESTAMP_RE = re.compile(
|
||
r"(?P<bracket>[<\[])"
|
||
r"(?P<year>\d{4})-(?P<month>\d{2})-(?P<day>\d{2})"
|
||
r"(?:\s+(?P<dayname>\w{2,3}))?"
|
||
r"(?:\s+(?P<hour>\d{2}):(?P<minute>\d{2})(?:-(?P<end_hour>\d{2}):(?P<end_minute>\d{2}))?)?"
|
||
r"[>\]]"
|
||
)
|
||
TODO_KEYWORDS = {"TODO", "STRT", "DONE", "WAIT"}
|
||
TODO_KEYWORD_RE = re.compile(r"^(" + "|".join(TODO_KEYWORDS) + r")\b\s*(.*)", re.DOTALL)
|
||
DEFAULT_CONFIG_PATH = ROOT_DIR / "config.yaml"
|
||
|
||
|
||
def excluded_dir_names() -> set[str]:
|
||
"""Directories to skip while scanning, from defaults plus DIRS_TO_EXCLUDE env var."""
|
||
excluded = {".git", ".venv", "venv", "__pycache__"}
|
||
raw = os.environ.get("DIRS_TO_EXCLUDE", "")
|
||
for name in raw.split(","):
|
||
name = name.strip()
|
||
if name:
|
||
excluded.add(name)
|
||
return excluded
|
||
|
||
|
||
def scan_org_files(base_dir: Path) -> list[OrgFile]:
|
||
"""Recursively find .org files under base_dir and return their contents."""
|
||
org_files: list[OrgFile] = []
|
||
excluded = excluded_dir_names()
|
||
|
||
for root, dirs, files in os.walk(base_dir):
|
||
dirs[:] = [d for d in dirs if d not in excluded]
|
||
root_path = Path(root)
|
||
for filename in files:
|
||
if filename.lower().endswith(".org"):
|
||
file_path = root_path / filename
|
||
try:
|
||
content = file_path.read_text(encoding="utf-8")
|
||
except UnicodeDecodeError:
|
||
content = file_path.read_text(encoding="utf-8", errors="replace")
|
||
|
||
org_files.append(
|
||
OrgFile(
|
||
path=file_path,
|
||
relative_path=normalize_relative_path(str(file_path.relative_to(base_dir))),
|
||
title=extract_org_title(content, file_path.stem),
|
||
file_id=extract_org_id(content),
|
||
content=content,
|
||
mtime=file_path.stat().st_mtime,
|
||
)
|
||
)
|
||
|
||
org_files.sort(key=lambda f: f.mtime, reverse=True)
|
||
|
||
if os.environ.get("ORGWEB_PIN_TODAY_FILE", "").lower() in ("1", "true", "yes"):
|
||
today = datetime.now().strftime("%Y-%m-%d")
|
||
pinned = []
|
||
rest = []
|
||
for f in org_files:
|
||
if today in f.relative_path or today in f.title:
|
||
f.pinned = True
|
||
pinned.append(f)
|
||
else:
|
||
rest.append(f)
|
||
org_files = pinned + rest
|
||
|
||
return org_files
|
||
|
||
|
||
def extract_org_title(content: str, fallback: str) -> str:
|
||
for line in content.splitlines():
|
||
match = TITLE_RE.match(line)
|
||
if match:
|
||
return match.group(1).strip()
|
||
return fallback
|
||
|
||
|
||
def extract_org_id(content: str) -> str | None:
|
||
for line in content.splitlines():
|
||
match = ID_RE.match(line)
|
||
if match:
|
||
return match.group(1).strip()
|
||
return None
|
||
|
||
|
||
def normalize_relative_path(path_value: str) -> str:
|
||
normalized = os.path.normpath(path_value.replace("\\", "/"))
|
||
if normalized == ".":
|
||
return ""
|
||
return normalized.lstrip("./")
|
||
|
||
|
||
def extract_org_link_targets(content: str) -> list[str]:
|
||
targets: list[str] = []
|
||
for match in ORG_LINK_RE.finditer(content):
|
||
targets.append(match.group(1).strip())
|
||
return targets
|
||
|
||
|
||
def resolve_link_target(
|
||
source_relative_path: str,
|
||
raw_target: str,
|
||
known_paths: set[str],
|
||
id_to_path: dict[str, str],
|
||
) -> str | None:
|
||
target = raw_target.strip()
|
||
if not target:
|
||
return None
|
||
if "::" in target:
|
||
target = target.split("::", 1)[0]
|
||
if "#" in target:
|
||
target = target.split("#", 1)[0]
|
||
if target.lower().startswith("id:"):
|
||
id_value = target[3:].strip().lower()
|
||
return id_to_path.get(id_value)
|
||
if target.startswith("file:"):
|
||
target = target[5:]
|
||
elif "://" in target:
|
||
return None
|
||
elif ":" in target and not target.endswith(".org"):
|
||
return None
|
||
|
||
source_dir = os.path.dirname(source_relative_path)
|
||
if target.startswith("/"):
|
||
candidate = normalize_relative_path(target.lstrip("/"))
|
||
else:
|
||
candidate = normalize_relative_path(os.path.join(source_dir, target))
|
||
|
||
if not candidate:
|
||
return None
|
||
if candidate in known_paths:
|
||
return candidate
|
||
if not candidate.endswith(".org"):
|
||
with_suffix = f"{candidate}.org"
|
||
if with_suffix in known_paths:
|
||
return with_suffix
|
||
return None
|
||
|
||
|
||
def find_backlinks(org_files: Iterable[OrgFile], selected_path: str) -> list[OrgFile]:
|
||
files = list(org_files)
|
||
known_paths = {f.relative_path for f in files}
|
||
id_to_path = {f.file_id.lower(): f.relative_path for f in files if f.file_id}
|
||
backlinks: list[OrgFile] = []
|
||
seen_sources: set[str] = set()
|
||
|
||
for source in files:
|
||
if source.relative_path == selected_path:
|
||
continue
|
||
link_targets = extract_org_link_targets(source.content)
|
||
for raw_target in link_targets:
|
||
resolved = resolve_link_target(source.relative_path, raw_target, known_paths, id_to_path)
|
||
if resolved == selected_path and source.relative_path not in seen_sources:
|
||
backlinks.append(source)
|
||
seen_sources.add(source.relative_path)
|
||
break
|
||
|
||
backlinks.sort(key=lambda f: (f.title.lower(), f.relative_path.lower()))
|
||
return backlinks
|
||
|
||
|
||
def build_backlink_counts(org_files: Iterable[OrgFile]) -> dict[str, int]:
|
||
files = list(org_files)
|
||
known_paths = {f.relative_path for f in files}
|
||
id_to_path = {f.file_id.lower(): f.relative_path for f in files if f.file_id}
|
||
counts = {f.relative_path: 0 for f in files}
|
||
|
||
for source in files:
|
||
seen_targets: set[str] = set()
|
||
for raw_target in extract_org_link_targets(source.content):
|
||
resolved = resolve_link_target(source.relative_path, raw_target, known_paths, id_to_path)
|
||
if not resolved or resolved == source.relative_path or resolved in seen_targets:
|
||
continue
|
||
if resolved in counts:
|
||
counts[resolved] += 1
|
||
seen_targets.add(resolved)
|
||
|
||
return counts
|
||
|
||
|
||
def _timestamp_match_to_sort_key(match: re.Match[str]) -> int | None:
|
||
try:
|
||
year = int(match.group(1))
|
||
month = int(match.group(2))
|
||
day = int(match.group(3))
|
||
hour = int(match.group(4) or "0")
|
||
minute = int(match.group(5) or "0")
|
||
except ValueError:
|
||
return None
|
||
return year * 100000000 + month * 1000000 + day * 10000 + hour * 100 + minute
|
||
|
||
|
||
def extract_created_sort_key(content: str) -> int | None:
|
||
for line in content.splitlines():
|
||
created_match = CREATED_LINE_RE.match(line)
|
||
if created_match:
|
||
ts_match = ORG_TIMESTAMP_RE.search(created_match.group(1))
|
||
if ts_match:
|
||
key = _timestamp_match_to_sort_key(ts_match)
|
||
if key is not None:
|
||
return key
|
||
|
||
first_ts = ORG_TIMESTAMP_RE.search(content)
|
||
if first_ts:
|
||
return _timestamp_match_to_sort_key(first_ts)
|
||
return None
|
||
|
||
|
||
def note_href(relative_path: str, edit_mode: bool, path_to_slug: dict[str, str]) -> str:
|
||
slug = path_to_slug.get(relative_path, relative_path)
|
||
href = "/n/" + quote(slug, safe="")
|
||
if edit_mode:
|
||
href += "?edit=1"
|
||
return href
|
||
|
||
|
||
def truncate_label(text: str, max_chars: int = 32) -> str:
|
||
if len(text) <= max_chars:
|
||
return text
|
||
if max_chars <= 3:
|
||
return "." * max_chars
|
||
return text[: max_chars - 3] + "..."
|
||
|
||
|
||
_MONTH_ABBR = [
|
||
"", "Jan", "Feb", "Mar", "Apr", "May", "Jun",
|
||
"Jul", "Aug", "Sep", "Oct", "Nov", "Dec",
|
||
]
|
||
|
||
|
||
def _format_timestamp(m: re.Match[str]) -> str:
|
||
"""Turn an org timestamp match into a styled <time> element."""
|
||
year = m.group("year")
|
||
month = int(m.group("month"))
|
||
day = int(m.group("day"))
|
||
dayname = m.group("dayname") or ""
|
||
hour = m.group("hour")
|
||
minute = m.group("minute")
|
||
end_hour = m.group("end_hour")
|
||
end_minute = m.group("end_minute")
|
||
is_active = m.group("bracket") == "<"
|
||
|
||
month_str = _MONTH_ABBR[month] if 1 <= month <= 12 else m.group("month")
|
||
date_part = f"{month_str} {day}, {year}"
|
||
if dayname:
|
||
date_part = f"{dayname} {date_part}"
|
||
|
||
time_part = ""
|
||
if hour is not None:
|
||
time_part = f"{hour}:{minute}"
|
||
if end_hour is not None:
|
||
time_part += f"–{end_hour}:{end_minute}"
|
||
|
||
iso_date = f"{year}-{m.group('month')}-{m.group('day')}"
|
||
if hour is not None:
|
||
iso_date += f"T{hour}:{minute}"
|
||
|
||
cls = "org-timestamp" if is_active else "org-timestamp org-timestamp-inactive"
|
||
inner = f"<span class='org-ts-date'>{html.escape(date_part)}</span>"
|
||
if time_part:
|
||
inner += f"<span class='org-ts-time'>{html.escape(time_part)}</span>"
|
||
|
||
return f"<time class='{cls}' datetime='{iso_date}'>{inner}</time>"
|
||
|
||
|
||
_ESCAPED_TS_RE = re.compile(
|
||
r"(?P<bracket>[\x00\[])"
|
||
r"(?P<year>\d{4})-(?P<month>\d{2})-(?P<day>\d{2})"
|
||
r"(?:\s+(?P<dayname>\w{2,3}))?"
|
||
r"(?:\s+(?P<hour>\d{2}):(?P<minute>\d{2})(?:-(?P<end_hour>\d{2}):(?P<end_minute>\d{2}))?)?"
|
||
r"[\x01\]]"
|
||
)
|
||
|
||
|
||
def _replace_timestamps(escaped_html: str) -> str:
|
||
"""Find org timestamps in already-escaped text and wrap them in <time> tags.
|
||
|
||
Because the text has been HTML-escaped, angle brackets appear as < / >.
|
||
We temporarily swap them to sentinel bytes so the regex can match uniformly.
|
||
"""
|
||
result = escaped_html.replace("<", "\x00").replace(">", "\x01")
|
||
|
||
def _sub(m: re.Match[str]) -> str:
|
||
raw = m.group(0).replace("\x00", "<").replace("\x01", ">")
|
||
ts_match = INLINE_TIMESTAMP_RE.match(raw)
|
||
if ts_match:
|
||
return _format_timestamp(ts_match)
|
||
return m.group(0).replace("\x00", "<").replace("\x01", ">")
|
||
|
||
result = _ESCAPED_TS_RE.sub(_sub, result)
|
||
return result.replace("\x00", "<").replace("\x01", ">")
|
||
|
||
|
||
_CODE_RE = re.compile(r'(?:^|(?<=[\s\(\[\{"\'\-\u2014/]))=(?!\s)([\s\S]+?)(?<!\s)=(?=[\s\)\}\]"\'\-\.,:;!\?\u2014/]|$)')
|
||
_VERBATIM_RE = re.compile(r'(?:^|(?<=[\s\(\[\{"\'\-\u2014/]))~(?!\s)([\s\S]+?)(?<!\s)~(?=[\s\)\}\]"\'\-\.,:;!\?\u2014/]|$)')
|
||
_BOLD_RE = re.compile(r'(?:^|(?<=[\s\(\[\{"\'\-\u2014/]))\*(?!\s)([\s\S]+?)(?<!\s)\*(?=[\s\)\}\]"\'\-\.,:;!\?\u2014/]|$)')
|
||
_ITALIC_RE = re.compile(r'(?:^|(?<=[\s\(\[\{"\'\-\u2014/]))/(?!\s)([\s\S]+?)(?<!\s)/(?=[\s\)\}\]"\'\-\.,:;!\?\u2014/]|$)')
|
||
_UNDERLINE_RE = re.compile(r'(?:^|(?<=[\s\(\[\{"\'\-\u2014/]))_(?!\s)([\s\S]+?)(?<!\s)_(?=[\s\)\}\]"\'\-\.,:;!\?\u2014/]|$)')
|
||
_STRIKE_RE = re.compile(r'(?:^|(?<=[\s\(\[\{"\'\-\u2014/]))\+(?!\s)([\s\S]+?)(?<!\s)\+(?=[\s\)\}\]"\'\-\.,:;!\?\u2014/]|$)')
|
||
|
||
|
||
def apply_inline_formatting(html_str: str) -> str:
|
||
"""Apply org inline text formatting (=code=, ~verbatim~, *bold*, /italic/, _underline_, +strike+) to text segments outside HTML tags."""
|
||
parts = re.split(r"(<[^>]+>)", html_str)
|
||
for i in range(0, len(parts), 2):
|
||
text = parts[i]
|
||
if not text:
|
||
continue
|
||
text = _CODE_RE.sub(r"<code class='org-code'>\1</code>", text)
|
||
text = _VERBATIM_RE.sub(r"<code class='org-code'>\1</code>", text)
|
||
text = _BOLD_RE.sub(r"<strong>\1</strong>", text)
|
||
text = _ITALIC_RE.sub(r"<em>\1</em>", text)
|
||
text = _UNDERLINE_RE.sub(r"<u>\1</u>", text)
|
||
text = _STRIKE_RE.sub(r"<del>\1</del>", text)
|
||
parts[i] = text
|
||
return "".join(parts)
|
||
|
||
|
||
def render_plain_text_with_links(text: str) -> str:
|
||
rendered_parts: list[str] = []
|
||
cursor = 0
|
||
for match in URL_RE.finditer(text):
|
||
start, end = match.span()
|
||
if start > cursor:
|
||
rendered_parts.append(html.escape(text[cursor:start]))
|
||
url = match.group(0)
|
||
safe_url = html.escape(url, quote=True)
|
||
rendered_parts.append(f"<a class='org-link' href='{safe_url}' target='_blank' rel='noopener noreferrer'>{safe_url}</a>")
|
||
cursor = end
|
||
if cursor < len(text):
|
||
rendered_parts.append(html.escape(text[cursor:]))
|
||
return _replace_timestamps("".join(rendered_parts))
|
||
|
||
|
||
def _render_heading_with_todo(
|
||
heading_text: str, render_line_fn: Callable[[str], str] | None = None
|
||
) -> tuple[str, str | None, str, list[str]]:
|
||
"""Parse a heading for TODO keyword and tags.
|
||
|
||
Returns (rendered_html, keyword_or_None, bare_heading_text, tags_list).
|
||
"""
|
||
if render_line_fn is None:
|
||
render_line_fn = html.escape
|
||
|
||
# Strip tags from the end first
|
||
tag_match = _ORG_TAG_RE.search(heading_text)
|
||
tags: list[str] = []
|
||
text_without_tags = heading_text
|
||
if tag_match:
|
||
tags = [t for t in tag_match.group(1).split(":") if t]
|
||
text_without_tags = heading_text[: tag_match.start()].strip()
|
||
|
||
m = TODO_KEYWORD_RE.match(text_without_tags)
|
||
if m:
|
||
keyword = m.group(1)
|
||
rest = m.group(2).strip()
|
||
css_class = f"org-todo org-todo-{keyword.lower()}"
|
||
rendered = f"<span class='{css_class}'>{html.escape(keyword)}</span> {render_line_fn(rest)}"
|
||
return rendered, keyword, rest, tags
|
||
return render_line_fn(text_without_tags), None, text_without_tags.strip(), tags
|
||
|
||
|
||
def render_line_with_links(
|
||
line: str, source_relative_path: str, known_paths: set[str], id_to_path: dict[str, str],
|
||
path_to_slug: dict[str, str],
|
||
) -> str:
|
||
rendered_parts: list[str] = []
|
||
cursor = 0
|
||
|
||
for match in ORG_LINK_RE.finditer(line):
|
||
start, end = match.span()
|
||
if start > cursor:
|
||
rendered_parts.append(render_plain_text_with_links(line[cursor:start]))
|
||
|
||
raw_target = match.group(1).strip()
|
||
label = (match.group(2) or "").strip()
|
||
resolved = resolve_link_target(source_relative_path, raw_target, known_paths, id_to_path)
|
||
if resolved:
|
||
safe_href = html.escape(note_href(resolved, False, path_to_slug), quote=True)
|
||
link_text = label if label else raw_target
|
||
rendered_parts.append(f"<a class='org-link' href='{safe_href}'>{html.escape(link_text)}</a>")
|
||
elif raw_target.startswith(("http://", "https://", "mailto:")) or "://" in raw_target:
|
||
safe_href = html.escape(raw_target, quote=True)
|
||
link_text = label if label else raw_target
|
||
rendered_parts.append(f"<a class='org-link' href='{safe_href}' target='_blank' rel='noopener noreferrer'>{html.escape(link_text)}</a>")
|
||
else:
|
||
link_text = label if label else raw_target
|
||
rendered_parts.append(html.escape(link_text))
|
||
|
||
cursor = end
|
||
|
||
if cursor < len(line):
|
||
rendered_parts.append(render_plain_text_with_links(line[cursor:]))
|
||
|
||
full_html = "".join(rendered_parts)
|
||
return apply_inline_formatting(full_html)
|
||
|
||
|
||
_BLOCK_START_RE = re.compile(r"^\s*#\+(BEGIN_SRC|BEGIN_EXAMPLE|BEGIN_QUOTE|BEGIN_EXPORT)\b\s*(.*)$", re.IGNORECASE)
|
||
_BLOCK_END_RE = re.compile(r"^\s*#\+(END_SRC|END_EXAMPLE|END_QUOTE|END_EXPORT)\b", re.IGNORECASE)
|
||
_TABLE_ROW_RE = re.compile(r"^\s*\|.*\|\s*$")
|
||
_TABLE_SEP_RE = re.compile(r"^\s*\|[-+=|]+\|\s*$")
|
||
_LIST_ITEM_RE = re.compile(r"^\s*(?:([-+])|(\d+)[\.\)])\s+(.*)$")
|
||
_HR_RE = re.compile(r"^\s*-{5,}\s*$")
|
||
_COMMENT_OR_KEYWORD_RE = re.compile(r"^\s*#(?:[+A-Za-z_]|\s)")
|
||
|
||
|
||
def render_org_to_html(
|
||
content: str, source_relative_path: str, known_paths: set[str], id_to_path: dict[str, str],
|
||
path_to_slug: dict[str, str],
|
||
show_webhook: bool = False,
|
||
) -> str:
|
||
"""Render org mode content to HTML, supporting headings, lists, tables, blocks, links, and inline markup."""
|
||
lines = content.splitlines()
|
||
html_lines: list[str] = []
|
||
section_stack: list[tuple[int, list[str], list[str]]] = []
|
||
i = 0
|
||
n = len(lines)
|
||
|
||
in_properties = False
|
||
property_items: list[tuple[str, str]] = []
|
||
|
||
def render_line(l: str) -> str:
|
||
return render_line_with_links(l, source_relative_path, known_paths, id_to_path, path_to_slug)
|
||
|
||
def current_container() -> list[str]:
|
||
return html_lines if not section_stack else section_stack[-1][2]
|
||
|
||
def emit(*parts: str) -> None:
|
||
current_container().extend(parts)
|
||
|
||
def close_sections(min_level: int) -> None:
|
||
"""Close open sections whose heading level is >= min_level."""
|
||
while section_stack and section_stack[-1][0] >= min_level:
|
||
level, heading_lines, body_lines = section_stack.pop()
|
||
section_html = (
|
||
"\n".join(heading_lines)
|
||
+ "\n"
|
||
+ "\n".join(body_lines)
|
||
+ "\n</div>\n</section>"
|
||
)
|
||
current_container().append(section_html)
|
||
|
||
while i < n:
|
||
line = lines[i]
|
||
stripped = line.strip()
|
||
|
||
# Check PROPERTIES drawer
|
||
if re.match(r"^\s*:PROPERTIES:\s*$", stripped, re.IGNORECASE):
|
||
in_properties = True
|
||
property_items = []
|
||
i += 1
|
||
while i < n:
|
||
s = lines[i].strip()
|
||
if re.match(r"^\s*:END:\s*$", s, re.IGNORECASE):
|
||
in_properties = False
|
||
if property_items:
|
||
dl_items = [
|
||
f"<dt>{html.escape(k)}</dt><dd>{html.escape(v)}</dd>"
|
||
for k, v in property_items
|
||
]
|
||
emit(f"<dl class='org-properties'>{''.join(dl_items)}</dl>")
|
||
i += 1
|
||
break
|
||
prop_match = PROPERTY_LINE_RE.match(s)
|
||
if prop_match:
|
||
property_items.append((prop_match.group(1), prop_match.group(2)))
|
||
i += 1
|
||
continue
|
||
|
||
if in_properties:
|
||
i += 1
|
||
continue
|
||
|
||
# Check TITLE
|
||
title_match = TITLE_RE.match(stripped)
|
||
if title_match:
|
||
emit(f"<h1>{html.escape(title_match.group(1))}</h1>")
|
||
i += 1
|
||
continue
|
||
|
||
# Check Code/Example/Quote/Export blocks
|
||
block_match = _BLOCK_START_RE.match(stripped)
|
||
if block_match:
|
||
block_kind = block_match.group(1).upper()
|
||
block_arg = block_match.group(2).strip()
|
||
block_lines: list[str] = []
|
||
i += 1
|
||
while i < n:
|
||
if _BLOCK_END_RE.match(lines[i].strip()):
|
||
i += 1
|
||
break
|
||
block_lines.append(lines[i])
|
||
i += 1
|
||
|
||
block_content = "\n".join(block_lines)
|
||
if block_kind == "BEGIN_SRC":
|
||
lang_cls = f" class='language-{html.escape(block_arg.lower())}'" if block_arg else ""
|
||
emit(f"<pre class='org-src-block'><code{lang_cls}>{html.escape(block_content)}</code></pre>")
|
||
elif block_kind == "BEGIN_EXAMPLE":
|
||
emit(f"<pre class='org-example'><code>{html.escape(block_content)}</code></pre>")
|
||
elif block_kind == "BEGIN_QUOTE":
|
||
quote_rendered = "\n".join(f"<p>{render_line(l)}</p>" for l in block_lines if l.strip())
|
||
emit(f"<blockquote class='org-quote'>{quote_rendered}</blockquote>")
|
||
# EXPORT is ignored
|
||
continue
|
||
|
||
# Check Table
|
||
if _TABLE_ROW_RE.match(line):
|
||
table_lines: list[str] = []
|
||
while i < n and _TABLE_ROW_RE.match(lines[i]):
|
||
table_lines.append(lines[i])
|
||
i += 1
|
||
|
||
rows: list[tuple[bool, list[str]]] = []
|
||
for tl in table_lines:
|
||
if _TABLE_SEP_RE.match(tl):
|
||
rows.append((True, []))
|
||
else:
|
||
parts = [c.strip() for c in tl.strip().split("|")]
|
||
if parts and parts[0] == "":
|
||
parts.pop(0)
|
||
if parts and parts[-1] == "":
|
||
parts.pop()
|
||
rows.append((False, [render_line(c) for c in parts]))
|
||
|
||
has_header = len(rows) >= 2 and not rows[0][0] and rows[1][0]
|
||
table_html: list[str] = ["<table class='org-table'>"]
|
||
|
||
if has_header:
|
||
table_html.append("<thead><tr>")
|
||
for cell in rows[0][1]:
|
||
table_html.append(f"<th>{cell}</th>")
|
||
table_html.append("</tr></thead>")
|
||
body_rows = rows[2:]
|
||
else:
|
||
body_rows = rows
|
||
|
||
table_html.append("<tbody>")
|
||
for is_sep, cells in body_rows:
|
||
if is_sep:
|
||
continue
|
||
table_html.append("<tr>")
|
||
for cell in cells:
|
||
table_html.append(f"<td>{cell}</td>")
|
||
table_html.append("</tr>")
|
||
table_html.append("</tbody></table>")
|
||
emit("".join(table_html))
|
||
continue
|
||
|
||
# Check Headings (* at column 0 or star + space)
|
||
if stripped.startswith("*"):
|
||
stars = len(stripped) - len(stripped.lstrip("*"))
|
||
if stars > 0 and len(stripped) > stars and stripped[stars] == " ":
|
||
level = min(stars + 1, 6)
|
||
close_sections(level)
|
||
heading_text = stripped[stars + 1 :]
|
||
rendered_title, keyword, bare_text, heading_tags = _render_heading_with_todo(
|
||
heading_text, render_line_fn=render_line
|
||
)
|
||
starts_collapsed = level >= 3
|
||
section_class = "org-section" + (" collapsed" if starts_collapsed else "")
|
||
toggle_glyph = "▸" if starts_collapsed else "▾"
|
||
toggle_expanded = "false" if starts_collapsed else "true"
|
||
toggle_title = "Expand section" if starts_collapsed else "Collapse section"
|
||
heading_lines: list[str] = [
|
||
f"<section class='{section_class}' data-level='{level}'>",
|
||
(
|
||
f"<h{level} class='org-heading' data-level='{level}'>"
|
||
f"<button class='org-collapse-toggle' type='button' aria-expanded='{toggle_expanded}' "
|
||
f"title='{toggle_title}'>{toggle_glyph}</button>"
|
||
f"<span class='org-heading-text'>{rendered_title}</span>"
|
||
f"</h{level}>"
|
||
),
|
||
]
|
||
if heading_tags:
|
||
tag_spans = "".join(
|
||
f"<span class='org-tag'>{html.escape(t)}</span>"
|
||
for t in heading_tags
|
||
)
|
||
heading_lines.append(f"<div class='org-tags'>{tag_spans}</div>")
|
||
if keyword and show_webhook:
|
||
safe_file = html.escape(source_relative_path, quote=True)
|
||
safe_heading = html.escape(bare_text, quote=True)
|
||
heading_lines.append(
|
||
f"<button class='webhook-send-btn' type='button' "
|
||
f"data-file='{safe_file}' data-heading='{safe_heading}' "
|
||
f"data-state='STRT'>Start work</button> "
|
||
f"<button class='webhook-send-btn' type='button' "
|
||
f"data-file='{safe_file}' data-heading='{safe_heading}' "
|
||
f"data-state='DONE'>End work</button>"
|
||
)
|
||
heading_lines.append("<div class='org-section-body'>")
|
||
section_stack.append((level, heading_lines, []))
|
||
i += 1
|
||
continue
|
||
|
||
# Check Horizontal Rule
|
||
if _HR_RE.match(stripped):
|
||
emit("<hr class='org-hr'>")
|
||
i += 1
|
||
continue
|
||
|
||
# Check Lists (Unordered & Ordered)
|
||
list_match = _LIST_ITEM_RE.match(line)
|
||
if list_match:
|
||
is_ordered = list_match.group(2) is not None
|
||
list_items: list[str] = []
|
||
tag_name = "ol" if is_ordered else "ul"
|
||
|
||
while i < n:
|
||
lm = _LIST_ITEM_RE.match(lines[i])
|
||
if lm:
|
||
item_ordered = lm.group(2) is not None
|
||
if item_ordered != is_ordered:
|
||
break
|
||
item_text = lm.group(3)
|
||
i += 1
|
||
while (
|
||
i < n
|
||
and lines[i].strip()
|
||
and not _LIST_ITEM_RE.match(lines[i])
|
||
and not lines[i].startswith("*")
|
||
and not _BLOCK_START_RE.match(lines[i].strip())
|
||
and not _TABLE_ROW_RE.match(lines[i])
|
||
):
|
||
item_text += " " + lines[i].strip()
|
||
i += 1
|
||
list_items.append(render_line(item_text))
|
||
else:
|
||
break
|
||
|
||
items_html = "".join(f"<li>{item}</li>" for item in list_items)
|
||
emit(f"<{tag_name} class='org-list'>{items_html}</{tag_name}>")
|
||
continue
|
||
|
||
# Skip comment or unrecognized directive lines (#+CREATED:, #+filetags:, # ...)
|
||
if _COMMENT_OR_KEYWORD_RE.match(stripped):
|
||
i += 1
|
||
continue
|
||
|
||
# Empty lines
|
||
if not stripped:
|
||
emit("<div class='spacer'></div>")
|
||
i += 1
|
||
continue
|
||
|
||
# Regular Paragraph (group consecutive soft-wrapped lines so
|
||
# multi-line org emphasis like *bold spanning lines* can match)
|
||
paragraph_lines: list[str] = [line]
|
||
i += 1
|
||
while i < n:
|
||
nxt = lines[i]
|
||
nxt_stripped = nxt.strip()
|
||
if not nxt_stripped:
|
||
break
|
||
if _BLOCK_START_RE.match(nxt_stripped):
|
||
break
|
||
if _TABLE_ROW_RE.match(nxt):
|
||
break
|
||
if _LIST_ITEM_RE.match(nxt):
|
||
break
|
||
if _HR_RE.match(nxt_stripped):
|
||
break
|
||
if _COMMENT_OR_KEYWORD_RE.match(nxt_stripped):
|
||
break
|
||
if TITLE_RE.match(nxt_stripped):
|
||
break
|
||
if re.match(r"^\s*:PROPERTIES:\s*$", nxt_stripped, re.IGNORECASE):
|
||
break
|
||
if nxt_stripped.startswith("*"):
|
||
stars = len(nxt_stripped) - len(nxt_stripped.lstrip("*"))
|
||
if stars > 0 and len(nxt_stripped) > stars and nxt_stripped[stars] == " ":
|
||
break
|
||
paragraph_lines.append(nxt)
|
||
i += 1
|
||
|
||
paragraph_text = "\n".join(p.rstrip() for p in paragraph_lines)
|
||
emit(f"<p>{render_line(paragraph_text)}</p>")
|
||
|
||
close_sections(1)
|
||
return "\n".join(html_lines)
|
||
|
||
|
||
def find_org_file(org_files: Iterable[OrgFile], relative_path: str | None) -> OrgFile | None:
|
||
if relative_path is None:
|
||
return None
|
||
for org_file in org_files:
|
||
if org_file.relative_path == relative_path:
|
||
return org_file
|
||
return None
|
||
|
||
|
||
def build_slugs(org_files: Iterable[OrgFile]) -> tuple[dict[str, str], dict[str, str], dict[str, str]]:
|
||
"""Return (path_to_slug, slug_to_path, id_lower_to_path) maps.
|
||
|
||
Slugs prefer the file's :ID: value; files without an ID (or with a
|
||
duplicate ID) fall back to their relative path. id_lower_to_path maps
|
||
lowercased IDs to paths for case-insensitive lookup.
|
||
"""
|
||
path_to_slug: dict[str, str] = {}
|
||
slug_to_path: dict[str, str] = {}
|
||
id_lower_to_path: dict[str, str] = {}
|
||
used_lower: set[str] = set()
|
||
for f in org_files:
|
||
if f.file_id and f.file_id.lower() not in used_lower:
|
||
slug = f.file_id
|
||
else:
|
||
slug = f.relative_path
|
||
if slug in slug_to_path:
|
||
continue
|
||
used_lower.add(slug.lower())
|
||
path_to_slug[f.relative_path] = slug
|
||
slug_to_path[slug] = f.relative_path
|
||
if f.file_id:
|
||
id_lower_to_path[f.file_id.lower()] = f.relative_path
|
||
return path_to_slug, slug_to_path, id_lower_to_path
|
||
|
||
|
||
def resolve_slug(slug: str, slug_to_path: dict[str, str], id_lower_to_path: dict[str, str]) -> str | None:
|
||
"""Resolve a URL slug back to a relative path (exact, then case-insensitive ID)."""
|
||
if slug in slug_to_path:
|
||
return slug_to_path[slug]
|
||
return id_lower_to_path.get(slug.lower())
|
||
|
||
|
||
_HEADING_RE = re.compile(r"^(\*+)\s+(.*)$")
|
||
_ORG_TAG_RE = re.compile(r"\s+:([\w@:]+):\s*$")
|
||
_DRAWER_RE = re.compile(r"^\s*:(\w+):\s*$")
|
||
_DRAWER_END_RE = re.compile(r"^\s*:END:\s*$", re.IGNORECASE)
|
||
_PROP_KV_RE = re.compile(r"^\s*:([A-Za-z_][\w-]*):\s+(.+?)\s*$")
|
||
_TS_INLINE_RE = re.compile(r"[<\[](\d{4}-\d{2}-\d{2}[^\]>]*)[\]>]")
|
||
_NOTE_TAKEN_RE = re.compile(
|
||
r"^\s*[-+]\s+Note taken on\s+\[([^\]]+)\]\s*(?:\\\\)?\s*$"
|
||
)
|
||
|
||
|
||
def extract_heading_data(content: str, heading_text: str, file_id: str = "", state: str = "STRT") -> dict | None:
|
||
"""Find a heading in *content* by its text and extract webhook-ready data."""
|
||
lines = content.splitlines()
|
||
target_idx: int | None = None
|
||
target_level = 0
|
||
|
||
for idx, raw_line in enumerate(lines):
|
||
hm = _HEADING_RE.match(raw_line)
|
||
if hm:
|
||
stars = len(hm.group(1))
|
||
rest = hm.group(2)
|
||
kw_match = TODO_KEYWORD_RE.match(rest)
|
||
bare_title = kw_match.group(2) if kw_match else rest
|
||
tag_match = _ORG_TAG_RE.search(bare_title)
|
||
bare_title_no_tags = bare_title[: tag_match.start()] if tag_match else bare_title
|
||
if bare_title_no_tags.strip() == heading_text.strip():
|
||
target_idx = idx
|
||
target_level = stars
|
||
break
|
||
|
||
if target_idx is None:
|
||
return None
|
||
|
||
heading_line = lines[target_idx]
|
||
hm = _HEADING_RE.match(heading_line)
|
||
full_rest = hm.group(2) if hm else ""
|
||
|
||
kw_match = TODO_KEYWORD_RE.match(full_rest)
|
||
heading_keyword = kw_match.group(1) if kw_match else ""
|
||
description = kw_match.group(2) if kw_match else full_rest
|
||
|
||
tag_match = _ORG_TAG_RE.search(description)
|
||
tags: list[str] = []
|
||
if tag_match:
|
||
tags = [t for t in tag_match.group(1).split(":") if t]
|
||
description = description[: tag_match.start()].strip()
|
||
|
||
sub_lines = lines[target_idx + 1 :]
|
||
end = len(sub_lines)
|
||
for i, sl in enumerate(sub_lines):
|
||
sm = _HEADING_RE.match(sl)
|
||
if sm and len(sm.group(1)) <= target_level:
|
||
end = i
|
||
break
|
||
sub_lines = sub_lines[:end]
|
||
|
||
properties: dict[str, str] = {}
|
||
drawers: dict[str, list[str]] = {}
|
||
body_lines: list[str] = []
|
||
timestamps: list[str] = []
|
||
in_drawer: str | None = None
|
||
drawer_lines: list[str] = []
|
||
properties_seen = False
|
||
|
||
for sl in sub_lines:
|
||
stripped = sl.strip()
|
||
if in_drawer is not None:
|
||
if _DRAWER_END_RE.match(stripped):
|
||
if in_drawer == "PROPERTIES":
|
||
if not properties_seen:
|
||
for dl in drawer_lines:
|
||
pm = _PROP_KV_RE.match(dl)
|
||
if pm:
|
||
properties[pm.group(1)] = pm.group(2)
|
||
properties_seen = True
|
||
else:
|
||
drawers[in_drawer] = list(drawer_lines)
|
||
in_drawer = None
|
||
drawer_lines = []
|
||
else:
|
||
drawer_lines.append(sl)
|
||
continue
|
||
|
||
dm = _DRAWER_RE.match(stripped)
|
||
if dm:
|
||
in_drawer = dm.group(1).upper()
|
||
drawer_lines = []
|
||
continue
|
||
|
||
for ts in _TS_INLINE_RE.findall(sl):
|
||
timestamps.append(ts)
|
||
|
||
body_lines.append(sl)
|
||
|
||
emacs_id = properties.get("ID", "") or file_id
|
||
notes = _extract_notes(body_lines)
|
||
|
||
return {
|
||
"description": description,
|
||
"labels": tags,
|
||
"state": state,
|
||
"timestamps": timestamps,
|
||
"notes": notes,
|
||
"drawers": drawers,
|
||
"emacs_id": emacs_id,
|
||
"updated_at": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
|
||
"source": "orgweb",
|
||
"properties": properties,
|
||
"body": "\n".join(body_lines).strip(),
|
||
}
|
||
|
||
|
||
def add_to_inbox(content: str, text: str) -> str:
|
||
"""Add a '** TODO <text>' item under a top-level '* Inbox' section.
|
||
|
||
The Inbox is kept as the first top-level heading in the file: if it is
|
||
missing it is created there (right after any file preamble such as
|
||
#+TITLE or #+CREATED), and if it already exists further down it is
|
||
moved to the top before the new task is added.
|
||
"""
|
||
text = text.strip()
|
||
if not text:
|
||
return content
|
||
new_task = f"** TODO {text}"
|
||
inbox_re = re.compile(r"^\* Inbox\s*$")
|
||
top_level_re = re.compile(r"^\*\s")
|
||
|
||
lines = content.rstrip("\n").splitlines()
|
||
|
||
def first_top_level() -> int:
|
||
for idx, line in enumerate(lines):
|
||
if top_level_re.match(line):
|
||
return idx
|
||
return len(lines)
|
||
|
||
def section_end(start: int) -> int:
|
||
for idx in range(start + 1, len(lines)):
|
||
if top_level_re.match(lines[idx]):
|
||
return idx
|
||
return len(lines)
|
||
|
||
inbox_idx: int | None = None
|
||
for idx, line in enumerate(lines):
|
||
if inbox_re.match(line):
|
||
inbox_idx = idx
|
||
break
|
||
|
||
insert_at = first_top_level()
|
||
|
||
if inbox_idx is None:
|
||
lines.insert(insert_at, "* Inbox")
|
||
inbox_idx = insert_at
|
||
elif inbox_idx != insert_at:
|
||
end = section_end(inbox_idx)
|
||
inbox_block = lines[inbox_idx:end]
|
||
del lines[inbox_idx:end]
|
||
lines[insert_at:insert_at] = inbox_block
|
||
inbox_idx = insert_at
|
||
|
||
lines.insert(section_end(inbox_idx), new_task)
|
||
return "\n".join(lines) + "\n"
|
||
|
||
|
||
def update_heading_keyword(content: str, heading_text: str, new_keyword: str) -> str | None:
|
||
"""Set the TODO keyword on a heading (identified by its bare text) in *content*.
|
||
|
||
Returns the updated content, or None if the heading could not be found.
|
||
"""
|
||
lines = content.splitlines()
|
||
for idx, raw_line in enumerate(lines):
|
||
hm = _HEADING_RE.match(raw_line)
|
||
if not hm:
|
||
continue
|
||
rest = hm.group(2)
|
||
kw_match = TODO_KEYWORD_RE.match(rest)
|
||
bare_title = kw_match.group(2) if kw_match else rest
|
||
tag_match = _ORG_TAG_RE.search(bare_title)
|
||
bare_title_no_tags = bare_title[: tag_match.start()] if tag_match else bare_title
|
||
if bare_title_no_tags.strip() != heading_text.strip():
|
||
continue
|
||
heading_marker = hm.group(1)
|
||
if kw_match:
|
||
new_rest = f"{new_keyword} {kw_match.group(2)}".rstrip()
|
||
else:
|
||
new_rest = f"{new_keyword} {rest}"
|
||
lines[idx] = f"{heading_marker} {new_rest}"
|
||
return "\n".join(lines)
|
||
return None
|
||
|
||
|
||
def _parse_org_ts_to_iso(raw: str) -> str:
|
||
"""Convert an org timestamp body like '2026-02-12 Thu 16:00' to ISO 8601."""
|
||
m = re.match(
|
||
r"(\d{4})-(\d{2})-(\d{2})(?:\s+\w{2,3})?(?:\s+(\d{2}):(\d{2}))?", raw
|
||
)
|
||
if not m:
|
||
return raw.strip()
|
||
parts = [m.group(1), "-", m.group(2), "-", m.group(3)]
|
||
if m.group(4) is not None:
|
||
parts += ["T", m.group(4), ":", m.group(5)]
|
||
return "".join(parts)
|
||
|
||
|
||
def _extract_notes(body_lines: list[str]) -> list[dict[str, str]]:
|
||
"""Extract '- Note taken on [TIMESTAMP]' entries and their content from body lines."""
|
||
notes: list[dict[str, str]] = []
|
||
i = 0
|
||
while i < len(body_lines):
|
||
nm = _NOTE_TAKEN_RE.match(body_lines[i])
|
||
if nm:
|
||
timestamp = _parse_org_ts_to_iso(nm.group(1))
|
||
i += 1
|
||
# skip one optional blank line after the note header
|
||
if i < len(body_lines) and not body_lines[i].strip():
|
||
i += 1
|
||
content_lines: list[str] = []
|
||
while i < len(body_lines):
|
||
if _NOTE_TAKEN_RE.match(body_lines[i]):
|
||
break
|
||
content_lines.append(body_lines[i])
|
||
i += 1
|
||
content = "\n".join(content_lines).strip()
|
||
notes.append({"timestamp": timestamp, "content": content})
|
||
else:
|
||
i += 1
|
||
return notes
|
||
|
||
|
||
def build_index_page(
|
||
org_files: Iterable[OrgFile],
|
||
selected_path: str | None = None,
|
||
edit_mode: bool = False,
|
||
status_message: str | None = None,
|
||
status_level: str = "success",
|
||
show_webhook: bool = False,
|
||
) -> str:
|
||
org_files = list(org_files)
|
||
|
||
if not org_files:
|
||
body = "<p>No .org files were found in this directory.</p>"
|
||
nav = ""
|
||
backlinks_html = "<p class='backlinks-empty'>Open a note to see backlinks.</p>"
|
||
else:
|
||
selected = find_org_file(org_files, selected_path) or org_files[0]
|
||
known_paths = {f.relative_path for f in org_files}
|
||
id_to_path = {f.file_id.lower(): f.relative_path for f in org_files if f.file_id}
|
||
path_to_slug, _, _ = build_slugs(org_files)
|
||
backlinks = find_backlinks(org_files, selected.relative_path)
|
||
backlink_counts = build_backlink_counts(org_files)
|
||
|
||
nav_items = []
|
||
for f in org_files:
|
||
active = "active" if f.relative_path == selected.relative_path else ""
|
||
safe_href = html.escape(note_href(f.relative_path, False, path_to_slug), quote=True)
|
||
safe_title = html.escape(truncate_label(f.title))
|
||
safe_path = html.escape(truncate_label(f.relative_path))
|
||
full_title = html.escape(f.title, quote=True)
|
||
full_path = html.escape(f.relative_path, quote=True)
|
||
searchable = html.escape(f"{f.title} {f.relative_path}".lower(), quote=True)
|
||
backlinks_count = backlink_counts.get(f.relative_path, 0)
|
||
created_key = extract_created_sort_key(f.content)
|
||
created_key_attr = "" if created_key is None else str(created_key)
|
||
modified_ts = str(int(f.mtime))
|
||
pin_icon = "\U0001F4CC " if f.pinned else ""
|
||
display_title = f"{pin_icon}TODAY" if f.pinned else safe_title
|
||
pinned_attr = "server" if f.pinned else ""
|
||
nav_items.append(
|
||
f"<a class='file-link {active}' data-search='{searchable}' "
|
||
f"data-backlinks='{backlinks_count}' data-created-ts='{created_key_attr}' "
|
||
f"data-modified-ts='{modified_ts}' data-pinned='{pinned_attr}' "
|
||
f"data-file-path='{full_path}' href='{safe_href}'>"
|
||
f"<span class='file-title' title='{full_title}'>{display_title}</span>"
|
||
f"<span class='file-path' title='{full_path}'>{safe_path}</span>"
|
||
f"<button class='pin-toggle' type='button' title='Pin/unpin this file' "
|
||
f"aria-label='Pin or unpin this file'>📌</button>"
|
||
"</a>"
|
||
)
|
||
|
||
nav = "\n".join(nav_items)
|
||
mode_toggle = (
|
||
f"<a class='mode-link' href='{html.escape(note_href(selected.relative_path, False, path_to_slug), quote=True)}'>Preview</a>"
|
||
if edit_mode
|
||
else f"<a class='mode-link' href='{html.escape(note_href(selected.relative_path, True, path_to_slug), quote=True)}'>Edit</a>"
|
||
)
|
||
status_html = ""
|
||
if status_message:
|
||
safe_message = html.escape(status_message)
|
||
safe_level = "error" if status_level == "error" else "success"
|
||
status_html = f"<p class='status status-{safe_level}'>{safe_message}</p>"
|
||
|
||
if backlinks:
|
||
backlink_items = []
|
||
for source in backlinks:
|
||
safe_href = html.escape(note_href(source.relative_path, edit_mode, path_to_slug), quote=True)
|
||
safe_title = html.escape(truncate_label(source.title))
|
||
safe_path = html.escape(truncate_label(source.relative_path))
|
||
full_title = html.escape(source.title, quote=True)
|
||
full_path = html.escape(source.relative_path, quote=True)
|
||
backlink_items.append(
|
||
f"<a class='backlink-item' href='{safe_href}'>"
|
||
f"<span class='file-title' title='{full_title}'>{safe_title}</span>"
|
||
f"<span class='file-path' title='{full_path}'>{safe_path}</span>"
|
||
"</a>"
|
||
)
|
||
backlinks_html = "\n".join(backlink_items)
|
||
else:
|
||
backlinks_html = "<p class='backlinks-empty'>No notes link to this note yet.</p>"
|
||
|
||
if edit_mode:
|
||
body = (
|
||
f"<h2>Editing {html.escape(selected.relative_path)}</h2>"
|
||
f"<div class='toolbar'>{mode_toggle}</div>"
|
||
f"{status_html}"
|
||
"<form class='editor-form' method='post' action='/edit'>"
|
||
f"<input type='hidden' name='file' value='{html.escape(selected.relative_path, quote=True)}'>"
|
||
f"<textarea class='editor-box' name='content'>{html.escape(selected.content)}</textarea>"
|
||
"<button class='submit-btn' type='submit'>Submit</button>"
|
||
"</form>"
|
||
)
|
||
else:
|
||
capture_form = (
|
||
"<form class='capture-form' method='post' action='/capture'>"
|
||
f"<input type='hidden' name='file' value='{html.escape(selected.relative_path, quote=True)}'>"
|
||
"<input class='capture-input' type='text' name='text' placeholder='Capture a task\u2026' autocomplete='off'>"
|
||
"<button class='capture-btn' type='submit'>Capture</button>"
|
||
"</form>"
|
||
)
|
||
body = (
|
||
f"<h2>{html.escape(selected.relative_path)}</h2>"
|
||
f"<div class='toolbar'>{mode_toggle}</div>"
|
||
f"{status_html}"
|
||
f"{capture_form}"
|
||
f"<article class='org-content'>{render_org_to_html(selected.content, selected.relative_path, known_paths, id_to_path, path_to_slug, show_webhook=show_webhook)}</article>"
|
||
)
|
||
|
||
template = TEMPLATE_PATH.read_text(encoding="utf-8")
|
||
return (
|
||
template.replace("{{NAV_ITEMS}}", nav)
|
||
.replace("{{MAIN_CONTENT}}", body)
|
||
.replace("{{BACKLINKS}}", backlinks_html)
|
||
)
|
||
|
||
|
||
def build_backlinks_page(org_files: Iterable[OrgFile], selected_path: str | None = None) -> str:
|
||
"""Return a minimal standalone HTML page containing only the backlinks for a note."""
|
||
org_files = list(org_files)
|
||
if not org_files:
|
||
items_html = "<p>No .org files found.</p>"
|
||
else:
|
||
selected = find_org_file(org_files, selected_path) or org_files[0]
|
||
path_to_slug, _, _ = build_slugs(org_files)
|
||
backlinks = find_backlinks(org_files, selected.relative_path)
|
||
if backlinks:
|
||
backlink_items = []
|
||
for source in backlinks:
|
||
safe_href = html.escape(note_href(source.relative_path, False, path_to_slug), quote=True)
|
||
safe_title = html.escape(source.title)
|
||
safe_path = html.escape(source.relative_path)
|
||
full_title = html.escape(source.title, quote=True)
|
||
full_path = html.escape(source.relative_path, quote=True)
|
||
backlink_items.append(
|
||
f"<a class='backlink-item' href='{safe_href}'>"
|
||
f"<span class='file-title' title='{full_title}'>{safe_title}</span>"
|
||
f"<span class='file-path' title='{full_path}'>{safe_path}</span>"
|
||
"</a>"
|
||
)
|
||
items_html = "\n".join(backlink_items)
|
||
else:
|
||
items_html = "<p class='backlinks-empty'>No notes link to this note yet.</p>"
|
||
|
||
return (
|
||
"<!doctype html>\n<html lang='en'>\n<head>\n"
|
||
" <meta charset='utf-8'>\n"
|
||
" <meta name='viewport' content='width=device-width, initial-scale=1'>\n"
|
||
" <title>Backlinks</title>\n"
|
||
" <link rel='stylesheet' href='/static/style.css'>\n"
|
||
"</head>\n<body>\n"
|
||
" <div class='backlinks-pane' style='border:none;height:100vh;'>\n"
|
||
" <h3>Backlinks</h3>\n"
|
||
f" <div class='backlinks-list'>{items_html}</div>\n"
|
||
" </div>\n"
|
||
" <script>(function(){"
|
||
"var t=new URLSearchParams(window.location.search).get('theme');"
|
||
"if(t==='dark')document.body.setAttribute('data-theme','dark');"
|
||
"})()</script>\n"
|
||
"</body>\n</html>"
|
||
)
|
||
|
||
|
||
def make_handler(base_dir: Path, webhook_url: str = "", webhook_token: str = ""):
|
||
class OrgRequestHandler(BaseHTTPRequestHandler):
|
||
def do_GET(self) -> None:
|
||
parsed = urlparse(self.path)
|
||
path = parsed.path
|
||
|
||
if path.startswith("/static/"):
|
||
self.serve_static(path)
|
||
return
|
||
|
||
if path == "/sw.js":
|
||
self.serve_service_worker()
|
||
return
|
||
|
||
org_files = scan_org_files(base_dir)
|
||
params = parse_qs(parsed.query)
|
||
|
||
backlinks_match = re.fullmatch(r"/n/([^/]+)/backlinks", path)
|
||
if backlinks_match:
|
||
_, slug_to_path, id_lower_to_path = build_slugs(org_files)
|
||
resolved = resolve_slug(unquote(backlinks_match.group(1)), slug_to_path, id_lower_to_path)
|
||
if resolved is None:
|
||
self.send_error(404, "Not found")
|
||
return
|
||
html_page = build_backlinks_page(org_files, selected_path=resolved)
|
||
encoded = html_page.encode("utf-8")
|
||
self.send_response(200)
|
||
self.send_header("Content-Type", "text/html; charset=utf-8")
|
||
self.send_header("Content-Length", str(len(encoded)))
|
||
self.end_headers()
|
||
self.wfile.write(encoded)
|
||
return
|
||
|
||
note_match = re.fullmatch(r"/n/([^/]+)", path)
|
||
if note_match:
|
||
_, slug_to_path, id_lower_to_path = build_slugs(org_files)
|
||
resolved = resolve_slug(unquote(note_match.group(1)), slug_to_path, id_lower_to_path)
|
||
if resolved is None:
|
||
self.send_error(404, "Not found")
|
||
return
|
||
self.render_index(org_files, params, selected_path=resolved)
|
||
return
|
||
|
||
if path == "/":
|
||
self.render_index(org_files, params, selected_path=None)
|
||
return
|
||
|
||
self.send_error(404, "Not found")
|
||
|
||
def render_index(self, org_files: list[OrgFile], params: dict[str, list[str]], selected_path: str | None) -> None:
|
||
edit_mode = params.get("edit", ["0"])[0] == "1"
|
||
saved = params.get("saved", ["0"])[0] == "1"
|
||
error = params.get("error", [""])[0]
|
||
status_message = None
|
||
status_level = "success"
|
||
if saved:
|
||
status_message = "Saved successfully."
|
||
elif error == "missing":
|
||
status_message = "Select a valid .org file first."
|
||
status_level = "error"
|
||
elif error == "empty":
|
||
status_message = "Enter some text to capture."
|
||
status_level = "error"
|
||
elif error == "write":
|
||
status_message = "Could not write this file."
|
||
status_level = "error"
|
||
|
||
html_page = build_index_page(
|
||
org_files,
|
||
selected_path=selected_path,
|
||
edit_mode=edit_mode,
|
||
status_message=status_message,
|
||
status_level=status_level,
|
||
show_webhook=bool(webhook_url and webhook_token),
|
||
)
|
||
encoded = html_page.encode("utf-8")
|
||
|
||
self.send_response(200)
|
||
self.send_header("Content-Type", "text/html; charset=utf-8")
|
||
self.send_header("Content-Length", str(len(encoded)))
|
||
self.end_headers()
|
||
self.wfile.write(encoded)
|
||
|
||
def do_POST(self) -> None:
|
||
parsed = urlparse(self.path)
|
||
if parsed.path == "/webhook":
|
||
self.handle_webhook()
|
||
return
|
||
if parsed.path == "/capture":
|
||
self.handle_capture()
|
||
return
|
||
if parsed.path != "/edit":
|
||
self.send_error(404, "Not found")
|
||
return
|
||
|
||
content_length = int(self.headers.get("Content-Length", "0"))
|
||
body = self.rfile.read(content_length).decode("utf-8", errors="replace")
|
||
params = parse_qs(body)
|
||
selected_path = params.get("file", [None])[0]
|
||
new_content = params.get("content", [""])[0]
|
||
|
||
org_files = scan_org_files(base_dir)
|
||
path_to_slug, _, _ = build_slugs(org_files)
|
||
selected = find_org_file(org_files, selected_path)
|
||
if selected is None:
|
||
self.redirect_with_query("/", {"edit": "1", "error": "missing"})
|
||
return
|
||
|
||
try:
|
||
selected.path.write_text(new_content, encoding="utf-8")
|
||
except OSError:
|
||
self.redirect_to_note(path_to_slug, selected.relative_path, {"edit": "1", "error": "write"})
|
||
return
|
||
|
||
self.redirect_to_note(path_to_slug, selected.relative_path, {"edit": "1", "saved": "1"})
|
||
|
||
def handle_capture(self) -> None:
|
||
content_length = int(self.headers.get("Content-Length", "0"))
|
||
body = self.rfile.read(content_length).decode("utf-8", errors="replace")
|
||
params = parse_qs(body)
|
||
selected_path = params.get("file", [None])[0]
|
||
text = params.get("text", [""])[0].strip()
|
||
|
||
org_files = scan_org_files(base_dir)
|
||
path_to_slug, _, _ = build_slugs(org_files)
|
||
selected = find_org_file(org_files, selected_path)
|
||
if selected is None:
|
||
self.redirect_with_query("/", {"error": "missing"})
|
||
return
|
||
if not text:
|
||
self.redirect_to_note(path_to_slug, selected.relative_path, {"error": "empty"})
|
||
return
|
||
|
||
updated = add_to_inbox(selected.content, text)
|
||
try:
|
||
selected.path.write_text(updated, encoding="utf-8")
|
||
except OSError:
|
||
self.redirect_to_note(path_to_slug, selected.relative_path, {"error": "write"})
|
||
return
|
||
|
||
self.redirect_to_note(path_to_slug, selected.relative_path, {"saved": "1"})
|
||
|
||
def handle_webhook(self) -> None:
|
||
if not webhook_url:
|
||
self.send_json_response(400, {"error": "No webhook_url configured"})
|
||
return
|
||
|
||
content_length = int(self.headers.get("Content-Length", "0"))
|
||
body = self.rfile.read(content_length).decode("utf-8", errors="replace")
|
||
try:
|
||
payload = json.loads(body)
|
||
except json.JSONDecodeError:
|
||
self.send_json_response(400, {"error": "Invalid JSON"})
|
||
return
|
||
|
||
file_path = payload.get("file", "")
|
||
heading_text = payload.get("heading", "")
|
||
if not file_path or not heading_text:
|
||
self.send_json_response(400, {"error": "Missing file or heading"})
|
||
return
|
||
|
||
state = payload.get("state", "STRT")
|
||
if state not in ("STRT", "DONE"):
|
||
self.send_json_response(400, {"error": "Invalid state"})
|
||
return
|
||
|
||
org_files = scan_org_files(base_dir)
|
||
org_file = find_org_file(org_files, file_path)
|
||
if org_file is None:
|
||
self.send_json_response(404, {"error": "File not found"})
|
||
return
|
||
|
||
data = extract_heading_data(org_file.content, heading_text, file_id=org_file.file_id or "", state=state)
|
||
if data is None:
|
||
self.send_json_response(404, {"error": "Heading not found"})
|
||
return
|
||
|
||
print(json.dumps(data))
|
||
encoded_data = json.dumps(data).encode("utf-8")
|
||
headers = {"Content-Type": "application/json"}
|
||
if webhook_token:
|
||
headers["Authorization"] = f"Token {webhook_token}"
|
||
|
||
new_keyword = "DONE" if state == "DONE" else "STRT"
|
||
updated_content = update_heading_keyword(org_file.content, heading_text, new_keyword)
|
||
file_updated = False
|
||
if updated_content is not None:
|
||
try:
|
||
org_file.path.write_text(updated_content, encoding="utf-8")
|
||
file_updated = True
|
||
except OSError:
|
||
file_updated = False
|
||
|
||
try:
|
||
req = Request(webhook_url, data=encoded_data, headers=headers, method="POST")
|
||
with urlopen(req, timeout=10) as resp:
|
||
resp_body = resp.read().decode("utf-8", errors="replace")
|
||
result: dict = {"ok": True, "response": resp_body}
|
||
if not file_updated:
|
||
result["error"] = "Could not update heading state in file"
|
||
self.send_json_response(200, result)
|
||
except Exception as exc:
|
||
self.send_json_response(502, {"error": f"Webhook failed: {exc}"})
|
||
|
||
def send_json_response(self, status: int, data: dict) -> None:
|
||
encoded = json.dumps(data).encode("utf-8")
|
||
self.send_response(status)
|
||
self.send_header("Content-Type", "application/json")
|
||
self.send_header("Content-Length", str(len(encoded)))
|
||
self.end_headers()
|
||
self.wfile.write(encoded)
|
||
|
||
def redirect_with_query(self, path: str, params: dict[str, str]) -> None:
|
||
location = f"{path}?{urlencode(params)}"
|
||
self.send_response(303)
|
||
self.send_header("Location", location)
|
||
self.send_header("Content-Length", "0")
|
||
self.end_headers()
|
||
|
||
def redirect_to_note(self, path_to_slug: dict[str, str], relative_path: str, params: dict[str, str]) -> None:
|
||
slug = path_to_slug.get(relative_path, relative_path)
|
||
location = f"/n/{quote(slug, safe='')}?{urlencode(params)}"
|
||
self.send_response(303)
|
||
self.send_header("Location", location)
|
||
self.send_header("Content-Length", "0")
|
||
self.end_headers()
|
||
|
||
def serve_static(self, request_path: str) -> None:
|
||
rel_path = request_path[len("/static/") :]
|
||
candidate = (STATIC_DIR / rel_path).resolve()
|
||
if STATIC_DIR.resolve() not in candidate.parents and candidate != STATIC_DIR.resolve():
|
||
self.send_error(403, "Forbidden")
|
||
return
|
||
if not candidate.is_file():
|
||
self.send_error(404, "Not found")
|
||
return
|
||
|
||
data = candidate.read_bytes()
|
||
content_type = mimetypes.guess_type(str(candidate))[0] or "application/octet-stream"
|
||
self.send_response(200)
|
||
self.send_header("Content-Type", content_type)
|
||
self.send_header("Content-Length", str(len(data)))
|
||
self.end_headers()
|
||
self.wfile.write(data)
|
||
|
||
def serve_service_worker(self) -> None:
|
||
candidate = STATIC_DIR / "sw.js"
|
||
if not candidate.is_file():
|
||
self.send_error(404, "Not found")
|
||
return
|
||
data = candidate.read_bytes()
|
||
self.send_response(200)
|
||
self.send_header("Content-Type", "application/javascript")
|
||
self.send_header("Service-Worker-Allowed", "/")
|
||
self.send_header("Content-Length", str(len(data)))
|
||
self.end_headers()
|
||
self.wfile.write(data)
|
||
|
||
def log_message(self, fmt: str, *args) -> None:
|
||
return
|
||
|
||
return OrgRequestHandler
|
||
|
||
|
||
def serve(base_dir: Path, host: str, port: int, open_browser: bool = True,
|
||
webhook_url: str = "", webhook_token: str = "") -> None:
|
||
handler_cls = make_handler(base_dir, webhook_url=webhook_url, webhook_token=webhook_token)
|
||
|
||
class ReusableHTTPServer(socketserver.ThreadingMixIn, HTTPServer):
|
||
daemon_threads = True
|
||
allow_reuse_address = True
|
||
|
||
server = ReusableHTTPServer((host, port), handler_cls)
|
||
actual_port = server.server_address[1]
|
||
url = f"http://{host}:{actual_port}/"
|
||
print(f"Serving org files from {base_dir}")
|
||
print(f"Open {url}")
|
||
|
||
if open_browser:
|
||
threading.Timer(0.4, lambda: webbrowser.open(url)).start()
|
||
|
||
try:
|
||
server.serve_forever()
|
||
except KeyboardInterrupt:
|
||
print("\nShutting down.")
|
||
finally:
|
||
server.server_close()
|
||
|
||
|
||
def parse_args() -> argparse.Namespace:
|
||
parser = argparse.ArgumentParser(description="Serve local .org files in a browser-friendly web view.")
|
||
parser.add_argument(
|
||
"--config",
|
||
default=str(DEFAULT_CONFIG_PATH),
|
||
help=f"Path to config YAML file (default: {DEFAULT_CONFIG_PATH})",
|
||
)
|
||
parser.add_argument("--dir", default="notes", help="Directory to scan for .org files (default: notes)")
|
||
parser.add_argument(
|
||
"--host",
|
||
default=None,
|
||
help="Host/IP to bind the web server (overrides config bind_addr)",
|
||
)
|
||
parser.add_argument(
|
||
"--port",
|
||
type=int,
|
||
default=None,
|
||
help="Port to bind (overrides config bind_port)",
|
||
)
|
||
parser.add_argument("--no-browser", action="store_true", help="Don't auto-open the browser")
|
||
return parser.parse_args()
|
||
|
||
|
||
def _strip_quotes(value: str) -> str:
|
||
if len(value) >= 2 and ((value[0] == value[-1] == "'") or (value[0] == value[-1] == '"')):
|
||
return value[1:-1]
|
||
return value
|
||
|
||
|
||
def load_runtime_config(config_path: Path) -> dict[str, str | int]:
|
||
config: dict[str, str] = {}
|
||
if config_path.exists():
|
||
text = config_path.read_text(encoding="utf-8")
|
||
for raw_line in text.splitlines():
|
||
line = raw_line.strip()
|
||
if not line or line.startswith("#"):
|
||
continue
|
||
if ":" not in line:
|
||
continue
|
||
key, value = line.split(":", 1)
|
||
key = key.strip()
|
||
value = value.strip()
|
||
if not key:
|
||
continue
|
||
if "#" in value:
|
||
value = value.split("#", 1)[0].strip()
|
||
config[key] = _strip_quotes(value)
|
||
|
||
bind_addr = config.get("bind_addr", "127.0.0.1")
|
||
bind_port_raw = config.get("bind_port", "8000")
|
||
try:
|
||
bind_port = int(bind_port_raw)
|
||
except ValueError as exc:
|
||
raise ValueError(f"Invalid bind_port in {config_path}: {bind_port_raw!r}") from exc
|
||
if not 1 <= bind_port <= 65535:
|
||
raise ValueError(f"bind_port out of range in {config_path}: {bind_port}")
|
||
|
||
webhook_url = os.environ.get("ORGWEB_WEBHOOK_URL", "")
|
||
webhook_token = os.environ.get("ORGWEB_WEBHOOK_TOKEN", "")
|
||
|
||
return {
|
||
"bind_addr": bind_addr,
|
||
"bind_port": bind_port,
|
||
"webhook_url": webhook_url,
|
||
"webhook_token": webhook_token,
|
||
}
|
||
|
||
|
||
def main() -> None:
|
||
args = parse_args()
|
||
config = load_runtime_config(Path(args.config))
|
||
base_dir = Path(args.dir).resolve()
|
||
host = args.host if args.host is not None else str(config["bind_addr"])
|
||
port = args.port if args.port is not None else int(config["bind_port"])
|
||
serve(
|
||
base_dir, host, port,
|
||
open_browser=not args.no_browser,
|
||
webhook_url=str(config.get("webhook_url", "")),
|
||
webhook_token=str(config.get("webhook_token", "")),
|
||
)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|