feature/limpieza-periodica
This commit is contained in:
@@ -0,0 +1,69 @@
|
||||
"""
|
||||
Claves Redis por job de import CSV (layouts_csv): las de storage_keys más extras por módulo.
|
||||
Usado por la limpieza periódica (victor "el senior de limpieza") para saber si un job sigue vivo en Redis.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, List
|
||||
|
||||
from . import storage as common_storage
|
||||
|
||||
# Prefijo en nombre de archivo `{prefix}_{uuid}.csv` -> job_type (vacío = solo `{uuid}.csv`)
|
||||
FILE_PREFIX_TO_JOB_TYPE: dict[str, str] = {
|
||||
"veh": "veh",
|
||||
"fa": "fa",
|
||||
"trl": "trl",
|
||||
"trp": "trp",
|
||||
"ped": "ped",
|
||||
"part": "part",
|
||||
"er": "er",
|
||||
"cb": "cb",
|
||||
"drv": "drv",
|
||||
"cls": "cls",
|
||||
"cp": "cp",
|
||||
"bom": "bom",
|
||||
"exp": "exp",
|
||||
"crreg": "crreg",
|
||||
}
|
||||
|
||||
|
||||
def all_redis_keys_for_layout_import(job_type: str, job_id: str) -> List[str]:
|
||||
"""Todas las claves Redis que pueden existir para un job (file, meta, error_lines, extras)."""
|
||||
file_k, meta_k, err_k = common_storage.storage_keys(job_type, job_id)
|
||||
keys: List[str] = [file_k, meta_k, err_k]
|
||||
if job_type == "veh":
|
||||
keys.append(f"veh_import_status:{job_id}")
|
||||
elif job_type == "trl":
|
||||
keys.append(f"trl_import_status:{job_id}")
|
||||
elif job_type == "trp":
|
||||
keys.append(f"trp_import_status:{job_id}")
|
||||
elif job_type == "drv":
|
||||
keys.append(f"drv_import_status:{job_id}")
|
||||
keys.append(f"drv_import_transporter_map:{job_id}")
|
||||
return keys
|
||||
|
||||
|
||||
def any_layout_import_key_exists(redis_client: Any, job_type: str, job_id: str) -> bool:
|
||||
for k in all_redis_keys_for_layout_import(job_type, job_id):
|
||||
if redis_client.exists(k):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def delete_extra_layout_import_keys_from_redis(redis_client: Any, job_type: str, job_id: str) -> None:
|
||||
"""Borra claves que delete_import_from_redis no cubre (status, mapas)."""
|
||||
extras: list[str] = []
|
||||
if job_type == "veh":
|
||||
extras.append(f"veh_import_status:{job_id}")
|
||||
elif job_type == "trl":
|
||||
extras.append(f"trl_import_status:{job_id}")
|
||||
elif job_type == "trp":
|
||||
extras.append(f"trp_import_status:{job_id}")
|
||||
elif job_type == "drv":
|
||||
extras.append(f"drv_import_status:{job_id}")
|
||||
extras.append(f"drv_import_transporter_map:{job_id}")
|
||||
if extras:
|
||||
try:
|
||||
redis_client.delete(*extras)
|
||||
except Exception:
|
||||
pass
|
||||
@@ -18,6 +18,7 @@ def dispatch_tracked_layouts_csv_commit(
|
||||
db: Session,
|
||||
current_user: dict[str, Any],
|
||||
redis_client: Any,
|
||||
layout_job_id: str,
|
||||
meta_redis_key: str,
|
||||
celery_task: Task,
|
||||
task_name: str,
|
||||
@@ -54,5 +55,6 @@ def dispatch_tracked_layouts_csv_commit(
|
||||
task_origin=task_origin,
|
||||
args=args,
|
||||
task_id=commit_id,
|
||||
meta_payload={"layout_import_job_id": layout_job_id},
|
||||
)
|
||||
return commit_id
|
||||
|
||||
172
backend/api/v1/modules/a76/layouts_csv/common/victor.py
Normal file
172
backend/api/v1/modules/a76/layouts_csv/common/victor.py
Normal file
@@ -0,0 +1,172 @@
|
||||
"""
|
||||
Limpieza periódica de archivos huérfanos de import CSV (layouts/imports/temp y errors).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
import time
|
||||
from typing import Optional, Tuple
|
||||
|
||||
from sqlalchemy import or_
|
||||
|
||||
from api.v1.modules.core.tasks_tracking.models import TaskRun, TaskStatus
|
||||
from core.celery_app import celery_app
|
||||
from core.database import CoreSessionLocal
|
||||
|
||||
from . import storage as common_storage
|
||||
from .import_redis_keys import (
|
||||
FILE_PREFIX_TO_JOB_TYPE,
|
||||
any_layout_import_key_exists,
|
||||
delete_extra_layout_import_keys_from_redis,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Evita carreras con uploads/commits recién creados (segundos)
|
||||
MIN_ORPHAN_FILE_AGE_SEC = int(os.getenv("LAYOUT_ORPHAN_MIN_AGE_SEC", "600"))
|
||||
|
||||
_UUID_RE = r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}"
|
||||
|
||||
|
||||
def _get_redis():
|
||||
import redis
|
||||
|
||||
url = os.getenv("VALKEY_URL", os.getenv("REDIS_URL", "redis://valkey:6379/0"))
|
||||
return redis.Redis.from_url(url, decode_responses=False)
|
||||
|
||||
|
||||
def parse_temp_csv_filename(name: str) -> Optional[Tuple[str, str]]:
|
||||
"""Devuelve (job_type, job_id) o None si no es un CSV de import conocido."""
|
||||
if not name.endswith(".csv"):
|
||||
return None
|
||||
base = name[:-4]
|
||||
m = re.match(rf"^({_UUID_RE})$", base, re.I)
|
||||
if m:
|
||||
return "", m.group(1)
|
||||
m = re.match(rf"^([a-z0-9]+)_({_UUID_RE})$", base, re.I)
|
||||
if not m:
|
||||
return None
|
||||
prefix, jid = m.group(1).lower(), m.group(2)
|
||||
jt = FILE_PREFIX_TO_JOB_TYPE.get(prefix)
|
||||
if jt is None:
|
||||
return None
|
||||
return jt, jid
|
||||
|
||||
|
||||
def parse_errors_jsonl_filename(name: str) -> Optional[Tuple[str, str]]:
|
||||
if not name.endswith(".jsonl"):
|
||||
return None
|
||||
base = name[:-6]
|
||||
m = re.match(rf"^({_UUID_RE})$", base, re.I)
|
||||
if m:
|
||||
return "", m.group(1)
|
||||
m = re.match(rf"^([a-z0-9]+)_({_UUID_RE})$", base, re.I)
|
||||
if not m:
|
||||
return None
|
||||
prefix, jid = m.group(1).lower(), m.group(2)
|
||||
jt = FILE_PREFIX_TO_JOB_TYPE.get(prefix)
|
||||
if jt is None:
|
||||
return None
|
||||
return jt, jid
|
||||
|
||||
|
||||
def _layout_import_protected_by_task_run(db, job_id: str) -> bool:
|
||||
row = (
|
||||
db.query(TaskRun)
|
||||
.filter(
|
||||
TaskRun.task_group == "layouts_csv",
|
||||
TaskRun.status.in_([TaskStatus.PENDING.value, TaskStatus.ACTIVE.value]),
|
||||
or_(
|
||||
TaskRun.task_id == job_id,
|
||||
TaskRun.meta_payload.contains({"layout_import_job_id": job_id}),
|
||||
),
|
||||
)
|
||||
.first()
|
||||
)
|
||||
return row is not None
|
||||
|
||||
|
||||
def _reference_mtime(job_type: str, job_id: str) -> Optional[float]:
|
||||
csv_path = common_storage.file_path_for_job(job_type, job_id)
|
||||
err_path = common_storage.error_path_for_job(job_type, job_id)
|
||||
mtimes = []
|
||||
if os.path.isfile(csv_path):
|
||||
mtimes.append(os.path.getmtime(csv_path))
|
||||
if os.path.isfile(err_path):
|
||||
mtimes.append(os.path.getmtime(err_path))
|
||||
if not mtimes:
|
||||
return None
|
||||
return max(mtimes)
|
||||
|
||||
|
||||
def _try_cleanup_layout_import_job(r, db, job_type: str, job_id: str) -> bool:
|
||||
if any_layout_import_key_exists(r, job_type, job_id):
|
||||
return False
|
||||
if _layout_import_protected_by_task_run(db, job_id):
|
||||
return False
|
||||
mtime = _reference_mtime(job_type, job_id)
|
||||
if mtime is None:
|
||||
return False
|
||||
if time.time() - mtime < MIN_ORPHAN_FILE_AGE_SEC:
|
||||
return False
|
||||
|
||||
file_path = common_storage.file_path_for_job(job_type, job_id)
|
||||
error_path = common_storage.error_path_for_job(job_type, job_id)
|
||||
meta_path = None
|
||||
if os.path.isfile(file_path):
|
||||
meta_path = file_path.replace(".csv", ".meta.json")
|
||||
|
||||
common_storage.cleanup_import_job(
|
||||
job_type,
|
||||
job_id,
|
||||
file_path=file_path if os.path.isfile(file_path) else None,
|
||||
error_path=error_path if os.path.isfile(error_path) else None,
|
||||
meta_path=meta_path if meta_path and os.path.isfile(meta_path) else None,
|
||||
)
|
||||
delete_extra_layout_import_keys_from_redis(r, job_type, job_id)
|
||||
return True
|
||||
|
||||
|
||||
def run_orphan_layout_import_cleanup() -> dict:
|
||||
"""Barrido síncrono; devuelve contadores para logs/resultado Celery."""
|
||||
removed_jobs: list[str] = []
|
||||
r = _get_redis()
|
||||
db = CoreSessionLocal()
|
||||
try:
|
||||
temp_dir = common_storage.upload_dir()
|
||||
os.makedirs(temp_dir, exist_ok=True)
|
||||
for name in os.listdir(temp_dir):
|
||||
parsed = parse_temp_csv_filename(name)
|
||||
if not parsed:
|
||||
continue
|
||||
job_type, job_id = parsed
|
||||
if _try_cleanup_layout_import_job(r, db, job_type, job_id):
|
||||
removed_jobs.append(f"{job_type or 'invoice'}:{job_id}")
|
||||
|
||||
err_dir = common_storage.error_dir()
|
||||
if os.path.isdir(err_dir):
|
||||
for name in os.listdir(err_dir):
|
||||
parsed = parse_errors_jsonl_filename(name)
|
||||
if not parsed:
|
||||
continue
|
||||
job_type, job_id = parsed
|
||||
csv_path = common_storage.file_path_for_job(job_type, job_id)
|
||||
if os.path.isfile(csv_path):
|
||||
continue
|
||||
if _try_cleanup_layout_import_job(r, db, job_type, job_id):
|
||||
rid = f"{job_type or 'invoice'}:{job_id}:errors_only"
|
||||
if rid not in removed_jobs:
|
||||
removed_jobs.append(rid)
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
if removed_jobs:
|
||||
logger.info("layout import cleanup removed %s job(s): %s", len(removed_jobs), removed_jobs[:20])
|
||||
return {"removed_count": len(removed_jobs), "removed": removed_jobs}
|
||||
|
||||
|
||||
@celery_app.task(name="cleanup_orphan_layout_imports")
|
||||
def cleanup_orphan_layout_imports():
|
||||
return run_orphan_layout_import_cleanup()
|
||||
Reference in New Issue
Block a user