apiVersion: v1
kind: ConfigMap
metadata:
name: argos-panel-config
namespace: argos-core
data:
app.py: |
import os, sqlite3, time
from fastapi import FastAPI, HTTPException
from fastapi.responses import HTMLResponse
from minio import Minio
from urllib.parse import urlparse
DB="/data/argos.db"
mc=Minio(os.getenv("MINIO_ENDPOINT","s3.argos.interna"),
access_key=os.getenv("MINIO_ACCESS_KEY"),
secret_key=os.getenv("MINIO_SECRET_KEY"),
secure=os.getenv("MINIO_SECURE","false").lower()=="true")
app=FastAPI()
def rows(limit=100, camera=None, since=None):
q="SELECT id, ts, edge, camera, label, s3url, thumb_s3 FROM events"
cond=[]; args=[]
if camera: cond.append("camera=?"); args.append(camera)
if since: cond.append("ts>=?"); args.append(int(since))
if cond: q+=" WHERE "+ " AND ".join(cond)
q+=" ORDER BY ts DESC LIMIT ?"; args.append(limit)
con=sqlite3.connect(DB); cur=con.cursor()
cur.execute(q, tuple(args)); r=cur.fetchall(); con.close()
return r
@app.get("/api/events")
def api_events(limit:int=100, camera:str=None, since:int=None):
return [dict(id=i, ts=t, edge=e, camera=c, label=l or "", s3url=s, thumb=th or "")
for (i,t,e,c,l,s,th) in rows(limit,camera,since)]
@app.get("/api/url/{event_id}")
def presign(event_id: str, expires: int = 600):
con=sqlite3.connect(DB); cur=con.cursor()
cur.execute("SELECT s3url FROM events WHERE id=?", (event_id,))
row=cur.fetchone(); con.close()
if not row: raise HTTPException(404, "Not found")
s3url=row[0]; p=urlparse(s3url); b=p.netloc; k=p.path.lstrip("/")
return {"url": mc.presigned_get_object(b, k, expires=expires)}
@app.get("/", response_class=HTMLResponse)
def index():
return """
ARGOS Panel
ARGOS – Alarmas
"""