feat: add admin endpoint to send digest emails for users
Some checks failed
Chore App Build, Test, and Push Docker Images / build-and-push (push) Failing after 1m37s
Some checks failed
Chore App Build, Test, and Push Docker Images / build-and-push (push) Failing after 1m37s
- Implemented a new endpoint `/admin/test/send-digest` in `admin_api.py` to trigger digest emails for specific users. - Added a script `send_digest.py` to facilitate sending digest emails via the admin API. - Enhanced the digest action handling in `digest_action_api.py` to support token peeking without consuming it. - Updated the `send_digests` function in `digest_scheduler.py` to utilize the new `send_digest_for_user` function for sending emails. - Introduced a new utility function `peek_token` in `digest_token.py` to validate tokens without consuming them. - Modified the `ParentView.vue` component to handle digest actions upon receiving a digest token in the URL. - Updated `.gitignore` to exclude sensitive certificate files.
This commit is contained in:
114
backend/scripts/send_digest.py
Normal file
114
backend/scripts/send_digest.py
Normal file
@@ -0,0 +1,114 @@
|
||||
"""
|
||||
Script to trigger a digest email for a specific user via the admin API.
|
||||
|
||||
Usage:
|
||||
python scripts/send_digest.py <email>
|
||||
|
||||
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 <email>')
|
||||
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()
|
||||
Reference in New Issue
Block a user