Export sessions to CSV and back up the database

EveProfitManager/core.py

@@ -3,6 +3,7 @@
 
 All ISK amounts are in millions, all times are in minutes.
 """
+import csv
 import hashlib
 import math
 import os
@@ -173,6 +174,9 @@ def __init__(self, path=None):
         self.db = sqlite3.connect(path)
         self.db.row_factory = sqlite3.Row
         self.db.execute("PRAGMA foreign_keys = ON")
+        self._migrate()
+
+    def _migrate(self):
         self.db.executescript(SCHEMA)
         columns = [r["name"] for r in self.db.execute("PRAGMA table_info(sessions)")]
         # Columns added after the first release, for databases created before them
@@ -495,6 +499,90 @@ def import_sessions_txt(self, path):
                 imported += 1
         return imported
 
+    # --- export and backup ----------------------------------------------
+
+    def export_csv(self, path, kind=None):
+        """
+        Write every run with its session to a CSV file, oldest session first.
+
+        A session with no runs yet gets one row with the run columns empty.
+
+        Args:
+            path (str | Path): File to write, replacing any existing one.
+            kind (str, optional): Only export ABYSS or INCURSION sessions.
+
+        Returns:
+            int: Number of rows written.
+        """
+        where, args = ("WHERE s.kind = ?", (kind,)) if kind else ("", ())
+        rows = self.db.execute(
+            "SELECT s.id AS session_id, s.kind, s.started_at, s.ended_at, s.weather, s.tier, s.imported, "
+            "r.id AS run_id, r.recorded_at, r.profit AS profit_millions, r.minutes, "
+            "r.isk_after AS isk_after_millions, r.count AS runs "
+            f"FROM sessions s LEFT JOIN runs r ON r.session_id = s.id {where} ORDER BY s.started_at, s.id, r.id",
+            args,
+        ).fetchall()
+        with open(path, "w", newline="", encoding="utf-8") as f:
+            out = csv.writer(f)
+            out.writerow(_CSV_COLUMNS)
+            out.writerows(tuple("" if row[c] is None else row[c] for c in _CSV_COLUMNS) for row in rows)
+        return len(rows)
+
+    def backup(self, path):
+        """
+        Save a consistent copy of the whole database, replacing any file at path.
+
+        Args:
+            path (str | Path): Where to put the copy.
+
+        Raises:
+            ValueError: If path is the live database itself.
+        """
+        live = self.db.execute("PRAGMA database_list").fetchone()["file"]
+        if live and os.path.exists(path) and os.path.samefile(live, path):
+            raise ValueError("Choose a different file than the live database")
+        target = sqlite3.connect(path)
+        try:
+            self.db.backup(target)
+        except sqlite3.Error:
+            target.close()
+            Path(path).unlink(missing_ok=True)  # don't leave half a backup that looks like a real one
+            raise
+        finally:
+            target.close()
+
+    def restore(self, path):
+        """
+        Replace all data with the contents of a backup made by backup().
+
+        The file is checked first, so a bad file leaves the current data alone.
+        Backups from older versions are brought up to the current schema.
+
+        Args:
+            path (str | Path): Backup file.
+
+        Raises:
+            ValueError: If path isn't an intact tracker database.
+        """
+        try:
+            source = sqlite3.connect(f"{Path(path).resolve().as_uri()}?mode=ro", uri=True)
+        except sqlite3.Error as err:
+            raise ValueError(f"Can't open {path}: {err}") from err
+        try:
+            try:
+                if source.execute("PRAGMA integrity_check").fetchone()[0] != "ok":
+                    raise ValueError("The backup file is damaged")
+                tables = {r[0] for r in source.execute("SELECT name FROM sqlite_master WHERE type = 'table'")}
+            except sqlite3.DatabaseError:
+                tables = set()
+            if not {"sessions", "runs"} <= tables:
+                raise ValueError("That isn't an AbyssTracker database")
+            source.backup(self.db)
+        finally:
+            source.close()
+        self._migrate()
+        self.db.commit()
+
 
 def parse_sessions_txt(text):
     """
@@ -583,6 +671,11 @@ def _now():
     return datetime.now().isoformat(timespec="seconds")
 
 
+_CSV_COLUMNS = (
+    "session_id", "kind", "started_at", "ended_at", "weather", "tier", "imported",
+    "run_id", "recorded_at", "profit_millions", "minutes", "isk_after_millions", "runs",
+)
+
 _SESSION_QUERY = (
     "SELECT s.*, COALESCE(SUM(r.profit), 0) AS profit, COALESCE(SUM(r.minutes), 0) AS minutes, "
     "COALESCE(SUM(r.count), 0) AS run_count FROM sessions s LEFT JOIN runs r ON r.session_id = s.id"

EveProfitManager/gui.py

@@ -4,10 +4,12 @@
 Run with:  python gui.py
 """
 import math
+import sqlite3
 import sys
 import threading
 import time
 import tkinter as tk
+from datetime import datetime
 from tkinter import filedialog, messagebox
 from tkinter import font as tkfont
 
@@ -205,9 +207,9 @@ def elapsed(self):
 
     def value_minutes(self):
         """Minutes left (countdown) or elapsed (stopwatch); None if the timer was never started."""
-        elapsed = self.elapsed()
-        if elapsed <= 0:
+        if not self.running and self.elapsed_before <= 0:  # not by elapsed(): a coarse clock reads 0 just after Start
             return None
+        elapsed = self.elapsed()
         if self.countdown:
             return round(max(0.0, self.countdown - elapsed) / 60, 2)
         return round(elapsed / 60, 2)
@@ -1365,10 +1367,16 @@ def __init__(self, master, app):
         self.toolbar = ctk.CTkFrame(app.header, fg_color="transparent")
         self.span = ctk.CTkLabel(self.toolbar, text="", font=font(13), text_color=t.soft)
         self.span.pack(side="left", padx=(0, 14))
-        ctk.CTkButton(
-            self.toolbar, text="Import Sessions.txt", width=160, height=34, fg_color="transparent", border_width=1,
-            border_color=t.line, hover_color=t.panel, text_color=INK, command=self.import_txt,
-        ).pack(side="left")
+        for text, width, command in (
+            ("Import Sessions.txt", 160, self.import_txt),
+            ("Export CSV", 100, self.export_csv),
+            ("Back up", 80, self.back_up),
+            ("Restore", 80, self.restore),
+        ):
+            ctk.CTkButton(
+                self.toolbar, text=text, width=width, height=34, fg_color="transparent", border_width=1,
+                border_color=t.line, hover_color=t.panel, text_color=INK, command=command,
+            ).pack(side="left", padx=(0, 8))
 
         self.filter = ChipGroup(self, list(self.FILTERS), "All", lambda _v: self.refresh())
         self.filter.set_theme(t)
@@ -1505,6 +1513,57 @@ def import_txt(self):
         messagebox.showinfo("Import", f"Imported {count} session{'s' if count != 1 else ''}.", parent=self)
         self.app.refresh()
 
+    def export_csv(self):
+        name = self.filter.get()
+        path = filedialog.asksaveasfilename(
+            parent=self, title="Export CSV", defaultextension=".csv", initialfile=f"abyss-tracker-{name.lower()}.csv",
+            filetypes=[("CSV files", "*.csv"), ("All files", "*.*")],
+        )
+        if not path:
+            return
+        try:
+            count = self.tracker.export_csv(path, self.FILTERS[name])
+        except OSError as err:
+            messagebox.showerror("Export failed", str(err), parent=self)
+            return
+        messagebox.showinfo("Export", f"Exported {count} row{'s' if count != 1 else ''}.", parent=self)
+
+    def back_up(self):
+        path = filedialog.asksaveasfilename(
+            parent=self, title="Back up tracker.db", defaultextension=".db",
+            initialfile=f"tracker-{datetime.now():%Y-%m-%d}.db",
+            filetypes=[("Database files", "*.db"), ("All files", "*.*")],
+        )
+        if not path:
+            return
+        try:
+            self.tracker.backup(path)
+        except (OSError, ValueError, sqlite3.Error) as err:
+            messagebox.showerror("Backup failed", str(err), parent=self)
+            return
+        messagebox.showinfo("Backup", "Backup saved.", parent=self)
+
+    def restore(self):
+        path = filedialog.askopenfilename(
+            parent=self, title="Restore from backup", filetypes=[("Database files", "*.db"), ("All files", "*.*")]
+        )
+        if not path or not messagebox.askyesno(
+            "Restore backup",
+            "Replace all sessions and runs with this backup?\n\nEverything logged since it was made is lost. "
+            "This can't be undone.",
+            parent=self,
+        ):
+            return
+        try:
+            self.tracker.restore(path)
+        except (OSError, ValueError, sqlite3.Error) as err:
+            messagebox.showerror("Restore failed", str(err), parent=self)
+            return
+        for kind in (ABYSS, INCURSION):  # a live session id may now point at nothing, or at a different session
+            self.app.pages[kind].resume()
+        messagebox.showinfo("Restore", "Backup restored.", parent=self)
+        self.app.refresh()
+
 
 def _date_span(first, last):
     """'01 Sep to 23 Oct 2026', with both years when they differ."""

EveProfitManager/test_export.py

@@ -0,0 +1,155 @@
+import csv
+import sqlite3
+import unittest
+from pathlib import Path
+from tempfile import TemporaryDirectory
+
+from core import ABYSS, INCURSION, Tracker
+
+
+class ExportTest(unittest.TestCase):
+    def setUp(self):
+        self.tmp = TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        self.dir = Path(self.tmp.name)
+        self.tracker = Tracker(":memory:")
+        self.addCleanup(self.tracker.close)
+
+    def rows(self, path):
+        with open(path, newline="", encoding="utf-8") as f:
+            return list(csv.DictReader(f))
+
+    def seed(self):
+        a = self.tracker.start_session(ABYSS, starting_isk=1000, weather="Dark", tier=3)
+        self.tracker.log_abyss_run(a, 1000, 1031, time_left=5)
+        self.tracker.log_abyss_run(a, 1031, 1050, time_left=8)
+        self.tracker.end_session(a)
+        i = self.tracker.start_session(INCURSION)
+        self.tracker.log_incursion_site(i, 93, 12)
+        self.tracker.end_session(i)
+        return a, i
+
+    def test_csv_has_one_row_per_run_with_its_session(self):
+        a, i = self.seed()
+        out = self.dir / "out.csv"
+        self.assertEqual(self.tracker.export_csv(out), 3)
+        rows = self.rows(out)
+        self.assertEqual(len(rows), 3)
+        first = rows[0]
+        self.assertEqual(first["session_id"], str(a))
+        self.assertEqual(first["kind"], ABYSS)
+        self.assertEqual(first["weather"], "Dark")
+        self.assertEqual(first["tier"], "3")
+        self.assertEqual(float(first["profit_millions"]), 31.0)
+        self.assertEqual(float(first["minutes"]), 15.0)
+        self.assertEqual(float(first["isk_after_millions"]), 1031.0)
+        self.assertEqual(rows[2]["kind"], INCURSION)
+        self.assertEqual(rows[2]["weather"], "")
+        self.assertEqual(rows[2]["isk_after_millions"], "")
+
+    def test_csv_can_be_limited_to_one_kind(self):
+        self.seed()
+        out = self.dir / "out.csv"
+        self.assertEqual(self.tracker.export_csv(out, kind=INCURSION), 1)
+        self.assertEqual({r["kind"] for r in self.rows(out)}, {INCURSION})
+
+    def test_csv_of_an_empty_tracker_is_just_the_header(self):
+        out = self.dir / "out.csv"
+        self.assertEqual(self.tracker.export_csv(out), 0)
+        self.assertEqual(self.rows(out), [])
+        self.assertIn("session_id", out.read_text(encoding="utf-8"))
+
+    def test_csv_keeps_a_session_that_has_no_runs_yet(self):
+        sid = self.tracker.start_session(INCURSION)
+        out = self.dir / "out.csv"
+        self.tracker.export_csv(out)
+        (row,) = self.rows(out)
+        self.assertEqual(row["session_id"], str(sid))
+        self.assertEqual(row["run_id"], "")
+
+
+class BackupTest(unittest.TestCase):
+    def setUp(self):
+        self.tmp = TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        self.dir = Path(self.tmp.name)
+        self.tracker = Tracker(self.dir / "tracker.db")
+        self.addCleanup(self.tracker.close)
+
+    def log(self, profit):
+        sid = self.tracker.start_session(INCURSION)
+        self.tracker.log_incursion_site(sid, profit, 10)
+        self.tracker.end_session(sid)
+
+    def test_backup_is_a_copy_that_opens_as_a_tracker(self):
+        self.log(50)
+        copy = self.dir / "backup.db"
+        self.tracker.backup(copy)
+        other = Tracker(copy)
+        self.assertEqual(other.totals().profit, 50)
+        other.close()
+
+    def test_backup_replaces_an_older_backup_file(self):
+        copy = self.dir / "backup.db"
+        self.log(50)
+        self.tracker.backup(copy)
+        self.log(25)
+        self.tracker.backup(copy)
+        other = Tracker(copy)
+        self.assertEqual(other.totals().profit, 75)
+        other.close()
+
+    def test_backup_onto_the_live_database_is_refused(self):
+        with self.assertRaises(ValueError):
+            self.tracker.backup(self.dir / "tracker.db")
+        self.assertEqual(self.tracker.totals().sessions, 0)
+
+    def test_restore_replaces_the_data_and_keeps_the_tracker_usable(self):
+        self.log(50)
+        copy = self.dir / "backup.db"
+        self.tracker.backup(copy)
+        self.log(25)
+        self.tracker.set_setting("abyss_weather", "Dark")
+        self.tracker.restore(copy)
+        self.assertEqual(self.tracker.totals().profit, 50)
+        self.assertEqual(self.tracker.totals().sessions, 1)
+        self.assertIsNone(self.tracker.get_setting("abyss_weather"))
+        self.log(5)  # still writable afterwards
+        self.assertEqual(self.tracker.totals().profit, 55)
+
+    def test_restore_from_a_database_older_than_the_current_schema(self):
+        old = self.dir / "old.db"
+        db = sqlite3.connect(old)
+        db.executescript(
+            "CREATE TABLE sessions (id INTEGER PRIMARY KEY, kind TEXT NOT NULL, started_at TEXT NOT NULL,"
+            " ended_at TEXT, starting_isk REAL, imported INTEGER NOT NULL DEFAULT 0);"
+            "CREATE TABLE runs (id INTEGER PRIMARY KEY, session_id INTEGER NOT NULL, recorded_at TEXT NOT NULL,"
+            " profit REAL NOT NULL, minutes REAL NOT NULL, isk_after REAL, count INTEGER NOT NULL DEFAULT 1);"
+            "INSERT INTO sessions (kind, started_at, ended_at) VALUES ('abyss', '2024-06-17T23:54:00', '2024-06-17T23:54:00');"
+            "INSERT INTO runs (session_id, recorded_at, profit, minutes) VALUES (1, '2024-06-17T23:54:00', 10, 19);"
+        )
+        db.commit()
+        db.close()
+        self.tracker.restore(old)
+        self.assertEqual(self.tracker.totals().profit, 10)
+        self.assertIsNone(self.tracker.session(1).weather)
+        self.tracker.start_session(ABYSS, starting_isk=1, weather="Dark", tier=1)
+
+    def test_restore_refuses_files_that_are_not_trackers(self):
+        self.log(50)
+        junk = self.dir / "junk.db"
+        junk.write_text("not a database at all" * 50)
+        other = self.dir / "other.db"
+        db = sqlite3.connect(other)
+        db.execute("CREATE TABLE unrelated (x)")
+        db.commit()
+        db.close()
+        for bad in (junk, other, self.dir / "missing.db"):
+            with self.assertRaises(ValueError):
+                self.tracker.restore(bad)
+        self.assertEqual(self.tracker.totals().profit, 50)
+        self.assertFalse((self.dir / "missing.db").exists())
+
+
+if __name__ == "__main__":
+    unittest.main()

README.md

@@ -35,7 +35,7 @@ This program is designed to help users track their profits and time spent during
     with the session's profit, ISK/hour, time and wallet below.
   - **History page:** net profit, ISK/hour and winning sessions, a profit-per-session chart with hover details,
     ISK/hour for each weather (and the best one), and every past session with its weather and tier,
-    filterable by Abyss/Incursion. Old `Sessions.txt` files can be imported.
+    filterable by Abyss/Incursion. Old `Sessions.txt` files can be imported, the current filter can be exported to CSV, and `tracker.db` can be backed up and restored with a click.
   - **Pocket mode:** a small always-on-top window to keep beside the EVE client, with the timer, the wallet (or
     payout) field and Log button, the session's profit, ISK/hour and runs, and the last three runs. It shares the
     timer and session with the full window; **Unpin** lets other windows cover it and **Expand** brings the full