|
#!/usr/bin/env python3 |
|
""" |
|
Obsidian Markdown -> KitchenOwl Recipe importer (household-scoped API) |
|
|
|
Supports: |
|
- YAML frontmatter (German keys) |
|
- Simple quantity parsing (e.g. "Zwiebel 2x", "1 Packung Nudeln", "Nudeln 1 Packung") |
|
- Filters out invalid/empty ingredient names to satisfy AddRecipe schema |
|
- Skip flag: kitchenOwl: true -> recipe will NOT be imported again |
|
|
|
KitchenOwl backend (your code) expects for POST /api/household/<id>/recipe: |
|
- name: string (required, not whitespace) |
|
- description: string (optional, but if present not None) |
|
- items: list of {name(required), description(default=""), optional(default=True)} |
|
- tags: list of strings (optional) |
|
- time/cook_time/prep_time/yields/visibility >= 0 (optional) |
|
|
|
Usage: |
|
pip install requests |
|
python kitchenowl_import_recipes.py \ |
|
--input "/path/to/Obsidian/Vault/Rezepte" \ |
|
--url "http://192.168.137.89" \ |
|
--token "YOUR_LIFETIME_OR_VALID_JWT" \ |
|
--household-id 1 \ |
|
--dry-run |
|
""" |
|
|
|
from __future__ import annotations |
|
|
|
import argparse |
|
import re |
|
import sys |
|
from dataclasses import dataclass |
|
from pathlib import Path |
|
from typing import Dict, List, Any, Optional, Tuple |
|
|
|
import requests |
|
|
|
|
|
# ----------------------------- |
|
# YAML FRONTMATTER (simple) |
|
# ----------------------------- |
|
def parse_frontmatter(md_text: str) -> Tuple[Dict[str, Any], str]: |
|
lines = md_text.splitlines() |
|
if not lines or lines[0].strip() != "---": |
|
return {}, md_text |
|
|
|
fm: Dict[str, Any] = {} |
|
i = 1 |
|
current_list_key: Optional[str] = None |
|
|
|
while i < len(lines): |
|
line = lines[i] |
|
|
|
if line.strip() == "---": |
|
body = "\n".join(lines[i + 1 :]) |
|
return fm, body |
|
|
|
# list item (e.g. "- pasta") |
|
list_match = re.match(r"^\s*-\s+(.*)", line) |
|
if list_match and current_list_key: |
|
if not isinstance(fm.get(current_list_key), list): |
|
fm[current_list_key] = [] |
|
item = list_match.group(1).strip() |
|
if item: |
|
fm[current_list_key].append(item) |
|
i += 1 |
|
continue |
|
|
|
# key: value |
|
kv_match = re.match(r"^\s*([\wäöüÄÖÜß\-]+)\s*:\s*(.*)", line) |
|
if kv_match: |
|
key = kv_match.group(1).strip() |
|
value = kv_match.group(2).strip() |
|
|
|
if value == "": |
|
# could be scalar empty OR start of list. We'll allow list lines after. |
|
fm[key] = "" |
|
current_list_key = key |
|
else: |
|
current_list_key = None |
|
v_lower = value.lower() |
|
if v_lower in ("true", "false"): |
|
fm[key] = (v_lower == "true") |
|
elif re.fullmatch(r"-?\d+", value): |
|
try: |
|
fm[key] = int(value) |
|
except ValueError: |
|
fm[key] = value |
|
else: |
|
fm[key] = value |
|
else: |
|
current_list_key = None |
|
|
|
i += 1 |
|
|
|
# if closing '---' missing, keep what we parsed |
|
return fm, md_text |
|
|
|
|
|
def fm_bool(fm: Dict[str, Any], key: str, default: bool = False) -> bool: |
|
v = fm.get(key, default) |
|
if isinstance(v, bool): |
|
return v |
|
if isinstance(v, int): |
|
return v != 0 |
|
if isinstance(v, str): |
|
return v.strip().lower() in ("true", "yes", "1", "on") |
|
return bool(v) |
|
|
|
|
|
# ----------------------------- |
|
# MARKDOWN PARSING |
|
# ----------------------------- |
|
TITLE_RE = re.compile(r"^\s*#\s+(.*)") |
|
H2_RE = re.compile(r"^\s*##\s+(.*)") |
|
BULLET_RE = re.compile(r"^\s*-\s+(.*)") |
|
|
|
|
|
@dataclass |
|
class ParsedRecipe: |
|
title: str |
|
ingredients: List[str] |
|
instructions: str |
|
fm: Dict[str, Any] |
|
source_file: str |
|
|
|
|
|
def parse_recipe(md_text: str, source_file: str) -> Optional[ParsedRecipe]: |
|
fm, body = parse_frontmatter(md_text) |
|
lines = body.splitlines() |
|
|
|
title = str(fm.get("title", "")).strip() |
|
ingredients: List[str] = [] |
|
instructions_lines: List[str] = [] |
|
section: Optional[str] = None |
|
|
|
for line in lines: |
|
title_match = TITLE_RE.match(line) |
|
if title_match and not title: |
|
title = title_match.group(1).strip() |
|
continue |
|
|
|
h2_match = H2_RE.match(line) |
|
if h2_match: |
|
header = h2_match.group(1).strip().lower() |
|
if header == "zutaten": |
|
section = "ingredients" |
|
elif header == "rezept": |
|
section = "instructions" |
|
else: |
|
section = None |
|
continue |
|
|
|
if section == "ingredients": |
|
m = BULLET_RE.match(line) |
|
if m: |
|
ingredients.append(m.group(1).strip()) |
|
elif section == "instructions": |
|
instructions_lines.append(line) |
|
|
|
if not title or title.isspace(): |
|
return None |
|
|
|
return ParsedRecipe( |
|
title=title, |
|
ingredients=ingredients, |
|
instructions="\n".join(instructions_lines).strip(), |
|
fm=fm, |
|
source_file=source_file, |
|
) |
|
|
|
|
|
# ----------------------------- |
|
# INGREDIENT PARSE |
|
# ----------------------------- |
|
def parse_ingredient_line(line: str) -> Tuple[str, str, bool]: |
|
s = (line or "").strip() |
|
if not s: |
|
return "", "", False |
|
|
|
optional = False |
|
|
|
# optional marker at the end |
|
m_opt = re.match(r"^(.*)\s+\[(optional|opt)\]\s*$", s, re.IGNORECASE) |
|
if m_opt: |
|
s = m_opt.group(1).strip() |
|
optional = True |
|
|
|
# Zwiebel 2x |
|
m = re.match(r"^(.*)\s+(\d+)\s*x$", s, re.IGNORECASE) |
|
if m: |
|
return m.group(1).strip(), f"{m.group(2)}x", optional |
|
|
|
# 2x Zwiebel |
|
m = re.match(r"^(\d+)\s*x?\s+(.*)$", s, re.IGNORECASE) |
|
if m: |
|
return m.group(2).strip(), f"{m.group(1)}x", optional |
|
|
|
# 1 Packung Nudeln |
|
m = re.match(r"^(\d+)\s+([A-Za-zäöüÄÖÜß]+)\s+(.*)$", s) |
|
if m: |
|
return m.group(3).strip(), f"{m.group(1)} {m.group(2)}", optional |
|
|
|
# Nudeln 1 Packung |
|
m = re.match(r"^(.*)\s+(\d+)\s+([A-Za-zäöüÄÖÜß]+)$", s) |
|
if m: |
|
return m.group(1).strip(), f"{m.group(2)} {m.group(3)}", optional |
|
|
|
return s, "", optional |
|
|
|
|
|
# ----------------------------- |
|
# API CLIENT |
|
# ----------------------------- |
|
class KitchenOwlClient: |
|
def __init__(self, base_url: str, token: str, timeout: int = 25): |
|
self.base_url = base_url.rstrip("/") |
|
self.timeout = timeout |
|
self.s = requests.Session() |
|
|
|
tok = token.strip() |
|
if tok.lower().startswith("bearer "): |
|
tok = tok.split(" ", 1)[1].strip() |
|
|
|
self.s.headers.update( |
|
{ |
|
"Authorization": f"Bearer {tok}", |
|
"Content-Type": "application/json", |
|
"Accept": "application/json", |
|
} |
|
) |
|
|
|
def create_recipe(self, household_id: int, payload: Dict[str, Any]) -> Dict[str, Any]: |
|
url = f"{self.base_url}/api/household/{household_id}/recipe" |
|
r = self.s.post(url, json=payload, timeout=self.timeout) |
|
|
|
if r.status_code in (200, 201): |
|
return r.json() |
|
|
|
print("\n---- SERVER RESPONSE ----") |
|
print("Status:", r.status_code) |
|
print("Content-Type:", r.headers.get("Content-Type")) |
|
try: |
|
print("JSON:", r.json()) |
|
except Exception: |
|
print("Text:", r.text) |
|
print("-------------------------\n") |
|
|
|
raise RuntimeError(f"{r.status_code}: {r.text}") |
|
|
|
|
|
# ----------------------------- |
|
# Helpers |
|
# ----------------------------- |
|
def as_int(v: Any, default: int) -> int: |
|
if v is None or v == "": |
|
return default |
|
if isinstance(v, int): |
|
return v |
|
s = str(v).strip() |
|
if not s: |
|
return default |
|
m = re.match(r"^(-?\d+)", s) |
|
if m: |
|
try: |
|
return int(m.group(1)) |
|
except ValueError: |
|
return default |
|
return default |
|
|
|
|
|
def get_tags(fm: Dict[str, Any]) -> List[str]: |
|
tags = fm.get("tags", []) |
|
if tags is None or tags == "": |
|
return [] |
|
if isinstance(tags, list): |
|
return [str(t).strip() for t in tags if str(t).strip()] |
|
s = str(tags) |
|
return [t.strip() for t in s.split(",") if t.strip()] |
|
|
|
|
|
# ----------------------------- |
|
# MAIN |
|
# ----------------------------- |
|
def main() -> None: |
|
parser = argparse.ArgumentParser() |
|
parser.add_argument("--input", required=True, help="Folder with .md recipes") |
|
parser.add_argument("--url", required=True, help="KitchenOwl base URL, e.g. http://192.168.137.89") |
|
parser.add_argument("--token", required=True, help="JWT token (lifetime ok). Do NOT include 'Bearer ' prefix.") |
|
parser.add_argument("--household-id", type=int, required=True) |
|
parser.add_argument("--dry-run", action="store_true") |
|
|
|
# defaults (used if YAML empty/missing) |
|
parser.add_argument("--default-portionen", type=int, default=1) |
|
parser.add_argument("--default-zeit", type=int, default=0) |
|
parser.add_argument("--default-vorbereitungszeit", type=int, default=0) |
|
parser.add_argument("--default-kochzeit", type=int, default=0) |
|
|
|
args = parser.parse_args() |
|
|
|
in_path = Path(args.input).expanduser().resolve() |
|
if not in_path.exists() or not in_path.is_dir(): |
|
print(f"Input path is not a directory: {in_path}", file=sys.stderr) |
|
sys.exit(2) |
|
|
|
client = KitchenOwlClient(args.url, args.token) |
|
|
|
parsed = 0 |
|
created = 0 |
|
skipped = 0 |
|
|
|
for file in sorted(in_path.rglob("*.md")): |
|
text = file.read_text(encoding="utf-8", errors="replace") |
|
recipe = parse_recipe(text, source_file=str(file.relative_to(in_path))) |
|
if not recipe: |
|
continue |
|
|
|
# Skip if marked as already transferred |
|
if fm_bool(recipe.fm, "kitchenOwl", default=False): |
|
skipped += 1 |
|
print(f"[SKIP] kitchenOwl: true -> {recipe.title} ({recipe.source_file})") |
|
continue |
|
|
|
parsed += 1 |
|
|
|
# YAML mappings (German keys) |
|
yields = as_int(recipe.fm.get("portionen"), args.default_portionen) |
|
time_total = as_int(recipe.fm.get("zeit"), args.default_zeit) |
|
prep_time = as_int(recipe.fm.get("vorbereitungszeit"), args.default_vorbereitungszeit) |
|
cook_time = as_int(recipe.fm.get("kochzeit"), args.default_kochzeit) |
|
tags = get_tags(recipe.fm) |
|
|
|
# IMPORTANT: backend schema rejects empty/whitespace item names |
|
items_payload: List[Dict[str, Any]] = [] |
|
for ing in recipe.ingredients: |
|
name, desc, optional = parse_ingredient_line(ing) |
|
name = (name or "").strip() |
|
if not name or name.isspace(): |
|
print(f"[WARN] Skipping invalid ingredient in {recipe.source_file}: {ing!r}") |
|
continue |
|
|
|
items_payload.append( |
|
{ |
|
"name": name, |
|
"description": (desc or ""), |
|
"optional": optional, |
|
} |
|
) |
|
|
|
description = recipe.instructions.strip() or "" |
|
|
|
payload: Dict[str, Any] = { |
|
"name": recipe.title, |
|
"description": description, |
|
"items": items_payload, |
|
"tags": tags, |
|
"yields": yields, |
|
"time": time_total, |
|
"prep_time": prep_time, |
|
"cook_time": cook_time, |
|
"visibility": 0, |
|
} |
|
|
|
if args.dry_run: |
|
print(f"\n[DRY] Would create recipe: {payload['name']}") |
|
print(f" yields={yields}, time={time_total}, prep={prep_time}, cook={cook_time}") |
|
print(f" tags={tags}") |
|
print(f" items={len(items_payload)}") |
|
print(f" from={recipe.source_file}") |
|
continue |
|
|
|
res = client.create_recipe(args.household_id, payload) |
|
created += 1 |
|
print(f"[OK] Created recipe: {payload['name']} (id={res.get('id')}) from {recipe.source_file}") |
|
|
|
print("\n--- Summary ---") |
|
print(f"Considered (not kitchenOwl:true): {parsed}") |
|
print(f"Skipped (kitchenOwl:true): {skipped}") |
|
print(f"Created recipes: {created}") |
|
|
|
|
|
if __name__ == "__main__": |
|
main() |