d23273590f
Multi-tenant system to fetch and store restaurant lists from Beli accounts. Each user can connect up to 3 Beli profiles; a nightly Cronicle job (1–5 AM, randomised) refreshes the 3 stalest profiles using the reverse-engineered Beli API (mirrored at git.torrtle.co/ryan/beli-api). Rankings (been) and bookmarks (want-to-try) are stored in Postgres and queryable via a new LangChain agent tool. Beli passwords and refresh tokens are stored Fernet-encrypted. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
234 lines
8.0 KiB
Python
234 lines
8.0 KiB
Python
import logging
|
|
import os
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import httpx
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _base_url() -> str:
|
|
return os.environ.get("CRONICLE_URL", "").rstrip("/")
|
|
|
|
|
|
def _api_key() -> str:
|
|
return os.environ.get("CRONICLE_API_KEY", "")
|
|
|
|
|
|
def _is_configured() -> bool:
|
|
return bool(_base_url() and _api_key())
|
|
|
|
|
|
async def _get_first_category_id() -> str:
|
|
async with httpx.AsyncClient() as client:
|
|
resp = await client.get(
|
|
f"{_base_url()}/api/app/get_categories/v1",
|
|
params={"api_key": _api_key()},
|
|
timeout=10,
|
|
)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
if data.get("code") != 0:
|
|
raise RuntimeError(f"Cronicle get_categories error: {data}")
|
|
rows = data.get("rows", [])
|
|
if not rows:
|
|
raise RuntimeError("Cronicle has no categories configured")
|
|
return rows[0]["id"]
|
|
|
|
|
|
def _make_trigger_script() -> str:
|
|
"""Node.js script that tries raggr-green then falls back to raggr-blue."""
|
|
cron_secret = os.environ.get("CRON_SECRET", "")
|
|
path = "/api/scheduled-messages/trigger"
|
|
return f"""#!/usr/bin/env node
|
|
const http = require('http');
|
|
function post(host) {{
|
|
return new Promise((resolve, reject) => {{
|
|
const req = http.request(
|
|
{{ hostname: host, port: 8080, path: '{path}', method: 'POST',
|
|
headers: {{'X-Cron-Secret': '{cron_secret}'}} }},
|
|
res => {{ res.resume(); resolve(res.statusCode); }}
|
|
);
|
|
req.setTimeout(300000, () => {{ req.destroy(); reject(new Error('timeout')); }});
|
|
req.on('error', reject);
|
|
req.end();
|
|
}});
|
|
}}
|
|
post('raggr-green').catch(() => post('raggr-blue')).catch(err => {{
|
|
console.error('trigger failed:', err.message);
|
|
process.exitCode = 1;
|
|
}});
|
|
"""
|
|
|
|
|
|
def _build_timing(scheduled_at: datetime, recurrence: str) -> dict:
|
|
"""Build a Cronicle timing object from a scheduled datetime and recurrence.
|
|
|
|
- none: one-shot (fires exactly once at the given UTC datetime)
|
|
- daily: fires every day at the same hour/minute
|
|
- weekly: fires every week on the same weekday and hour/minute
|
|
- monthly: fires every month on the same day-of-month and hour/minute
|
|
"""
|
|
utc = scheduled_at.astimezone(timezone.utc)
|
|
# Round up to the next minute if there are sub-minute seconds so Cronicle
|
|
# fires after the message is due, not before.
|
|
if utc.second or utc.microsecond:
|
|
utc = (utc + timedelta(minutes=1)).replace(second=0, microsecond=0)
|
|
h, m = utc.hour, utc.minute
|
|
|
|
if recurrence == "daily":
|
|
return {"hours": [h], "minutes": [m]}
|
|
elif recurrence == "weekly":
|
|
# weekday: 0=Sunday in Cronicle (Python isoweekday: Mon=1..Sun=7)
|
|
wd = utc.isoweekday() % 7 # convert to 0=Sunday
|
|
return {"weekdays": [wd], "hours": [h], "minutes": [m]}
|
|
elif recurrence == "monthly":
|
|
return {"days": [utc.day], "hours": [h], "minutes": [m]}
|
|
else:
|
|
# one-shot
|
|
return {
|
|
"years": [utc.year],
|
|
"months": [utc.month],
|
|
"days": [utc.day],
|
|
"hours": [utc.hour],
|
|
"minutes": [utc.minute],
|
|
}
|
|
|
|
|
|
def _make_beli_refresh_script() -> str:
|
|
"""Node.js script that triggers the Beli nightly refresh endpoint."""
|
|
cron_secret = os.environ.get("CRON_SECRET", "")
|
|
path = "/api/beli/cron/refresh"
|
|
return f"""#!/usr/bin/env node
|
|
const http = require('http');
|
|
function post(host) {{
|
|
return new Promise((resolve, reject) => {{
|
|
const req = http.request(
|
|
{{ hostname: host, port: 8080, path: '{path}', method: 'POST',
|
|
headers: {{'X-Cron-Secret': '{cron_secret}'}} }},
|
|
res => {{ res.resume(); resolve(res.statusCode); }}
|
|
);
|
|
req.setTimeout(30000, () => {{ req.destroy(); reject(new Error('timeout')); }});
|
|
req.on('error', reject);
|
|
req.end();
|
|
}});
|
|
}}
|
|
post('raggr-green').catch(() => post('raggr-blue')).catch(err => {{
|
|
console.error('beli refresh trigger failed:', err.message);
|
|
process.exitCode = 1;
|
|
}});
|
|
"""
|
|
|
|
|
|
async def ensure_beli_nightly_event() -> str | None:
|
|
"""Create (once) a daily 1 AM Cronicle event that triggers Beli nightly refresh.
|
|
|
|
Safe to call on every startup — checks for an existing event by title first.
|
|
Returns the Cronicle event ID, or None if Cronicle is not configured.
|
|
"""
|
|
if not _is_configured():
|
|
return None
|
|
|
|
title = "Beli Nightly Refresh"
|
|
try:
|
|
category_id = await _get_first_category_id()
|
|
|
|
# Check whether an event with this title already exists.
|
|
async with httpx.AsyncClient() as client:
|
|
resp = await client.get(
|
|
f"{_base_url()}/api/app/get_schedule/v1",
|
|
params={"api_key": _api_key()},
|
|
timeout=10,
|
|
)
|
|
resp.raise_for_status()
|
|
existing = resp.json().get("rows", [])
|
|
|
|
if any(e.get("title") == title for e in existing):
|
|
logger.info("Cronicle: Beli nightly event already registered")
|
|
return next(e["id"] for e in existing if e.get("title") == title)
|
|
|
|
script = _make_beli_refresh_script()
|
|
async with httpx.AsyncClient() as client:
|
|
resp = await client.post(
|
|
f"{_base_url()}/api/app/create_event/v1",
|
|
json={
|
|
"api_key": _api_key(),
|
|
"title": title,
|
|
"enabled": 1,
|
|
"plugin": "shellplug",
|
|
"target": "allgrp",
|
|
"category": category_id,
|
|
"timing": {"hours": [1], "minutes": [0]},
|
|
"timezone": "UTC",
|
|
"timeout": 60,
|
|
"params": {"script": script},
|
|
"notes": "Auto-created: triggers Beli nightly restaurant refresh",
|
|
},
|
|
timeout=10,
|
|
)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
if data.get("code") != 0:
|
|
raise RuntimeError(f"Cronicle create_event error: {data}")
|
|
event_id = data["id"]
|
|
logger.info("Cronicle: created Beli nightly event %s", event_id)
|
|
return event_id
|
|
|
|
except Exception:
|
|
logger.exception("Failed to register Beli nightly Cronicle event")
|
|
return None
|
|
|
|
|
|
async def create_message_event(
|
|
msg_id: str, scheduled_at: datetime, recurrence: str = "none"
|
|
) -> str | None:
|
|
"""Create a Cronicle event for a scheduled message.
|
|
|
|
Uses one-shot timing for one-time messages, recurring timing otherwise.
|
|
Returns the Cronicle event ID, or None if Cronicle is not configured.
|
|
"""
|
|
if not _is_configured():
|
|
return None
|
|
|
|
try:
|
|
category_id = await _get_first_category_id()
|
|
timing = _build_timing(scheduled_at, recurrence)
|
|
script = _make_trigger_script()
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
resp = await client.post(
|
|
f"{_base_url()}/api/app/create_event/v1",
|
|
json={
|
|
"api_key": _api_key(),
|
|
"title": f"Scheduled Message {msg_id}",
|
|
"enabled": 1,
|
|
"plugin": "shellplug",
|
|
"target": "allgrp",
|
|
"category": category_id,
|
|
"timing": timing,
|
|
"timezone": "UTC",
|
|
"timeout": 360,
|
|
"params": {"script": script},
|
|
"notes": f"Auto-created for scheduled message {msg_id} (recurrence={recurrence})",
|
|
},
|
|
timeout=10,
|
|
)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
if data.get("code") != 0:
|
|
raise RuntimeError(f"Cronicle create_event error: {data}")
|
|
event_id = data["id"]
|
|
logger.info(
|
|
"Cronicle: created %s event %s for message %s",
|
|
recurrence,
|
|
event_id,
|
|
msg_id,
|
|
)
|
|
return event_id
|
|
|
|
except Exception:
|
|
logger.exception(
|
|
"Failed to create Cronicle event for message %s — will not fire", msg_id
|
|
)
|
|
return None
|