File size: 1,808 Bytes
c49b21b |
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 |
import os
import tempfile
def _is_writable(path: str) -> bool:
try:
if not os.path.exists(path):
os.makedirs(path, exist_ok=True)
test_fd, test_path = tempfile.mkstemp(prefix='.wtest_', dir=path)
os.close(test_fd)
os.unlink(test_path)
return True
except Exception:
return False
def _detect_data_dir() -> str:
# 1) Respect DATA_DIR env only if writable
env = os.getenv('DATA_DIR')
if env and _is_writable(env):
return env
# 2) Prefer /data if writable (Spaces)
if _is_writable('/data'):
return '/data'
# 3) Local dev fallback: /app/data if writable
if _is_writable('/app/data'):
return '/app/data'
# 4) Final fallback: /tmp
return '/tmp'
DATA_DIR = _detect_data_dir()
# Logs: prefer DATA_DIR/logs, fallback to /tmp/logs
_preferred_logs = os.getenv('LOG_DIR') or os.path.join(DATA_DIR, 'logs')
try:
os.makedirs(_preferred_logs, exist_ok=True)
# sanity: try to write
if not _is_writable(_preferred_logs):
raise PermissionError("Log dir not writable")
except Exception:
_preferred_logs = '/tmp/logs'
os.makedirs(_preferred_logs, exist_ok=True)
LOG_DIR = _preferred_logs
# Path for scheduler's last_run marker
def _compute_last_run_path(base_dir: str) -> str:
candidates = [
os.path.join(base_dir, 'deployment', 'last_run.txt'),
os.path.join(base_dir, 'last_run.txt'),
'/tmp/last_run.txt',
]
for p in candidates:
try:
os.makedirs(os.path.dirname(p), exist_ok=True)
# test write
with open(p, 'a'):
pass
return p
except Exception:
continue
return '/tmp/last_run.txt'
LAST_RUN_PATH = _compute_last_run_path(DATA_DIR)
|