admin/db.py - main - amne/dca-agent

admin/db.py
661 строка25 KB
amne
feat: Phase 2 — pipeline_steps tracking + API
11 авг 2026, 14:52
11 авг 2026, 14:522b38c25
import asyncio
import aiosqlite
import json
import os
import secrets
import time
import logging
from typing import Optional
from pathlib import Path

logger = logging.getLogger("dca-admin")

DB_PATH = os.environ.get("DCA_DB_PATH", str(Path(__file__).parent / "admin.db"))

SCHEMA = """
CREATE TABLE IF NOT EXISTS clients (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT NOT NULL,
    slug TEXT UNIQUE NOT NULL,
    api_key TEXT UNIQUE NOT NULL,
    is_active INTEGER DEFAULT 1,
    tariff TEXT DEFAULT 'basic',
    price_per_call REAL DEFAULT 20.0,
    created_at REAL,
    settings TEXT DEFAULT '{}'
);

CREATE TABLE IF NOT EXISTS pipelines (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT NOT NULL,
    nexara_url TEXT DEFAULT '',
    nexara_key TEXT DEFAULT '',
    llm_url TEXT DEFAULT 'https://openrouter.ai/api/v1',
    llm_key TEXT DEFAULT '',
    llm_model TEXT DEFAULT 'deepseek/deepseek-v4-pro',
    llm_temperature REAL DEFAULT 0.2,
    normalize_template TEXT DEFAULT '',
    report_template TEXT DEFAULT '',
    created_at REAL,
    updated_at REAL
);

CREATE TABLE IF NOT EXISTS pipeline_configs (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    client_id INTEGER NOT NULL REFERENCES clients(id),
    pipeline_id INTEGER REFERENCES pipelines(id),
    fallback_llm_url TEXT DEFAULT '',
    fallback_llm_key TEXT DEFAULT '',
    max_file_size_mb INTEGER DEFAULT 50,
    max_calls_per_day INTEGER DEFAULT 100,
    updated_at REAL,
    UNIQUE(client_id)
);

CREATE TABLE IF NOT EXISTS call_stats (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    client_id INTEGER NOT NULL REFERENCES clients(id),
    filename TEXT DEFAULT '',
    duration_sec REAL DEFAULT 0,
    tokens_in INTEGER DEFAULT 0,
    tokens_out INTEGER DEFAULT 0,
    status TEXT DEFAULT 'done',
    error TEXT DEFAULT '',
    created_at REAL
);

CREATE TABLE IF NOT EXISTS heartbeats (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    client_id INTEGER NOT NULL REFERENCES clients(id),
    status TEXT,
    queue_size INTEGER DEFAULT 0,
    processed_today INTEGER DEFAULT 0,
    uptime_seconds INTEGER DEFAULT 0,
    received_at REAL
);

CREATE TABLE IF NOT EXISTS admin_users (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    email TEXT UNIQUE NOT NULL,
    password_hash TEXT NOT NULL,
    display_name TEXT DEFAULT '',
    created_at REAL
);

CREATE INDEX IF NOT EXISTS idx_stats_client ON call_stats(client_id);
CREATE INDEX IF NOT EXISTS idx_stats_created ON call_stats(created_at);
CREATE INDEX IF NOT EXISTS idx_heartbeats_client ON heartbeats(client_id);

CREATE TABLE IF NOT EXISTS call_results (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    client_id INTEGER NOT NULL REFERENCES clients(id),
    filename TEXT NOT NULL,
    transcript TEXT DEFAULT '',
    dialog TEXT DEFAULT '',
    normalized TEXT DEFAULT '',
    report TEXT DEFAULT '',
    duration_sec REAL DEFAULT 0,
    created_at REAL
);
CREATE INDEX IF NOT EXISTS idx_results_client ON call_results(client_id);
CREATE INDEX IF NOT EXISTS idx_results_filename ON call_results(filename);
CREATE INDEX IF NOT EXISTS idx_results_created ON call_results(created_at);

CREATE TABLE IF NOT EXISTS transcript_cache (
    filename TEXT PRIMARY KEY,
    client_id INTEGER NOT NULL,
    transcript_json TEXT NOT NULL,
    dialog TEXT NOT NULL,
    duration REAL DEFAULT 0,
    created_at REAL
);
CREATE INDEX IF NOT EXISTS idx_tcache_client ON transcript_cache(client_id);

CREATE TABLE IF NOT EXISTS normalize_cache (
    filename TEXT NOT NULL,
    prompt_hash TEXT NOT NULL,
    client_id INTEGER NOT NULL,
    normalized_text TEXT NOT NULL,
    tokens_in INTEGER DEFAULT 0,
    tokens_out INTEGER DEFAULT 0,
    created_at REAL,
    UNIQUE(filename, prompt_hash)
);

CREATE TABLE IF NOT EXISTS report_cache (
    filename TEXT NOT NULL,
    prompt_hash TEXT NOT NULL,
    client_id INTEGER NOT NULL,
    report_text TEXT NOT NULL,
    duration REAL DEFAULT 0,
    tokens_in INTEGER DEFAULT 0,
    tokens_out INTEGER DEFAULT 0,
    created_at REAL,
    UNIQUE(filename, prompt_hash)
);

CREATE TABLE IF NOT EXISTS pipeline_steps (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    filename TEXT NOT NULL,
    client_id INTEGER NOT NULL,
    step TEXT NOT NULL,
    status TEXT DEFAULT 'pending',
    attempts INTEGER DEFAULT 0,
    error TEXT DEFAULT '',
    next_retry_at REAL,
    created_at REAL,
    updated_at REAL,
    UNIQUE(filename, step)
);
CREATE INDEX IF NOT EXISTS idx_steps_lookup ON pipeline_steps(status, step, next_retry_at);
"""


async def init_db():
    import bcrypt as _bcrypt
    async with aiosqlite.connect(DB_PATH) as db:
        await db.executescript(SCHEMA)
        cur = await db.execute("SELECT COUNT(*) FROM admin_users")
        count = (await cur.fetchone())[0]
        if count == 0:
            # Generate a random default password and hash it with bcrypt
            raw_password = secrets.token_hex(16)
            hashed = _bcrypt.hashpw(raw_password.encode(), _bcrypt.gensalt()).decode()
            await db.execute(
                "INSERT INTO admin_users (email, password_hash, display_name, created_at) VALUES (?, ?, ?, ?)",
                ("admin@dca.local", hashed, "Администратор", time.time())
            )
            await db.commit()
            logger.info(f"[db] created default admin, password: {raw_password}")
            logger.info(f"[db] IMPORTANT: change this password immediately via login + admin settings")
    return True


# --- Client CRUD ---

_TRANSLIT = {
    'а':'a','б':'b','в':'v','г':'g','д':'d','е':'e','ё':'e','ж':'zh','з':'z',
    'и':'i','й':'y','к':'k','л':'l','м':'m','н':'n','о':'o','п':'p','р':'r',
    'с':'s','т':'t','у':'u','ф':'f','х':'h','ц':'ts','ч':'ch','ш':'sh','щ':'sch',
    'ъ':'','ы':'y','ь':'','э':'e','ю':'yu','я':'ya',
}


def _make_slug(name: str) -> str:
    """Generate URL-safe slug from clinic name (RU/EN)."""
    import re as _re
    s = name.lower().strip()
    # Transliterate Cyrillic
    s = ''.join(_TRANSLIT.get(c, c) for c in s)
    # Replace spaces and separators with hyphen
    s = s.replace(' ', '-').replace('_', '-').replace('/', '-')
    # Keep only a-z, 0-9, hyphen
    s = _re.sub(r'[^a-z0-9-]', '', s)
    # Collapse multiple hyphens
    s = _re.sub(r'-+', '-', s).strip('-')
    return s or 'clinic'


async def create_client(name: str, slug: str = "", tariff: str = "basic", price_per_call: float = 20.0) -> dict:
    api_key = f"dca_{secrets.token_hex(24)}"
    slug = _make_slug(slug) if slug else _make_slug(name)
    async with aiosqlite.connect(DB_PATH) as db:
        cur_def = await db.execute("SELECT id FROM pipelines ORDER BY id LIMIT 1")
        default_pipe = await cur_def.fetchone()
        default_pid = default_pipe[0] if default_pipe else None
        cur = await db.execute(
            "INSERT INTO clients (name, slug, api_key, tariff, price_per_call, created_at, settings) VALUES (?, ?, ?, ?, ?, ?, ?)",
            (name, slug, api_key, tariff, price_per_call, time.time(), json.dumps({}))
        )
        client_id = cur.lastrowid
        await db.execute(
            "INSERT INTO pipeline_configs (client_id, pipeline_id, updated_at) VALUES (?, ?, ?)",
            (client_id, default_pid, time.time())
        )
        await db.commit()
    return {"id": client_id, "name": name, "slug": slug, "api_key": api_key}


async def list_clients() -> list:
    async with aiosqlite.connect(DB_PATH) as db:
        db.row_factory = aiosqlite.Row
        cur = await db.execute(
            """SELECT c.id, c.name, c.slug, c.api_key, c.is_active, c.tariff, c.price_per_call, c.created_at,
                      pc.llm_model, pc.max_calls_per_day
               FROM clients c LEFT JOIN pipeline_configs pc ON pc.client_id = c.id
               ORDER BY c.id"""
        )
        rows = await cur.fetchall()
    return [dict(r) for r in rows]


async def get_client(client_id: int) -> Optional[dict]:
    async with aiosqlite.connect(DB_PATH) as db:
        db.row_factory = aiosqlite.Row
        cur = await db.execute("SELECT * FROM clients WHERE id = ?", (client_id,))
        row = await cur.fetchone()
    return dict(row) if row else None


async def authenticate_client(client_id: str, api_key: str) -> Optional[dict]:
    async with aiosqlite.connect(DB_PATH) as db:
        db.row_factory = aiosqlite.Row
        cur = await db.execute(
            "SELECT * FROM clients WHERE id = ? AND api_key = ? AND is_active = 1",
            (client_id, api_key)
        )
        row = await cur.fetchone()
    return dict(row) if row else None


# --- Pipeline Config ---

async def get_pipeline_config(client_id: int) -> Optional[dict]:
    """Get resolved config: pipeline + client overrides."""
    async with aiosqlite.connect(DB_PATH) as db:
        db.row_factory = aiosqlite.Row
        cur = await db.execute("SELECT * FROM pipeline_configs WHERE client_id = ?", (client_id,))
        cfg = await cur.fetchone()
        if not cfg:
            return None
        cfg = dict(cfg)
        if cfg.get("pipeline_id"):
            cur = await db.execute("SELECT * FROM pipelines WHERE id = ?", (cfg["pipeline_id"],))
            pipe = await cur.fetchone()
            if pipe:
                pipe = dict(pipe)
                return {
                    "id": cfg["id"],
                    "client_id": client_id,
                    "pipeline_id": cfg["pipeline_id"],
                    "pipeline_name": pipe["name"],
                    "nexara_url": pipe["nexara_url"],
                    "nexara_key": pipe["nexara_key"],
                    "llm_url": pipe["llm_url"],
                    "llm_key": pipe["llm_key"],
                    "llm_model": pipe["llm_model"],
                    "llm_temperature": pipe["llm_temperature"],
                    "normalize_template": pipe["normalize_template"],
                    "report_template": pipe["report_template"],
                    "fallback_llm_url": cfg.get("fallback_llm_url", ""),
                    "fallback_llm_key": cfg.get("fallback_llm_key", ""),
                    "max_file_size_mb": cfg["max_file_size_mb"],
                    "max_calls_per_day": cfg["max_calls_per_day"],
                    "updated_at": cfg["updated_at"],
                }
        return cfg


async def update_pipeline_config(client_id: int, data: dict) -> Optional[dict]:
    fields = []
    values = []
    for k in ["pipeline_id", "fallback_llm_url", "fallback_llm_key",
              "max_file_size_mb", "max_calls_per_day"]:
        if k in data:
            fields.append(f"{k} = ?")
            values.append(data[k])
    if not fields:
        return await get_pipeline_config(client_id)
    fields.append("updated_at = ?")
    values.append(time.time())
    values.append(client_id)
    async with aiosqlite.connect(DB_PATH) as db:
        await db.execute(
            f"UPDATE pipeline_configs SET {', '.join(fields)} WHERE client_id = ?",
            values
        )
        await db.commit()
    return await get_pipeline_config(client_id)


# --- Pipelines (admin-level) ---

async def list_pipelines() -> list:
    async with aiosqlite.connect(DB_PATH) as db:
        db.row_factory = aiosqlite.Row
        cur = await db.execute("SELECT * FROM pipelines ORDER BY id")
        rows = await cur.fetchall()
    return [dict(r) for r in rows]


async def get_pipeline(pipeline_id: int) -> Optional[dict]:
    async with aiosqlite.connect(DB_PATH) as db:
        db.row_factory = aiosqlite.Row
        cur = await db.execute("SELECT * FROM pipelines WHERE id = ?", (pipeline_id,))
        row = await cur.fetchone()
    return dict(row) if row else None


async def create_pipeline(data: dict) -> dict:
    async with aiosqlite.connect(DB_PATH) as db:
        now = time.time()
        cur = await db.execute(
            """INSERT INTO pipelines (name, nexara_url, nexara_key, llm_url, llm_key, llm_model,
               llm_temperature, normalize_template, report_template, created_at, updated_at)
               VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
            (data["name"], data.get("nexara_url", ""), data.get("nexara_key", ""),
             data.get("llm_url", "https://openrouter.ai/api/v1"), data.get("llm_key", ""),
             data.get("llm_model", "deepseek/deepseek-v4-pro"), data.get("llm_temperature", 0.2),
             data.get("normalize_template", ""), data.get("report_template", ""), now, now)
        )
        pid = cur.lastrowid
        await db.commit()
    return {"id": pid, "name": data["name"]}


async def update_pipeline(pipeline_id: int, data: dict) -> Optional[dict]:
    fields = []
    values = []
    for k in ["name", "nexara_url", "nexara_key", "llm_url", "llm_key", "llm_model",
              "llm_temperature", "normalize_template", "report_template"]:
        if k in data:
            fields.append(f"{k} = ?")
            values.append(data[k])
    if not fields:
        return await get_pipeline(pipeline_id)
    fields.append("updated_at = ?")
    values.append(time.time())
    values.append(pipeline_id)
    async with aiosqlite.connect(DB_PATH) as db:
        await db.execute(
            f"UPDATE pipelines SET {', '.join(fields)} WHERE id = ?", values
        )
        await db.commit()
    return await get_pipeline(pipeline_id)


async def copy_pipeline(pipeline_id: int, new_name: str) -> Optional[dict]:
    """Clone a pipeline with all fields under a new name."""
    src = await get_pipeline(pipeline_id)
    if not src:
        return None
    src["name"] = new_name
    return await create_pipeline(src)


async def delete_pipeline(pipeline_id: int) -> bool:
    async with aiosqlite.connect(DB_PATH) as db:
        # Unlink from clients
        await db.execute("UPDATE pipeline_configs SET pipeline_id = NULL WHERE pipeline_id = ?", (pipeline_id,))
        await db.execute("DELETE FROM pipelines WHERE id = ?", (pipeline_id,))
        await db.commit()
    return True


# --- Call Stats (billing only) ---

async def save_call_stat(client_id: int, stat: dict):
    async with aiosqlite.connect(DB_PATH) as db:
        await db.execute(
            """INSERT INTO call_stats 
               (client_id, filename, duration_sec, tokens_in, tokens_out, status, error, created_at)
               VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
            (client_id, stat.get("filename", ""), stat.get("duration_sec", 0),
             stat.get("tokens_in", 0), stat.get("tokens_out", 0),
             stat.get("status", "done"), stat.get("error", ""), time.time())
        )
        await db.commit()


async def get_client_stats(client_id: int) -> dict:
    async with aiosqlite.connect(DB_PATH) as db:
        cur = await db.execute(
            """SELECT COUNT(*) as total_calls,
                      SUM(duration_sec) as total_sec,
                      SUM(tokens_in) as total_tokens_in,
                      SUM(tokens_out) as total_tokens_out,
                      SUM(CASE WHEN date(created_at, 'unixepoch') = date('now') THEN 1 ELSE 0 END) as today_calls,
                      SUM(CASE WHEN date(created_at, 'unixepoch') = date('now') THEN duration_sec ELSE 0 END) as today_sec,
                      SUM(CASE WHEN date(created_at, 'unixepoch') = date('now') THEN tokens_in + tokens_out ELSE 0 END) as today_tokens
               FROM call_stats WHERE client_id = ?""",
            (client_id,)
        )
        row = await cur.fetchone()
    return {
        "total_calls": row[0] or 0,
        "total_min": round((row[1] or 0) / 60, 1),
        "total_tokens_in": row[2] or 0,
        "total_tokens_out": row[3] or 0,
        "today_calls": row[4] or 0,
        "today_min": round((row[5] or 0) / 60, 1),
        "today_tokens": row[6] or 0,
    }


async def list_call_stats(client_id: int, limit: int = 50, offset: int = 0) -> list:
    async with aiosqlite.connect(DB_PATH) as db:
        db.row_factory = aiosqlite.Row
        cur = await db.execute(
            """SELECT id, filename, duration_sec, tokens_in, tokens_out, status, error, created_at
               FROM call_stats WHERE client_id = ?
               ORDER BY created_at DESC LIMIT ? OFFSET ?""",
            (client_id, limit, offset)
        )
        rows = await cur.fetchall()
    return [dict(r) for r in rows]


async def save_call_result(client_id: int, filename: str, result: dict):
    """Store full processing result (transcript, dialog, normalized, report)."""
    import json as _json
    async with aiosqlite.connect(DB_PATH) as db:
        await db.execute(
            """INSERT INTO call_results (client_id, filename, transcript, dialog, normalized, report, duration_sec, created_at)
               VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
            (client_id, filename,
             _json.dumps(result.get("transcript", {}), ensure_ascii=False),
             result.get("dialog", ""),
             result.get("normalized", ""),
             result.get("report", ""),
             result.get("duration", 0),
             time.time())
        )
        await db.commit()


async def list_call_results(client_id: int, limit: int = 50, offset: int = 0) -> list:
    async with aiosqlite.connect(DB_PATH) as db:
        db.row_factory = aiosqlite.Row
        cur = await db.execute(
            """SELECT id, filename, duration_sec, substr(report, 1, 200) as report_preview, created_at
               FROM call_results WHERE client_id = ?
               ORDER BY created_at DESC LIMIT ? OFFSET ?""",
            (client_id, limit, offset)
        )
        rows = await cur.fetchall()
    return [dict(r) for r in rows]


async def get_call_result(client_id: int, result_id: int) -> Optional[dict]:
    async with aiosqlite.connect(DB_PATH) as db:
        db.row_factory = aiosqlite.Row
        cur = await db.execute(
            "SELECT * FROM call_results WHERE id = ? AND client_id = ?",
            (result_id, client_id)
        )
        row = await cur.fetchone()
    return dict(row) if row else None


async def get_today_call_count(client_id: int) -> int:
    """Count successful calls today for daily limit enforcement."""
    async with aiosqlite.connect(DB_PATH) as db:
        cur = await db.execute(
            """SELECT COUNT(*) FROM call_stats
               WHERE client_id = ? AND status = 'done'
               AND date(created_at, 'unixepoch') = date('now')""",
            (client_id,)
        )
        row = await cur.fetchone()
    return row[0] or 0


# --- Heartbeats ---

async def save_heartbeat(client_id: int, data: dict):
    async with aiosqlite.connect(DB_PATH) as db:
        await db.execute(
            """INSERT INTO heartbeats (client_id, status, queue_size, processed_today, uptime_seconds, received_at)
               VALUES (?, ?, ?, ?, ?, ?)""",
            (client_id, data.get("status"), data.get("queue_size", 0),
             data.get("processed_today", 0), data.get("uptime_seconds", 0), time.time())
        )
        await db.commit()


async def get_latest_heartbeat(client_id: int) -> Optional[dict]:
    async with aiosqlite.connect(DB_PATH) as db:
        db.row_factory = aiosqlite.Row
        cur = await db.execute(
            "SELECT * FROM heartbeats WHERE client_id = ? ORDER BY received_at DESC LIMIT 1",
            (client_id,)
        )
        row = await cur.fetchone()
    return dict(row) if row else None


async def cleanup_old_heartbeats(days: int = 30):
    """Delete heartbeats older than N days."""
    cutoff = time.time() - (days * 86400)
    async with aiosqlite.connect(DB_PATH) as db:
        cur = await db.execute("DELETE FROM heartbeats WHERE received_at < ?", (cutoff,))
        await db.commit()
        return cur.rowcount


# --- Transcript cache ---

async def get_transcript_cache(filename: str) -> Optional[dict]:
    """Get cached Nexara transcript by filename. Returns None if not cached."""
    async with aiosqlite.connect(DB_PATH) as db:
        db.row_factory = aiosqlite.Row
        cur = await db.execute(
            "SELECT transcript_json, dialog, duration FROM transcript_cache WHERE filename = ?",
            (filename,)
        )
        row = await cur.fetchone()
    if not row:
        return None
    import json as _json
    return {
        "transcript": _json.loads(row["transcript_json"]),
        "dialog": row["dialog"],
        "duration": row["duration"],
    }


async def save_transcript_cache(client_id: int, filename: str, transcript: dict, dialog: str, duration: float):
    """Cache Nexara transcript so retries don't re-run STT."""
    import json as _json
    async with aiosqlite.connect(DB_PATH) as db:
        await db.execute(
            """INSERT OR REPLACE INTO transcript_cache (filename, client_id, transcript_json, dialog, duration, created_at)
               VALUES (?, ?, ?, ?, ?, ?)""",
            (filename, client_id, _json.dumps(transcript, ensure_ascii=False), dialog, duration, time.time())
        )
        await db.commit()


# --- Normalize cache ---

async def get_normalize_cache(filename: str, prompt_hash: str) -> Optional[dict]:
    async with aiosqlite.connect(DB_PATH) as db:
        db.row_factory = aiosqlite.Row
        cur = await db.execute(
            "SELECT normalized_text, tokens_in, tokens_out FROM normalize_cache WHERE filename=? AND prompt_hash=?",
            (filename, prompt_hash)
        )
        row = await cur.fetchone()
    if not row:
        return None
    return {"normalized": row["normalized_text"], "tokens_in": row["tokens_in"], "tokens_out": row["tokens_out"]}


async def save_normalize_cache(client_id: int, filename: str, prompt_hash: str, normalized: str, tokens_in: int, tokens_out: int):
    async with aiosqlite.connect(DB_PATH) as db:
        await db.execute(
            "INSERT OR REPLACE INTO normalize_cache (filename, prompt_hash, client_id, normalized_text, tokens_in, tokens_out, created_at) VALUES (?,?,?,?,?,?,?)",
            (filename, prompt_hash, client_id, normalized, tokens_in, tokens_out, time.time())
        )
        await db.commit()


# --- Report cache ---

async def get_report_cache(filename: str, prompt_hash: str) -> Optional[dict]:
    async with aiosqlite.connect(DB_PATH) as db:
        db.row_factory = aiosqlite.Row
        cur = await db.execute(
            "SELECT report_text, duration, tokens_in, tokens_out FROM report_cache WHERE filename=? AND prompt_hash=?",
            (filename, prompt_hash)
        )
        row = await cur.fetchone()
    if not row:
        return None
    return {"report": row["report_text"], "duration": row["duration"], "tokens_in": row["tokens_in"], "tokens_out": row["tokens_out"]}


async def save_report_cache(client_id: int, filename: str, prompt_hash: str, report: str, tokens_in: int, tokens_out: int, duration: float):
    async with aiosqlite.connect(DB_PATH) as db:
        await db.execute(
            "INSERT OR REPLACE INTO report_cache (filename, prompt_hash, client_id, report_text, duration, tokens_in, tokens_out, created_at) VALUES (?,?,?,?,?,?,?,?)",
            (filename, prompt_hash, client_id, report, duration, tokens_in, tokens_out, time.time())
        )
        await db.commit()


# --- Pipeline steps tracking ---

async def init_pipeline_steps(filename: str, client_id: int):
    """Create pending steps for a new file. Idempotent."""
    now = time.time()
    async with aiosqlite.connect(DB_PATH) as db:
        for step in ['stt', 'normalize', 'report']:
            await db.execute(
                """INSERT OR IGNORE INTO pipeline_steps
                   (filename, client_id, step, status, attempts, created_at, updated_at)
                   VALUES (?, ?, ?, 'pending', 0, ?, ?)""",
                (filename, client_id, step, now, now)
            )
        await db.commit()

async def mark_step_status(filename: str, step: str, status: str, error: str = ''):
    now = time.time()
    async with aiosqlite.connect(DB_PATH) as db:
        if status == 'processing':
            await db.execute(
                "UPDATE pipeline_steps SET status='processing', updated_at=? WHERE filename=? AND step=?",
                (now, filename, step)
            )
        elif status == 'done':
            await db.execute(
                "UPDATE pipeline_steps SET status='done', error='', updated_at=? WHERE filename=? AND step=?",
                (now, filename, step)
            )
        elif status == 'error':
            await db.execute(
                "UPDATE pipeline_steps SET status='error', attempts=attempts+1, error=?, updated_at=? WHERE filename=? AND step=?",
                (error, now, filename, step)
            )
        await db.commit()

async def get_pipeline_steps(filename: str) -> list:
    """Get all step statuses for a file."""
    async with aiosqlite.connect(DB_PATH) as db:
        db.row_factory = aiosqlite.Row
        cur = await db.execute(
            "SELECT step, status, attempts, error, updated_at FROM pipeline_steps WHERE filename=? ORDER BY CASE step WHEN 'stt' THEN 1 WHEN 'normalize' THEN 2 WHEN 'report' THEN 3 END",
            (filename,)
        )
        rows = await cur.fetchall()
    return [dict(r) for r in rows]

async def reset_step_retry(filename: str, step: str):
    """Reset a failed step back to pending for retry."""
    now = time.time()
    async with aiosqlite.connect(DB_PATH) as db:
        await db.execute(
            "UPDATE pipeline_steps SET status='pending', next_retry_at=?, updated_at=? WHERE filename=? AND step=?",
            (now, now, filename, step)
        )
        await db.commit()