api/ads_manager.py

import os
import sqlite3
import logging
from fastapi import APIRouter, HTTPException
from typing import List, Dict, Any

_logger = logging.getLogger("ads_manager")
router = APIRouter(prefix="/api/ads", tags=["ads"])

DB_PATH = os.getenv("ADS_DB_PATH", "/app/data/ads/ads-client.db")

def get_db_connection():
    if not os.path.exists(DB_PATH):
        _logger.warning(f"Database Ads non trovato in {DB_PATH}")
        return None
    try:
        conn = sqlite3.connect(DB_PATH)
        conn.row_factory = sqlite3.Row
        return conn
    except Exception as e:
        _logger.error(f"Errore connessione DB Ads: {e}")
        return None

@router.get("/health")
async def ads_health():
    conn = get_db_connection()
    if not conn:
        return {"status": "error", "message": "Database non disponibile"}
    try:
        cursor = conn.cursor()
        cursor.execute("SELECT count(*) FROM http_cache")
        count = cursor.fetchone()[0]
        conn.close()
        return {"status": "ok", "record_count": count}
    except Exception as e:
        return {"status": "error", "message": str(e)}

@router.get("/cache")
async def get_ads_cache(limit: int = 10):
    conn = get_db_connection()
    if not conn:
        raise HTTPException(status_code=503, detail="Database Ads non disponibile")
    try:
        cursor = conn.cursor()
        cursor.execute("SELECT * FROM http_cache LIMIT ?", (limit,))
        rows = [dict(row) for row in cursor.fetchall()]
        conn.close()
        return {"ok": True, "data": rows}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))