Merge pull request #398 from arabcoders/dev

Re-implemented Archive feature

.vscode/settings.json

@@ -16,6 +16,7 @@
     "ahash",
     "aiocron",
     "anyio",
+    "Archiver",
     "arrowless",
     "attl",
     "autonumber",

app/library/Archiver.py

@@ -0,0 +1,377 @@
+import logging
+import os
+import threading
+import time
+from pathlib import Path
+
+from app.library.Singleton import ThreadSafe
+
+LOG: logging.Logger = logging.getLogger("Archiver")
+
+
+class _Entry:
+    """
+    Internal cache entry for a single archive file.
+
+    Attributes:
+        ids (set[str]): Cached IDs contained in the archive file.
+        size (int): Last known file size from os.stat.
+        mtime (float): Last known modification time from os.stat.
+        loaded (bool): Whether the entry has been loaded from disk at least once.
+
+    """
+
+    def __init__(self) -> None:
+        self.ids: set[str] = set()
+        self.size: int = -1
+        self.mtime: float = -1.0
+        self.loaded: bool = False
+
+
+class Archiver(metaclass=ThreadSafe):
+    """
+    Global archive cache for yt-dlp download archive files.
+
+    Caches IDs per file in memory for fast membership checks. When read stat
+    checks are enabled, the cache is refreshed on read if the file's size or
+    mtime changed. The add and delete operations write to disk and then update
+    the in-memory cache and metadata accordingly.
+    """
+
+    def __init__(self) -> None:
+        if getattr(self, "_initialized", False):
+            return
+
+        self._cache: dict[str, _Entry] = {}
+        self._locks: dict[str, threading.RLock] = {}
+        self._global_lock = threading.RLock()
+        self._stats_check: bool = False
+        self._initialized = True
+
+    @classmethod
+    def get_instance(cls) -> "Archiver":
+        return cls()
+
+    def _key(self, file: str | Path) -> str:
+        """
+        Return a normalized absolute path key for the archive file.
+
+        Args:
+            file (str|Path): The archive file path.
+
+        Returns:
+            str: The absolute, resolved file path to be used as cache key.
+
+        """
+        p: Path = Path(file) if not isinstance(file, Path) else file
+        return str(p.resolve())
+
+    def _get_lock(self, key: str) -> threading.RLock:
+        """
+        Get or create a per-file lock.
+
+        Args:
+            key (str): The normalized file key as returned by _key().
+
+        Returns:
+            threading.RLock: The lock for this key.
+
+        """
+        with self._global_lock:
+            if key not in self._locks:
+                self._locks[key] = threading.RLock()
+            return self._locks[key]
+
+    def invalidate(self, file: str | Path) -> None:
+        """
+        Drop any cached entry for the given archive file.
+
+        Args:
+            file (str|Path): The archive file path.
+
+        """
+        key: str = self._key(file)
+        with self._global_lock:
+            self._cache.pop(key, None)
+
+    def _stat(self, key: str) -> os.stat_result | None:
+        """
+        Safely stat a file path key.
+
+        Args:
+            key (str): The normalized file key.
+
+        Returns:
+            os.stat_result|None: The stat result, or None on error.
+
+        """
+        try:
+            return os.stat(key)
+        except OSError:
+            return None
+
+    def _ensure_loaded(self, key: str) -> _Entry:
+        """
+        Ensure a cache entry is present and up to date.
+
+        When read stat checks are enabled, the file is reloaded if its size or
+        modification time differs from the last cached values. Otherwise the
+        existing cache entry is used.
+
+        Args:
+            key (str): The normalized file key.
+
+        Returns:
+            _Entry: The cache entry for the file.
+
+        """
+        entry: _Entry | None = self._cache.get(key)
+        if entry and entry.loaded:
+            if not self._stats_check:
+                return entry
+
+            st: os.stat_result | None = self._stat(key)
+
+            if not st:
+                self._cache[key] = _Entry()
+                return self._cache[key]
+
+            if entry.size == st.st_size and entry.mtime == st.st_mtime:
+                return entry
+
+        lock: threading.RLock = self._get_lock(key)
+        with lock:
+            entry = self._cache.get(key) or _Entry()
+            st = self._stat(key) if self._stats_check else None
+            if self._stats_check and not st:
+                entry.ids = set()
+                entry.size = -1
+                entry.mtime = -1
+                entry.loaded = True
+                self._cache[key] = entry
+                return entry
+
+            if self._stats_check and st and entry.loaded and entry.size == st.st_size and entry.mtime == st.st_mtime:
+                return entry
+
+            start: float = time.perf_counter()
+            ids: set[str] = set()
+            try:
+                with open(key, encoding="utf-8") as f:
+                    for line in f:
+                        s = line.strip()
+                        if not s or len(s.split()) < 2:
+                            continue
+                        ids.add(s)
+            except OSError as e:
+                LOG.error(f"Failed to read archive file '{key}': {e!s}")
+                ids = set()
+
+            try:
+                elapsed_ms: float = (time.perf_counter() - start) * 1000.0
+                LOG.debug(f"_ensure_loaded took {elapsed_ms:.2f}ms (loaded={len(ids)})")
+            except Exception:
+                pass
+
+            entry.ids = ids
+            if st:
+                entry.size = st.st_size
+                entry.mtime = st.st_mtime
+            entry.loaded = True
+            self._cache[key] = entry
+            return entry
+
+    def read(self, file: str | Path, ids: list[str] | None = None) -> list[str]:
+        """
+        Read IDs from the archive cache, loading once if needed.
+
+        Args:
+            file (str|Path): The archive file path.
+            ids (list[str]|None): Optional IDs to filter by; when None or empty,
+                all cached IDs are returned (order not guaranteed).
+
+        Returns:
+            list[str]: The IDs present in the archive, optionally filtered.
+
+        """
+        if not file:
+            return []
+
+        key: str = self._key(file)
+        entry: _Entry = self._ensure_loaded(key)
+
+        if not ids:
+            return list(entry.ids)
+
+        ids_set: set[str] = {s.strip() for s in ids if str(s).strip() and len(str(s).strip().split()) >= 2}
+        if not ids_set:
+            return []
+
+        return [s for s in (str(x).strip() for x in ids) if s and len(s.split()) >= 2 and s in entry.ids]
+
+    def has(self, file: str | Path) -> bool:
+        """
+        Check if the archive contains any IDs.
+
+        Args:
+            file (str|Path): The archive file path.
+
+        Returns:
+            bool: True if the archive has at least one ID, False otherwise.
+
+        """
+        if not file:
+            return False
+
+        key: str = self._key(file)
+        entry: _Entry = self._ensure_loaded(key)
+        return bool(entry.ids)
+
+    def add(self, file: str | Path, ids: list[str], skip_check: bool = False) -> bool:
+        """
+        Append IDs to an archive and update the cache.
+
+        Args:
+            file (str|Path): The archive file path.
+            ids (list[str]): IDs to append; invalid or duplicate IDs are ignored.
+            skip_check (bool): If True, do not check for existing IDs in the cache.
+
+        Returns:
+            bool: True if any new IDs were appended, False otherwise.
+
+        """
+        if not file or not ids:
+            return False
+
+        key: str = self._key(file)
+        lock: threading.RLock = self._get_lock(key)
+        with lock:
+            entry: _Entry = self._ensure_loaded(key)
+
+            new_ids: list[str] = []
+            for raw in ids:
+                s: str = str(raw).strip()
+                if not s or len(s.split()) < 2:
+                    continue
+                if not skip_check and s in entry.ids:
+                    continue
+                if s in new_ids:
+                    continue
+                new_ids.append(s)
+
+            if not new_ids:
+                return False
+
+            path = Path(key)
+            try:
+                if not path.parent.exists():
+                    path.parent.mkdir(parents=True, exist_ok=True)
+
+                try:
+                    from yt_dlp.utils import locked_file
+
+                    with locked_file(str(path), "a", encoding="utf-8") as f:
+                        f.write("".join(f"{x}\n" for x in new_ids))
+                except Exception:
+                    with path.open("a", encoding="utf-8") as f:
+                        f.write("".join(f"{x}\n" for x in new_ids))
+
+            except OSError as e:
+                LOG.error(f"Failed to write to archive file '{key}': {e!s}")
+                return False
+
+            entry.ids.update(new_ids)
+            st: os.stat_result | None = self._stat(key)
+            if st:
+                entry.size, entry.mtime = st.st_size, st.st_mtime
+
+            return True
+
+    def delete(self, file: str | Path, ids: list[str]) -> bool:
+        """
+        Remove IDs from an archive and update the cache.
+
+        Args:
+            file (str|Path): The archive file path.
+            ids (list[str]): IDs to remove; invalid lines are ignored.
+
+        Returns:
+            bool: True on success (including no-op), False on error.
+
+        """
+        if not file or not ids:
+            return False
+
+        key: str = self._key(file)
+        lock: threading.RLock = self._get_lock(key)
+        with lock:
+            path = Path(key)
+            if not path.exists():
+                return False
+
+            entry: _Entry = self._cache.get(key) or _Entry()
+            remove_ids: set[str] = {x.strip() for x in ids if str(x).strip() and len(str(x).strip().split()) >= 2}
+            if not remove_ids:
+                return True
+
+            kept_lines: list[str] = []
+            changed = False
+            try:
+                with path.open("r", encoding="utf-8") as f:
+                    for line in f:
+                        s: str = line.strip()
+                        if not s or len(s.split()) < 2:
+                            changed = True
+                            continue
+                        if s in remove_ids:
+                            changed = True
+                            continue
+                        kept_lines.append(line)
+            except OSError as e:
+                LOG.error(f"Failed reading archive for delete '{key}': {e!s}")
+                return False
+
+            if not changed:
+                return True
+
+            try:
+                try:
+                    from yt_dlp.utils import locked_file
+
+                    with locked_file(str(path), "w", encoding="utf-8") as f:
+                        f.writelines(kept_lines)
+                except Exception:
+                    with path.open("w", encoding="utf-8") as f:
+                        f.writelines(kept_lines)
+            except OSError as e:
+                LOG.error(f"Failed writing archive after delete '{key}': {e!s}")
+                return False
+
+            if entry.loaded:
+                entry.ids.difference_update(remove_ids)
+
+            st: os.stat_result | None = self._stat(key)
+            if st:
+                entry.size, entry.mtime = st.st_size, st.st_mtime
+
+            self._cache[key] = entry
+
+            return True
+
+    # Global configuration
+    @classmethod
+    def set_skip_read_stat_checks(cls, skip: bool = True) -> None:
+        """
+        Control os.stat checks on read paths for external change detection.
+
+        When skip is True, Archiver assumes exclusive control of archive files
+        and skips stat() comparisons during reads. Writes always refresh
+        metadata regardless of this setting.
+
+        Args:
+            skip (bool): If True, skip read-time stat checks.
+
+        """
+        inst = cls.get_instance()
+        with inst._global_lock:
+            inst._stats_check = not skip

app/library/Utils.py

@@ -1357,141 +1357,51 @@ def list_folders(path: Path, base: Path, depth_limit: int) -> list[str]:
 
 def archive_add(file: str | Path, ids: list[str], skip_check: bool = False) -> bool:
     """
-    Add IDs to an archive file.
+    Add IDs to an archive file (delegates to the global Archiver).
 
     Args:
         file (str|Path): The archive file path.
         ids (list[str]): List of IDs to add.
         skip_check (bool): If True, skip checking for existing IDs.
 
+    Returns:
+        bool: True if any new IDs were appended, False otherwise.
+
     """
-    if not ids or not file:
-        return False
+    from app.library.Archiver import Archiver
 
-    path: Path = Path(file) if not isinstance(file, Path) else file
-    existing_ids: set[str] = set()
-
-    if not skip_check and path.exists():
-        with path.open("r", encoding="utf-8") as f:
-            for line in f:
-                s = line.strip()
-
-                if not s or len(s.split()) < 2:
-                    continue
-
-                existing_ids.add(s)
-
-    new_ids: list[str] = []
-    for raw in ids:
-        s: str = str(raw).strip()
-
-        if not s or len(s.split()) < 2 or s in existing_ids or s in new_ids:
-            continue
-
-        new_ids.append(s)
-
-    if not new_ids:
-        return False
-
-    try:
-        if not path.parent.exists():
-            path.parent.mkdir(parents=True, exist_ok=True)
-
-        with path.open("a", encoding="utf-8") as f:
-            f.write("".join(f"{x}\n" for x in new_ids))
-
-        return True
-    except OSError as e:
-        LOG.error(f"Failed to write to archive file '{path!s}'. {e!s}")
-        return False
+    return Archiver.get_instance().add(file, ids, skip_check)
 
 
 def archive_read(file: str | Path, ids: list[str] | None = None) -> list[str]:
     """
-    Read IDs from an archive file with optional filtering.
-
-    - If `ids` is empty, return all IDs present in the archive file.
-    - If `ids` is provided, return only the ids that exist in the archive file,
+    Read IDs from an archive file with optional filtering (delegates to Archiver).
 
     Args:
         file (str|Path): The archive file path.
-        ids (list[str]): Optional list of IDs to query; empty list returns all.
+        ids (list[str]|None): Optional list of IDs to query; None/empty returns all.
 
     Returns:
-        list[str]: List of ids found in the archive file filtered by `ids` if provided.
+        list[str]: IDs present in the archive, optionally filtered.
 
     """
-    if not file:
-        return []
+    from app.library.Archiver import Archiver
 
-    path: Path = Path(file) if not isinstance(file, Path) else file
-    if not file or not path.exists():
-        return []
-
-    ids_set: set[str] | None = (
-        {s.strip() for s in ids if str(s).strip() and len(str(s).strip().split()) >= 2} if ids else None
-    )
-
-    found: list[str] = []
-    with path.open("r", encoding="utf-8") as f:
-        for line in f:
-            s: str = line.strip()
-
-            if not s or len(s.split()) < 2:
-                continue
-
-            if ids_set is None or s in ids_set:
-                found.append(s)
-
-    return found
+    return Archiver.get_instance().read(file, ids)
 
 
 def archive_delete(file: str | Path, ids: list[str]) -> bool:
     """
-    Delete the given IDs from an archive file.
+    Delete IDs from an archive file (delegates to Archiver).
 
     Args:
         file (str|Path): The archive file path.
         ids (list[str]): List of IDs to remove.
 
     Returns:
-        bool: True if deletion succeeded (or nothing to do), False on error.
+        bool: True on success (including no-op), False on error.
 
     """
-    if not file or not ids:
-        return False
+    from app.library.Archiver import Archiver
 
-    path: Path = Path(file) if not isinstance(file, Path) else file
-
-    if not path.exists():
-        return False
-
-    remove_ids: set[str] = {x.strip() for x in ids if str(x).strip() and len(str(x).strip().split()) >= 2}
-    if not remove_ids:
-        return True
-
-    changed = False
-    kept_lines: list[str] = []
-    removed_ids: list[str] = []
-    with path.open("r", encoding="utf-8") as f:
-        for line in f:
-            s: str = line.strip()
-
-            if not s or len(s.split()) < 2:
-                changed = True
-                continue
-
-            if s in remove_ids:
-                changed = True
-                removed_ids.append(s)
-                continue
-
-            kept_lines.append(line)
-
-    if not changed:
-        return True
-
-    with path.open("w", encoding="utf-8") as f:
-        f.writelines(kept_lines)
-
-    return True
+    return Archiver.get_instance().delete(file, ids)

app/library/ytdlp.py

@@ -1,20 +1,99 @@
+import logging
 from typing import Any
 
 import yt_dlp
 
 import app.postprocessors  # noqa: F401
 
+LOG: logging.Logger = logging.getLogger(__name__)
+
+
+class _ArchiveProxy:
+    """
+    Proxy for yt-dlp's self.archive that delegates to our Archiver.
+
+    Supports membership checks (`id in proxy`) and `.add(id)`.
+    """
+
+    def __init__(self, file: str | None):
+        self._file: str | None = file
+
+    def __contains__(self, item: str) -> bool:
+        if not self._file or not item:
+            return False
+
+        try:
+            from app.library.Archiver import Archiver
+
+            status: bool = item in Archiver.get_instance().read(self._file, [item])
+            LOG.debug(f"ArchiveProxy: '{item}' in '{self._file}': {'yes' if status else 'no'}.")
+            return status
+        except Exception:
+            return False
+
+    def add(self, item: str) -> bool:
+        if not self._file or not item:
+            return False
+
+        try:
+            from app.library.Archiver import Archiver
+
+            status: bool = Archiver.get_instance().add(self._file, [item])
+            LOG.debug(f"ArchiveProxy: Added '{item}' to '{self._file}': {'yes' if status else 'no'}.")
+            return status
+        except Exception:
+            return False
+
+    def __bool__(self) -> bool:
+        return bool(self._file)
+
 
 class YTDLP(yt_dlp.YoutubeDL):
     _interrupted = False
 
-    def _delete_downloaded_files(self, *args, **kwargs):
+    def __init__(self, params=None, auto_init=True):
+        # Avoid yt-dlp preloading the archive file by stripping the param first
+        orig_file = None
+        patched_params = None
+        if params is not None:
+            try:
+                orig_file: str | None = params.get("download_archive")
+                patched_params: dict = dict(params)
+                if "download_archive" in patched_params:
+                    patched_params.pop("download_archive", None)
+            except Exception:
+                patched_params = params
+
+        super().__init__(params=patched_params, auto_init=auto_init)
+
+        # Restore param and replace upstream archive set with our proxy
+        if orig_file is not None:
+            try:
+                self.params["download_archive"] = orig_file
+            except Exception:
+                pass
+
+        self.archive = _ArchiveProxy(orig_file)
+
+    def _delete_downloaded_files(self, *args, **kwargs) -> None:
         if self._interrupted:
             self.to_screen("[info] Cancelled — skipping temp cleanup.")
             return None
 
         return super()._delete_downloaded_files(*args, **kwargs)
 
+    def record_download_archive(self, info_dict) -> None:
+        if not self.params.get("download_archive"):
+            return
+
+        if not (archive_id := self._make_archive_id(info_dict)):
+            return
+
+        assert archive_id
+
+        self.write_debug(f"Adding to archive: {archive_id}")
+        self.archive.add(archive_id)
+
 
 def ytdlp_options() -> list[dict[str, Any]]:
     """
@@ -57,4 +136,3 @@ def ytdlp_options() -> list[dict[str, Any]]:
             opt["description"] = "No description available from yt-dlp."
 
     return opts
-