Spaces:
Running
Running
File size: 2,678 Bytes
29bfc1f 49be4be 29bfc1f 49be4be 29bfc1f 49be4be 29bfc1f 49be4be 3341f00 29bfc1f 3341f00 29bfc1f 3341f00 29bfc1f | 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 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 | import asyncio
import json
from src.core.config import SUPABASE_URL, SUPABASE_SERVICE_KEY
from loguru import logger as _loguru
_http_session = None
#Removing the Default Loging Template
_loguru.remove()
#Adding the Custom Loging Template
_loguru.add(
lambda msg: print(msg, end=""),
format="<green>{time:HH:mm:ss}</green> | <level>{level:<8}</level> | {message}",
level="DEBUG",
colorize=True,
)
_log_fn = _loguru.log
# A coroutine to initialize the HTTP session
async def init_logging_session():
global _http_session
if SUPABASE_URL and SUPABASE_SERVICE_KEY:
import aiohttp
_http_session = aiohttp.ClientSession(
headers={
"Content-Type": "application/json",
"apikey": SUPABASE_SERVICE_KEY,
"Authorization": f"Bearer {SUPABASE_SERVICE_KEY}",
"Prefer": "return=minimal",
}
)
# Coroutine to close the HTTP session
async def close_logging_session():
global _http_session
if _http_session:
await _http_session.close()
async def _supabase_log(level: str, event: str, data: dict) -> None:
if not _http_session:
return
try:
import aiohttp
row = {
"level": level.upper(),
"event": event,
"user_id": str(data.get("user_id", "anonymous")),
"ip": str(data.get("ip", "")),
"mode": str(data.get("mode", "")),
"page": str(data.get("page", "")),
"duration_ms": int(data["duration_ms"]) if "duration_ms" in data else None,
"error": str(data["error"]) if "error" in data else None,
"data": data,
}
async with _http_session.post(
f"{SUPABASE_URL}/rest/v1/app_logs",
json=row,
timeout=aiohttp.ClientTimeout(total=5),
) as r:
if r.status not in (200, 201):
body = await r.text()
_log_fn("WARNING", f"Supabase log failed {r.status}: {body[:200]}")
except Exception as exc:
_log_fn("DEBUG", f"Supabase log push skipped: {exc}")
_LEVEL_ALIASES = {"WARN": "WARNING", "ERR": "ERROR", "FATAL": "CRITICAL"}
def log(level: str, event: str, **data) -> None:
norm_level = _LEVEL_ALIASES.get(level.upper(), level.upper())
_log_fn(norm_level, f"[{event}] {json.dumps(data, default=str)}")
try:
loop = asyncio.get_running_loop()
loop.create_task(_supabase_log(norm_level, event, data))
except RuntimeError:
pass
def warn(msg: str) -> None:
_log_fn("WARNING", msg) |