Fix Mealie shopping refresh flow with bulk recipe import and Bilka outputs

2026-04-22 16:12:23 +02:00
parent 83f8908a3f
commit ef9200f65b
8 changed files with 949 additions and 15 deletions
@@ -1,5 +1,9 @@
 #!/usr/bin/env python3
-"""Merge Mealie shopping items with a local Google Keep base list.
+"""Build and merge shopping list for Bilka ToGo.
+
+Flow:
+1) Build Mealie shopping list from meal plan entries in the Friday-Thursday window.
+2) Merge those Mealie shopping items with a local Google Keep base list.

 Output:
 - /www/bilka_togo_checklist.md
@@ -13,10 +17,13 @@ import json
 import re
 import urllib.request
 from collections import defaultdict
+from datetime import date, timedelta
 from pathlib import Path


 ROOT_CANDIDATES = [Path('/Volumes/homeassistant'), Path('/config')]
+MEALIE_BASE_URL = 'http://10.0.0.142:9925'
+TARGET_SHOPPING_LIST_NAME = 'Bilka ToGo'
 CATEGORY_RULES = {
    'frugt & grønt': ['banan', 'aeble', 'æble', 'citron', 'lime', 'tomat', 'salat', 'agurk', 'gulerod', 'kartoffel', 'log', 'løg', 'hvidlog', 'hvidløg', 'broccoli', 'spidskal', 'spidskål', 'avocado', 'peberfrugt'],
    'kød & fisk': ['kylling', 'oksekød', 'hakket', 'ribeye', 'bacon', 'laks', 'fisk', 'skinke', 'pølse'],
@@ -53,6 +60,30 @@ def api_get(url: str, token: str) -> dict:
    return json.loads(raw) if raw else {}


+def api_request(
+    base_url: str,
+    path: str,
+    token: str,
+    method: str = 'GET',
+    payload: dict | list | None = None,
+    timeout: int = 90,
+):
+    data = None
+    headers = {'Authorization': token}
+    if payload is not None:
+        data = json.dumps(payload).encode('utf-8')
+        headers['Content-Type'] = 'application/json'
+    req = urllib.request.Request(f'{base_url}{path}', headers=headers, data=data, method=method)
+    with urllib.request.urlopen(req, timeout=timeout) as resp:
+        raw = resp.read()
+    if not raw:
+        return {}
+    try:
+        return json.loads(raw)
+    except json.JSONDecodeError:
+        return {}
+
+
 def normalize_name(value: str) -> str:
    value = value.lower().strip()
    value = value.replace('å', 'aa').replace('æ', 'ae').replace('ø', 'oe')
@@ -82,9 +113,83 @@ def read_keep_items(keep_path: Path) -> list[str]:
    return items


-def fetch_mealie_items(base_url: str, token: str) -> list[dict]:
-    data = api_get(f'{base_url}/api/households/shopping/items?perPage=500', token)
-    return data.get('items', []) or []
+def friday_to_thursday_window(today: date) -> tuple[date, date]:
+    days_until_friday = (4 - today.weekday()) % 7
+    start = today + timedelta(days=days_until_friday)
+    end = start + timedelta(days=6)
+    return start, end
+
+
+def ensure_shopping_list(base_url: str, token: str, name: str) -> str:
+    lists = api_request(base_url, '/api/households/shopping/lists?perPage=200', token).get('items', []) or []
+    for shopping_list in lists:
+        if (shopping_list.get('name') or '').strip().lower() == name.lower():
+            return shopping_list['id']
+
+    created = api_request(
+        base_url,
+        '/api/households/shopping/lists',
+        token,
+        method='POST',
+        payload={'name': name},
+    )
+    return created['id']
+
+
+def clear_shopping_list_items(base_url: str, token: str, shopping_list_id: str) -> None:
+    items = api_request(base_url, '/api/households/shopping/items?perPage=1000', token).get('items', []) or []
+    for item in items:
+        if item.get('shoppingListId') == shopping_list_id and item.get('id'):
+            api_request(base_url, f"/api/households/shopping/items/{item['id']}", token, method='DELETE')
+
+
+def get_mealplan_recipe_ids(base_url: str, token: str, start_date: date, end_date: date) -> list[str]:
+    entries: list[dict] = []
+    current = start_date
+    while current <= end_date:
+        path = (
+            f"/api/households/mealplans?start_date={current.isoformat()}"
+            f"&end_date={current.isoformat()}&perPage=100"
+        )
+        day_payload = api_request(base_url, path, token)
+        day_items = day_payload.get('items', []) if isinstance(day_payload, dict) else []
+        entries.extend(day_items or [])
+        current = current + timedelta(days=1)
+
+    result: list[str] = []
+    seen: set[str] = set()
+    for entry in entries:
+        recipe_id = entry.get('recipeId')
+        if not recipe_id:
+            continue
+        if recipe_id in seen:
+            continue
+        seen.add(recipe_id)
+        result.append(recipe_id)
+    return result
+
+
+def add_recipes_to_shopping_list(base_url: str, token: str, shopping_list_id: str, recipe_ids: list[str]) -> int:
+    if not recipe_ids:
+        return 0
+
+    payload = [{'recipeId': rid, 'recipeIncrementQuantity': 1} for rid in recipe_ids]
+    api_request(
+        base_url,
+        f'/api/households/shopping/lists/{shopping_list_id}/recipe',
+        token,
+        method='POST',
+        payload=payload,
+    )
+    return len(recipe_ids)
+
+
+def fetch_mealie_items(base_url: str, token: str, shopping_list_id: str | None = None) -> list[dict]:
+    data = api_request(base_url, '/api/households/shopping/items?perPage=1000', token)
+    items = data.get('items', []) or []
+    if not shopping_list_id:
+        return items
+    return [item for item in items if item.get('shoppingListId') == shopping_list_id]


 def extract_item_name(item: dict) -> str:
@@ -135,7 +240,7 @@ def merge_items(mealie_items: list[dict], keep_items: list[str]) -> list[dict]:
    return result


-def write_outputs(root: Path, items: list[dict]) -> None:
+def write_outputs(root: Path, items: list[dict], start_date: date, end_date: date) -> None:
    www = root / 'www'
    www.mkdir(parents=True, exist_ok=True)

@@ -150,6 +255,7 @@ def write_outputs(root: Path, items: list[dict]) -> None:
        '# Bilka ToGo - Kryds-af-liste',
        '',
        'Gå listen igennem derhjemme først, og bestil kun de varer du mangler.',
+        f'Plan-vindue: {start_date.isoformat()} til {end_date.isoformat()}',
        '',
    ]

@@ -167,6 +273,7 @@ def write_outputs(root: Path, items: list[dict]) -> None:
        '# Bilka ToGo - Klar til bestilling',
        '',
        'Kryds af hvad I allerede har i huset, og bestil resten.',
+        f'Plan-vindue: {start_date.isoformat()} til {end_date.isoformat()}',
        '',
    ]

@@ -184,13 +291,25 @@ def main() -> None:
    token = read_bearer_token(root / 'secrets.yaml')
    keep_path = root / 'dokumenter' / 'google_keep_indkoeb.txt'

-    mealie_items = fetch_mealie_items('http://10.0.0.142:9925', token)
+    start_date, end_date = friday_to_thursday_window(date.today())
+
+    shopping_list_id = ensure_shopping_list(MEALIE_BASE_URL, token, TARGET_SHOPPING_LIST_NAME)
+    clear_shopping_list_items(MEALIE_BASE_URL, token, shopping_list_id)
+
+    recipe_ids = get_mealplan_recipe_ids(MEALIE_BASE_URL, token, start_date, end_date)
+    added_recipes = add_recipes_to_shopping_list(MEALIE_BASE_URL, token, shopping_list_id, recipe_ids)
+
+    mealie_items = fetch_mealie_items(MEALIE_BASE_URL, token, shopping_list_id=shopping_list_id)
    keep_items = read_keep_items(keep_path)

    merged = merge_items(mealie_items, keep_items)
-    write_outputs(root, merged)
+    write_outputs(root, merged, start_date, end_date)

-    print(f'OK: merged {len(merged)} items')
+    print(
+        'OK: '
+        f'window={start_date.isoformat()}..{end_date.isoformat()} '
+        f'recipes_added={added_recipes} merged_items={len(merged)}'
+    )


 if __name__ == '__main__':