Merge pull request 'AI-2.0' (#6) from AI-2.0 into main

Reviewed-on: #6
This commit is contained in:
Don Harper 2026-06-14 00:25:28 -05:00
commit 15b7669fc2
12 changed files with 559 additions and 147 deletions

168
PROJECT_DOCUMENTATION.md Normal file
View file

@ -0,0 +1,168 @@
# Pen Tracker Project Documentation
## Overview
`pen-tracker` is a small Python package for tracking fountain pens and inks using CSV storage. It provides a single command-line interface (`pen-tracker`) for interactive data management, plus direct CLI commands for adding, listing, exporting, and maintaining the collection.
This repository is structured as a standard Python package with source under `src/pen_tracker`, package metadata in `pyproject.toml`, and a small test suite under `tests`.
## Key goals
- Maintain a pen collection with fields such as make, model, nib, ink, purchase date, and notes.
- Maintain an ink collection separately with vendor, name, color, purchase date, size, and notes.
- Store data in CSV files for compatibility and simplicity.
- Use a single CLI entry point for developer and automation workflows.
## Repository layout
- `pyproject.toml` — packaging metadata, dependencies, and entry points.
- `README.md` — user-facing usage documentation.
- `PROJECT_DOCUMENTATION.md` — developer/AI reference documentation.
- `import_csv_to_sqlite.py` — standalone CSV-to-SQLite migration helper.
- `src/pen_tracker/engine.py` — domain logic, data models, SQLite persistence, history tracking, and storage defaults.
- `src/pen_tracker/cli.py` — command-line interface, interactive menus, and command parsing.
- `tests/test_engine.py` — basic test coverage for core engine behavior.
## Core components
### `src/pen_tracker/engine.py`
Primary responsibilities:
- Define `Pen` and `Ink` dataclasses.
- Provide `PenTracker` and `InkTracker` classes that:
- Decide storage path from environment or default XDG data directories.
- Load CSV files into Python objects.
- Normalize data on load.
- Persist sorted CSV output.
Important details:
- Default pen storage path: `~/.local/share/pen-tracker/pens.csv`
- Default ink storage path: `~/.local/share/pen-tracker/inks.csv`
- Environment overrides:
- `PEN_TRACKER_CSV` for pen storage.
- `INK_TRACKER_CSV` for ink storage.
- `PenTracker.load_data()` maps CSV headers like `Date-Purchased` and `Current-Ink` to dataclass fields like `Date_Purchased` and `Current_Ink`.
- CSV headers are written using the original hyphenated naming convention.
### `src/pen_tracker/cli.py`
Primary responsibilities:
- Parse command-line arguments with `argparse`.
- Provide commands for:
- `add` — add a new pen from CLI inputs.
- `list` — display all pens.
- `export` — export pens to JSON.
- `add-ink` — add a new ink.
- `list-inks` — list saved inks.
- Provide an interactive text menu when no command is supplied.
- Display one-based IDs for pens and inks.
- Validate user input and dates.
## Usage patterns
### Installation
```bash
pip install .
```
### Run interactive CLI
```bash
pen-tracker
```
### Add a pen via CLI
```bash
pen-tracker add --make "Pilot" --model "Metropolitan" --nib "F"
```
### List saved pens
```bash
pen-tracker list
```
### Export pens to JSON
```bash
pen-tracker export --output my_pens.json
```
### Add an ink via CLI
```bash
pen-tracker add-ink --vendor "Diamine" --name "Majestic Blue" --color "Blue"
```
### List inks
```bash
pen-tracker list-inks
```
## Data model
### Pen fields
- `Make`
- `Model`
- `Date_Purchased` / `Date-Purchased`
- `Vendor`
- `Nib`
- `Nib_Material` / `Nib-Material`
- `Body`
- `Cap`
- `Post`
- `Current_Ink` / `Current-Ink`
- `Inked_date` / `Inked-date`
- `Notes`
### Ink fields
- `Vendor`
- `Name`
- `Color`
- `Purchased`
- `Size`
- `Notes`
## Packaging and scripts
The package is configured with `pyproject.toml` and uses `setuptools.build_meta` as the build backend. There are no runtime dependencies.
Entry point:
- `pen-tracker = pen_tracker.cli:main`
## Developer notes
- The CLI is the only supported interface. The repository no longer includes any TUI or graphical code.
- Data is now persisted in SQLite, with default path `~/.local/share/pen-tracker/pen_tracker.db`.
- Ink changes for each pen are recorded in `pen_ink_history`, including transitions to `N/A`.
- When modifying storage columns, update both `engine.py` and `cli.py` to maintain field mappings and database behavior.
- One-based IDs are used consistently for user-facing selection and display.
- SQLite persistence always sorts data before saving.
## Testing
- `tests/test_engine.py` contains test coverage for the core engine layer.
- Add tests for new fields, storage behavior, and CLI data validation if extending features.
## Notes for AI agents
- Focus on `src/pen_tracker/engine.py` for the data model and persistence logic.
- Focus on `src/pen_tracker/cli.py` for user-facing behavior, commands, and interactive menu flow.
- The default storage paths are derived from XDG conventions and environment variables.
- There is no GUI, web service, or database; all data is stored in CSV files.
- The project version is `0.5.0` and the package name is `pen-tracker`.
## Extension suggestions
- Add unit tests for the CLI parser and interactive menu flows.
- Add a dedicated `docs/` folder if more formal design docs or API references are needed.
- Add JSON schema validation or stronger date handling if data robustness is required.

View file

@ -10,9 +10,9 @@ pip install .
## Data Storage
Pen data is stored in `~/.local/share/pen-tracker/pens.csv` by default, following XDG Base Directory specification. The directory is created automatically if it doesn't exist.
Data is stored in a SQLite database by default at `~/.local/share/pen-tracker/pen_tracker.db`, following the XDG Base Directory specification. The directory is created automatically if it doesn't exist.
You can override the location by setting the `PEN_TRACKER_CSV` environment variable.
You can override the location by setting the `PEN_TRACKER_DB` environment variable or using `--db` when running the CLI.
## Usage
@ -25,11 +25,20 @@ pen-tracker
Command-line mode:
```bash
pen-tracker add --make "Pilot" --model "Metropolitan" --nib "F"
pen-tracker --db ~/.local/share/pen-tracker/pen_tracker.db add --make "Pilot" --model "Metropolitan" --nib "F"
pen-tracker list
pen-tracker export --output my_pens.json
```
### Import CSV data
If you are migrating from the old CSV storage format, run:
```bash
python import_csv_to_sqlite.py --db ~/.local/share/pen-tracker/pen_tracker.db
```
The import script will initialize the SQLite database if it does not already exist.
## Features
- Track fountain pens with details like make, model, nib, ink, etc.

Binary file not shown.

84
import_csv_to_sqlite.py Executable file
View file

@ -0,0 +1,84 @@
#!/usr/bin/env python3
import argparse
import csv
import os
import sys
from typing import Dict, List
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'src'))
from pen_tracker.engine import Ink, InkTracker, Pen, PenTracker
def get_default_pens_csv() -> str:
env_path = os.getenv('PEN_TRACKER_CSV')
if env_path:
return env_path
data_home = os.getenv('XDG_DATA_HOME', os.path.expanduser('~/.local/share'))
app_data_dir = os.path.join(data_home, 'pen-tracker')
os.makedirs(app_data_dir, exist_ok=True)
return os.path.join(app_data_dir, 'pens.csv')
def get_default_inks_csv() -> str:
env_path = os.getenv('INK_TRACKER_CSV')
if env_path:
return env_path
data_home = os.getenv('XDG_DATA_HOME', os.path.expanduser('~/.local/share'))
app_data_dir = os.path.join(data_home, 'pen-tracker')
os.makedirs(app_data_dir, exist_ok=True)
return os.path.join(app_data_dir, 'inks.csv')
def load_csv_rows(path: str, fieldnames: List[str]) -> List[Dict[str, str]]:
if not os.path.exists(path):
return []
with open(path, mode='r', encoding='utf-8-sig', newline='') as f:
reader = csv.DictReader(f)
if reader.fieldnames:
reader.fieldnames = [h.strip() for h in reader.fieldnames]
rows: List[Dict[str, str]] = []
for row in reader:
clean_row = {k.strip(): (v.strip() if v else 'N/A') for k, v in row.items() if k is not None}
for field in fieldnames:
clean_row.setdefault(field, 'N/A')
rows.append(clean_row)
return rows
def import_data(db_path: str, pens_csv: str, inks_csv: str) -> None:
ink_tracker = InkTracker(db_path)
pen_tracker = PenTracker(db_path)
ink_rows = load_csv_rows(inks_csv, ink_tracker.headers)
inks = [Ink(**row) for row in ink_rows]
ink_tracker.inks = inks
ink_tracker.save_data()
pen_rows = load_csv_rows(pens_csv, pen_tracker.headers)
pens = [Pen(**{pen_tracker.key_map.get(k, k): v for k, v in row.items()}) for row in pen_rows]
pen_tracker.pens = pens
pen_tracker.save_data()
print(f'Imported {len(ink_tracker.inks)} inks from {inks_csv}')
print(f'Imported {len(pen_tracker.pens)} pens from {pens_csv}')
print(f'SQLite database initialized at: {db_path}')
def main() -> None:
parser = argparse.ArgumentParser(description='Import CSV data into pen-tracker SQLite database')
parser.add_argument('--db', default=None, help='Path to the SQLite database file')
parser.add_argument('--pens-csv', default=None, help='Path to pens CSV file')
parser.add_argument('--inks-csv', default=None, help='Path to inks CSV file')
args = parser.parse_args()
db_path = args.db
pens_csv = args.pens_csv or get_default_pens_csv()
inks_csv = args.inks_csv or get_default_inks_csv()
import_data(db_path, pens_csv, inks_csv)
if __name__ == '__main__':
main()

View file

@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "pen-tracker"
version = "0.5.0"
version = "0.5.1"
authors = [
{ name="Don Harper", email="don@donharper.org" },
]

View file

@ -1,6 +1,6 @@
Metadata-Version: 2.4
Name: pen-tracker
Version: 0.5.0
Version: 0.5.1
Summary: A fountain pen collection tracker.
Author-email: Don Harper <don@donharper.org>
Requires-Python: >=3.8
@ -20,9 +20,9 @@ pip install .
## Data Storage
Pen data is stored in `~/.local/share/pen-tracker/pens.csv` by default, following XDG Base Directory specification. The directory is created automatically if it doesn't exist.
Data is stored in a SQLite database by default at `~/.local/share/pen-tracker/pen_tracker.db`, following the XDG Base Directory specification. The directory is created automatically if it doesn't exist.
You can override the location by setting the `PEN_TRACKER_CSV` environment variable.
You can override the location by setting the `PEN_TRACKER_DB` environment variable or using `--db` when running the CLI.
## Usage
@ -35,11 +35,20 @@ pen-tracker
Command-line mode:
```bash
pen-tracker add --make "Pilot" --model "Metropolitan" --nib "F"
pen-tracker --db ~/.local/share/pen-tracker/pen_tracker.db add --make "Pilot" --model "Metropolitan" --nib "F"
pen-tracker list
pen-tracker export --output my_pens.json
```
### Import CSV data
If you are migrating from the old CSV storage format, run:
```bash
python import_csv_to_sqlite.py --db ~/.local/share/pen-tracker/pen_tracker.db
```
The import script will initialize the SQLite database if it does not already exist.
## Features
- Track fountain pens with details like make, model, nib, ink, etc.

View file

@ -1,5 +1,4 @@
import os
import csv
import argparse
import json
from datetime import datetime
@ -148,6 +147,23 @@ class CLITracker(PenTracker):
field = self.key_map.get(header, header)
value = getattr(pen, field)
print(f"{header:<20}: {value}")
history = []
if pen.id is not None:
history = self.get_ink_history(pen.id)
print(""*45)
print("\nInk Change History:")
if history:
print(f"{'When':<20} | {'Ink':<20}")
print("-" * 45)
for row in history:
changed_at = row.get('changed_at', 'N/A')
ink_name = row.get('ink_name', 'N/A')
print(f"{changed_at:<20} | {ink_name:<20}")
else:
print("No ink history available.")
print(""*45)
input("\nPress Enter to return to menu...")
@ -192,7 +208,7 @@ class CLITracker(PenTracker):
new_ink = Ink(**new_ink_data)
self.ink_tracker.inks.append(new_ink)
self.ink_tracker.save_data()
print("\n[✔] Ink added successfully to inks.csv!")
print("\n[✔] Ink added successfully!")
def view_all_inks(self):
if not self.ink_tracker.inks:
@ -237,7 +253,7 @@ def clear_screen():
def main():
# This is the entry point defined in pyproject.toml
parser = argparse.ArgumentParser(description="Fountain Pen Tracker")
parser.add_argument('--csv', default=None, help='Path to CSV file')
parser.add_argument('--db', default=None, help='Path to SQLite database file')
subparsers = parser.add_subparsers(dest='command', help='Commands')
# Add command
@ -275,7 +291,7 @@ def main():
args = parser.parse_args()
tracker = CLITracker(args.csv)
tracker = CLITracker(args.db)
if args.command == 'add':
# Validate dates
@ -331,7 +347,7 @@ def main():
else:
# Interactive mode
while True:
print("\n🖋️ FOUNTAIN PEN TRACKER (CSV Edition)")
print("\n🖋️ FOUNTAIN PEN TRACKER (SQLite Edition)")
print("1. View Collection Summary")
print("2. Add New Pen")
print("3. Edit a Pen")

View file

@ -1,163 +1,278 @@
import csv
import os
import sqlite3
import logging
from dataclasses import dataclass, asdict
from typing import List
from dataclasses import dataclass
from typing import Any, Dict, List, Optional, Sequence, Set, Tuple
logger = logging.getLogger(__name__)
@dataclass
class Ink:
Vendor: str = "N/A"
Name: str = "N/A"
Color: str = "N/A"
Purchased: str = "N/A"
Size: str = "N/A"
Notes: str = "N/A"
@dataclass
class Pen:
Make: str = "N/A"
Model: str = "N/A"
Date_Purchased: str = "N/A"
Vendor: str = "N/A"
Nib: str = "N/A"
Nib_Material: str = "N/A"
Body: str = "N/A"
Cap: str = "N/A"
Post: str = "N/A"
Current_Ink: str = "N/A"
Inked_date: str = "N/A"
Notes: str = "N/A"
class InkTracker:
def __init__(self, storage_file: str = None):
if storage_file is None:
storage_file = os.getenv('INK_TRACKER_CSV')
if storage_file is None:
def get_default_db_path(db_path: Optional[str] = None) -> str:
if db_path:
return db_path
env_path = os.getenv('PEN_TRACKER_DB')
if env_path:
return env_path
data_home = os.getenv('XDG_DATA_HOME', os.path.expanduser('~/.local/share'))
app_data_dir = os.path.join(data_home, 'pen-tracker')
os.makedirs(app_data_dir, exist_ok=True)
storage_file = os.path.join(app_data_dir, 'inks.csv')
self.storage_file = storage_file
return os.path.join(app_data_dir, 'pen_tracker.db')
class SqliteStore:
def __init__(self, db_path: Optional[str] = None):
self.db_path = get_default_db_path(db_path)
self.conn = sqlite3.connect(self.db_path)
self.conn.row_factory = sqlite3.Row
self.conn.execute('PRAGMA foreign_keys = ON')
self.initialize_db()
def initialize_db(self):
self.conn.execute(
'''CREATE TABLE IF NOT EXISTS inks (
id INTEGER PRIMARY KEY,
vendor TEXT NOT NULL,
name TEXT NOT NULL,
color TEXT NOT NULL,
purchased TEXT NOT NULL,
size TEXT NOT NULL,
notes TEXT NOT NULL
)'''
)
self.conn.execute(
'''CREATE TABLE IF NOT EXISTS pens (
id INTEGER PRIMARY KEY,
make TEXT NOT NULL,
model TEXT NOT NULL,
date_purchased TEXT NOT NULL,
vendor TEXT NOT NULL,
nib TEXT NOT NULL,
nib_material TEXT NOT NULL,
body TEXT NOT NULL,
cap TEXT NOT NULL,
post TEXT NOT NULL,
current_ink TEXT NOT NULL,
inked_date TEXT NOT NULL,
notes TEXT NOT NULL
)'''
)
self.conn.execute(
'''CREATE TABLE IF NOT EXISTS pen_ink_history (
id INTEGER PRIMARY KEY,
pen_id INTEGER NOT NULL,
ink_name TEXT NOT NULL,
changed_at TEXT NOT NULL DEFAULT (datetime('now')),
FOREIGN KEY(pen_id) REFERENCES pens(id) ON DELETE CASCADE
)'''
)
self.conn.commit()
def execute(self, query: str, params: Tuple[Any, ...] = ()):
return self.conn.execute(query, params)
def executemany(self, query: str, seq_of_params: Sequence[Tuple[Any, ...]]):
return self.conn.executemany(query, seq_of_params)
def fetchall(self, query: str, params: Tuple[Any, ...] = ()) -> List[Dict[str, Any]]:
cursor = self.conn.execute(query, params)
return [dict(row) for row in cursor.fetchall()]
def commit(self):
self.conn.commit()
def close(self):
self.conn.close()
@dataclass
class Ink:
Vendor: str = 'N/A'
Name: str = 'N/A'
Color: str = 'N/A'
Purchased: str = 'N/A'
Size: str = 'N/A'
Notes: str = 'N/A'
id: Optional[int] = None
@dataclass
class Pen:
Make: str = 'N/A'
Model: str = 'N/A'
Date_Purchased: str = 'N/A'
Vendor: str = 'N/A'
Nib: str = 'N/A'
Nib_Material: str = 'N/A'
Body: str = 'N/A'
Cap: str = 'N/A'
Post: str = 'N/A'
Current_Ink: str = 'N/A'
Inked_date: str = 'N/A'
Notes: str = 'N/A'
id: Optional[int] = None
class InkTracker:
def __init__(self, db_path: Optional[str] = None):
self.db = SqliteStore(db_path)
self.headers = ['Vendor', 'Name', 'Color', 'Purchased', 'Size', 'Notes']
self.inks: List[Ink] = self.load_data()
def _sort_inks(self):
"""Sorts the inks list by Vendor, then by Name alphabetically."""
self.inks.sort(key=lambda x: (x.Vendor.lower(), x.Name.lower()))
def load_data(self) -> List[Ink]:
"""Loads data from the CSV file."""
if not os.path.exists(self.storage_file):
self._create_empty_csv()
return []
inks = []
try:
with open(self.storage_file, mode='r', encoding='utf-8-sig') as f:
reader = csv.DictReader(f)
if reader.fieldnames:
reader.fieldnames = [h.strip() for h in reader.fieldnames]
for row in reader:
if None in row:
extras = row.pop(None)
if extras:
extra_text = ",".join(extras).strip()
if extra_text:
row['Notes'] = (row.get('Notes') or '')
if row['Notes']:
row['Notes'] += ", "
row['Notes'] += extra_text
clean_row = {
k.strip(): (v.strip() if v else "N/A")
for k, v in row.items() if k is not None
}
inks.append(Ink(**clean_row))
inks.sort(key=lambda x: (x.Vendor.lower(), x.Name.lower()))
except Exception as e:
logger.error(f"Error loading inks CSV: {e}")
rows = self.db.fetchall(
'SELECT * FROM inks ORDER BY vendor COLLATE NOCASE, name COLLATE NOCASE'
)
inks: List[Ink] = []
for row in rows:
inks.append(
Ink(
Vendor=row['vendor'],
Name=row['name'],
Color=row['color'],
Purchased=row['purchased'],
Size=row['size'],
Notes=row['notes'],
id=row['id'],
)
)
return inks
def _create_empty_csv(self):
"""Creates the CSV file with headers if it is missing."""
with open(self.storage_file, mode='w', newline='', encoding='utf-8') as f:
writer = csv.DictWriter(f, fieldnames=self.headers)
writer.writeheader()
def save_data(self):
"""Sorts the data before writing to ensure persistent alphabetical order."""
self._sort_inks()
with open(self.storage_file, mode='w', newline='', encoding='utf-8') as f:
writer = csv.DictWriter(f, fieldnames=self.headers)
writer.writeheader()
self.db.execute('DELETE FROM inks')
for ink in self.inks:
writer.writerow(asdict(ink))
cursor = self.db.execute(
'INSERT INTO inks (vendor, name, color, purchased, size, notes) VALUES (?, ?, ?, ?, ?, ?)',
(ink.Vendor, ink.Name, ink.Color, ink.Purchased, ink.Size, ink.Notes),
)
ink.id = cursor.lastrowid
self.db.commit()
self.inks = self.load_data()
class PenTracker:
def __init__(self, storage_file: str = None):
if storage_file is None:
storage_file = os.getenv('PEN_TRACKER_CSV')
if storage_file is None:
data_home = os.getenv('XDG_DATA_HOME', os.path.expanduser('~/.local/share'))
app_data_dir = os.path.join(data_home, 'pen-tracker')
os.makedirs(app_data_dir, exist_ok=True)
storage_file = os.path.join(app_data_dir, 'pens.csv')
self.storage_file = storage_file
def __init__(self, db_path: Optional[str] = None):
self.db = SqliteStore(db_path)
self.headers = [
'Make', 'Model', 'Date-Purchased', 'Vendor', 'Nib',
'Nib-Material', 'Body', 'Cap', 'Post',
'Current-Ink', 'Inked-date', 'Notes'
'Current-Ink', 'Inked-date', 'Notes',
]
self.key_map = {
'Date-Purchased': 'Date_Purchased',
'Inked-date': 'Inked_date',
'Nib-Material': 'Nib_Material',
'Current-Ink': 'Current_Ink'
'Current-Ink': 'Current_Ink',
}
self.reverse_key_map = {v: k for k, v in self.key_map.items()}
self.pens: List[Pen] = self.load_data()
def _sort_pens(self):
"""Sorts the pens list by Make, then by Model alphabetically."""
self.pens.sort(key=lambda x: (x.Make.lower(), x.Model.lower()))
def load_data(self) -> List[Pen]:
"""Loads data from the CSV file."""
if not os.path.exists(self.storage_file):
self._create_empty_csv()
return []
pens = []
try:
# Standard, clean way to open the file
with open(self.storage_file, mode='r', encoding='utf-8-sig') as f:
reader = csv.DictReader(f)
for row in reader:
# Strip whitespace from keys and values, handle empty values
clean_row = {k.strip(): (v.strip() if v else "N/A") for k, v in row.items()}
# Map keys to dataclass field names
mapped_row = {self.key_map.get(k, k): v for k, v in clean_row.items()}
pens.append(Pen(**mapped_row))
pens.sort(key=lambda x: (x.Make.lower(), x.Model.lower()))
except Exception as e:
logger.error(f"Error loading CSV: {e}")
rows = self.db.fetchall(
'SELECT * FROM pens ORDER BY make COLLATE NOCASE, model COLLATE NOCASE'
)
pens: List[Pen] = []
for row in rows:
pens.append(
Pen(
Make=row['make'],
Model=row['model'],
Date_Purchased=row['date_purchased'],
Vendor=row['vendor'],
Nib=row['nib'],
Nib_Material=row['nib_material'],
Body=row['body'],
Cap=row['cap'],
Post=row['post'],
Current_Ink=row['current_ink'],
Inked_date=row['inked_date'],
Notes=row['notes'],
id=row['id'],
)
)
return pens
def _create_empty_csv(self):
"""Creates the CSV file with headers if it is missing."""
with open(self.storage_file, mode='w', newline='', encoding='utf-8') as f:
writer = csv.DictWriter(f, fieldnames=self.headers)
writer.writeheader()
def _insert_pen(self, pen: Pen) -> int:
cursor = self.db.execute(
'INSERT INTO pens (make, model, date_purchased, vendor, nib, nib_material, body, cap, post, current_ink, inked_date, notes) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)',
(
pen.Make,
pen.Model,
pen.Date_Purchased,
pen.Vendor,
pen.Nib,
pen.Nib_Material,
pen.Body,
pen.Cap,
pen.Post,
pen.Current_Ink,
pen.Inked_date,
pen.Notes,
),
)
row_id = cursor.lastrowid
return int(row_id) if row_id is not None else 0
def _update_pen(self, pen: Pen):
self.db.execute(
'UPDATE pens SET make = ?, model = ?, date_purchased = ?, vendor = ?, nib = ?, nib_material = ?, body = ?, cap = ?, post = ?, current_ink = ?, inked_date = ?, notes = ? WHERE id = ?',
(
pen.Make,
pen.Model,
pen.Date_Purchased,
pen.Vendor,
pen.Nib,
pen.Nib_Material,
pen.Body,
pen.Cap,
pen.Post,
pen.Current_Ink,
pen.Inked_date,
pen.Notes,
pen.id,
),
)
def _record_history(self, pen_id: int, ink_name: str):
self.db.execute(
'INSERT INTO pen_ink_history (pen_id, ink_name) VALUES (?, ?)',
(pen_id, ink_name),
)
def get_ink_history(self, pen_id: int) -> List[Dict[str, str]]:
return self.db.fetchall(
'SELECT * FROM pen_ink_history WHERE pen_id = ? ORDER BY changed_at',
(pen_id,),
)
def save_data(self):
"""Sorts the data before writing to ensure persistent alphabetical order."""
self._sort_pens()
with open(self.storage_file, mode='w', newline='', encoding='utf-8') as f:
writer = csv.DictWriter(f, fieldnames=self.headers)
writer.writeheader()
existing_rows = {row['id']: row for row in self.db.fetchall('SELECT * FROM pens')}
seen_ids: Set[int] = set()
for pen in self.pens:
row = asdict(pen)
# Map back to CSV keys
csv_row = {k.replace('_', '-'): v for k, v in row.items()}
writer.writerow(csv_row)
if pen.id is None:
pen.id = self._insert_pen(pen)
self._record_history(pen.id, pen.Current_Ink)
else:
old_row = existing_rows.pop(pen.id, None)
if old_row is None:
pen.id = self._insert_pen(pen)
self._record_history(pen.id, pen.Current_Ink)
else:
if old_row['current_ink'] != pen.Current_Ink:
self._record_history(pen.id, pen.Current_Ink)
self._update_pen(pen)
seen_ids.add(pen.id)
for orphan_id in existing_rows.keys():
self.db.execute('DELETE FROM pens WHERE id = ?', (orphan_id,))
self.db.commit()
self.pens = self.load_data()

View file

@ -1,11 +1,11 @@
import unittest
import tempfile
import os
from pen_tracker.engine import PenTracker, Pen, InkTracker, Ink
import tempfile
import unittest
from pen_tracker.engine import Ink, InkTracker, Pen, PenTracker
class TestPenTracker(unittest.TestCase):
def setUp(self):
self.temp_file = tempfile.NamedTemporaryFile(delete=False, suffix='.csv')
self.temp_file = tempfile.NamedTemporaryFile(delete=False, suffix='.db')
self.temp_file.close()
self.tracker = PenTracker(self.temp_file.name)
@ -33,11 +33,28 @@ class TestPenTracker(unittest.TestCase):
self.assertEqual(new_tracker.pens[0].Make, "A")
self.assertEqual(new_tracker.pens[1].Make, "Z")
def test_pen_ink_history_records_changes(self):
pen = Pen(Make="Pilot", Model="Metropolitan", Nib="F", Current_Ink="Blue")
self.tracker.pens.append(pen)
self.tracker.save_data()
persisted = self.tracker.pens[0]
history = self.tracker.get_ink_history(persisted.id)
self.assertEqual(len(history), 1)
self.assertEqual(history[0]['ink_name'], "Blue")
persisted.Current_Ink = "N/A"
self.tracker.save_data()
history = self.tracker.get_ink_history(persisted.id)
self.assertEqual(len(history), 2)
self.assertEqual(history[-1]['ink_name'], "N/A")
class TestInkTracker(unittest.TestCase):
def setUp(self):
self.temp_file = tempfile.NamedTemporaryFile(delete=False, suffix='.csv')
self.temp_file = tempfile.NamedTemporaryFile(delete=False, suffix='.db')
self.temp_file.close()
self.tracker = InkTracker(self.temp_file.name)
@ -64,20 +81,14 @@ class TestInkTracker(unittest.TestCase):
self.assertEqual(new_tracker.inks[0].Vendor, "A")
self.assertEqual(new_tracker.inks[1].Vendor, "Z")
def test_loads_row_with_extra_trailing_field(self):
csv_content = (
"Vendor,Name,Color,Purchased,Size,Notes\n"
"Waterman,Intense Black,Black,,\"Cartridge,International Short\",\n"
"Pilot,Black Cartridge,Black,2024-01-01,Cartridge,Good ink\n"
)
with open(self.temp_file.name, 'w', encoding='utf-8') as f:
f.write(csv_content)
def test_save_and_load_ink(self):
self.tracker.inks.append(Ink(Vendor='Waterman', Name='Intense Black', Color='Black', Notes='Good ink'))
self.tracker.save_data()
new_tracker = InkTracker(self.temp_file.name)
self.assertEqual(len(new_tracker.inks), 2)
self.assertEqual(new_tracker.inks[0].Vendor, "Pilot")
self.assertEqual(new_tracker.inks[1].Vendor, "Waterman")
self.assertEqual(new_tracker.inks[1].Notes, "N/A")
self.assertEqual(len(new_tracker.inks), 1)
self.assertEqual(new_tracker.inks[0].Vendor, 'Waterman')
self.assertEqual(new_tracker.inks[0].Name, 'Intense Black')
if __name__ == '__main__':
unittest.main()