""" SQLite data access layer for the Video Production Pipeline. Drop-in replacement for database_pg.py — same function signatures, same return types. Uses aiosqlite for async operations. Zero external dependencies (no Postgres needed). Tables: projects – one row per pipeline run images – one row per generated image (with optional BLOB data) videos – one row per generated video (with optional BLOB data) audios – one row per generated audio (with optional BLOB data) characters – character templates settings – key-value settings store grok_accounts – multi-account rotation for Grok world_styles – custom world styles clothing_styles – custom clothing styles production_rules – custom production rules """ from __future__ import annotations import base64 import json from datetime import datetime, timezone from pathlib import Path from typing import Any, Optional import aiosqlite _DB_PATH: Path | None = None _db: aiosqlite.Connection | None = None SCHEMA = """ CREATE TABLE IF NOT EXISTS projects ( id TEXT PRIMARY KEY, status TEXT NOT NULL DEFAULT 'pending', mode TEXT NOT NULL DEFAULT 'plan', platform TEXT NOT NULL DEFAULT 'tiktok', character TEXT NOT NULL DEFAULT '', story_preview TEXT NOT NULL DEFAULT '', title TEXT NOT NULL DEFAULT '', created_at TEXT NOT NULL, result_json TEXT ); CREATE TABLE IF NOT EXISTS images ( id INTEGER PRIMARY KEY AUTOINCREMENT, project_id TEXT NOT NULL REFERENCES projects(id) ON DELETE CASCADE, scene_number INTEGER NOT NULL, image_prompt TEXT NOT NULL DEFAULT '', file_path TEXT NOT NULL DEFAULT '', image_data BLOB, mime_type TEXT NOT NULL DEFAULT 'image/png', file_size_bytes INTEGER, status TEXT NOT NULL DEFAULT 'pending', created_at TEXT NOT NULL, updated_at TEXT NOT NULL, UNIQUE(project_id, scene_number) ); CREATE TABLE IF NOT EXISTS videos ( id INTEGER PRIMARY KEY AUTOINCREMENT, project_id TEXT NOT NULL REFERENCES projects(id) ON DELETE CASCADE, scene_number INTEGER NOT NULL, video_prompt TEXT NOT NULL DEFAULT '', file_path TEXT NOT NULL DEFAULT '', video_data BLOB, mime_type TEXT NOT NULL DEFAULT 'video/mp4', file_size_bytes INTEGER, duration_seconds REAL, status TEXT NOT NULL DEFAULT 'pending', created_at TEXT NOT NULL, updated_at TEXT NOT NULL, UNIQUE(project_id, scene_number) ); CREATE TABLE IF NOT EXISTS audios ( id INTEGER PRIMARY KEY AUTOINCREMENT, project_id TEXT NOT NULL REFERENCES projects(id) ON DELETE CASCADE, scene_number INTEGER NOT NULL, tts_text TEXT NOT NULL DEFAULT '', tts_provider TEXT NOT NULL DEFAULT 'elevenlabs', file_path TEXT NOT NULL DEFAULT '', audio_data BLOB, mime_type TEXT NOT NULL DEFAULT 'audio/wav', file_size_bytes INTEGER, duration_seconds REAL, status TEXT NOT NULL DEFAULT 'pending', created_at TEXT NOT NULL, updated_at TEXT NOT NULL, UNIQUE(project_id, scene_number) ); CREATE TABLE IF NOT EXISTS characters ( id TEXT PRIMARY KEY, name TEXT NOT NULL UNIQUE, physical_description TEXT NOT NULL DEFAULT '', style_keywords TEXT NOT NULL DEFAULT '[]', negative_keywords TEXT NOT NULL DEFAULT '[]', hard_prohibitions TEXT NOT NULL DEFAULT '[]', camera_settings TEXT NOT NULL DEFAULT '', lighting_settings TEXT NOT NULL DEFAULT '', pose_rule TEXT NOT NULL DEFAULT '', environment_rule TEXT NOT NULL DEFAULT '', clothing_rule TEXT NOT NULL DEFAULT '', anatomical_highlight_rules TEXT NOT NULL DEFAULT '', consequence_philosophy TEXT NOT NULL DEFAULT '', shot_variety_rules TEXT NOT NULL DEFAULT '', thumbnail_path TEXT NOT NULL DEFAULT '', is_default INTEGER NOT NULL DEFAULT 0, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS settings ( key TEXT PRIMARY KEY, value TEXT NOT NULL DEFAULT '', description TEXT ); CREATE TABLE IF NOT EXISTS grok_accounts ( id INTEGER PRIMARY KEY AUTOINCREMENT, label TEXT NOT NULL DEFAULT '', sso_token TEXT NOT NULL, sso_rw_token TEXT NOT NULL, user_id TEXT NOT NULL, is_active INTEGER NOT NULL DEFAULT 1, last_used_at TEXT, usage_count INTEGER NOT NULL DEFAULT 0, daily_usage_count INTEGER NOT NULL DEFAULT 0, daily_usage_reset_at TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS world_styles ( id TEXT PRIMARY KEY, name TEXT NOT NULL, preset_type TEXT NOT NULL DEFAULT 'custom', background_rules TEXT NOT NULL DEFAULT '', floor_rules TEXT NOT NULL DEFAULT '', secondary_characters_rules TEXT NOT NULL DEFAULT '', lighting_rules TEXT NOT NULL DEFAULT '', props_rules TEXT NOT NULL DEFAULT '', architecture_allowed INTEGER NOT NULL DEFAULT 1, is_preset INTEGER NOT NULL DEFAULT 0, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS clothing_styles ( id TEXT PRIMARY KEY, name TEXT NOT NULL, preset_type TEXT NOT NULL DEFAULT 'custom', clothing_rules TEXT NOT NULL DEFAULT '', opacity_rules TEXT NOT NULL DEFAULT '', max_pieces INTEGER NOT NULL DEFAULT 2, is_preset INTEGER NOT NULL DEFAULT 0, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS production_rules ( id TEXT PRIMARY KEY, name TEXT NOT NULL, camera_settings TEXT NOT NULL DEFAULT '', pose_rule TEXT NOT NULL DEFAULT '', shot_intercalation_rules TEXT NOT NULL DEFAULT '', death_scene_rules TEXT NOT NULL DEFAULT '', child_rules TEXT NOT NULL DEFAULT '', zombie_rules TEXT NOT NULL DEFAULT '', consequence_philosophy TEXT NOT NULL DEFAULT '', scale_rule TEXT NOT NULL DEFAULT '', is_preset INTEGER NOT NULL DEFAULT 0, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ); """ async def _auto_migrate(db: aiosqlite.Connection) -> None: """Add columns that may be missing from older database schemas. Uses 'ALTER TABLE ADD COLUMN' which is a no-op if the column already exists (we catch the 'duplicate column name' error and skip). """ migrations = [ # videos: add blob + metadata columns ("videos", "video_data", "BLOB"), ("videos", "mime_type", "TEXT NOT NULL DEFAULT 'video/mp4'"), ("videos", "file_size_bytes", "INTEGER"), ("videos", "duration_seconds", "REAL"), # images: add blob + metadata columns (may already exist) ("images", "image_data", "BLOB"), ("images", "mime_type", "TEXT NOT NULL DEFAULT 'image/png'"), ("images", "file_size_bytes", "INTEGER"), # characters: add newer fields ("characters", "consequence_philosophy", "TEXT NOT NULL DEFAULT ''"), ("characters", "shot_variety_rules", "TEXT NOT NULL DEFAULT ''"), ] migrated = 0 for table, column, coltype in migrations: try: await db.execute(f"ALTER TABLE {table} ADD COLUMN {column} {coltype}") migrated += 1 except Exception: pass # Column already exists if migrated: await db.commit() async def init_db(db_path: str | Path | None = None) -> None: """Initialise the database (create tables, enable WAL + FK).""" global _DB_PATH, _db if _db is not None: return if db_path is None: project_root = Path(__file__).resolve().parent.parent data_dir = project_root / "data" data_dir.mkdir(parents=True, exist_ok=True) db_path = data_dir / "pipeline.db" _DB_PATH = Path(db_path) _DB_PATH.parent.mkdir(parents=True, exist_ok=True) _db = await aiosqlite.connect(str(_DB_PATH)) _db.row_factory = aiosqlite.Row await _db.execute("PRAGMA journal_mode=WAL") await _db.execute("PRAGMA foreign_keys=ON") await _db.executescript(SCHEMA) await _db.commit() # Auto-migrate: add columns that may be missing from older schemas await _auto_migrate(_db) async def get_db() -> aiosqlite.Connection: """Return the shared database connection.""" if _db is None: await init_db() return _db async def close_db() -> None: """Close the database connection.""" global _db if _db is not None: await _db.close() _db = None def _now() -> str: return datetime.now(timezone.utc).isoformat() # --------------------------------------------------------------------------- # Projects # --------------------------------------------------------------------------- async def insert_project( *, id: str, status: str = "running", mode: str = "plan", platform: str = "tiktok", character: str = "", story_preview: str = "", title: str = "", ) -> None: db = await get_db() await db.execute( """INSERT INTO projects (id, status, mode, platform, character, story_preview, title, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)""", (id, status, mode, platform, character, story_preview, title, _now()), ) await db.commit() async def update_project_status(project_id: str, status: str) -> None: db = await get_db() await db.execute( "UPDATE projects SET status = ? WHERE id = ?", (status, project_id) ) await db.commit() async def update_project_result( project_id: str, result: dict[str, Any], title: str = "" ) -> None: db = await get_db() params: list[Any] = [json.dumps(result)] sql = "UPDATE projects SET result_json = ?" if title: sql += ", title = ?" params.append(title) sql += " WHERE id = ?" params.append(project_id) await db.execute(sql, params) await db.commit() async def get_project(project_id: str) -> dict[str, Any] | None: db = await get_db() cur = await db.execute("SELECT * FROM projects WHERE id = ?", (project_id,)) row = await cur.fetchone() if row is None: return None proj = dict(row) if proj.get("result_json"): proj["result"] = json.loads(proj["result_json"]) else: proj["result"] = None del proj["result_json"] return proj async def delete_project(project_id: str) -> bool: db = await get_db() cur = await db.execute("DELETE FROM projects WHERE id = ?", (project_id,)) await db.commit() return cur.rowcount > 0 async def list_projects() -> list[dict[str, Any]]: db = await get_db() cur = await db.execute( """SELECT id, status, mode, platform, character, story_preview, title, created_at FROM projects ORDER BY created_at DESC""" ) rows = await cur.fetchall() return [dict(r) for r in rows] # --------------------------------------------------------------------------- # Images # --------------------------------------------------------------------------- async def upsert_image( *, project_id: str, scene_number: int, image_prompt: str = "", file_path: str = "", image_data: bytes | None = None, mime_type: str = "image/png", status: str = "generated", ) -> None: db = await get_db() now = _now() file_size = len(image_data) if image_data else None await db.execute( """INSERT INTO images (project_id, scene_number, image_prompt, file_path, image_data, mime_type, file_size_bytes, status, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(project_id, scene_number) DO UPDATE SET image_prompt = excluded.image_prompt, file_path = excluded.file_path, image_data = COALESCE(excluded.image_data, images.image_data), mime_type = excluded.mime_type, file_size_bytes = COALESCE(excluded.file_size_bytes, images.file_size_bytes), status = excluded.status, updated_at = excluded.updated_at""", (project_id, scene_number, image_prompt, file_path, image_data, mime_type, file_size, status, now, now), ) await db.commit() async def update_image_status(project_id: str, scene_number: int, status: str) -> None: db = await get_db() await db.execute( """UPDATE images SET status = ?, updated_at = ? WHERE project_id = ? AND scene_number = ?""", (status, _now(), project_id, scene_number), ) await db.commit() async def update_image_path(project_id: str, scene_number: int, file_path: str) -> None: db = await get_db() await db.execute( """UPDATE images SET file_path = ?, status = 'generated', updated_at = ? WHERE project_id = ? AND scene_number = ?""", (file_path, _now(), project_id, scene_number), ) await db.commit() async def update_image_data( project_id: str, scene_number: int, image_data: bytes, mime_type: str = "image/png" ) -> None: """Store image binary data in the database.""" db = await get_db() await db.execute( """UPDATE images SET image_data = ?, mime_type = ?, file_size_bytes = ?, updated_at = ? WHERE project_id = ? AND scene_number = ?""", (image_data, mime_type, len(image_data), _now(), project_id, scene_number), ) await db.commit() async def get_project_images(project_id: str) -> list[dict[str, Any]]: db = await get_db() cur = await db.execute( """SELECT id, project_id, scene_number, image_prompt, file_path, mime_type, file_size_bytes, status, created_at, updated_at FROM images WHERE project_id = ? ORDER BY scene_number""", (project_id,), ) rows = await cur.fetchall() return [dict(r) for r in rows] async def get_image_data( project_id: str, scene_number: int ) -> tuple[bytes | None, str]: """Get image binary data and mime type.""" db = await get_db() cur = await db.execute( """SELECT image_data, mime_type FROM images WHERE project_id = ? AND scene_number = ?""", (project_id, scene_number), ) row = await cur.fetchone() if row: return row["image_data"], row["mime_type"] or "image/png" return None, "image/png" async def get_image_data_base64( project_id: str, scene_number: int ) -> tuple[str | None, str]: """Get image as Base64 string.""" data, mime_type = await get_image_data(project_id, scene_number) if data: return base64.b64encode(data).decode(), mime_type return None, mime_type async def delete_image(project_id: str, scene_number: int) -> bool: """Delete an image record for a given project and scene.""" db = await get_db() cur = await db.execute( "DELETE FROM images WHERE project_id = ? AND scene_number = ?", (project_id, scene_number), ) await db.commit() return cur.rowcount > 0 # --------------------------------------------------------------------------- # Videos # --------------------------------------------------------------------------- async def upsert_video( *, project_id: str, scene_number: int, video_prompt: str = "", file_path: str = "", video_data: bytes | None = None, mime_type: str = "video/mp4", duration_seconds: float | None = None, status: str = "generated", ) -> None: db = await get_db() now = _now() file_size = len(video_data) if video_data else None await db.execute( """INSERT INTO videos (project_id, scene_number, video_prompt, file_path, video_data, mime_type, file_size_bytes, duration_seconds, status, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(project_id, scene_number) DO UPDATE SET video_prompt = excluded.video_prompt, file_path = excluded.file_path, video_data = COALESCE(excluded.video_data, videos.video_data), mime_type = excluded.mime_type, file_size_bytes = COALESCE(excluded.file_size_bytes, videos.file_size_bytes), duration_seconds = COALESCE(excluded.duration_seconds, videos.duration_seconds), status = excluded.status, updated_at = excluded.updated_at""", (project_id, scene_number, video_prompt, file_path, video_data, mime_type, file_size, duration_seconds, status, now, now), ) await db.commit() async def get_project_videos(project_id: str) -> list[dict[str, Any]]: db = await get_db() cur = await db.execute( """SELECT id, project_id, scene_number, video_prompt, file_path, mime_type, file_size_bytes, duration_seconds, status, created_at, updated_at FROM videos WHERE project_id = ? ORDER BY scene_number""", (project_id,), ) rows = await cur.fetchall() return [dict(r) for r in rows] async def update_video_status(project_id: str, scene_number: int, status: str) -> None: db = await get_db() await db.execute( """UPDATE videos SET status = ?, updated_at = ? WHERE project_id = ? AND scene_number = ?""", (status, _now(), project_id, scene_number), ) await db.commit() async def get_video_data( project_id: str, scene_number: int ) -> tuple[bytes | None, str]: """Get video binary data and mime type.""" db = await get_db() cur = await db.execute( """SELECT video_data, mime_type FROM videos WHERE project_id = ? AND scene_number = ?""", (project_id, scene_number), ) row = await cur.fetchone() if row: return row["video_data"], row["mime_type"] or "video/mp4" return None, "video/mp4" # --------------------------------------------------------------------------- # Audios # --------------------------------------------------------------------------- async def upsert_audio( *, project_id: str, scene_number: int, tts_text: str = "", tts_provider: str = "elevenlabs", file_path: str = "", audio_data: bytes | None = None, mime_type: str = "audio/wav", duration_seconds: float | None = None, status: str = "generated", ) -> None: db = await get_db() now = _now() file_size = len(audio_data) if audio_data else None await db.execute( """INSERT INTO audios (project_id, scene_number, tts_text, tts_provider, file_path, audio_data, mime_type, file_size_bytes, duration_seconds, status, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(project_id, scene_number) DO UPDATE SET tts_text = excluded.tts_text, tts_provider = excluded.tts_provider, file_path = excluded.file_path, audio_data = COALESCE(excluded.audio_data, audios.audio_data), mime_type = excluded.mime_type, file_size_bytes = COALESCE(excluded.file_size_bytes, audios.file_size_bytes), duration_seconds = COALESCE(excluded.duration_seconds, audios.duration_seconds), status = excluded.status, updated_at = excluded.updated_at""", (project_id, scene_number, tts_text, tts_provider, file_path, audio_data, mime_type, file_size, duration_seconds, status, now, now), ) await db.commit() async def get_project_audios(project_id: str) -> list[dict[str, Any]]: db = await get_db() cur = await db.execute( """SELECT id, project_id, scene_number, tts_text, tts_provider, file_path, mime_type, file_size_bytes, duration_seconds, status, created_at, updated_at FROM audios WHERE project_id = ? ORDER BY scene_number""", (project_id,), ) rows = await cur.fetchall() return [dict(r) for r in rows] # --------------------------------------------------------------------------- # Characters # --------------------------------------------------------------------------- def _char_row_to_dict(row: dict[str, Any]) -> dict[str, Any]: """Convert a character DB row to a dict with parsed JSON fields.""" d = dict(row) for key in ("style_keywords", "negative_keywords", "hard_prohibitions"): val = d.get(key, "[]") try: d[key] = json.loads(val) if isinstance(val, str) else val except (json.JSONDecodeError, TypeError): d[key] = [] # Convert is_default from int to bool for API compat if "is_default" in d: d["is_default"] = bool(d["is_default"]) return d async def insert_character( *, id: str, name: str, physical_description: str = "", style_keywords: list[str] | None = None, negative_keywords: list[str] | None = None, hard_prohibitions: list[str] | None = None, camera_settings: str = "", lighting_settings: str = "", pose_rule: str = "", environment_rule: str = "", clothing_rule: str = "", anatomical_highlight_rules: str = "", consequence_philosophy: str = "", shot_variety_rules: str = "", thumbnail_path: str = "", is_default: bool = False, ) -> None: db = await get_db() now = _now() await db.execute( """INSERT OR IGNORE INTO characters (id, name, physical_description, style_keywords, negative_keywords, hard_prohibitions, camera_settings, lighting_settings, pose_rule, environment_rule, clothing_rule, anatomical_highlight_rules, consequence_philosophy, shot_variety_rules, thumbnail_path, is_default, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", ( id, name, physical_description, json.dumps(style_keywords or []), json.dumps(negative_keywords or []), json.dumps(hard_prohibitions or []), camera_settings, lighting_settings, pose_rule, environment_rule, clothing_rule, anatomical_highlight_rules, consequence_philosophy, shot_variety_rules, thumbnail_path, 1 if is_default else 0, now, now, ), ) await db.commit() async def update_character(char_id: str, updates: dict[str, Any]) -> bool: """Update one or more fields on a character.""" db = await get_db() for key in ("style_keywords", "negative_keywords", "hard_prohibitions"): if key in updates and isinstance(updates[key], list): updates[key] = json.dumps(updates[key]) if "is_default" in updates: updates["is_default"] = 1 if updates["is_default"] else 0 updates["updated_at"] = _now() set_clause = ", ".join(f"{k} = ?" for k in updates) vals = list(updates.values()) + [char_id] cur = await db.execute(f"UPDATE characters SET {set_clause} WHERE id = ?", vals) await db.commit() return cur.rowcount > 0 async def delete_character(char_id: str) -> bool: db = await get_db() cur = await db.execute("DELETE FROM characters WHERE id = ?", (char_id,)) await db.commit() return cur.rowcount > 0 async def get_character(char_id: str) -> dict[str, Any] | None: db = await get_db() cur = await db.execute("SELECT * FROM characters WHERE id = ?", (char_id,)) row = await cur.fetchone() return _char_row_to_dict(row) if row else None async def get_character_by_name(name: str) -> dict[str, Any] | None: db = await get_db() cur = await db.execute("SELECT * FROM characters WHERE name = ?", (name,)) row = await cur.fetchone() return _char_row_to_dict(row) if row else None async def list_characters() -> list[dict[str, Any]]: db = await get_db() cur = await db.execute( "SELECT * FROM characters ORDER BY is_default DESC, name ASC" ) rows = await cur.fetchall() return [_char_row_to_dict(r) for r in rows] async def character_count() -> int: db = await get_db() cur = await db.execute("SELECT COUNT(*) FROM characters") row = await cur.fetchone() return row[0] if row else 0 # --------------------------------------------------------------------------- # Settings CRUD # --------------------------------------------------------------------------- async def get_setting(key: str) -> str | None: """Get a setting value by key.""" db = await get_db() cur = await db.execute("SELECT value FROM settings WHERE key = ?", (key,)) row = await cur.fetchone() return row["value"] if row else None async def set_setting(key: str, value: str, description: str | None = None) -> None: """Set a setting value (upsert).""" db = await get_db() if description: await db.execute( """INSERT INTO settings (key, value, description) VALUES (?, ?, ?) ON CONFLICT (key) DO UPDATE SET value = excluded.value, description = excluded.description""", (key, value, description), ) else: await db.execute( """INSERT INTO settings (key, value) VALUES (?, ?) ON CONFLICT (key) DO UPDATE SET value = excluded.value""", (key, value), ) await db.commit() async def get_all_settings() -> dict[str, str]: """Get all settings as a dict.""" db = await get_db() cur = await db.execute("SELECT key, value FROM settings") rows = await cur.fetchall() return {row["key"]: row["value"] for row in rows} async def delete_setting(key: str) -> None: """Delete a setting by key.""" db = await get_db() await db.execute("DELETE FROM settings WHERE key = ?", (key,)) await db.commit() # --------------------------------------------------------------------------- # Grok Accounts (multi-cookie rotation) # --------------------------------------------------------------------------- async def insert_grok_account(*, label: str, sso_token: str, sso_rw_token: str, user_id: str) -> dict[str, Any]: """Insert a new Grok account and return it.""" db = await get_db() now = _now() cur = await db.execute( """INSERT INTO grok_accounts (label, sso_token, sso_rw_token, user_id, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?)""", (label, sso_token, sso_rw_token, user_id, now, now), ) await db.commit() row_id = cur.lastrowid cur2 = await db.execute("SELECT * FROM grok_accounts WHERE id = ?", (row_id,)) row = await cur2.fetchone() return dict(row) async def list_grok_accounts() -> list[dict[str, Any]]: """List all Grok accounts.""" db = await get_db() cur = await db.execute("SELECT * FROM grok_accounts ORDER BY created_at") rows = await cur.fetchall() return [dict(r) for r in rows] async def get_next_grok_account() -> dict[str, Any] | None: """Pick the least-recently-used active Grok account (rotation). Resets daily_usage_count when date has changed. """ db = await get_db() today = datetime.now(timezone.utc).strftime("%Y-%m-%d") cur = await db.execute( """SELECT * FROM grok_accounts WHERE is_active = 1 ORDER BY last_used_at IS NULL DESC, last_used_at ASC, usage_count ASC LIMIT 1""" ) row = await cur.fetchone() if row is None: return None account = dict(row) # Update usage new_daily = 1 if (account.get("daily_usage_reset_at") or "") < today else account["daily_usage_count"] + 1 await db.execute( """UPDATE grok_accounts SET last_used_at = ?, usage_count = usage_count + 1, daily_usage_count = ?, daily_usage_reset_at = ?, updated_at = ? WHERE id = ?""", (_now(), new_daily, today, _now(), account["id"]), ) await db.commit() # Re-fetch cur2 = await db.execute("SELECT * FROM grok_accounts WHERE id = ?", (account["id"],)) row2 = await cur2.fetchone() return dict(row2) if row2 else account async def update_grok_account(account_id: int, **fields) -> dict[str, Any] | None: """Update a Grok account. Accepts: label, sso_token, sso_rw_token, user_id, is_active.""" allowed = {"label", "sso_token", "sso_rw_token", "user_id", "is_active"} updates = {k: v for k, v in fields.items() if k in allowed} if not updates: return None db = await get_db() if "is_active" in updates: updates["is_active"] = 1 if updates["is_active"] else 0 updates["updated_at"] = _now() set_clause = ", ".join(f"{k} = ?" for k in updates) vals = list(updates.values()) + [account_id] await db.execute(f"UPDATE grok_accounts SET {set_clause} WHERE id = ?", vals) await db.commit() cur = await db.execute("SELECT * FROM grok_accounts WHERE id = ?", (account_id,)) row = await cur.fetchone() return dict(row) if row else None async def delete_grok_account(account_id: int) -> bool: """Delete a Grok account. Returns True if deleted.""" db = await get_db() cur = await db.execute("DELETE FROM grok_accounts WHERE id = ?", (account_id,)) await db.commit() return cur.rowcount > 0 # --------------------------------------------------------------------------- # World Styles CRUD # --------------------------------------------------------------------------- async def list_world_styles() -> list[dict[str, Any]]: """List all world styles (custom only, presets come from presets.py).""" db = await get_db() cur = await db.execute( "SELECT * FROM world_styles WHERE is_preset = 0 ORDER BY name" ) rows = await cur.fetchall() return [dict(r) for r in rows] async def get_world_style(style_id: str) -> dict[str, Any] | None: """Get a world style by ID.""" db = await get_db() cur = await db.execute("SELECT * FROM world_styles WHERE id = ?", (style_id,)) row = await cur.fetchone() return dict(row) if row else None async def insert_world_style( *, id: str, name: str, preset_type: str = "custom", background_rules: str = "", floor_rules: str = "", secondary_characters_rules: str = "", lighting_rules: str = "", props_rules: str = "", architecture_allowed: bool = True, ) -> None: """Insert a new world style.""" db = await get_db() now = _now() await db.execute( """INSERT INTO world_styles (id, name, preset_type, background_rules, floor_rules, secondary_characters_rules, lighting_rules, props_rules, architecture_allowed, is_preset, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 0, ?, ?)""", (id, name, preset_type, background_rules, floor_rules, secondary_characters_rules, lighting_rules, props_rules, 1 if architecture_allowed else 0, now, now), ) await db.commit() async def delete_world_style(style_id: str) -> bool: """Delete a world style.""" db = await get_db() cur = await db.execute( "DELETE FROM world_styles WHERE id = ? AND is_preset = 0", (style_id,) ) await db.commit() return cur.rowcount > 0 # --------------------------------------------------------------------------- # Clothing Styles CRUD # --------------------------------------------------------------------------- async def list_clothing_styles() -> list[dict[str, Any]]: """List all clothing styles (custom only, presets come from presets.py).""" db = await get_db() cur = await db.execute( "SELECT * FROM clothing_styles WHERE is_preset = 0 ORDER BY name" ) rows = await cur.fetchall() return [dict(r) for r in rows] async def get_clothing_style(style_id: str) -> dict[str, Any] | None: """Get a clothing style by ID.""" db = await get_db() cur = await db.execute("SELECT * FROM clothing_styles WHERE id = ?", (style_id,)) row = await cur.fetchone() return dict(row) if row else None async def insert_clothing_style( *, id: str, name: str, preset_type: str = "custom", clothing_rules: str = "", opacity_rules: str = "", max_pieces: int = 2, ) -> None: """Insert a new clothing style.""" db = await get_db() now = _now() await db.execute( """INSERT INTO clothing_styles (id, name, preset_type, clothing_rules, opacity_rules, max_pieces, is_preset, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, 0, ?, ?)""", (id, name, preset_type, clothing_rules, opacity_rules, max_pieces, now, now), ) await db.commit() async def delete_clothing_style(style_id: str) -> bool: """Delete a clothing style.""" db = await get_db() cur = await db.execute( "DELETE FROM clothing_styles WHERE id = ? AND is_preset = 0", (style_id,) ) await db.commit() return cur.rowcount > 0 # --------------------------------------------------------------------------- # Production Rules CRUD # --------------------------------------------------------------------------- async def list_production_rules() -> list[dict[str, Any]]: """List all production rules (custom only, presets come from presets.py).""" db = await get_db() cur = await db.execute( "SELECT * FROM production_rules WHERE is_preset = 0 ORDER BY name" ) rows = await cur.fetchall() return [dict(r) for r in rows] async def get_production_rules(rules_id: str) -> dict[str, Any] | None: """Get production rules by ID.""" db = await get_db() cur = await db.execute("SELECT * FROM production_rules WHERE id = ?", (rules_id,)) row = await cur.fetchone() return dict(row) if row else None async def insert_production_rules( *, id: str, name: str, camera_settings: str = "", pose_rule: str = "", shot_intercalation_rules: str = "", death_scene_rules: str = "", child_rules: str = "", zombie_rules: str = "", consequence_philosophy: str = "", scale_rule: str = "", ) -> None: """Insert new production rules.""" db = await get_db() now = _now() await db.execute( """INSERT INTO production_rules (id, name, camera_settings, pose_rule, shot_intercalation_rules, death_scene_rules, child_rules, zombie_rules, consequence_philosophy, scale_rule, is_preset, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, ?, ?)""", (id, name, camera_settings, pose_rule, shot_intercalation_rules, death_scene_rules, child_rules, zombie_rules, consequence_philosophy, scale_rule, now, now), ) await db.commit() async def delete_production_rules(rules_id: str) -> bool: """Delete production rules.""" db = await get_db() cur = await db.execute( "DELETE FROM production_rules WHERE id = ? AND is_preset = 0", (rules_id,) ) await db.commit() return cur.rowcount > 0