build-small-hackathon/Off-Grid-Field-Repair-Logbook

1from __future__ import annotations2 3import json4import os5import shutil6import sqlite37from contextlib import contextmanager8from dataclasses import dataclass9from datetime import datetime, timezone10from pathlib import Path11from typing import Any, Iterable, Iterator12 13APP_NAME = "p3_field_repair_logbook"14DEFAULT_DATA_DIR = Path("data")15DEFAULT_DB_PATH = DEFAULT_DATA_DIR / "sqlite" / f"{APP_NAME}.sqlite3"16DEFAULT_ARTIFACTS_DIR = DEFAULT_DATA_DIR / "artifacts" / APP_NAME17DEFAULT_DEMO_PACKS_DIR = DEFAULT_DATA_DIR / "demo_packs" / APP_NAME18 19 20SCHEMA = """21PRAGMA foreign_keys = ON;22 23CREATE TABLE IF NOT EXISTS manuals (24    id INTEGER PRIMARY KEY AUTOINCREMENT,25    title TEXT NOT NULL,26    source_url TEXT NOT NULL,27    license_name TEXT NOT NULL,28    attribution TEXT NOT NULL,29    created_at TEXT NOT NULL,30    is_demo INTEGER NOT NULL DEFAULT 1,31    body_text TEXT NOT NULL,32    UNIQUE(title, source_url)33);34 35CREATE TABLE IF NOT EXISTS manual_sections (36    id INTEGER PRIMARY KEY AUTOINCREMENT,37    manual_id INTEGER NOT NULL REFERENCES manuals(id) ON DELETE CASCADE,38    section_slug TEXT NOT NULL,39    section_title TEXT NOT NULL,40    section_order INTEGER NOT NULL,41    page_hint TEXT,42    content TEXT NOT NULL,43    citation TEXT NOT NULL,44    is_demo INTEGER NOT NULL DEFAULT 1,45    UNIQUE(manual_id, section_slug)46);47 48CREATE TABLE IF NOT EXISTS jobs (49    id INTEGER PRIMARY KEY AUTOINCREMENT,50    created_at TEXT NOT NULL,51    job_title TEXT NOT NULL,52    technician TEXT NOT NULL,53    location TEXT NOT NULL,54    equipment_type TEXT NOT NULL,55    severity TEXT NOT NULL,56    symptom TEXT NOT NULL,57    notes TEXT NOT NULL,58    photo_path TEXT,59    photo_caption TEXT,60    response_json TEXT NOT NULL,61    resolution_status TEXT NOT NULL,62    linked_section_ids TEXT NOT NULL,63    linked_job_ids TEXT NOT NULL,64    is_demo INTEGER NOT NULL DEFAULT 065);66 67CREATE TABLE IF NOT EXISTS settings (68    key TEXT PRIMARY KEY,69    value TEXT NOT NULL70);71"""72 73 74def utc_now() -> str:75    return datetime.now(timezone.utc).isoformat(timespec="seconds")76 77 78@dataclass79class DbPaths:80    db_path: Path = DEFAULT_DB_PATH81    artifacts_dir: Path = DEFAULT_ARTIFACTS_DIR82    demo_packs_dir: Path = DEFAULT_DEMO_PACKS_DIR83 84 85def ensure_dirs(paths: DbPaths | None = None) -> DbPaths:86    paths = paths or DbPaths()87    paths.db_path.parent.mkdir(parents=True, exist_ok=True)88    paths.artifacts_dir.mkdir(parents=True, exist_ok=True)89    paths.demo_packs_dir.mkdir(parents=True, exist_ok=True)90    return paths91 92 93@contextmanager94def connect(db_path: str | Path | None = None):95    paths = ensure_dirs(DbPaths(db_path=Path(db_path) if db_path else DEFAULT_DB_PATH))96    conn = sqlite3.connect(paths.db_path, check_same_thread=False)97    conn.row_factory = sqlite3.Row98    conn.execute("PRAGMA journal_mode=WAL")99    conn.execute("PRAGMA foreign_keys=ON")100    try:101        yield conn102    finally:103        conn.close()104 105def init_db(db_path: str | Path | None = None) -> None:106    with connect(db_path) as conn:107        conn.executescript(SCHEMA)108        conn.commit()109 110 111def reset_db(db_path: str | Path | None = None) -> None:112    with connect(db_path) as conn:113        for table in ["manual_sections", "manuals", "jobs", "settings"]:114            conn.execute(f"DELETE FROM {table}")115        conn.commit()116 117 118def set_setting(key: str, value: Any, db_path: str | Path | None = None) -> None:119    with connect(db_path) as conn:120        conn.execute(121            "INSERT INTO settings(key, value) VALUES (?, ?) ON CONFLICT(key) DO UPDATE SET value=excluded.value",122            (key, json.dumps(value)),123        )124        conn.commit()125 126 127def get_setting(key: str, default: Any = None, db_path: str | Path | None = None) -> Any:128    with connect(db_path) as conn:129        row = conn.execute("SELECT value FROM settings WHERE key = ?", (key,)).fetchone()130    if row is None:131        return default132    try:133        return json.loads(row["value"])134    except Exception:135        return row["value"]136 137 138def copy_into_artifacts(src: str | Path, artifacts_dir: str | Path | None = None) -> str:139    dest_root = Path(artifacts_dir) if artifacts_dir else DEFAULT_ARTIFACTS_DIR140    dest_root.mkdir(parents=True, exist_ok=True)141    src_path = Path(src)142    dest = dest_root / src_path.name143    shutil.copy2(src_path, dest)144    return str(dest)145 146 147def iter_rows(conn: sqlite3.Connection, query: str, params: Iterable[Any] = ()) -> Iterator[sqlite3.Row]:148    for row in conn.execute(query, tuple(params)):149        yield row150