Spaces:
Paused
Paused
File size: 2,054 Bytes
6774799 |
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 |
import os
import shutil
import tempfile
import time
import uuid
import streamlit as st
from config import SESSION_TTL_SECONDS
def init_file_manager():
base_dir = get_base_sessions_dir()
os.makedirs(base_dir, exist_ok=True)
session_dir = get_session_dir()
os.makedirs(session_dir, exist_ok=True)
touch()
cleanup_expired_sessions()
def get_base_sessions_dir():
return os.path.join(tempfile.gettempdir(), "pycaps-sessions")
def get_session_dir():
return os.path.join(get_base_sessions_dir(), st.session_state.session_id)
def get_path(*relative_path):
return os.path.join(get_session_dir(), *relative_path)
def get_random_file_name(ext):
return get_path(f"{uuid.uuid4()}.{ext}")
def create_temp_file(suffix="", prefix="tmp", text=False):
fd, path = tempfile.mkstemp(suffix=suffix, prefix=prefix, dir=get_session_dir(), text=text)
# Abre y devuelve el archivo como objeto file abierto (similar a NamedTemporaryFile)
mode = "w+" if text else "w+b"
file_obj = os.fdopen(fd, mode)
return file_obj
def create_temp_dir(suffix="", prefix="tmp"):
return tempfile.mkdtemp(suffix=suffix, prefix=prefix, dir=get_session_dir())
def touch():
keepalive_path = os.path.join(get_session_dir(), ".keepalive")
with open(keepalive_path, "w") as f:
f.write(str(time.time()))
def cleanup_expired_sessions():
now = time.time()
for name in os.listdir(get_base_sessions_dir()):
path = os.path.join(get_base_sessions_dir(), name)
if not os.path.isdir(path):
continue
try:
keepalive_file = os.path.join(path, ".keepalive")
if os.path.exists(keepalive_file):
last_touched = os.path.getmtime(keepalive_file)
else:
last_touched = os.path.getmtime(path)
if now - last_touched > SESSION_TTL_SECONDS:
shutil.rmtree(path)
except Exception as e:
pass
def delete_current_session_dir():
shutil.rmtree(get_session_dir(), ignore_errors=True)
|