317 lines
11 KiB
Python
317 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""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
|
|
- /www/bilka_togo_checklist_bilka.md
|
|
- /www/bilka_togo_checklist.json
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
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'],
|
|
'mejeri & æg': ['mælk', 'yoghurt', 'smør', 'ost', 'feta', 'mozzarella', 'æg', 'fløde', 'creme fraiche'],
|
|
'kolonial': ['ris', 'pasta', 'mel', 'havregryn', 'olie', 'eddike', 'krydder', 'bønne', 'linse', 'dåse', 'sukker', 'salt', 'tomatpuré', 'tomatpure', 'nori', 'soja'],
|
|
'frost': ['frost', 'frossen', 'is', 'edamame'],
|
|
'husholdning': ['toiletpapir', 'køkkenrulle', 'kokkenrulle', 'sæbe', 'sæb', 'opvasketabs', 'vaskemiddel', 'affaldspose'],
|
|
}
|
|
|
|
|
|
def detect_root() -> Path:
|
|
# Prefer repository-relative root so the script works both in HA (/config)
|
|
# and when run locally from the mounted workspace (/Volumes/homeassistant).
|
|
local_root = Path(__file__).resolve().parent.parent
|
|
if (local_root / 'secrets.yaml').exists():
|
|
return local_root
|
|
for root in ROOT_CANDIDATES:
|
|
if (root / 'secrets.yaml').exists():
|
|
return root
|
|
raise RuntimeError('Could not locate Home Assistant config root')
|
|
|
|
|
|
def read_bearer_token(secrets_path: Path) -> str:
|
|
for line in secrets_path.read_text().splitlines():
|
|
if line.strip().startswith('mealie_bearer_token:'):
|
|
return line.split(':', 1)[1].strip().strip('"')
|
|
raise RuntimeError('mealie_bearer_token not found in secrets.yaml')
|
|
|
|
|
|
def api_get(url: str, token: str) -> dict:
|
|
req = urllib.request.Request(url, headers={'Authorization': token})
|
|
with urllib.request.urlopen(req, timeout=20) as resp:
|
|
raw = resp.read()
|
|
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')
|
|
value = re.sub(r'\s+', ' ', value)
|
|
value = re.sub(r'[^a-z0-9 ]', '', value)
|
|
return value
|
|
|
|
|
|
def classify_category(name: str) -> str:
|
|
key = normalize_name(name)
|
|
for category, needles in CATEGORY_RULES.items():
|
|
for needle in needles:
|
|
if needle in key:
|
|
return category
|
|
return 'andet'
|
|
|
|
|
|
def read_keep_items(keep_path: Path) -> list[str]:
|
|
if not keep_path.exists():
|
|
return []
|
|
items: list[str] = []
|
|
for line in keep_path.read_text().splitlines():
|
|
stripped = line.strip()
|
|
if not stripped or stripped.startswith('#'):
|
|
continue
|
|
items.append(stripped)
|
|
return items
|
|
|
|
|
|
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:
|
|
display = (item.get('display') or '').strip()
|
|
if display:
|
|
return display
|
|
food = item.get('food') or {}
|
|
return (food.get('name') or '').strip()
|
|
|
|
|
|
def merge_items(mealie_items: list[dict], keep_items: list[str]) -> list[dict]:
|
|
merged: dict[str, dict] = {}
|
|
|
|
for item in mealie_items:
|
|
name = extract_item_name(item)
|
|
if not name:
|
|
continue
|
|
norm = normalize_name(name)
|
|
if norm not in merged:
|
|
merged[norm] = {
|
|
'name': name,
|
|
'category': classify_category(name),
|
|
'sources': set(),
|
|
}
|
|
merged[norm]['sources'].add('Mealie')
|
|
|
|
for name in keep_items:
|
|
norm = normalize_name(name)
|
|
if norm not in merged:
|
|
merged[norm] = {
|
|
'name': name,
|
|
'category': classify_category(name),
|
|
'sources': set(),
|
|
}
|
|
merged[norm]['sources'].add('Google Keep')
|
|
|
|
result = []
|
|
for _, row in merged.items():
|
|
result.append(
|
|
{
|
|
'name': row['name'],
|
|
'category': row['category'],
|
|
'sources': sorted(row['sources']),
|
|
}
|
|
)
|
|
|
|
result.sort(key=lambda x: (x['category'], normalize_name(x['name'])))
|
|
return result
|
|
|
|
|
|
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)
|
|
|
|
payload = {'count': len(items), 'items': items}
|
|
(www / 'bilka_togo_checklist.json').write_text(json.dumps(payload, ensure_ascii=False, indent=2))
|
|
|
|
grouped: dict[str, list[dict]] = defaultdict(list)
|
|
for item in items:
|
|
grouped[item['category']].append(item)
|
|
|
|
lines = [
|
|
'# 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()}',
|
|
'',
|
|
]
|
|
|
|
for category in sorted(grouped.keys()):
|
|
lines.append(f'## {category.title()}')
|
|
for item in grouped[category]:
|
|
sources = '/'.join(item['sources'])
|
|
lines.append(f"- [ ] {item['name']} ({sources})")
|
|
lines.append('')
|
|
|
|
(www / 'bilka_togo_checklist.md').write_text('\n'.join(lines))
|
|
|
|
# Bilka-ready checklist: same grouped list, but without source metadata.
|
|
bilka_lines = [
|
|
'# 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()}',
|
|
'',
|
|
]
|
|
|
|
for category in sorted(grouped.keys()):
|
|
bilka_lines.append(f'## {category.title()}')
|
|
for item in grouped[category]:
|
|
bilka_lines.append(f"- [ ] {item['name']}")
|
|
bilka_lines.append('')
|
|
|
|
(www / 'bilka_togo_checklist_bilka.md').write_text('\n'.join(bilka_lines))
|
|
|
|
|
|
def main() -> None:
|
|
root = detect_root()
|
|
token = read_bearer_token(root / 'secrets.yaml')
|
|
keep_path = root / 'dokumenter' / 'google_keep_indkoeb.txt'
|
|
|
|
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, start_date, end_date)
|
|
|
|
print(
|
|
'OK: '
|
|
f'window={start_date.isoformat()}..{end_date.isoformat()} '
|
|
f'recipes_added={added_recipes} merged_items={len(merged)}'
|
|
)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|