Add Beli restaurant recommendation integration
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>
This commit is contained in:
@@ -157,3 +157,7 @@ OBSIDIAN_VAULT_PATH=/app/data/obsidian
|
||||
# LangGraph checkpoint store (SQLite). Must be on a persistent volume.
|
||||
# Both blue and green containers share this volume via docker-compose.
|
||||
CHECKPOINT_DB_PATH=/app/checkpoints/checkpoints.db
|
||||
|
||||
# Beli restaurant recommendations
|
||||
# Beli account credentials for each profile are stored encrypted in the database.
|
||||
# No additional env vars needed — auth is done via the Beli API at login time.
|
||||
|
||||
@@ -25,6 +25,7 @@ import blueprints.whatsapp
|
||||
import blueprints.imessage
|
||||
import blueprints.scheduled_messages
|
||||
import blueprints.cron
|
||||
import blueprints.beli
|
||||
import blueprints.users.models
|
||||
from config.db import TORTOISE_CONFIG
|
||||
from metrics import render_metrics
|
||||
@@ -66,6 +67,7 @@ app.register_blueprint(blueprints.whatsapp.whatsapp_blueprint)
|
||||
app.register_blueprint(blueprints.imessage.imessage_blueprint)
|
||||
app.register_blueprint(blueprints.scheduled_messages.scheduled_messages_blueprint)
|
||||
app.register_blueprint(blueprints.cron.cron_blueprint)
|
||||
app.register_blueprint(blueprints.beli.beli_blueprint)
|
||||
|
||||
|
||||
async def _obsidian_sync_loop():
|
||||
@@ -106,6 +108,10 @@ async def lifespan():
|
||||
app.main_agent = build_agent(checkpointer)
|
||||
logging.info(f"Agent checkpointer ready at {checkpoint_path}")
|
||||
|
||||
from utils.cronicle import ensure_beli_nightly_event
|
||||
|
||||
await ensure_beli_nightly_event()
|
||||
|
||||
watcher_task = None
|
||||
if os.getenv("OBSIDIAN_CONTINUOUS_SYNC") == "true":
|
||||
watcher_task = asyncio.create_task(_obsidian_sync_loop())
|
||||
|
||||
@@ -0,0 +1,163 @@
|
||||
import hmac
|
||||
import logging
|
||||
import os
|
||||
|
||||
from quart import Blueprint, jsonify, request
|
||||
from quart_jwt_extended import get_jwt_identity, jwt_refresh_token_required
|
||||
|
||||
from .logic import MAX_PROFILES_PER_USER, refresh_profile, schedule_nightly_refresh
|
||||
from .models import BeliProfile, BeliRestaurant, BeliStatus
|
||||
|
||||
beli_blueprint = Blueprint("beli", __name__, url_prefix="/api/beli")
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _check_cron_secret() -> bool:
|
||||
expected = os.environ.get("CRON_SECRET", "")
|
||||
if not expected:
|
||||
return False
|
||||
provided = request.headers.get("X-Cron-Secret", "")
|
||||
return hmac.compare_digest(provided.encode(), expected.encode())
|
||||
|
||||
|
||||
def _serialize_profile(profile: BeliProfile) -> dict:
|
||||
return {
|
||||
"id": str(profile.id),
|
||||
"beli_email": profile.beli_email,
|
||||
"beli_user_id": profile.beli_user_id,
|
||||
"enabled": profile.enabled,
|
||||
"last_refreshed_at": (
|
||||
profile.last_refreshed_at.isoformat() if profile.last_refreshed_at else None
|
||||
),
|
||||
"last_error": profile.last_error,
|
||||
"created_at": profile.created_at.isoformat(),
|
||||
}
|
||||
|
||||
|
||||
def _serialize_restaurant(r: BeliRestaurant) -> dict:
|
||||
return {
|
||||
"id": str(r.id),
|
||||
"profile_id": str(r.profile_id),
|
||||
"beli_business_id": r.beli_business_id,
|
||||
"name": r.name,
|
||||
"city": r.city,
|
||||
"neighborhood": r.neighborhood,
|
||||
"cuisine": r.cuisine,
|
||||
"price_range": r.price_range,
|
||||
"status": r.status.value,
|
||||
"rank_index": r.rank_index,
|
||||
"score": float(r.score) if r.score is not None else None,
|
||||
"notes": r.notes,
|
||||
"tags": r.tags,
|
||||
"fetched_at": r.fetched_at.isoformat(),
|
||||
}
|
||||
|
||||
|
||||
@beli_blueprint.route("/profiles", methods=["GET"])
|
||||
@jwt_refresh_token_required
|
||||
async def list_profiles():
|
||||
user_id = get_jwt_identity()
|
||||
profiles = await BeliProfile.filter(user_id=user_id).order_by("created_at").all()
|
||||
return jsonify([_serialize_profile(p) for p in profiles])
|
||||
|
||||
|
||||
@beli_blueprint.route("/profiles", methods=["POST"])
|
||||
@jwt_refresh_token_required
|
||||
async def create_profile():
|
||||
user_id = get_jwt_identity()
|
||||
|
||||
count = await BeliProfile.filter(user_id=user_id).count()
|
||||
if count >= MAX_PROFILES_PER_USER:
|
||||
return (
|
||||
jsonify(
|
||||
{"error": f"Maximum {MAX_PROFILES_PER_USER} Beli profiles per user"}
|
||||
),
|
||||
400,
|
||||
)
|
||||
|
||||
data = await request.get_json()
|
||||
if not data:
|
||||
return jsonify({"error": "Invalid payload"}), 400
|
||||
|
||||
email = (data.get("beli_email") or "").strip()
|
||||
password = (data.get("beli_password") or "").strip()
|
||||
if not email or not password:
|
||||
return jsonify({"error": "beli_email and beli_password are required"}), 400
|
||||
|
||||
if await BeliProfile.exists(user_id=user_id, beli_email=email):
|
||||
return jsonify({"error": "A profile with this email already exists"}), 409
|
||||
|
||||
from utils.beli_service import authenticate
|
||||
|
||||
try:
|
||||
session = await authenticate(email, password)
|
||||
except Exception as e:
|
||||
return jsonify({"error": f"Beli authentication failed: {e}"}), 400
|
||||
|
||||
profile = await BeliProfile.create(
|
||||
user_id=user_id,
|
||||
beli_email=email,
|
||||
beli_password=password,
|
||||
beli_user_id=session["user_id"],
|
||||
beli_refresh_token=session["refresh_token"],
|
||||
enabled=True,
|
||||
)
|
||||
return jsonify(_serialize_profile(profile)), 201
|
||||
|
||||
|
||||
@beli_blueprint.route("/profiles/<profile_id>", methods=["DELETE"])
|
||||
@jwt_refresh_token_required
|
||||
async def delete_profile(profile_id: str):
|
||||
user_id = get_jwt_identity()
|
||||
profile = await BeliProfile.get_or_none(id=profile_id, user_id=user_id)
|
||||
if not profile:
|
||||
return jsonify({"error": "Not found"}), 404
|
||||
|
||||
await BeliRestaurant.filter(profile=profile).delete()
|
||||
await profile.delete()
|
||||
return jsonify({"status": "deleted"})
|
||||
|
||||
|
||||
@beli_blueprint.route("/profiles/<profile_id>/refresh", methods=["POST"])
|
||||
@jwt_refresh_token_required
|
||||
async def refresh_profile_endpoint(profile_id: str):
|
||||
user_id = get_jwt_identity()
|
||||
profile = await BeliProfile.get_or_none(id=profile_id, user_id=user_id)
|
||||
if not profile:
|
||||
return jsonify({"error": "Not found"}), 404
|
||||
|
||||
result = await refresh_profile(str(profile.id))
|
||||
return jsonify(result)
|
||||
|
||||
|
||||
@beli_blueprint.route("/restaurants", methods=["GET"])
|
||||
@jwt_refresh_token_required
|
||||
async def list_restaurants():
|
||||
user_id = get_jwt_identity()
|
||||
profile_ids = [
|
||||
p.id for p in await BeliProfile.filter(user_id=user_id).only("id").all()
|
||||
]
|
||||
|
||||
qs = BeliRestaurant.filter(profile_id__in=profile_ids)
|
||||
|
||||
status_filter = request.args.get("status")
|
||||
city_filter = request.args.get("city")
|
||||
if status_filter:
|
||||
try:
|
||||
qs = qs.filter(status=BeliStatus(status_filter))
|
||||
except ValueError:
|
||||
return jsonify({"error": f"Invalid status: {status_filter}"}), 400
|
||||
if city_filter:
|
||||
qs = qs.filter(city__icontains=city_filter)
|
||||
|
||||
restaurants = await qs.order_by("rank_index", "name").all()
|
||||
return jsonify([_serialize_restaurant(r) for r in restaurants])
|
||||
|
||||
|
||||
@beli_blueprint.route("/cron/refresh", methods=["POST"])
|
||||
async def cron_nightly_refresh():
|
||||
if not _check_cron_secret():
|
||||
return jsonify({"error": "Unauthorized"}), 401
|
||||
|
||||
result = await schedule_nightly_refresh()
|
||||
return jsonify(result)
|
||||
@@ -0,0 +1,148 @@
|
||||
import asyncio
|
||||
import logging
|
||||
import random
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from utils.beli_service import (
|
||||
RANKING_CATEGORIES,
|
||||
REQUEST_SPACING,
|
||||
authenticate,
|
||||
extract_business,
|
||||
fetch_bookmarks,
|
||||
fetch_user_rankings,
|
||||
is_token_expired,
|
||||
refresh_access_token,
|
||||
)
|
||||
|
||||
from .models import BeliProfile, BeliRestaurant, BeliStatus
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
MAX_PROFILES_PER_USER = 3
|
||||
NIGHTLY_BATCH_SIZE = 3
|
||||
NIGHTLY_WINDOW_SECONDS = 4 * 3600 # spread refreshes across a 4-hour window
|
||||
|
||||
|
||||
async def _get_access_token(profile: BeliProfile) -> str:
|
||||
"""Return a live access token, refreshing or re-authenticating as needed."""
|
||||
if profile.beli_refresh_token and not is_token_expired(profile.beli_refresh_token):
|
||||
try:
|
||||
token = await refresh_access_token(profile.beli_refresh_token)
|
||||
return token
|
||||
except Exception:
|
||||
logger.warning(
|
||||
"Refresh token failed for profile %s, falling back to password",
|
||||
profile.id,
|
||||
)
|
||||
|
||||
session = await authenticate(profile.beli_email, profile.beli_password)
|
||||
profile.beli_user_id = session["user_id"]
|
||||
profile.beli_refresh_token = session["refresh_token"]
|
||||
await profile.save()
|
||||
return session["access_token"]
|
||||
|
||||
|
||||
async def refresh_profile(profile_id: str) -> dict:
|
||||
"""Fetch the latest rankings and bookmarks for a profile and upsert them."""
|
||||
profile = await BeliProfile.get_or_none(id=profile_id)
|
||||
if not profile or not profile.enabled:
|
||||
return {"skipped": True}
|
||||
|
||||
try:
|
||||
access_token = await _get_access_token(profile)
|
||||
|
||||
upserted = 0
|
||||
|
||||
# Fetch all ranking categories (restaurants the user has been to and ranked)
|
||||
for category in RANKING_CATEGORIES:
|
||||
await asyncio.sleep(REQUEST_SPACING)
|
||||
results = await fetch_user_rankings(
|
||||
access_token, profile.beli_user_id, category
|
||||
)
|
||||
for item in results:
|
||||
biz = extract_business(item)
|
||||
if not biz:
|
||||
continue
|
||||
await BeliRestaurant.update_or_create(
|
||||
profile=profile,
|
||||
beli_business_id=biz["beli_business_id"],
|
||||
status=BeliStatus.BEEN,
|
||||
defaults={
|
||||
"name": biz["name"],
|
||||
"city": biz["city"],
|
||||
"neighborhood": biz["neighborhood"],
|
||||
"cuisine": biz["cuisine"],
|
||||
"price_range": biz["price_range"],
|
||||
"rank_index": biz["rank_index"],
|
||||
"score": biz["score"],
|
||||
"notes": biz["notes"],
|
||||
"tags": biz["tags"],
|
||||
},
|
||||
)
|
||||
upserted += 1
|
||||
|
||||
# Fetch bookmarks (want-to-try list)
|
||||
await asyncio.sleep(REQUEST_SPACING)
|
||||
bookmarks = await fetch_bookmarks(access_token)
|
||||
for item in bookmarks:
|
||||
biz = extract_business(item)
|
||||
if not biz:
|
||||
continue
|
||||
await BeliRestaurant.update_or_create(
|
||||
profile=profile,
|
||||
beli_business_id=biz["beli_business_id"],
|
||||
status=BeliStatus.WANT_TO_TRY,
|
||||
defaults={
|
||||
"name": biz["name"],
|
||||
"city": biz["city"],
|
||||
"neighborhood": biz["neighborhood"],
|
||||
"cuisine": biz["cuisine"],
|
||||
"price_range": biz["price_range"],
|
||||
"rank_index": None,
|
||||
"score": None,
|
||||
"notes": biz["notes"],
|
||||
"tags": biz["tags"],
|
||||
},
|
||||
)
|
||||
upserted += 1
|
||||
|
||||
profile.last_refreshed_at = datetime.now(timezone.utc)
|
||||
profile.last_error = None
|
||||
await profile.save()
|
||||
|
||||
logger.info(
|
||||
"Beli profile %s refreshed: %d restaurants upserted", profile_id, upserted
|
||||
)
|
||||
return {"ok": True, "upserted": upserted}
|
||||
|
||||
except Exception as e:
|
||||
profile.last_error = str(e)
|
||||
await profile.save()
|
||||
logger.error("Failed to refresh Beli profile %s: %s", profile_id, e)
|
||||
return {"error": str(e)}
|
||||
|
||||
|
||||
async def _refresh_after_delay(profile_id: str, delay: float) -> None:
|
||||
await asyncio.sleep(delay)
|
||||
await refresh_profile(profile_id)
|
||||
|
||||
|
||||
async def schedule_nightly_refresh() -> dict:
|
||||
"""Pick the NIGHTLY_BATCH_SIZE stalest profiles and schedule them with random delays."""
|
||||
profiles = (
|
||||
await BeliProfile.filter(enabled=True)
|
||||
.order_by("last_refreshed_at")
|
||||
.limit(NIGHTLY_BATCH_SIZE)
|
||||
.all()
|
||||
)
|
||||
|
||||
if not profiles:
|
||||
logger.info("No enabled Beli profiles to refresh")
|
||||
return {"scheduled": 0}
|
||||
|
||||
for profile in profiles:
|
||||
delay = random.uniform(0, NIGHTLY_WINDOW_SECONDS)
|
||||
asyncio.create_task(_refresh_after_delay(str(profile.id), delay))
|
||||
logger.info("Scheduled Beli refresh for profile %s in %.0fs", profile.id, delay)
|
||||
|
||||
return {"scheduled": len(profiles)}
|
||||
@@ -0,0 +1,53 @@
|
||||
import enum
|
||||
import uuid
|
||||
|
||||
from tortoise import fields
|
||||
from tortoise.models import Model
|
||||
|
||||
from blueprints.email.crypto_service import EncryptedTextField
|
||||
|
||||
|
||||
class BeliStatus(str, enum.Enum):
|
||||
BEEN = "been"
|
||||
WANT_TO_TRY = "want_to_try"
|
||||
|
||||
|
||||
class BeliProfile(Model):
|
||||
id = fields.UUIDField(primary_key=True, default=uuid.uuid4)
|
||||
user = fields.ForeignKeyField("models.User", related_name="beli_profiles")
|
||||
beli_email = fields.CharField(max_length=255)
|
||||
beli_password = EncryptedTextField()
|
||||
beli_user_id = fields.CharField(max_length=255, null=True)
|
||||
# refresh token lasts 7 days; we store it encrypted to avoid re-authenticating each run
|
||||
beli_refresh_token = EncryptedTextField(null=True)
|
||||
last_refreshed_at = fields.DatetimeField(null=True)
|
||||
last_error = fields.TextField(null=True)
|
||||
enabled = fields.BooleanField(default=True)
|
||||
created_at = fields.DatetimeField(auto_now_add=True)
|
||||
updated_at = fields.DatetimeField(auto_now=True)
|
||||
|
||||
class Meta:
|
||||
table = "beli_profiles"
|
||||
unique_together = (("user", "beli_email"),)
|
||||
|
||||
|
||||
class BeliRestaurant(Model):
|
||||
id = fields.UUIDField(primary_key=True, default=uuid.uuid4)
|
||||
profile = fields.ForeignKeyField("models.BeliProfile", related_name="restaurants")
|
||||
beli_business_id = fields.CharField(max_length=255)
|
||||
name = fields.CharField(max_length=500)
|
||||
city = fields.CharField(max_length=255, null=True)
|
||||
neighborhood = fields.CharField(max_length=255, null=True)
|
||||
cuisine = fields.JSONField(default=list)
|
||||
price_range = fields.CharField(max_length=10, null=True)
|
||||
status = fields.CharEnumField(enum_type=BeliStatus, max_length=20)
|
||||
rank_index = fields.IntField(null=True)
|
||||
score = fields.DecimalField(max_digits=5, decimal_places=2, null=True)
|
||||
notes = fields.TextField(null=True)
|
||||
tags = fields.JSONField(default=list)
|
||||
fetched_at = fields.DatetimeField(auto_now=True)
|
||||
created_at = fields.DatetimeField(auto_now_add=True)
|
||||
|
||||
class Meta:
|
||||
table = "beli_restaurants"
|
||||
unique_together = (("profile", "beli_business_id", "status"),)
|
||||
@@ -1288,6 +1288,86 @@ if google_gmail_enabled:
|
||||
tools.extend([gmail_search_emails, gmail_get_email])
|
||||
|
||||
|
||||
@tool
|
||||
async def beli_search_restaurants(
|
||||
status: str = "",
|
||||
city: str = "",
|
||||
cuisine: str = "",
|
||||
name: str = "",
|
||||
) -> str:
|
||||
"""Search restaurant recommendations from Beli profiles.
|
||||
|
||||
Use this tool when the user asks about:
|
||||
- Restaurant recommendations or suggestions
|
||||
- Places they've been to
|
||||
- Restaurants on their want-to-try list
|
||||
- Dining recommendations in a specific city or cuisine
|
||||
- What restaurants they've rated or reviewed on Beli
|
||||
|
||||
Args:
|
||||
status: Filter by status — 'been' (ranked places) or 'want_to_try' (bookmarks). Leave blank for all.
|
||||
city: Filter by city name (partial match). Leave blank for all cities.
|
||||
cuisine: Filter by cuisine type (partial match). Leave blank for all cuisines.
|
||||
name: Filter by restaurant name (partial match). Leave blank for all.
|
||||
|
||||
Returns:
|
||||
Formatted list of matching restaurants with ratings, cuisine, and location.
|
||||
"""
|
||||
from blueprints.beli.models import BeliRestaurant, BeliStatus
|
||||
|
||||
try:
|
||||
qs = BeliRestaurant.all()
|
||||
|
||||
if status:
|
||||
try:
|
||||
qs = qs.filter(status=BeliStatus(status))
|
||||
except ValueError:
|
||||
return f"Invalid status '{status}'. Valid values: been, want_to_try."
|
||||
|
||||
if city:
|
||||
qs = qs.filter(city__icontains=city)
|
||||
if name:
|
||||
qs = qs.filter(name__icontains=name)
|
||||
|
||||
restaurants = await qs.order_by("rank_index", "name").all()
|
||||
|
||||
if cuisine:
|
||||
cuisine_lower = cuisine.lower()
|
||||
restaurants = [
|
||||
r
|
||||
for r in restaurants
|
||||
if any(cuisine_lower in c.lower() for c in (r.cuisine or []))
|
||||
]
|
||||
|
||||
if not restaurants:
|
||||
return "No matching restaurants found in your Beli lists."
|
||||
|
||||
lines = []
|
||||
for r in restaurants:
|
||||
score = f" — ★{r.score}" if r.score is not None else ""
|
||||
rank = f" #{r.rank_index}" if r.rank_index is not None else ""
|
||||
location = ", ".join(filter(None, [r.neighborhood, r.city]))
|
||||
cuisines = ", ".join(r.cuisine) if r.cuisine else ""
|
||||
price = f" ({r.price_range})" if r.price_range else ""
|
||||
status_label = r.status.value.replace("_", " ").title()
|
||||
line = f"• {r.name}{price}{rank}{score} [{status_label}]"
|
||||
if location:
|
||||
line += f" — {location}"
|
||||
if cuisines:
|
||||
line += f" ({cuisines})"
|
||||
if r.notes:
|
||||
line += f"\n Note: {r.notes}"
|
||||
lines.append(line)
|
||||
|
||||
return f"Found {len(lines)} restaurant(s):\n\n" + "\n".join(lines)
|
||||
|
||||
except Exception as e:
|
||||
return f"Error searching Beli restaurants: {e}"
|
||||
|
||||
|
||||
tools.append(beli_search_restaurants)
|
||||
|
||||
|
||||
@RunnableLambda
|
||||
def _build_prompt(state, config: RunnableConfig = None) -> list:
|
||||
"""Inject the per-request system prompt, replacing any prior system messages."""
|
||||
|
||||
@@ -18,6 +18,7 @@ TORTOISE_CONFIG = {
|
||||
"blueprints.email.models",
|
||||
"blueprints.scheduled_messages.models",
|
||||
"blueprints.cron.models",
|
||||
"blueprints.beli.models",
|
||||
"aerich.models",
|
||||
],
|
||||
"default_connection": "default",
|
||||
|
||||
@@ -0,0 +1,131 @@
|
||||
from tortoise import BaseDBAsyncClient
|
||||
|
||||
RUN_IN_TRANSACTION = True
|
||||
|
||||
|
||||
async def upgrade(db: BaseDBAsyncClient) -> str:
|
||||
return """
|
||||
CREATE TABLE IF NOT EXISTS "cron_jobs" (
|
||||
"id" UUID NOT NULL PRIMARY KEY,
|
||||
"name" VARCHAR(255) NOT NULL,
|
||||
"description" TEXT,
|
||||
"prompt" TEXT NOT NULL,
|
||||
"recipient" VARCHAR(255) NOT NULL,
|
||||
"channel" VARCHAR(20) NOT NULL,
|
||||
"enabled" BOOL NOT NULL DEFAULT True,
|
||||
"last_run_at" TIMESTAMPTZ,
|
||||
"last_status" VARCHAR(50),
|
||||
"last_error" TEXT,
|
||||
"created_at" TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
"updated_at" TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
|
||||
);
|
||||
COMMENT ON COLUMN "cron_jobs"."channel" IS 'IMESSAGE: imessage\nEMAIL: email';
|
||||
CREATE TABLE IF NOT EXISTS "beli_profiles" (
|
||||
"id" UUID NOT NULL PRIMARY KEY,
|
||||
"beli_email" VARCHAR(255) NOT NULL,
|
||||
"beli_password" TEXT NOT NULL,
|
||||
"beli_user_id" VARCHAR(255),
|
||||
"beli_refresh_token" TEXT,
|
||||
"last_refreshed_at" TIMESTAMPTZ,
|
||||
"last_error" TEXT,
|
||||
"enabled" BOOL NOT NULL DEFAULT True,
|
||||
"created_at" TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
"updated_at" TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
"user_id" UUID NOT NULL REFERENCES "users" ("id") ON DELETE CASCADE,
|
||||
CONSTRAINT "uid_beli_profil_user_id_f4bba0" UNIQUE ("user_id", "beli_email")
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS "beli_restaurants" (
|
||||
"id" UUID NOT NULL PRIMARY KEY,
|
||||
"beli_business_id" VARCHAR(255) NOT NULL,
|
||||
"name" VARCHAR(500) NOT NULL,
|
||||
"city" VARCHAR(255),
|
||||
"neighborhood" VARCHAR(255),
|
||||
"cuisine" JSONB NOT NULL,
|
||||
"price_range" VARCHAR(10),
|
||||
"status" VARCHAR(20) NOT NULL,
|
||||
"rank_index" INT,
|
||||
"score" DECIMAL(5,2),
|
||||
"notes" TEXT,
|
||||
"tags" JSONB NOT NULL,
|
||||
"fetched_at" TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
"created_at" TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
"profile_id" UUID NOT NULL REFERENCES "beli_profiles" ("id") ON DELETE CASCADE,
|
||||
CONSTRAINT "uid_beli_restau_profile_921fe6" UNIQUE ("profile_id", "beli_business_id", "status")
|
||||
);
|
||||
COMMENT ON COLUMN "beli_restaurants"."status" IS 'BEEN: been\nWANT_TO_TRY: want_to_try';"""
|
||||
|
||||
|
||||
async def downgrade(db: BaseDBAsyncClient) -> str:
|
||||
return """
|
||||
DROP TABLE IF EXISTS "beli_profiles";
|
||||
DROP TABLE IF EXISTS "cron_jobs";
|
||||
DROP TABLE IF EXISTS "beli_restaurants";"""
|
||||
|
||||
|
||||
MODELS_STATE = (
|
||||
"eJztXe9zmzgT/lcYf+rN5DJx0vTuMu/cjJ3Q1m1sd2Lner26w8ggx1xAcPxo6un0f38lAT"
|
||||
"YIgY1tbJTqw11qoRXwaCXtPrsS31u2Y0DLP7120Ffo+SAwHdS6Ur63ELAh/gf3+onSAq67"
|
||||
"ukoKAjC1qICeqkmvgKkfeEAP8MUZsHyIiwzo657pxjdDoWWRQkfHFU30sCoKkflfCLXAeY"
|
||||
"DBHHr4wucvuNhEBvwG/eSn+6jNTGgZmec2DXJvWq4FC5eW3d/3bl7TmuR2U013rNBGq9ru"
|
||||
"Ipg7aFk9DE3jlMiQaw8QQQ8E0Ei9BnnK+LWTouiJcUHghXD5qMaqwIAzEFoEjNb/ZiHSCQ"
|
||||
"YKvRP538s/WxXgwVATaE0UECy+/4jeavXOtLRFbnX9tnP34uLVL/QtHT948OhFikjrBxUE"
|
||||
"AYhEKa4rIOnfHJTXc+DxoUzqM2DiB90GxqRgheNKhxIgE4C2Q61lg2+aBdFDMMc/zy8vS2"
|
||||
"D8q3NHkcS1KJQO1utI6wfxpfPoGoF0BaHuQfLKGgjyQN7gK4FpQz6YWUkGUiMWPU3+0VCA"
|
||||
"8TsYQ2Qt4kFQgu+411dH407/A3kT2/f/syhEnbFKrpzT0gVT+uIV0xXLRpSPvfFbhfxU/h"
|
||||
"kOVFb3l/XG/7TIM4EwcDTkPGnASI3XpDQBJtOxoWts2bFZSdmxR+3Y+OFTA3YOEIJWlWkv"
|
||||
"JbLVzBc/VaXuaz3B6Q7rBTPznW0y8Z0Vz3tn7LQX+tDTqq3DKZEdFuNtsNxtKGy/9BIDZv"
|
||||
"bIXXkJGHnwXjseNB/Qe7igEPbwcwCk85bc2HS7j5tpKmir0tVTeOBpadOllQK/HX4nGESD"
|
||||
"sTO67tyoLYrhFOiPT8AztAIwbej74AH6eUC7seTr93fQWhq4fCzTZnA/arGZc3MRthQr59"
|
||||
"xJYZRBL3/JPrfZEoDwexvxvcmdStBZ40ukQNzMpdDSHSldC6FdiwB+4xhOY1zKhzKpL4pr"
|
||||
"UWYQqX+PM7ZQspC+6Hf+/iVjD90OB2+S6qmF9/p22JV+xnM0R/N+hu9C8MgzBog9qqLQzp"
|
||||
"kCmd5NiR956LTuR+rdlUJW9Aka9frdzpXim/YUTNB4OLzFaDgOtaGrWq/tTazXdrH12s5Z"
|
||||
"r6aNVxntES6qeAEZoYP5AfskQC7b5xsgiWsVQkmvZbGM12wN+0iGVYlNyksKiWo9tFLaLK"
|
||||
"pmlHBE92mhCOpw6QzJvaPjxXLmjUVxrQfGUZeqnlidfgfxcPvQdrxFi+NupK6WehnUzbRJ"
|
||||
"RVN6F8/Au8A3CSCq5GCkRKSPIX2Mn8rHkLGMZ9GxuVjG0Vj4Zk2GotDwDbYCd+Dh67b+iu"
|
||||
"y+9RaftPTEt/RIP1ZNU0nLiGLrHYBTcIHvPzkeRzO7JgLegg9mWooBc7oIIm+qcQxNCXhd"
|
||||
"YvemV3cMWbc36Nx94pvOSf203nY/jdUOgy60gVkpq2ApsB8VXT/i96mg7bPNSNkyVjZHyz"
|
||||
"7NQeDjKV1DoT0tosT5YHJE90ImHhbVi01AvSjG9ILDdMcka3VIOaIS0mTcahCR1+PNpY5j"
|
||||
"QYBKxnxalp1R4xBJHetTVeunwpQ6HN5mptRuj2Ub7vtdFc8HFGZcyYysy95gzAV3bgMdm1"
|
||||
"2PkEPVrplQGVkBNbb9apOZlXU6UxPrK1ZjHdPQNT+c/gt1DgFQDCgrJyCYtRhS2CWfa67n"
|
||||
"fDWNalNqTvBw1mnLcnRg7WDeM6HETabVy+Jp9TI3rVoGcLUHzwldTh7Xu9FwwIeUEWMAvU"
|
||||
"f4LT8bph6cKJbpB19qg/fzl3pcJ/Lm5UQvy+kytippQBK9z5IPlETvM+3YZTZnjmhblwqb"
|
||||
"26G1n3zYJjv5+TzYbHrKKui7PRbZOHPzNH8jKCL7GOi6ExKxnQBRSVudqCmBIfH1OTRC7I"
|
||||
"tp+8khHyXtiZlAnoZmCi2T2Koz09oVlS5u6kPUkmCA1BnboEOoxQluqAkxWBzdoEN5s/BG"
|
||||
"izanxPqN/waAkO0KQIYS50ScthiUNpGZoAkaBQ5e4hQXeD40FPpQCq34ZAZz5eLsVwMsFA"
|
||||
"+S+vhhTpVbEz36SuAo13PPscFNd4Jmjqd8xf4l/uND4OlzRQcumJqWGeBZmz6ZDMNsoct7"
|
||||
"DsMkXCAP0PUZnlxgheC5a+EPtqBidmVhjh7RutwoYHBZEjC4zAcMZngWIUY01mPOElWMJi"
|
||||
"snJqS16Cb2QQoBLdm944gI56Hz64j3XNUvT2TE9MgF8cCT1y7lVqaOsdCqbmvLCAmyveHQ"
|
||||
"w4JiNA9sTvR8DbCJkASWn89LrWx8R/xoFS03jqggIMszXyTLui19Dr+5Jm5ti47NSorZsY"
|
||||
"J05EaLdcxyVkyOzkrJ/OgEkDyGlVOkxSWMT5hU6aySNClbOoNxEbGY6oM1/GImVLApzxjL"
|
||||
"EJpwZj6EHqWGFcLy9fqdD6QYQUp3+UWsY4UWUhzkUsqDBqEb8TNGLCREurdwsaIpSSbrSa"
|
||||
"oNxYdBgG/vn0wQYTeTZrA6B6F/qoxC13W8wFdsrGKma8Gkhq+40KOb3SVH2RCOMlba6nRQ"
|
||||
"TlAylQljYfquBRZa1Qx8Vk76DqmjJ1xtjhW7CpwZIVHYtUOBSebnPJg9VEAcZGQYLM36bJ"
|
||||
"M//rjYAUp62Nav5+2Xv738/eLVy99xFfooy5LfSsDNZ9lSDLbZWZMTlMrIKGPhHhs1sUOK"
|
||||
"Oa1cC6Kge2hiy8SONrZHvnJ0tzT9PiN3wNT75cLf4Mx7C/iBBj3P4WQ0F6tsVkqQZV6eqy"
|
||||
"D5QpluKztWnqsgz1Vo7rkKDLGxj4RdsbDM5kwtkK5FrNwegBjh1kbLxgSCpHb6OIVMEYOc"
|
||||
"BW8dicx03HoeeYwvP/oKkVNczyGTjk8zT0kjkNKuME0U53nkyi0QHrlPYMb/+QqxaSNhsv"
|
||||
"ZiGds9iQpxYzpuCxrL1Fg8E8bM8Qw3GHpQIW/3iB+EpLn6EXusmNj3gzbE7VtRw0SAWs0K"
|
||||
"PfwSV5csckNYZOrRUKXdJl0rL70H265ZHlKDTLmNQsG0S5JM5JA3Tgq5Qp7o4SjDsx1M8z"
|
||||
"0ThgkGBdHgQgBzcj8leqR5qIeEeNLiZYJjxBSCWCT+U2JJh2SMwvYzNNuAnKSPPElLJuV5"
|
||||
"MikyD+uZ5WHVndQgfhZWbg8wx4/m7RMudqT5u5TlwZZC+5ke1E3X5B5iXhx/zwiJEh0+RE"
|
||||
"Z+2Ucd139EZ9cPPO4P1BZZcUedN+qVkpy/N0Fqv9O7vVKWB0Ye//OP8vz9OuLER9gX/Cxz"
|
||||
"7FbrZXV3gpUV06EQxIHYyDUsCvps+H20pfQBz9qLrbooSHL6QR3c9AZv8hZIK75ypbgQGb"
|
||||
"g/J2ikDsZXio8nygl6jWd99eaKRhagMUHXncG1ekuLdPKuVnyG6PGXA2yWhJ4HY/y36aZs"
|
||||
"CwfsqrvljU8HsQIznUSKMbAOwivxDe6ST1eKgXtkMUEfVfU9+fkE4SP53R8Oxm9JgY3Xur"
|
||||
"m1aEbnhD7UsDLyVuvSZLmMnDynljmnlgTvEnesihmUExRk2ZY5c5IQlDlzsmMLmd5k2E0X"
|
||||
"Vb9XyQpKvjeFyR4oX/Gz6HIq0iS+99pz0Dtn2uLQvMmlkzJ2V8eVtH+dqSR1xSd1q+6nkt"
|
||||
"uo8ltOU09Wwa5mxKRVzbWqXc+x3Uq07UpCFB09NKYykCMDOcIGcrb8jNFRPmAkzC5KL0Rb"
|
||||
"uLeMqExJa0LecFnwoSwDfLeow7GjcjV83UfuLpZMqSTUJFP683Vs1Y/51MlVpb/EweGrmA"
|
||||
"91FHNWuY+D7Je3+rzcQExvFNnLXySbtc2aswOblQK/ggGUlZK+LfNFnV0OCMq1IAq6h7aJ"
|
||||
"KFCFpzis0d3ioxyaanIeSnk9OMOzx7zo07ybanCuGUFgPrQWR4xABNZWdiC3AUksNIFYkL"
|
||||
"5wHSNG0pn7pTMlt/AsXFDJLTzTjpUnl8mTy5p8cplHjk8KPbDz94YJM3W3bEwsZGs9tIsB"
|
||||
"poDUy0K3htdjOm3f1J67Ihjp7aahbyLo+7GyxaEzyfYdhe1jOqMSb8LIisJNHYA8+bnSAm"
|
||||
"v51KtuBpxc6JIP2MX1BXGUD6GE2IKZTx1v7jiVxjYrJxFd6mRokhkvD+a70XBQoJYrEQbH"
|
||||
"e4Rf8LNh6sGJYpl+8KWuoZ5ajqahaQUm8k/JDWtakQgW5TwPS+kwSxdpIJ/HaupQwyYKb/"
|
||||
"NdsSozYkJqcnuTybVdPLe2c1OreBus+RmXXVUdXClTCNEEfewMxtp4qI3vyP5cbMpiI1QL"
|
||||
"vIZsycUa+KjRF8yjXngMYFZoq8P/jqC7ez7+z9exJ86hrKBu2sDiA7eUYcmqSOg0Fm4kfi"
|
||||
"Xg3KjXvX7n9sXlyTnD4yY6+TKneMgJeEdPFkcdlgKCzJWHDjgE4IGDZ/H6n9SXi//2i/8M"
|
||||
"Bvp28dCspOSum8Vdy2DTs+jXfLApZvwqxiWyUjI0kSZOd4xOMHmejcVvbZAiqyNN2hXegd"
|
||||
"jXnLc4fHx85aSMhwerOo3ZEF78Rc0Nz8WPO3A3Mr0J7kgxh/4Vej5343IxJ5ESEZPurYVa"
|
||||
"I0OjAohxdTEBbG/El7dL+PI2hy8vOqGzhJssPKHzYO5Jbavu3hyRo27k+PF/70kbqw=="
|
||||
)
|
||||
@@ -0,0 +1,157 @@
|
||||
"""
|
||||
Beli API client.
|
||||
|
||||
Reverse-engineered from ProjectBarks/beli-api (mirrored at git.torrtle.co/ryan/beli-api).
|
||||
Hosts and auth flow discovered from observed traffic — may break without notice.
|
||||
|
||||
Auth: POST /api/token/ → {access, refresh} (access: 20 min, refresh: 7 days)
|
||||
The API rejects requests without a browser-looking User-Agent and Origin header.
|
||||
"""
|
||||
|
||||
import base64
|
||||
import json
|
||||
import logging
|
||||
import random
|
||||
import time
|
||||
|
||||
import httpx
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
HOST_ONBOARD = "https://backoffice-service-onboarding-t57o3dxfca-nn.a.run.app"
|
||||
HOST_API = "https://backoffice-service-t57o3dxfca-nn.a.run.app"
|
||||
|
||||
RANKING_CATEGORIES = ["restaurants", "coffee", "bars", "bakeries"]
|
||||
|
||||
# 350 ms between requests — the API throttles bursts
|
||||
REQUEST_SPACING = 0.35
|
||||
|
||||
_USER_AGENTS = [
|
||||
"Mozilla/5.0 (iPhone; CPU iPhone OS 18_7 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148",
|
||||
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36",
|
||||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36",
|
||||
"Mozilla/5.0 (Linux; Android 16; SM-S928U) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/149.0.7827.91 Mobile Safari/537.36",
|
||||
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/18.1 Safari/605.1.15",
|
||||
]
|
||||
|
||||
|
||||
def _pick_ua() -> str:
|
||||
return random.choice(_USER_AGENTS)
|
||||
|
||||
|
||||
def _base_headers(access_token: str | None = None, ua: str | None = None) -> dict:
|
||||
h = {
|
||||
"User-Agent": ua or _pick_ua(),
|
||||
"Origin": "https://localhost",
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
if access_token:
|
||||
h["Authorization"] = f"Bearer {access_token}"
|
||||
return h
|
||||
|
||||
|
||||
def _decode_jwt_claims(token: str) -> dict:
|
||||
payload = token.split(".")[1]
|
||||
payload += "=" * (-len(payload) % 4)
|
||||
return json.loads(base64.urlsafe_b64decode(payload))
|
||||
|
||||
|
||||
def is_token_expired(token: str | None, skew_seconds: int = 60) -> bool:
|
||||
if not token:
|
||||
return True
|
||||
try:
|
||||
return _decode_jwt_claims(token)["exp"] - skew_seconds <= time.time()
|
||||
except Exception:
|
||||
return True
|
||||
|
||||
|
||||
async def authenticate(email: str, password: str) -> dict:
|
||||
"""Login with email+password. Returns {access_token, refresh_token, user_id}."""
|
||||
ua = _pick_ua()
|
||||
async with httpx.AsyncClient() as client:
|
||||
resp = await client.post(
|
||||
f"{HOST_ONBOARD}/api/token/",
|
||||
json={"email": email, "password": password},
|
||||
headers=_base_headers(ua=ua),
|
||||
timeout=15,
|
||||
)
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
|
||||
claims = _decode_jwt_claims(data["access"])
|
||||
return {
|
||||
"access_token": data["access"],
|
||||
"refresh_token": data["refresh"],
|
||||
"user_id": claims["user_id"],
|
||||
}
|
||||
|
||||
|
||||
async def refresh_access_token(refresh_token: str) -> str:
|
||||
"""Exchange a refresh token for a new access token."""
|
||||
async with httpx.AsyncClient() as client:
|
||||
resp = await client.post(
|
||||
f"{HOST_ONBOARD}/api/token/refresh/",
|
||||
json={"refresh": refresh_token},
|
||||
headers=_base_headers(),
|
||||
timeout=15,
|
||||
)
|
||||
resp.raise_for_status()
|
||||
return resp.json()["access"]
|
||||
|
||||
|
||||
async def fetch_user_rankings(
|
||||
access_token: str, user_id: str, category: str
|
||||
) -> list[dict]:
|
||||
"""Fetch a user's ranked restaurants for one category."""
|
||||
async with httpx.AsyncClient() as client:
|
||||
resp = await client.get(
|
||||
f"{HOST_API}/api/get-ranking/",
|
||||
params={"user": user_id, "category": category},
|
||||
headers=_base_headers(access_token),
|
||||
timeout=30,
|
||||
)
|
||||
resp.raise_for_status()
|
||||
return resp.json().get("results") or []
|
||||
|
||||
|
||||
async def fetch_bookmarks(access_token: str) -> list[dict]:
|
||||
"""Fetch the authenticated user's bookmarked (want-to-try) restaurants."""
|
||||
async with httpx.AsyncClient() as client:
|
||||
resp = await client.get(
|
||||
f"{HOST_API}/api/get-bookmark/",
|
||||
headers=_base_headers(access_token),
|
||||
timeout=30,
|
||||
)
|
||||
resp.raise_for_status()
|
||||
return resp.json().get("results") or []
|
||||
|
||||
|
||||
def extract_business(item: dict) -> dict | None:
|
||||
"""Pull restaurant fields from a ranking or bookmark result item.
|
||||
|
||||
The API returns results with the business either nested under a 'business'
|
||||
key or flattened at the top level — handle both.
|
||||
"""
|
||||
biz = item.get("business") or item
|
||||
biz_id = biz.get("id") or item.get("business_id")
|
||||
if not biz_id:
|
||||
return None
|
||||
|
||||
price_key = biz.get("price_key")
|
||||
if not price_key:
|
||||
price_int = biz.get("price")
|
||||
if price_int:
|
||||
price_key = "$" * price_int
|
||||
|
||||
return {
|
||||
"beli_business_id": str(biz_id),
|
||||
"name": biz.get("name", ""),
|
||||
"city": biz.get("city"),
|
||||
"neighborhood": biz.get("neighborhood") or biz.get("borough"),
|
||||
"cuisine": biz.get("cuisines") or [],
|
||||
"price_range": price_key,
|
||||
"rank_index": item.get("rank") or item.get("rank_index"),
|
||||
"score": item.get("score"),
|
||||
"notes": item.get("notes"),
|
||||
"tags": item.get("tags") or [],
|
||||
}
|
||||
@@ -95,6 +95,90 @@ def _build_timing(scheduled_at: datetime, recurrence: str) -> dict:
|
||||
}
|
||||
|
||||
|
||||
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:
|
||||
|
||||
Reference in New Issue
Block a user