48 lines
1.3 KiB
Python
48 lines
1.3 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Iterator
|
|
|
|
from .constants import MUSIC_EXTENSIONS
|
|
from .db import connect
|
|
from .hashing import fast_hash
|
|
|
|
|
|
def iter_music(root: Path) -> Iterator[Path]:
|
|
for path in root.rglob("*"):
|
|
if not path.is_file() or path.is_symlink():
|
|
continue
|
|
if path.name.startswith(".") or ".dedupe" in path.parts:
|
|
continue
|
|
if path.suffix.lower() in MUSIC_EXTENSIONS:
|
|
yield path
|
|
|
|
|
|
def index_file(conn, path: Path) -> None:
|
|
st = path.stat()
|
|
fh = fast_hash(path)
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO audio_tracks(file_path, file_size, fast_hash, inode, device, status)
|
|
VALUES (?, ?, ?, ?, ?, 'active')
|
|
ON CONFLICT(file_path) DO UPDATE SET
|
|
file_size=excluded.file_size,
|
|
fast_hash=excluded.fast_hash,
|
|
inode=excluded.inode,
|
|
device=excluded.device,
|
|
updated_at=CURRENT_TIMESTAMP,
|
|
status='active'
|
|
""",
|
|
(str(path), st.st_size, fh, st.st_ino, st.st_dev),
|
|
)
|
|
|
|
|
|
def scan(root: Path, db_path: Path) -> int:
|
|
count = 0
|
|
with connect(db_path) as conn:
|
|
for path in iter_music(root):
|
|
index_file(conn, path)
|
|
count += 1
|
|
return count
|