339 lines
13 KiB
Python
Executable File
339 lines
13 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
|
||
import json
|
||
import os
|
||
import sys
|
||
import tempfile
|
||
import time
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from urllib.request import Request, urlopen
|
||
from urllib.error import URLError
|
||
import base64
|
||
|
||
docstring = """
|
||
This script fetches recipes from the Nextcloud Cookbook API, merges them
|
||
with any existing notes already written in the output Org file (cook dates,
|
||
tips, etc.), and regenerates the file while preserving the org-roam header
|
||
and all hand-written notes.
|
||
|
||
It runs on a cron schedule (e.g. nightly) and emits a JSON summary line on
|
||
stdout on success, or on stderr on failure, matching the same format used by
|
||
calibre_org_sync.
|
||
|
||
JSON summary format:
|
||
{"script": "cookbook_org_sync", "status": "success", "duration_ms": 42,
|
||
"output": "/home/zaine/master-folder/org_files/org_roam/20260307233451-recipes_cookbook.org",
|
||
"timestamp": "2026-03-08T00:00:02.123456"}
|
||
"""
|
||
|
||
# ─────────────────────────────────────────────────────────────
|
||
# Configuration – edit these values
|
||
# ─────────────────────────────────────────────────────────────
|
||
|
||
NEXTCLOUD_BASE_URL = "https://nextcloud.zainezq.com"
|
||
NEXTCLOUD_USERNAME = os.environ.get("NEXTCLOUD_USER", "") # or hard-code
|
||
NEXTCLOUD_PASSWORD = os.environ.get("NEXTCLOUD_PASS", "") # or hard-code
|
||
|
||
RECIPES_ENDPOINT = "/apps/cookbook/api/v1/recipes"
|
||
|
||
OUTPUT_ORG = Path(
|
||
"/home/zaine/master-folder/org_files/org_roam/"
|
||
"20260307233451-recipes_cookbook.org"
|
||
)
|
||
|
||
# ─────────────────────────────────────────────────────────────
|
||
# API fetch
|
||
# ─────────────────────────────────────────────────────────────
|
||
|
||
def fetch_recipes():
|
||
url = NEXTCLOUD_BASE_URL.rstrip("/") + RECIPES_ENDPOINT
|
||
credentials = base64.b64encode(
|
||
f"{NEXTCLOUD_USERNAME}:{NEXTCLOUD_PASSWORD}".encode()
|
||
).decode()
|
||
|
||
req = Request(url, headers={
|
||
"Authorization": f"Basic {credentials}",
|
||
"Accept": "application/json",
|
||
"OCS-APIRequest": "true",
|
||
})
|
||
|
||
try:
|
||
with urlopen(req, timeout=30) as resp:
|
||
raw = resp.read()
|
||
except URLError as exc:
|
||
raise RuntimeError(f"Failed to reach Nextcloud API: {exc}") from exc
|
||
|
||
try:
|
||
data = json.loads(raw)
|
||
except json.JSONDecodeError as exc:
|
||
raise RuntimeError(f"Invalid JSON from API: {exc}") from exc
|
||
|
||
if not isinstance(data, list):
|
||
raise RuntimeError(f"Unexpected API response shape: {type(data)}")
|
||
|
||
return data
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────
|
||
# Organise recipes by category
|
||
# ─────────────────────────────────────────────────────────────
|
||
|
||
def group_by_category(recipes):
|
||
"""Returns {category: [recipe, ...]} sorted alphabetically."""
|
||
grouped = {}
|
||
for recipe in recipes:
|
||
cat = (recipe.get("category") or "Uncategorised").strip()
|
||
grouped.setdefault(cat, []).append(recipe)
|
||
|
||
for cat in grouped:
|
||
grouped[cat].sort(key=lambda r: r.get("name", "").lower())
|
||
|
||
return grouped
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────
|
||
# Preserve existing hand-written notes from the current file
|
||
# ─────────────────────────────────────────────────────────────
|
||
|
||
def parse_existing_notes(path):
|
||
"""
|
||
Walks the current org file and collects the PROPERTIES drawer content
|
||
for each recipe heading (matched by RECIPE_ID property).
|
||
|
||
Returns {recipe_id_str: {"cooked_dates": [...], "tips": str, "extra_props": [...]}}
|
||
"""
|
||
if not path.exists():
|
||
return {}
|
||
|
||
lines = path.read_text(encoding="utf-8").splitlines()
|
||
notes = {}
|
||
|
||
i = 0
|
||
while i < len(lines):
|
||
line = lines[i]
|
||
|
||
# Detect a level-2 recipe heading "** Some Title :keywords:"
|
||
if line.startswith("** "):
|
||
recipe_id = None
|
||
cooked_dates = []
|
||
tips_lines = []
|
||
extra_props = []
|
||
|
||
# Look for the PROPERTIES drawer immediately following
|
||
j = i + 1
|
||
in_props = False
|
||
while j < len(lines):
|
||
l = lines[j].strip()
|
||
if l == ":PROPERTIES:":
|
||
in_props = True
|
||
j += 1
|
||
continue
|
||
if in_props:
|
||
if l == ":END:":
|
||
in_props = False
|
||
j += 1
|
||
break
|
||
if l.startswith(":RECIPE_ID:"):
|
||
recipe_id = l.split(":", 2)[2].strip()
|
||
elif l.startswith(":COOKED:"):
|
||
# may be repeated
|
||
val = l.split(":", 2)[2].strip()
|
||
if val:
|
||
cooked_dates.append(val)
|
||
elif l.startswith(":TIPS:"):
|
||
tips_lines.append(l.split(":", 2)[2].strip())
|
||
elif l and not l.startswith(":CATEGORY:") \
|
||
and not l.startswith(":KEYWORDS:") \
|
||
and not l.startswith(":CREATED:") \
|
||
and not l.startswith(":MODIFIED:") \
|
||
and not l.startswith(":URL:"):
|
||
# Preserve unknown/custom properties
|
||
extra_props.append(l)
|
||
j += 1
|
||
continue
|
||
break # stop if we hit something that is not the drawer
|
||
|
||
if recipe_id:
|
||
notes[recipe_id] = {
|
||
"cooked_dates": cooked_dates,
|
||
"tips": " ".join(tips_lines),
|
||
"extra_props": extra_props,
|
||
}
|
||
|
||
i += 1
|
||
|
||
return notes
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────
|
||
# Org header handling
|
||
# ─────────────────────────────────────────────────────────────
|
||
|
||
def read_org_roam_header(path):
|
||
"""
|
||
Returns the org-roam PROPERTIES block (up to and including :END:).
|
||
Raises RuntimeError if the file exists but has no such block.
|
||
Returns None if the file does not exist.
|
||
"""
|
||
if not path.exists():
|
||
return None
|
||
|
||
lines = path.read_text(encoding="utf-8").splitlines()
|
||
for i, line in enumerate(lines):
|
||
if line.strip() == ":END:":
|
||
return "\n".join(lines[: i + 1]).rstrip()
|
||
|
||
raise RuntimeError("No org-roam PROPERTIES drawer found in output file")
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────
|
||
# Org body generation
|
||
# ─────────────────────────────────────────────────────────────
|
||
|
||
def emit_org_body(grouped, existing_notes):
|
||
lines = []
|
||
|
||
lines.append("#+title: Recipes Cookbook")
|
||
lines.append("#+AUTHOR: Auto-generated")
|
||
lines.append("#+filetags: :recipes:cooking:org:index:")
|
||
lines.append("#+STARTUP: content")
|
||
lines.append(f"#+PROPERTY: GENERATED_AT {datetime.now().isoformat()}")
|
||
lines.append("")
|
||
lines.append(
|
||
"# How to add notes: under any recipe heading, add :COOKED: <date> "
|
||
"to record when you made it, and :TIPS: your tip text to add notes."
|
||
)
|
||
lines.append(
|
||
"# Multiple :COOKED: lines are supported. "
|
||
"These are preserved across regenerations."
|
||
)
|
||
lines.append("")
|
||
|
||
for category in sorted(grouped.keys(), key=str.lower):
|
||
lines.append(f"* {category}")
|
||
lines.append("")
|
||
|
||
for recipe in grouped[category]:
|
||
rid = str(recipe.get("recipe_id") or recipe.get("id", ""))
|
||
name = recipe.get("name", "Unnamed Recipe")
|
||
keywords_raw = recipe.get("keywords", "") or ""
|
||
keywords = [k.strip() for k in keywords_raw.split(",") if k.strip()]
|
||
date_created = recipe.get("dateCreated", "")
|
||
date_modified = recipe.get("dateModified", "")
|
||
recipe_url = (
|
||
f"{NEXTCLOUD_BASE_URL.rstrip('/')}"
|
||
f"/apps/cookbook/webapp/recipes/{rid}"
|
||
)
|
||
|
||
# org tag string from keywords (spaces → underscores)
|
||
tag_str = ":".join(k.replace(" ", "_") for k in keywords)
|
||
tag_block = f":{tag_str}:" if tag_str else ""
|
||
|
||
lines.append(f"** {name} {tag_block}".rstrip())
|
||
lines.append(":PROPERTIES:")
|
||
lines.append(f":RECIPE_ID: {rid}")
|
||
lines.append(f":CATEGORY: {category}")
|
||
if keywords:
|
||
lines.append(f":KEYWORDS: {keywords_raw}")
|
||
lines.append(f":CREATED: {date_created}")
|
||
lines.append(f":MODIFIED: {date_modified}")
|
||
lines.append(f":URL: {recipe_url}")
|
||
|
||
# Re-inject preserved notes
|
||
note = existing_notes.get(rid, {})
|
||
for cd in note.get("cooked_dates", []):
|
||
lines.append(f":COOKED: {cd}")
|
||
if note.get("tips"):
|
||
lines.append(f":TIPS: {note['tips']}")
|
||
for ep in note.get("extra_props", []):
|
||
lines.append(ep)
|
||
|
||
lines.append(":END:")
|
||
|
||
# Placeholder reminder if never cooked
|
||
if not note.get("cooked_dates"):
|
||
lines.append(
|
||
"# Add :COOKED: <yyyy-mm-dd> in the drawer above "
|
||
"when you first make this."
|
||
)
|
||
|
||
lines.append("")
|
||
|
||
return "\n".join(lines)
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────
|
||
# Atomic write
|
||
# ─────────────────────────────────────────────────────────────
|
||
|
||
def atomic_write(path, content):
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
with tempfile.NamedTemporaryFile(
|
||
"w", delete=False, dir=str(path.parent), encoding="utf-8"
|
||
) as tmp:
|
||
tmp.write(content)
|
||
temp_name = tmp.name
|
||
os.replace(temp_name, path)
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────
|
||
# Main
|
||
# ─────────────────────────────────────────────────────────────
|
||
|
||
def main():
|
||
if not NEXTCLOUD_USERNAME or not NEXTCLOUD_PASSWORD:
|
||
raise RuntimeError(
|
||
"Nextcloud credentials not set. "
|
||
"Export NEXTCLOUD_USER and NEXTCLOUD_PASS environment variables."
|
||
)
|
||
|
||
# 1. Read existing org-roam header (must already exist)
|
||
header = read_org_roam_header(OUTPUT_ORG)
|
||
if header is None:
|
||
raise RuntimeError(
|
||
f"{OUTPUT_ORG} does not exist. "
|
||
"Create it first with the org-roam ID block."
|
||
)
|
||
|
||
# 2. Scrape any hand-written notes from the current file
|
||
existing_notes = parse_existing_notes(OUTPUT_ORG)
|
||
|
||
# 3. Fetch recipes from Nextcloud
|
||
recipes = fetch_recipes()
|
||
|
||
# 4. Group by category
|
||
grouped = group_by_category(recipes)
|
||
|
||
# 5. Generate new body and combine with header
|
||
body = emit_org_body(grouped, existing_notes)
|
||
content = header + "\n\n" + body
|
||
|
||
# 6. Write atomically
|
||
atomic_write(OUTPUT_ORG, content)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
start = time.time()
|
||
try:
|
||
main()
|
||
duration_ms = int((time.time() - start) * 1000)
|
||
print(json.dumps({
|
||
"script": "cookbook_org_sync",
|
||
"status": "success",
|
||
"duration_ms": duration_ms,
|
||
"output": str(OUTPUT_ORG),
|
||
"timestamp": datetime.now().isoformat(),
|
||
}))
|
||
sys.exit(0)
|
||
|
||
except Exception as exc:
|
||
duration_ms = int((time.time() - start) * 1000)
|
||
print(json.dumps({
|
||
"script": "cookbook_org_sync",
|
||
"status": "failure",
|
||
"duration_ms": duration_ms,
|
||
"error": str(exc),
|
||
}), file=sys.stderr)
|
||
sys.exit(1)
|