105 lines
3 KiB
Python
105 lines
3 KiB
Python
import sqlite3
|
|
import secrets
|
|
from datetime import datetime, timezone
|
|
from flask import current_app, g
|
|
|
|
SCHEMA = """
|
|
CREATE TABLE IF NOT EXISTS requests (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
name TEXT NOT NULL,
|
|
email TEXT NOT NULL,
|
|
hobbies TEXT,
|
|
notable_facts TEXT,
|
|
style_genre TEXT,
|
|
extra_requests TEXT,
|
|
status TEXT DEFAULT 'pending',
|
|
suno_title TEXT,
|
|
suno_style TEXT,
|
|
suno_lyrics TEXT,
|
|
song_a_path TEXT,
|
|
song_b_path TEXT,
|
|
customer_approved TEXT DEFAULT 'none',
|
|
approval_notified_at TIMESTAMP,
|
|
preview_sent_at TIMESTAMP,
|
|
delivery_sent_at TIMESTAMP,
|
|
square_payment_ref TEXT,
|
|
admin_alert_email TEXT,
|
|
player_token TEXT NOT NULL UNIQUE,
|
|
revision_note TEXT
|
|
);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_requests_status ON requests(status);
|
|
CREATE INDEX IF NOT EXISTS idx_requests_token ON requests(player_token);
|
|
"""
|
|
|
|
def get_db():
|
|
if 'db' not in g:
|
|
g.db = sqlite3.connect(current_app.config['DATABASE'])
|
|
g.db.row_factory = sqlite3.Row
|
|
return g.db
|
|
|
|
def close_db(e=None):
|
|
db = g.pop('db', None)
|
|
if db is not None:
|
|
db.close()
|
|
|
|
def init_db():
|
|
db = sqlite3.connect(current_app.config['DATABASE'])
|
|
db.executescript(SCHEMA)
|
|
db.commit()
|
|
db.close()
|
|
|
|
def new_token():
|
|
return secrets.token_urlsafe(32)
|
|
|
|
def now_utc():
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
def create_request(name, email, hobbies, notable_facts, style_genre, extra_requests):
|
|
db = get_db()
|
|
cur = db.execute(
|
|
"""INSERT INTO requests
|
|
(name, email, hobbies, notable_facts, style_genre, extra_requests, player_token)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?)""",
|
|
(name, email, hobbies, notable_facts, style_genre, extra_requests, new_token())
|
|
)
|
|
db.commit()
|
|
return cur.lastrowid
|
|
|
|
def get_request_by_id(request_id):
|
|
db = get_db()
|
|
row = db.execute('SELECT * FROM requests WHERE id = ?', (request_id,)).fetchone()
|
|
return dict(row) if row else None
|
|
|
|
def get_request_by_token(token):
|
|
db = get_db()
|
|
row = db.execute('SELECT * FROM requests WHERE player_token = ?', (token,)).fetchone()
|
|
return dict(row) if row else None
|
|
|
|
def list_requests(status=None):
|
|
db = get_db()
|
|
if status:
|
|
rows = db.execute('SELECT * FROM requests WHERE status = ? ORDER BY created_at DESC', (status,)).fetchall()
|
|
else:
|
|
rows = db.execute('SELECT * FROM requests ORDER BY created_at DESC').fetchall()
|
|
return [dict(r) for r in rows]
|
|
|
|
def update_request(request_id, **fields):
|
|
if not fields:
|
|
return
|
|
db = get_db()
|
|
cols = ', '.join(f'{k} = ?' for k in fields)
|
|
vals = list(fields.values()) + [request_id]
|
|
db.execute(f'UPDATE requests SET {cols} WHERE id = ?', vals)
|
|
db.commit()
|
|
|
|
def delete_request(request_id):
|
|
db = get_db()
|
|
db.execute('DELETE FROM requests WHERE id = ?', (request_id,))
|
|
db.commit()
|
|
|
|
def reset_all_requests():
|
|
db = get_db()
|
|
db.execute('DELETE FROM requests')
|
|
db.commit()
|