Ethanfel/8-cut
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: DB schema v2 — store start_time and output_path, add get_markers

Browse Source 
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Ethanfel
2026-04-06 13:08:49 +02:00
parent c34f3b4039
commit 63d48a968c
2 changed files with 94 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main.py
+46 -12
View File
@@ -60,31 +60,50 @@ def _normalize_filename(filename: str) -> str:
class ProcessedDB:
_SCHEMA_VERSION = 2 # bump when schema changes
def __init__(self, db_path: str | None = None):
if db_path is None:
db_path = str(Path.home() / ".8cut.db")
try:
self._con = sqlite3.connect(db_path)
self._con.execute(
"CREATE TABLE IF NOT EXISTS processed "
"(filename TEXT NOT NULL UNIQUE, processed_at TEXT NOT NULL)"
)
self._con.execute(
"CREATE INDEX IF NOT EXISTS idx_filename ON processed(filename)"
)
self._con.commit()
self._migrate()
self._enabled = True
except Exception as e:
print(f"8-cut: DB unavailable: {e}", file=sys.stderr)
self._con = None
self._enabled = False
def add(self, filename: str) -> None:
def _migrate(self) -> None:
"""Create or recreate table if schema is outdated."""
cols = {
row[1]
for row in self._con.execute("PRAGMA table_info(processed)").fetchall()
}
needs_recreate = "start_time" not in cols or "output_path" not in cols
if needs_recreate:
self._con.execute("DROP TABLE IF EXISTS processed")
self._con.execute(
"CREATE TABLE IF NOT EXISTS processed ("
" id INTEGER PRIMARY KEY AUTOINCREMENT,"
" filename TEXT NOT NULL,"
" start_time REAL NOT NULL,"
" output_path TEXT NOT NULL,"
" processed_at TEXT NOT NULL"
")"
)
self._con.execute(
"CREATE INDEX IF NOT EXISTS idx_filename ON processed(filename)"
)
self._con.commit()
def add(self, filename: str, start_time: float, output_path: str) -> None:
if not self._enabled:
return
self._con.execute(
"INSERT OR REPLACE INTO processed (filename, processed_at) VALUES (?, ?)",
(filename, datetime.now(timezone.utc).isoformat()),
"INSERT INTO processed (filename, start_time, output_path, processed_at)"
" VALUES (?, ?, ?, ?)",
(filename, start_time, output_path, datetime.now(timezone.utc).isoformat()),
)
self._con.commit()
@@ -104,6 +123,21 @@ class ProcessedDB:
best_ratio, best_match = ratio, stored
return best_match
def get_markers(self, filename: str) -> list[tuple[float, int, str]]:
"""Return [(start_time, marker_number, output_path), ...] for the best
fuzzy match of filename, sorted by start_time. Empty list if no match."""
if not self._enabled:
return []
match = self.find_similar(filename)
if match is None:
return []
rows = self._con.execute(
"SELECT start_time, output_path FROM processed"
" WHERE filename = ? ORDER BY start_time",
(match,),
).fetchall()
return [(t, i + 1, p) for i, (t, p) in enumerate(rows)]
class ExportWorker(QThread):
finished = pyqtSignal(str) # output path
@@ -518,7 +552,7 @@ class MainWindow(QMainWindow):
self._export_worker.start()
def _on_export_done(self, path: str):
self._db.add(os.path.basename(self._file_path))
self._db.add(os.path.basename(self._file_path), self._cursor, path)
self._export_counter += 1
self._update_next_label()
self._btn_export.setEnabled(True)
tests/test_utils.py
+48 -5
View File
@@ -64,7 +64,7 @@ def test_db_add_and_find_exact():
path = f.name
try:
db = ProcessedDB(path)
db.add("video.mp4")
db.add("video.mp4", 12.5, "/out/clip_001.mp4")
assert db.find_similar("video.mp4") == "video.mp4"
finally:
os.unlink(path)
@@ -74,7 +74,7 @@ def test_db_find_similar_resolution_variant():
path = f.name
try:
db = ProcessedDB(path)
db.add("episode_s01e01_2160p.mkv")
db.add("episode_s01e01_2160p.mkv", 0.0, "/out/ep_001.mp4")
assert db.find_similar("episode_s01e01_1080p.mkv") == "episode_s01e01_2160p.mkv"
finally:
os.unlink(path)
@@ -84,12 +84,55 @@ def test_db_find_similar_no_match():
path = f.name
try:
db = ProcessedDB(path)
db.add("alpha.mp4")
db.add("alpha.mp4", 0.0, "/out/alpha_001.mp4")
assert db.find_similar("completely_different_zzzz.mp4") is None
finally:
os.unlink(path)
def test_db_disabled_survives_bad_path():
db = ProcessedDB("/no/such/directory/8cut.db")
db.add("x.mp4") # must not raise
assert db.find_similar("x.mp4") is None # gracefully returns None
db.add("x.mp4", 0.0, "/out/x_001.mp4") # must not raise
assert db.find_similar("x.mp4") is None
def test_db_get_markers_returns_sorted():
with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
path = f.name
try:
db = ProcessedDB(path)
db.add("video.mp4", 30.0, "/out/clip_002.mp4")
db.add("video.mp4", 10.0, "/out/clip_001.mp4")
db.add("video.mp4", 50.0, "/out/clip_003.mp4")
markers = db.get_markers("video.mp4")
assert len(markers) == 3
assert markers[0] == (10.0, 1, "/out/clip_001.mp4")
assert markers[1] == (30.0, 2, "/out/clip_002.mp4")
assert markers[2] == (50.0, 3, "/out/clip_003.mp4")
finally:
os.unlink(path)
def test_db_get_markers_fuzzy_match():
with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
path = f.name
try:
db = ProcessedDB(path)
db.add("show_2160p.mkv", 5.0, "/out/s_001.mp4")
markers = db.get_markers("show_1080p.mkv")
assert len(markers) == 1
assert markers[0][0] == 5.0
assert markers[0][2] == "/out/s_001.mp4"
finally:
os.unlink(path)
def test_db_get_markers_no_match():
with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
path = f.name
try:
db = ProcessedDB(path)
markers = db.get_markers("nothing.mp4")
assert markers == []
finally:
os.unlink(path)
def test_db_get_markers_disabled():
db = ProcessedDB("/no/such/directory/8cut.db")
assert db.get_markers("x.mp4") == []