""" Script to trigger a digest email for a specific user via the admin API. Usage: python scripts/send_digest.py The script logs in as an admin using the ADMIN_EMAIL and ADMIN_PASSWORD environment variables (defaults to localhost:5000). Environment variables: ADMIN_EMAIL - Admin account email (required) ADMIN_PASSWORD - Admin account password (required) API_BASE_URL - Base URL of the backend API (default: http://localhost:5000) """ import sys import os import json import urllib.request import urllib.error API_BASE_URL = os.environ.get('API_BASE_URL', 'http://localhost:5000').rstrip('/') def _post(path: str, body: dict, cookie: str | None = None) -> tuple[int, dict]: url = f"{API_BASE_URL}{path}" data = json.dumps(body).encode() headers = {'Content-Type': 'application/json'} if cookie: headers['Cookie'] = cookie req = urllib.request.Request(url, data=data, headers=headers, method='POST') try: with urllib.request.urlopen(req) as resp: return resp.status, json.loads(resp.read()) except urllib.error.HTTPError as e: try: return e.code, json.loads(e.read()) except Exception: return e.code, {} def main() -> None: if len(sys.argv) != 2: print('Usage: python scripts/send_digest.py ') sys.exit(1) target_email = sys.argv[1].strip() admin_email = os.environ.get('ADMIN_EMAIL', '').strip() admin_password = os.environ.get('ADMIN_PASSWORD', '').strip() if not admin_email or not admin_password: print('Error: ADMIN_EMAIL and ADMIN_PASSWORD environment variables are required') sys.exit(1) # Log in to get a session cookie status, body = _post('/auth/login', {'email': admin_email, 'password': admin_password}) if status != 200: print(f'Login failed ({status}): {body.get("error", body)}') sys.exit(1) # urllib doesn't automatically store cookies; we need to capture Set-Cookie manually. # Re-do the login request to grab the cookie header. login_url = f"{API_BASE_URL}/auth/login" login_data = json.dumps({'email': admin_email, 'password': admin_password}).encode() login_req = urllib.request.Request( login_url, data=login_data, headers={'Content-Type': 'application/json'}, method='POST', ) try: with urllib.request.urlopen(login_req) as resp: raw_cookies = resp.headers.get_all('Set-Cookie') or [] except urllib.error.HTTPError: print('Login failed (could not retrieve cookie)') sys.exit(1) # Extract access_token cookie value cookie_header = '; '.join( part.split(';')[0] for part in raw_cookies ) if not cookie_header: print('Login succeeded but no cookies were returned') sys.exit(1) # Call the send-digest endpoint status, body = _post( '/admin/test/send-digest', {'email': target_email}, cookie=cookie_header, ) if status == 200: items_sent = body.get('items_sent', 0) if items_sent == 0: print(f'No pending items found for {target_email} — no email sent.') else: print(f'Digest sent to {target_email} with {items_sent} item(s).') elif status == 404 and body.get('code') == 'NOT_FOUND': print('Error: This endpoint is disabled in the current environment (DB_ENV=production).') sys.exit(1) elif status == 404 and body.get('code') == 'USER_NOT_FOUND': print(f'Error: No user found with email {target_email}') sys.exit(1) elif status == 403: print(f'Error: {admin_email} is not an admin account.') sys.exit(1) else: print(f'Error ({status}): {body.get("error", body)}') sys.exit(1) if __name__ == '__main__': main()