Spaces:
Running
Running
File size: 2,223 Bytes
a8784d9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 | """Temp-file registry so 'Clear all' genuinely frees disk, not just the UI.
Every PDF/ZIP scratch file goes through `new_temp`, which records the path. A
later `clear_all` unlinks every recorded file. Thread-safe (a min worker pool
on HF still shares this process). UI-free so `src/` stays testable.
Note: this clears the *file cache* we create. Gradio's own request queue is
per-request and transient (a handler can't flush other users' pending events),
and the priority queue in `queue_manager` is built and drained within a single
`process_pdf` call - neither leaves persistent state to clear.
"""
from __future__ import annotations
import os
import tempfile
import threading
_lock = threading.Lock()
_tracked: set[str] = set()
def new_temp(suffix: str = "") -> str:
"""Create a tracked temp file and return its path (handle closed)."""
fd, path = tempfile.mkstemp(suffix=suffix)
os.close(fd)
with _lock:
_tracked.add(path)
return path
def register(path: str) -> None:
"""Track an externally created path so clear_all() will remove it."""
with _lock:
_tracked.add(path)
def discard(path: str) -> bool:
"""Unlink one tracked file early (e.g. an input PDF after it's loaded).
Returns True if the file was removed. Untracks regardless so a vanished
file doesn't linger in the registry.
"""
with _lock:
_tracked.discard(path)
try:
os.remove(path)
return True
except OSError:
return False
def clear_all() -> int:
"""Unlink every tracked temp file. Returns count actually removed."""
removed = 0
with _lock:
# list() snapshot is required: we mutate _tracked (discard) in-loop.
# Iterating the set directly -> "Set changed size during iteration".
for path in list(_tracked): # NOSONAR python:S7504 false positive
try:
os.remove(path)
removed += 1
except FileNotFoundError:
pass
except OSError:
continue # leave it tracked; retry on next clear
_tracked.discard(path)
return removed
def tracked_count() -> int:
with _lock:
return len(_tracked)
|