84 lines
2.1 KiB
Python
84 lines
2.1 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import sqlite3
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
from .config import DB_PATH
|
|
|
|
|
|
def connect() -> sqlite3.Connection:
|
|
DB_PATH.parent.mkdir(parents=True, exist_ok=True)
|
|
conn = sqlite3.connect(DB_PATH)
|
|
conn.row_factory = sqlite3.Row
|
|
return conn
|
|
|
|
|
|
def init_db() -> None:
|
|
conn = connect()
|
|
conn.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS analyses (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
title TEXT,
|
|
filename TEXT,
|
|
created_at TEXT,
|
|
step INTEGER DEFAULT 1,
|
|
parsed_path TEXT,
|
|
settings_json TEXT,
|
|
result_path TEXT
|
|
)
|
|
"""
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
|
|
def create_analysis(title: str, filename: str, parsed_path: str) -> int:
|
|
conn = connect()
|
|
cur = conn.execute(
|
|
"""
|
|
INSERT INTO analyses (title, filename, created_at, step, parsed_path)
|
|
VALUES (?, ?, ?, 1, ?)
|
|
""",
|
|
(title, filename, datetime.now().strftime("%Y-%m-%d %H:%M:%S"), parsed_path),
|
|
)
|
|
conn.commit()
|
|
aid = cur.lastrowid
|
|
conn.close()
|
|
return int(aid)
|
|
|
|
|
|
def list_analyses() -> list[dict]:
|
|
conn = connect()
|
|
rows = conn.execute("SELECT * FROM analyses ORDER BY id DESC").fetchall()
|
|
conn.close()
|
|
return [dict(r) for r in rows]
|
|
|
|
|
|
def get_analysis(aid: int) -> dict | None:
|
|
conn = connect()
|
|
row = conn.execute("SELECT * FROM analyses WHERE id=?", (aid,)).fetchone()
|
|
conn.close()
|
|
return dict(row) if row else None
|
|
|
|
|
|
def update_analysis(aid: int, **fields) -> None:
|
|
if not fields:
|
|
return
|
|
keys = ", ".join(f"{k}=?" for k in fields)
|
|
conn = connect()
|
|
conn.execute(f"UPDATE analyses SET {keys} WHERE id=?", (*fields.values(), aid))
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
|
|
def load_json(path: str):
|
|
return json.loads(Path(path).read_text(encoding="utf-8"))
|
|
|
|
|
|
def save_json(path: str, obj) -> None:
|
|
Path(path).parent.mkdir(parents=True, exist_ok=True)
|
|
Path(path).write_text(json.dumps(obj, ensure_ascii=False), encoding="utf-8")
|