app_kit/storage.py

from __future__ import annotations

import json
import os
import shutil
import sqlite3
from contextlib import contextmanager
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterable, Iterator

APP_NAME = "p3_field_repair_logbook"
DEFAULT_DATA_DIR = Path("data")
DEFAULT_DB_PATH = DEFAULT_DATA_DIR / "sqlite" / f"{APP_NAME}.sqlite3"
DEFAULT_ARTIFACTS_DIR = DEFAULT_DATA_DIR / "artifacts" / APP_NAME
DEFAULT_DEMO_PACKS_DIR = DEFAULT_DATA_DIR / "demo_packs" / APP_NAME


SCHEMA = """
PRAGMA foreign_keys = ON;

CREATE TABLE IF NOT EXISTS manuals (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    title TEXT NOT NULL,
    source_url TEXT NOT NULL,
    license_name TEXT NOT NULL,
    attribution TEXT NOT NULL,
    created_at TEXT NOT NULL,
    is_demo INTEGER NOT NULL DEFAULT 1,
    body_text TEXT NOT NULL,
    UNIQUE(title, source_url)
);

CREATE TABLE IF NOT EXISTS manual_sections (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    manual_id INTEGER NOT NULL REFERENCES manuals(id) ON DELETE CASCADE,
    section_slug TEXT NOT NULL,
    section_title TEXT NOT NULL,
    section_order INTEGER NOT NULL,
    page_hint TEXT,
    content TEXT NOT NULL,
    citation TEXT NOT NULL,
    is_demo INTEGER NOT NULL DEFAULT 1,
    UNIQUE(manual_id, section_slug)
);

CREATE TABLE IF NOT EXISTS jobs (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    created_at TEXT NOT NULL,
    job_title TEXT NOT NULL,
    technician TEXT NOT NULL,
    location TEXT NOT NULL,
    equipment_type TEXT NOT NULL,
    severity TEXT NOT NULL,
    symptom TEXT NOT NULL,
    notes TEXT NOT NULL,
    photo_path TEXT,
    photo_caption TEXT,
    response_json TEXT NOT NULL,
    resolution_status TEXT NOT NULL,
    linked_section_ids TEXT NOT NULL,
    linked_job_ids TEXT NOT NULL,
    is_demo INTEGER NOT NULL DEFAULT 0
);

CREATE TABLE IF NOT EXISTS settings (
    key TEXT PRIMARY KEY,
    value TEXT NOT NULL
);
"""


def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat(timespec="seconds")


@dataclass
class DbPaths:
    db_path: Path = DEFAULT_DB_PATH
    artifacts_dir: Path = DEFAULT_ARTIFACTS_DIR
    demo_packs_dir: Path = DEFAULT_DEMO_PACKS_DIR


def ensure_dirs(paths: DbPaths | None = None) -> DbPaths:
    paths = paths or DbPaths()
    paths.db_path.parent.mkdir(parents=True, exist_ok=True)
    paths.artifacts_dir.mkdir(parents=True, exist_ok=True)
    paths.demo_packs_dir.mkdir(parents=True, exist_ok=True)
    return paths


@contextmanager
def connect(db_path: str | Path | None = None):
    paths = ensure_dirs(DbPaths(db_path=Path(db_path) if db_path else DEFAULT_DB_PATH))
    conn = sqlite3.connect(paths.db_path, check_same_thread=False)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute("PRAGMA foreign_keys=ON")
    try:
        yield conn
    finally:
        conn.close()

def init_db(db_path: str | Path | None = None) -> None:
    with connect(db_path) as conn:
        conn.executescript(SCHEMA)
        conn.commit()


def reset_db(db_path: str | Path | None = None) -> None:
    with connect(db_path) as conn:
        for table in ["manual_sections", "manuals", "jobs", "settings"]:
            conn.execute(f"DELETE FROM {table}")
        conn.commit()


def set_setting(key: str, value: Any, db_path: str | Path | None = None) -> None:
    with connect(db_path) as conn:
        conn.execute(
            "INSERT INTO settings(key, value) VALUES (?, ?) ON CONFLICT(key) DO UPDATE SET value=excluded.value",
            (key, json.dumps(value)),
        )
        conn.commit()


def get_setting(key: str, default: Any = None, db_path: str | Path | None = None) -> Any:
    with connect(db_path) as conn:
        row = conn.execute("SELECT value FROM settings WHERE key = ?", (key,)).fetchone()
    if row is None:
        return default
    try:
        return json.loads(row["value"])
    except Exception:
        return row["value"]


def copy_into_artifacts(src: str | Path, artifacts_dir: str | Path | None = None) -> str:
    dest_root = Path(artifacts_dir) if artifacts_dir else DEFAULT_ARTIFACTS_DIR
    dest_root.mkdir(parents=True, exist_ok=True)
    src_path = Path(src)
    dest = dest_root / src_path.name
    shutil.copy2(src_path, dest)
    return str(dest)


def iter_rows(conn: sqlite3.Connection, query: str, params: Iterable[Any] = ()) -> Iterator[sqlite3.Row]:
    for row in conn.execute(query, tuple(params)):
        yield row