apiVersion: v1
kind: ConfigMap
metadata:
name: argos-panel-config
namespace: argos-core
data:
app.py: |
import os
from fastapi import FastAPI, HTTPException, Query
from fastapi.responses import HTMLResponse, StreamingResponse, JSONResponse
from minio import Minio
from urllib.parse import quote
from datetime import timezone
import io
BUCKET = os.getenv("MINIO_BUCKET", "argos")
endpoint = os.getenv("MINIO_ENDPOINT", "minio.argos-core.svc.cluster.local:9000")
access_key = os.getenv("MINIO_ACCESS_KEY")
secret_key = os.getenv("MINIO_SECRET_KEY")
secure = os.getenv("MINIO_SECURE", "false").lower() == "true"
mc = Minio(endpoint, access_key=access_key, secret_key=secret_key, secure=secure)
app = FastAPI()
@app.get("/health")
def health():
# simple check contra MinIO
try:
mc.list_buckets()
return {"ok": True}
except Exception as e:
return {"ok": False, "error": str(e)}
def _iter_events(prefix: str | None, limit: int):
# list_objects devuelve en orden “no garantizado”; recolectamos y ordenamos por last_modified desc
objs = []
for obj in mc.list_objects(BUCKET, prefix=prefix, recursive=True):
# omitimos directorios “lógicos” y thumbs por defecto en el listado principal
if obj.is_dir:
continue
if "/thumbs/" in obj.object_name:
continue
objs.append(obj)
objs.sort(key=lambda o: o.last_modified or 0, reverse=True)
return objs[:limit]
@app.get("/api/events")
def api_events(prefix: str | None = Query(default=None), limit: int = 50):
data = []
for o in _iter_events(prefix, limit):
ts = None
if o.last_modified:
ts = o.last_modified.astimezone(timezone.utc).isoformat()
# si existe thumb, la inferimos (cam/... → cam/thumbs/...)
thumb = None
parts = o.object_name.split("/")
if len(parts) >= 2:
maybe = f"{parts[0]}/thumbs/" + parts[-1].rsplit(".",1)[0] + ".jpg"
try:
mc.stat_object(BUCKET, maybe)
thumb = maybe
except:
pass
data.append({
"key": o.object_name,
"size": o.size,
"last_modified": ts,
"thumb": thumb
})
return JSONResponse(data)
@app.get("/file")
def file(key: str):
try:
resp = mc.get_object(BUCKET, key)
except Exception as e:
raise HTTPException(status_code=404, detail=f"not found: {e}")
# stream al cliente sin exponer MinIO
def gen():
for d in resp.stream(32*1024):
yield d
# tipo de contenido básico por extensión
ctype = "application/octet-stream"
k = key.lower()
if k.endswith(".mp4"): ctype = "video/mp4"
elif k.endswith(".jpg") or k.endswith(".jpeg"): ctype = "image/jpeg"
return StreamingResponse(gen(), media_type=ctype)
@app.get("/view", response_class=HTMLResponse)
def view(key: str):
keyq = quote(key)
return f"""
Ver {key}
{key}
Descargar
Volver
"""
@app.get("/", response_class=HTMLResponse)
def home(prefix: str | None = None, limit: int = 20):
# HTML mínimo que consume /api/events y pinta thumbs/links
p = f"&prefix={quote(prefix)}" if prefix else ""
items = []
for e in _iter_events(prefix, limit):
thumb_html = ""
# intentar thumb
parts = e.object_name.split("/")
if len(parts) >= 2:
maybe = f"{parts[0]}/thumbs/" + parts[-1].rsplit(".",1)[0] + ".jpg"
try:
mc.stat_object(BUCKET, maybe)
thumb_html = f'})
'
except:
pass
items.append(f"""
{thumb_html}
{e.object_name}
""")
items_html = "\n".join(items) or "Sin eventos aún.
"
return f"""
Argos Panel
Argos Panel
"""