Link-Log/scripts/seed_demo_data.py

#!/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()