Added one-off seed_demo_data.py for test functionality
This commit is contained in:
@@ -0,0 +1,179 @@
|
||||
#!/usr/bin/env python3
|
||||
## Copyright © 2026 Olaf Kolkman
|
||||
## SPDX-License-Identifier: GPL-3.0-or-later
|
||||
"""One-off script: seed the running LinkLog API with random users and links.
|
||||
|
||||
Uses the HTTP API exclusively for user/link creation. The only direct
|
||||
database access is flipping `email_verified` for freshly created accounts,
|
||||
since there is no mailbox available in this environment to click the real
|
||||
verification link that the API would otherwise send.
|
||||
"""
|
||||
|
||||
import random
|
||||
import string
|
||||
import subprocess
|
||||
import sys
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
import requests
|
||||
|
||||
BASE_URL = 'http://localhost:5469'
|
||||
ADMIN_EMAIL = 'bootstrap_admin@example.invalid'
|
||||
ADMIN_PASSWORD = 'Bootstrap-Admin-Pass1!'
|
||||
CONTAINER_NAME = 'testlog-app'
|
||||
|
||||
NUM_USERS = 5
|
||||
LINKS_PER_USER = 500
|
||||
|
||||
FIRST_NAMES = ['ava', 'liam', 'noah', 'emma', 'olivia', 'mia', 'ethan', 'sofia', 'lucas', 'zoe']
|
||||
SITES = [
|
||||
('https://news.ycombinator.com', 'Hacker News'),
|
||||
('https://www.theguardian.com', 'The Guardian'),
|
||||
('https://arstechnica.com', 'Ars Technica'),
|
||||
('https://www.wired.com', 'Wired'),
|
||||
('https://github.com', 'GitHub'),
|
||||
('https://www.nature.com', 'Nature'),
|
||||
('https://www.bbc.com', 'BBC'),
|
||||
('https://techcrunch.com', 'TechCrunch'),
|
||||
('https://www.nytimes.com', 'NY Times'),
|
||||
('https://www.reddit.com', 'Reddit'),
|
||||
('https://stackoverflow.com', 'Stack Overflow'),
|
||||
('https://www.smithsonianmag.com', 'Smithsonian'),
|
||||
('https://www.economist.com', 'The Economist'),
|
||||
('https://www.nationalgeographic.com', 'Nat Geo'),
|
||||
('https://www.theverge.com', 'The Verge'),
|
||||
]
|
||||
PATH_WORDS = ['article', 'story', 'post', 'thread', 'blog', 'notes', 'guide', 'review', 'update', 'deep-dive']
|
||||
HASHTAG_POOL = ['#tech', '#science', '#news', '#opensource', '#ai', '#climate', '#music',
|
||||
'#photography', '#food', '#travel', '#fediverse', '#security', '#culture', '#space']
|
||||
COMMENT_TEMPLATES = [
|
||||
"Interesting take on {topic}.",
|
||||
"Worth a read if you're into {topic}.",
|
||||
"Not sure I agree with this on {topic}, but good points.",
|
||||
"Bookmarking this for later - {topic}.",
|
||||
"Great deep dive into {topic}.",
|
||||
"", # some links have no comment
|
||||
]
|
||||
|
||||
session = requests.Session()
|
||||
|
||||
|
||||
def rand_suffix(n=6):
|
||||
return ''.join(random.choices(string.ascii_lowercase + string.digits, k=n))
|
||||
|
||||
|
||||
def create_bootstrap_admin_if_needed():
|
||||
resp = session.post(f'{BASE_URL}/api/auth/login', json={'email': ADMIN_EMAIL, 'password': ADMIN_PASSWORD})
|
||||
if resp.status_code == 200:
|
||||
return resp.json()['access_token']
|
||||
raise RuntimeError(f'Bootstrap admin login failed: {resp.status_code} {resp.text}')
|
||||
|
||||
|
||||
def mark_email_verified(user_id: str):
|
||||
code = (
|
||||
"from backend.app.database import get_connection\n"
|
||||
f"with get_connection() as conn:\n"
|
||||
f" conn.execute('UPDATE users SET email_verified = 1 WHERE id = ?', ('{user_id}',))\n"
|
||||
" conn.commit()\n"
|
||||
)
|
||||
subprocess.run(
|
||||
['docker', 'exec', '-w', '/app', CONTAINER_NAME, 'python3', '-c', code],
|
||||
check=True, capture_output=True, text=True,
|
||||
)
|
||||
|
||||
|
||||
def create_user(admin_token: str, index: int) -> dict:
|
||||
name = f'{random.choice(FIRST_NAMES)}{index}_{rand_suffix()}'
|
||||
email = f'{name}@example-seed.invalid'
|
||||
password = f'Passw0rd-{rand_suffix(8)}!'
|
||||
resp = session.post(
|
||||
f'{BASE_URL}/api/admin/users',
|
||||
headers={'Authorization': f'Bearer {admin_token}'},
|
||||
json={'username': name, 'email': email, 'password': password, 'is_admin': False},
|
||||
)
|
||||
if resp.status_code == 201:
|
||||
user = resp.json()
|
||||
elif resp.status_code == 503:
|
||||
# User row was created but the verification email failed to send (no SMTP egress here).
|
||||
list_resp = session.get(f'{BASE_URL}/api/admin/users', headers={'Authorization': f'Bearer {admin_token}'})
|
||||
list_resp.raise_for_status()
|
||||
user = next(u for u in list_resp.json() if u['username'] == name)
|
||||
else:
|
||||
raise RuntimeError(f'Failed to create user {name}: {resp.status_code} {resp.text}')
|
||||
|
||||
mark_email_verified(user['id'])
|
||||
return {'id': user['id'], 'username': name, 'email': email, 'password': password}
|
||||
|
||||
|
||||
def login_user(email: str, password: str) -> str:
|
||||
resp = session.post(f'{BASE_URL}/api/auth/login', json={'email': email, 'password': password})
|
||||
resp.raise_for_status()
|
||||
return resp.json()['access_token']
|
||||
|
||||
|
||||
def random_link_payload():
|
||||
base_url, site_name = random.choice(SITES)
|
||||
url = f'{base_url}/{random.choice(PATH_WORDS)}/{rand_suffix(8)}'
|
||||
title = f'{site_name}: {random.choice(PATH_WORDS).replace("-", " ").title()} #{random.randint(1000, 9999)}'
|
||||
topic = random.choice(HASHTAG_POOL).lstrip('#')
|
||||
comment = random.choice(COMMENT_TEMPLATES).format(topic=topic)
|
||||
tags = random.sample(HASHTAG_POOL, k=random.randint(1, 4))
|
||||
timestamp = (datetime.now(timezone.utc) - timedelta(days=random.randint(0, 730),
|
||||
seconds=random.randint(0, 86400))).isoformat()
|
||||
return {
|
||||
'title': title,
|
||||
'url': url,
|
||||
'comment': comment,
|
||||
'tags': tags,
|
||||
'timestamp': timestamp,
|
||||
'post_to_mastodon': False,
|
||||
}
|
||||
|
||||
|
||||
def create_links_for_user(token: str, username: str, count: int):
|
||||
headers = {'Authorization': f'Bearer {token}'}
|
||||
created = 0
|
||||
for i in range(count):
|
||||
payload = random_link_payload()
|
||||
resp = session.post(f'{BASE_URL}/api/links', headers=headers, json=payload)
|
||||
if resp.status_code == 401:
|
||||
# access token expired mid-run; caller handles refresh via re-login
|
||||
raise PermissionError('token_expired')
|
||||
if resp.status_code not in (200, 201):
|
||||
print(f' ! link {i} failed for {username}: {resp.status_code} {resp.text[:200]}', file=sys.stderr)
|
||||
continue
|
||||
created += 1
|
||||
if created % 100 == 0:
|
||||
print(f' {username}: {created}/{count} links created')
|
||||
return created
|
||||
|
||||
|
||||
def main():
|
||||
print('Logging in as bootstrap admin...')
|
||||
admin_token = create_bootstrap_admin_if_needed()
|
||||
|
||||
users = []
|
||||
print(f'Creating {NUM_USERS} users...')
|
||||
for i in range(1, NUM_USERS + 1):
|
||||
user = create_user(admin_token, i)
|
||||
users.append(user)
|
||||
print(f' created user: {user["username"]} <{user["email"]}>')
|
||||
|
||||
for user in users:
|
||||
print(f'Seeding {LINKS_PER_USER} links for {user["username"]}...')
|
||||
token = login_user(user['email'], user['password'])
|
||||
try:
|
||||
total = create_links_for_user(token, user['username'], LINKS_PER_USER)
|
||||
except PermissionError:
|
||||
print(f' token expired, re-logging in for {user["username"]}')
|
||||
token = login_user(user['email'], user['password'])
|
||||
total = create_links_for_user(token, user['username'], LINKS_PER_USER)
|
||||
print(f' done: {total}/{LINKS_PER_USER} links created for {user["username"]}')
|
||||
|
||||
print('\nSummary:')
|
||||
for user in users:
|
||||
print(f' {user["username"]} / {user["email"]} / password: {user["password"]}')
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
Reference in New Issue
Block a user