"""Admin API — Private endpoints for site owner.""" import logging from datetime import datetime, timedelta from fastapi import APIRouter, Depends, HTTPException from sqlalchemy import select, func, text, Integer from sqlalchemy.ext.asyncio import AsyncSession from app.auth.models import User from app.auth.router import current_active_user from app.database import get_db from app.reports.models import SearchHistory, ReportCache, MonitorTask logger = logging.getLogger(__name__) router = APIRouter(prefix="/admin", tags=["admin"]) # Auth async def current_active_superuser(user: User = Depends(current_active_user)) -> User: if not user.is_superuser: raise HTTPException(status_code=403, detail="Not enough permissions") return user @router.get("/stats") async def get_stats( user: User = Depends(current_active_superuser), db: AsyncSession = Depends(get_db), ): """Estadísticas funcionales completas para el dashboard admin.""" try: now = datetime.utcnow() day_ago = now - timedelta(hours=24) week_ago = now - timedelta(days=7) month_ago = now - timedelta(days=30) # ── USUARIOS ────────────────────────────────────────────── result = await db.execute(select(func.count(User.id))) total_users = result.scalar() or 0 result = await db.execute(select(func.count(User.id)).where(User.is_active == True)) active_users = result.scalar() or 0 result = await db.execute(select(User.plan, func.count(User.id)).group_by(User.plan)) plans = {row[0] or "free": row[1] for row in result.all()} result = await db.execute(select(func.sum(User.credits))) total_credits = result.scalar() or 0 result = await db.execute(select(func.count(User.id)).where(User.created_at >= week_ago)) new_users_week = result.scalar() or 0 result = await db.execute(select(func.count(User.id)).where(User.created_at >= month_ago)) new_users_month = result.scalar() or 0 # ── INFORMES (BÚSQUEDAS) ───────────────────────────────── result = await db.execute(select(func.count(SearchHistory.id))) total_reports = result.scalar() or 0 result = await db.execute( select(func.count(SearchHistory.id)).where(SearchHistory.created_at >= day_ago) ) reports_24h = result.scalar() or 0 result = await db.execute( select(func.count(SearchHistory.id)).where(SearchHistory.created_at >= week_ago) ) reports_week = result.scalar() or 0 result = await db.execute( select(func.count(SearchHistory.id)).where(SearchHistory.created_at >= month_ago) ) reports_month = result.scalar() or 0 # Informes por tipo result = await db.execute( select(SearchHistory.type, func.count(SearchHistory.id)).group_by(SearchHistory.type) ) report_types = {row[0]: row[1] for row in result.all()} # Top identificadores buscados (últimos 7 días) result = await db.execute( select( SearchHistory.identifier, SearchHistory.type, SearchHistory.name, func.count(SearchHistory.id).label("count"), ) .where(SearchHistory.created_at >= week_ago) .group_by(SearchHistory.identifier, SearchHistory.type, SearchHistory.name) .order_by(func.count(SearchHistory.id).desc()) .limit(10) ) top_searches = [ {"identifier": r[0], "type": r[1], "name": r[2], "count": r[3]} for r in result.all() ] # Informes por día (últimos 7 días) result = await db.execute( select( func.date(SearchHistory.created_at).label("day"), func.count(SearchHistory.id).label("count"), ) .where(SearchHistory.created_at >= week_ago) .group_by(func.date(SearchHistory.created_at)) .order_by(func.date(SearchHistory.created_at)) ) daily_reports = [{"date": str(r[0]), "count": r[1]} for r in result.all()] # ── REVENUE ESTIMADO ───────────────────────────────────── # Precios por plan (ARS/mes) PLAN_PRICES = {"free": 0, "basic": 4999, "pro": 14999, "enterprise": 29999} revenue_monthly = sum(PLAN_PRICES.get(p, 0) * count for p, count in plans.items()) revenue_per_user = revenue_monthly / max(total_users, 1) # ── CONVERSIÓN ─────────────────────────────────────────── paid_users = sum(count for p, count in plans.items() if p != "free") conversion_rate = (paid_users / max(total_users, 1)) * 100 # ── CACHE ──────────────────────────────────────────────── result = await db.execute(select(func.count(ReportCache.id))) cache_entries = result.scalar() or 0 result = await db.execute(select(func.sum(ReportCache.hit_count))) total_cache_hits = result.scalar() or 0 result = await db.execute(select(func.count(ReportCache.id)).where(ReportCache.hit_count > 0)) cache_hits_count = result.scalar() or 0 cache_hit_rate = (cache_hits_count / max(cache_entries, 1)) * 100 # ── MONITOREO ──────────────────────────────────────────── result = await db.execute( select(func.count(MonitorTask.id)).where(MonitorTask.active == True) ) active_monitors = result.scalar() or 0 # ── SCRAPERS (telemetría en memoria) ────────────────────── from app.utils.telemetry import get_scraper_health, get_scrapers_alerts scrapers = get_scraper_health() alerts = get_scrapers_alerts() scrapers_ok = sum(1 for s in scrapers if s.get("status") in ("ok", "empty")) scrapers_error = sum(1 for s in scrapers if s.get("status") == "error") scrapers_blocked = sum(1 for s in scrapers if s.get("status") == "blocked") # ── LOGIN HISTORY ──────────────────────────────────────── from app.auth.models import LoginHistory result = await db.execute( select( func.count(LoginHistory.id).label("total"), func.sum(func.cast(LoginHistory.success, Integer)).label("success"), ).where(LoginHistory.created_at >= day_ago) ) login_stats = result.one() logins_24h = { "total": login_stats.total or 0, "successful": int(login_stats.success or 0), "failed": (login_stats.total or 0) - int(login_stats.success or 0), } return { "users": { "total": total_users, "active": active_users, "new_this_week": new_users_week, "new_this_month": new_users_month, "by_plan": plans, "total_credits": total_credits, }, "reports": { "total": total_reports, "last_24h": reports_24h, "last_week": reports_week, "last_month": reports_month, "by_type": report_types, "daily": daily_reports, "top_searches": top_searches, }, "revenue": { "estimated_monthly_ars": revenue_monthly, "per_user_ars": round(revenue_per_user, 2), "paid_users": paid_users, "conversion_rate_pct": round(conversion_rate, 1), }, "cache": { "entries": cache_entries, "total_hits": total_cache_hits, "hit_rate_pct": round(cache_hit_rate, 1), }, "scrapers": { "total": len(scrapers), "operational": scrapers_ok, "with_error": scrapers_error, "blocked": scrapers_blocked, "alerts": len(alerts), }, "monitors": { "active": active_monitors, }, "security": { "logins_24h": logins_24h, }, } except Exception as e: logger.error(f"Error fetching admin stats: {e}") raise HTTPException(status_code=500, detail="Error fetching stats") @router.get("/users") async def get_users( limit: int = 50, offset: int = 0, user: User = Depends(current_active_superuser), db: AsyncSession = Depends(get_db), ): """List all users.""" try: result = await db.execute( select(User).order_by(User.created_at.desc()).limit(limit).offset(offset) ) users = result.scalars().all() # Count total count_result = await db.execute(select(func.count(User.id))) total = count_result.scalar() or 0 return { "total": total, "users": [ { "id": str(u.id), "email": u.email, "full_name": u.full_name, "credits": u.credits, "plan": u.plan or "free", "is_active": u.is_active, "is_superuser": u.is_superuser, "created_at": u.created_at.isoformat() if u.created_at else None, } for u in users ], } except Exception as e: logger.error(f"Error fetching users: {e}") raise HTTPException(status_code=500, detail="Error fetching users") @router.get("/searches") async def get_searches( days: int = 7, user: User = Depends(current_active_superuser), db: AsyncSession = Depends(get_db), ): """Search analytics for the last N days.""" try: since = datetime.utcnow() - timedelta(days=days) # Searches per day result = await db.execute( select( func.date(SearchHistory.created_at).label("day"), func.count(SearchHistory.id).label("count"), ) .where(SearchHistory.created_at >= since) .group_by(func.date(SearchHistory.created_at)) .order_by(func.date(SearchHistory.created_at)) ) daily = [{"date": str(row[0]), "count": row[1]} for row in result.all()] # Top searched identifiers result = await db.execute( select( SearchHistory.identifier, SearchHistory.type, func.count(SearchHistory.id).label("count"), ) .where(SearchHistory.created_at >= since) .group_by(SearchHistory.identifier, SearchHistory.type) .order_by(func.count(SearchHistory.id).desc()) .limit(10) ) top_searches = [ {"identifier": row[0], "type": row[1], "count": row[2]} for row in result.all() ] # Unique users searching result = await db.execute( select(func.count(func.distinct(SearchHistory.user_id))) .where(SearchHistory.created_at >= since) ) unique_users = result.scalar() or 0 return { "period_days": days, "daily": daily, "top_searches": top_searches, "unique_users": unique_users, } except Exception as e: logger.error(f"Error fetching search analytics: {e}") raise HTTPException(status_code=500, detail="Error fetching search analytics") @router.get("/scrapers") async def get_scrapers(user: User = Depends(current_active_superuser)): """Scraper health status from in-memory telemetry.""" try: from app.utils.telemetry import get_scraper_health results = [] for scraper in get_scraper_health(): success_rate = scraper.get("success_rate_24h") avg_latency = scraper.get("avg_latency_ms_24h") if avg_latency is None: avg_latency = scraper.get("latency_ms") results.append({ "name": scraper.get("name"), "source": scraper.get("fuente", ""), "description": scraper.get("descripcion", ""), "status": scraper.get("status", "unknown"), "success_rate": (success_rate / 100) if success_rate is not None else None, "avg_latency_ms": avg_latency, "records_24h": scraper.get("records_found_last", 0), "last_check": scraper.get("last_seen"), }) return {"scrapers": results} except Exception as e: logger.error(f"Error fetching scraper health: {e}") raise HTTPException(status_code=500, detail="Error fetching scraper health") @router.get("/login-history") async def get_login_history( limit: int = 50, user: User = Depends(current_active_superuser), db: AsyncSession = Depends(get_db), ): """Historial de intentos de login (éxito/fallo, IP, timestamp).""" try: from app.auth.models import LoginHistory result = await db.execute( select(LoginHistory).order_by(LoginHistory.created_at.desc()).limit(limit) ) logs = result.scalars().all() count_result = await db.execute(select(func.count(LoginHistory.id))) total = count_result.scalar() or 0 # Estadísticas de las últimas 24h day_ago = datetime.utcnow() - timedelta(hours=24) result = await db.execute( select( func.count(LoginHistory.id).label("total"), func.sum(func.cast(LoginHistory.success, Integer)).label("success_count"), ).where(LoginHistory.created_at >= day_ago) ) stats_24h = result.one() success_24h = stats_24h.success_count or 0 total_24h = stats_24h.total or 0 return { "total": total, "stats_24h": { "total_attempts": total_24h, "successful": int(success_24h), "failed": total_24h - int(success_24h), }, "logs": [ { "id": l.id, "email": l.email, "success": l.success, "ip_address": l.ip_address, "user_agent": l.user_agent[:100] if l.user_agent else None, "failure_reason": l.failure_reason, "created_at": l.created_at.isoformat() if l.created_at else None, } for l in logs ], } except Exception as e: logger.error(f"Error fetching login history: {e}") raise HTTPException(status_code=500, detail="Error fetching login history")