diff --git a/CHANGELOG.md b/CHANGELOG.md
deleted file mode 100644
index 86bf338..0000000
--- a/CHANGELOG.md
+++ /dev/null
@@ -1,39 +0,0 @@
-# 0.1.6
-
- * Fix shortcuts for next/previous day to not collide with Normal text (Ctrl+N)
-
-# 0.1.5
-
- * Refactor schema to support versioning of pages. Add HistoryDialog and diff with ability to revert.
-
-# 0.1.4
-
- * Add auto-lock of app (configurable in Settings, defaults to 15 minutes)
- * Add 'Report a bug' to Help nav
-
-# 0.1.3
-
- * Fix bold toggle
- * Improvements to preview size in search results
- * Make URLs highlighted and clickable (Ctrl+click)
- * Explain the purpose of the encryption key for first-time use
- * Support saving the encryption key to the settings file to avoid being prompted (off by default)
- * Abbreviated toolbar symbols to keep things tidier. Add tooltips
- * Add ability to export the database to different formats
- * Add Documentation/Help menu
-
-# 0.1.2
-
- * Switch from Markdown to HTML via QTextEdit, with a toolbar
- * Add search ability
- * Fix Settings shortcut and change nav menu from 'File' to 'Application'
-
-# 0.1.1
-
- * Add ability to change the key
- * Add ability to jump to today's date
- * Add shortcut for Settings (Ctrl+E) so as not to collide with Ctrl+S (Save)
-
-# 0.1.0
-
- * Initial release.
diff --git a/README.md b/README.md
index ea1cbdc..8b20e14 100644
--- a/README.md
+++ b/README.md
@@ -9,7 +9,7 @@ It uses [SQLCipher bindings](https://pypi.org/project/sqlcipher3-wheels) as a dr
for SQLite3. This means that the underlying database for the notebook is encrypted at rest.
To increase security, the SQLCipher key is requested when the app is opened, and is not written
-to disk unless the user configures it to be in the settings.
+to disk.
There is deliberately no network connectivity or syncing intended.
@@ -19,23 +19,24 @@ There is deliberately no network connectivity or syncing intended.
## Features
- * Data is encrypted at rest
- * Encryption key is prompted for and never stored, unless user chooses to via Settings
* Every 'page' is linked to the calendar day
- * All changes are version controlled, with ability to view/diff versions and revert
- * Text is HTML with basic styling
- * Search
+ * Basic markdown
* Automatic periodic saving (or explicitly save)
+ * Navigating from one day to the next automatically saves
+ * Basic keyboard shortcuts
* Transparent integrity checking of the database when it opens
- * Automatic locking of the app after a period of inactivity (default 15 min)
- * Rekey the database (change the password)
- * Export the database to json, txt, html or csv
+
+
+## Yet to do
+
+ * Search
+ * Taxonomy/tagging
+ * Ability to change the SQLCipher key
+ * Export to other formats (plaintext, json, sql etc)
## How to install
-Make sure you have `libxcb-cursor0` installed (it may be called something else on non-Debian distributions).
-
### From source
* Clone this repo or download the tarball from the releases page
@@ -47,7 +48,7 @@ Make sure you have `libxcb-cursor0` installed (it may be called something else o
* Download the whl and run it
-### From PyPi/pip
+### From PyPi
* `pip install bouquin`
diff --git a/bouquin/db.py b/bouquin/db.py
index df5aa62..1ea60fa 100644
--- a/bouquin/db.py
+++ b/bouquin/db.py
@@ -1,23 +1,15 @@
from __future__ import annotations
-import csv
-import html
-import json
-import os
-
from dataclasses import dataclass
from pathlib import Path
-from sqlcipher3 import dbapi2 as sqlite
-from typing import List, Sequence, Tuple
-Entry = Tuple[str, str]
+from sqlcipher3 import dbapi2 as sqlite
@dataclass
class DBConfig:
path: Path
key: str
- idle_minutes: int = 15 # 0 = never lock
class DBManager:
@@ -26,17 +18,14 @@ class DBManager:
self.conn: sqlite.Connection | None = None
def connect(self) -> bool:
- """
- Open, decrypt and install schema on the database.
- """
# Ensure parent dir exists
self.cfg.path.parent.mkdir(parents=True, exist_ok=True)
self.conn = sqlite.connect(str(self.cfg.path))
- self.conn.row_factory = sqlite.Row
cur = self.conn.cursor()
cur.execute(f"PRAGMA key = '{self.cfg.key}';")
- cur.execute("PRAGMA foreign_keys = ON;")
- cur.execute("PRAGMA journal_mode = WAL;").fetchone()
+ cur.execute("PRAGMA cipher_compatibility = 4;")
+ cur.execute("PRAGMA journal_mode = WAL;")
+ self.conn.commit()
try:
self._integrity_ok()
except Exception:
@@ -47,18 +36,15 @@ class DBManager:
return True
def _integrity_ok(self) -> bool:
- """
- Runs the cipher_integrity_check PRAGMA on the database.
- """
cur = self.conn.cursor()
cur.execute("PRAGMA cipher_integrity_check;")
rows = cur.fetchall()
- # OK: nothing returned
+ # OK
if not rows:
return
- # Not OK: rows of problems returned
+ # Not OK
details = "; ".join(str(r[0]) for r in rows if r and r[0] is not None)
raise sqlite.IntegrityError(
"SQLCipher integrity check failed"
@@ -66,384 +52,39 @@ class DBManager:
)
def _ensure_schema(self) -> None:
- """
- Install the expected schema on the database.
- We also handle upgrades here.
- """
cur = self.conn.cursor()
- # Always keep FKs on
- cur.execute("PRAGMA foreign_keys = ON;")
-
- # Create new versioned schema if missing (< 0.1.5)
- cur.executescript(
+ cur.execute(
"""
- CREATE TABLE IF NOT EXISTS pages (
- date TEXT PRIMARY KEY, -- yyyy-MM-dd
- current_version_id INTEGER,
- FOREIGN KEY(current_version_id) REFERENCES versions(id) ON DELETE SET NULL
+ CREATE TABLE IF NOT EXISTS entries (
+ date TEXT PRIMARY KEY, -- ISO yyyy-MM-dd
+ content TEXT NOT NULL
);
-
- CREATE TABLE IF NOT EXISTS versions (
- id INTEGER PRIMARY KEY,
- date TEXT NOT NULL, -- FK to pages.date
- version_no INTEGER NOT NULL, -- 1,2,3… per date
- created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')),
- note TEXT,
- content TEXT NOT NULL,
- FOREIGN KEY(date) REFERENCES pages(date) ON DELETE CASCADE
- );
-
- CREATE UNIQUE INDEX IF NOT EXISTS ux_versions_date_ver ON versions(date, version_no);
- CREATE INDEX IF NOT EXISTS ix_versions_date_created ON versions(date, created_at);
"""
)
-
- # If < 0.1.5 'entries' table exists and nothing has been migrated yet, try to migrate.
- pre_0_1_5 = cur.execute(
- "SELECT 1 FROM sqlite_master WHERE type='table' AND name='entries';"
- ).fetchone()
- pages_empty = cur.execute("SELECT 1 FROM pages LIMIT 1;").fetchone() is None
-
- if pre_0_1_5 and pages_empty:
- # Seed pages and versions (all as version 1)
- cur.execute("INSERT OR IGNORE INTO pages(date) SELECT date FROM entries;")
- cur.execute(
- "INSERT INTO versions(date, version_no, content) "
- "SELECT date, 1, content FROM entries;"
- )
- # Point head to v1 for each page
- cur.execute(
- """
- UPDATE pages
- SET current_version_id = (
- SELECT v.id FROM versions v
- WHERE v.date = pages.date AND v.version_no = 1
- );
- """
- )
- cur.execute("DROP TABLE IF EXISTS entries;")
+ cur.execute("PRAGMA user_version = 1;")
self.conn.commit()
- def rekey(self, new_key: str) -> None:
- """
- Change the SQLCipher passphrase in-place, then reopen the connection
- with the new key to verify.
- """
- if self.conn is None:
- raise RuntimeError("Database is not connected")
- cur = self.conn.cursor()
- # Change the encryption key of the currently open database
- cur.execute(f"PRAGMA rekey = '{new_key}';")
- self.conn.commit()
-
- # Close and reopen with the new key to verify and restore PRAGMAs
- self.conn.close()
- self.conn = None
- self.cfg.key = new_key
- if not self.connect():
- raise sqlite.Error("Re-open failed after rekey")
-
def get_entry(self, date_iso: str) -> str:
- """
- Get a single entry by its date.
- """
cur = self.conn.cursor()
- row = cur.execute(
- """
- SELECT v.content
- FROM pages p
- JOIN versions v ON v.id = p.current_version_id
- WHERE p.date = ?;
- """,
- (date_iso,),
- ).fetchone()
+ cur.execute("SELECT content FROM entries WHERE date = ?;", (date_iso,))
+ row = cur.fetchone()
return row[0] if row else ""
def upsert_entry(self, date_iso: str, content: str) -> None:
- """
- Insert or update an entry.
- """
- # Make a new version and set it as current
- self.save_new_version(date_iso, content, note=None, set_current=True)
-
- def search_entries(self, text: str) -> list[str]:
- """
- Search for entries by term. This only works against the latest
- version of the page.
- """
cur = self.conn.cursor()
- pattern = f"%{text}%"
- rows = cur.execute(
+ cur.execute(
"""
- SELECT p.date, v.content
- FROM pages AS p
- JOIN versions AS v
- ON v.id = p.current_version_id
- WHERE TRIM(v.content) <> ''
- AND v.content LIKE LOWER(?) ESCAPE '\\'
- ORDER BY p.date DESC;
+ INSERT INTO entries(date, content) VALUES(?, ?)
+ ON CONFLICT(date) DO UPDATE SET content = excluded.content;
""",
- (pattern,),
- ).fetchall()
- return [(r[0], r[1]) for r in rows]
+ (date_iso, content),
+ )
+ self.conn.commit()
def dates_with_content(self) -> list[str]:
- """
- Find all entries and return the dates of them.
- This is used to mark the calendar days in bold if they contain entries.
- """
cur = self.conn.cursor()
- rows = cur.execute(
- """
- SELECT p.date
- FROM pages p
- JOIN versions v ON v.id = p.current_version_id
- WHERE TRIM(v.content) <> ''
- ORDER BY p.date;
- """
- ).fetchall()
- return [r[0] for r in rows]
-
- # ------------------------- Versioning logic here ------------------------#
- def save_new_version(
- self,
- date_iso: str,
- content: str,
- note: str | None = None,
- set_current: bool = True,
- ) -> tuple[int, int]:
- """
- Append a new version for this date. Returns (version_id, version_no).
- If set_current=True, flips the page head to this new version.
- """
- if self.conn is None:
- raise RuntimeError("Database is not connected")
- with self.conn: # transaction
- cur = self.conn.cursor()
- # Ensure page row exists
- cur.execute("INSERT OR IGNORE INTO pages(date) VALUES (?);", (date_iso,))
- # Next version number
- row = cur.execute(
- "SELECT COALESCE(MAX(version_no), 0) AS maxv FROM versions WHERE date=?;",
- (date_iso,),
- ).fetchone()
- next_ver = int(row["maxv"]) + 1
- # Insert the version
- cur.execute(
- "INSERT INTO versions(date, version_no, content, note) "
- "VALUES (?,?,?,?);",
- (date_iso, next_ver, content, note),
- )
- ver_id = cur.lastrowid
- if set_current:
- cur.execute(
- "UPDATE pages SET current_version_id=? WHERE date=?;",
- (ver_id, date_iso),
- )
- return ver_id, next_ver
-
- def list_versions(self, date_iso: str) -> list[dict]:
- """
- Returns history for a given date (newest first), including which one is current.
- Each item: {id, version_no, created_at, note, is_current}
- """
- cur = self.conn.cursor()
- rows = cur.execute(
- """
- SELECT v.id, v.version_no, v.created_at, v.note,
- CASE WHEN v.id = p.current_version_id THEN 1 ELSE 0 END AS is_current
- FROM versions v
- LEFT JOIN pages p ON p.date = v.date
- WHERE v.date = ?
- ORDER BY v.version_no DESC;
- """,
- (date_iso,),
- ).fetchall()
- return [dict(r) for r in rows]
-
- def get_version(
- self,
- *,
- date_iso: str | None = None,
- version_no: int | None = None,
- version_id: int | None = None,
- ) -> dict | None:
- """
- Fetch a specific version by (date, version_no) OR by version_id.
- Returns a dict with keys: id, date, version_no, created_at, note, content.
- """
- cur = self.conn.cursor()
- if version_id is not None:
- row = cur.execute(
- "SELECT id, date, version_no, created_at, note, content "
- "FROM versions WHERE id=?;",
- (version_id,),
- ).fetchone()
- else:
- if date_iso is None or version_no is None:
- raise ValueError(
- "Provide either version_id OR (date_iso and version_no)"
- )
- row = cur.execute(
- "SELECT id, date, version_no, created_at, note, content "
- "FROM versions WHERE date=? AND version_no=?;",
- (date_iso, version_no),
- ).fetchone()
- return dict(row) if row else None
-
- def revert_to_version(
- self,
- date_iso: str,
- *,
- version_no: int | None = None,
- version_id: int | None = None,
- ) -> None:
- """
- Point the page head (pages.current_version_id) to an existing version.
- Fast revert: no content is rewritten.
- """
- if self.conn is None:
- raise RuntimeError("Database is not connected")
- cur = self.conn.cursor()
-
- if version_id is None:
- if version_no is None:
- raise ValueError("Provide version_no or version_id")
- row = cur.execute(
- "SELECT id FROM versions WHERE date=? AND version_no=?;",
- (date_iso, version_no),
- ).fetchone()
- if row is None:
- raise ValueError("Version not found for this date")
- version_id = int(row["id"])
- else:
- # Ensure that version_id belongs to the given date
- row = cur.execute(
- "SELECT date FROM versions WHERE id=?;", (version_id,)
- ).fetchone()
- if row is None or row["date"] != date_iso:
- raise ValueError("version_id does not belong to the given date")
-
- with self.conn:
- cur.execute(
- "UPDATE pages SET current_version_id=? WHERE date=?;",
- (version_id, date_iso),
- )
-
- # ------------------------- Export logic here ------------------------#
- def get_all_entries(self) -> List[Entry]:
- """
- Get all entries. Used for exports.
- """
- cur = self.conn.cursor()
- rows = cur.execute(
- """
- SELECT p.date, v.content
- FROM pages p
- JOIN versions v ON v.id = p.current_version_id
- ORDER BY p.date;
- """
- ).fetchall()
- return [(r[0], r[1]) for r in rows]
-
- def export_json(
- self, entries: Sequence[Entry], file_path: str, pretty: bool = True
- ) -> None:
- """
- Export to json.
- """
- data = [{"date": d, "content": c} for d, c in entries]
- with open(file_path, "w", encoding="utf-8") as f:
- if pretty:
- json.dump(data, f, ensure_ascii=False, indent=2)
- else:
- json.dump(data, f, ensure_ascii=False, separators=(",", ":"))
-
- def export_csv(self, entries: Sequence[Entry], file_path: str) -> None:
- # utf-8-sig adds a BOM so Excel opens as UTF-8 by default.
- with open(file_path, "w", encoding="utf-8-sig", newline="") as f:
- writer = csv.writer(f)
- writer.writerow(["date", "content"]) # header
- writer.writerows(entries)
-
- def export_txt(
- self,
- entries: Sequence[Entry],
- file_path: str,
- separator: str = "\n\n— — — — —\n\n",
- strip_html: bool = True,
- ) -> None:
- import re, html as _html
-
- # Precompiled patterns
- STYLE_SCRIPT_RE = re.compile(r"(?is)<(script|style)[^>]*>.*?\1>")
- COMMENT_RE = re.compile(r"", re.S)
- BR_RE = re.compile(r"(?i)
")
- BLOCK_END_RE = re.compile(r"(?i)(p|div|section|article|li|h[1-6])\\s*>")
- TAG_RE = re.compile(r"<[^>]+>")
- WS_ENDS_RE = re.compile(r"[ \\t]+\\n")
- MULTINEWLINE_RE = re.compile(r"\\n{3,}")
-
- def _strip(s: str) -> str:
- # 1) Remove ",
- "