app/api/v1/admin/cache.py

from typing import List

from fastapi import APIRouter, Depends, HTTPException, Query, Request

from app.core.auth import verify_app_key
from app.core.batch import create_task, expire_task
from app.services.grok.batch_services.assets import ListService, DeleteService
from app.services.token.manager import get_token_manager
router = APIRouter()


@router.get("/cache", dependencies=[Depends(verify_app_key)])
async def cache_stats(request: Request):
    """获取缓存统计"""
    from app.services.grok.utils.cache import CacheService

    try:
        cache_service = CacheService()
        image_stats = cache_service.get_stats("image")
        video_stats = cache_service.get_stats("video")

        mgr = await get_token_manager()
        pools = mgr.pools
        accounts = []
        for pool_name, pool in pools.items():
            for info in pool.list():
                raw_token = (
                    info.token[4:] if info.token.startswith("sso=") else info.token
                )
                masked = (
                    f"{raw_token[:8]}...{raw_token[-16:]}"
                    if len(raw_token) > 24
                    else raw_token
                )
                accounts.append(
                    {
                        "token": raw_token,
                        "token_masked": masked,
                        "pool": pool_name,
                        "status": info.status,
                        "last_asset_clear_at": info.last_asset_clear_at,
                    }
                )

        scope = request.query_params.get("scope")
        selected_token = request.query_params.get("token")
        tokens_param = request.query_params.get("tokens")
        selected_tokens = []
        if tokens_param:
            selected_tokens = [t.strip() for t in tokens_param.split(",") if t.strip()]

        online_stats = {
            "count": 0,
            "status": "unknown",
            "token": None,
            "last_asset_clear_at": None,
        }
        online_details = []
        account_map = {a["token"]: a for a in accounts}
        if selected_tokens:
            total = 0
            raw_results = await ListService.fetch_assets_details(
                selected_tokens,
                account_map,
            )
            for token, res in raw_results.items():
                if res.get("ok"):
                    data = res.get("data", {})
                    detail = data.get("detail")
                    total += data.get("count", 0)
                else:
                    account = account_map.get(token)
                    detail = {
                        "token": token,
                        "token_masked": account["token_masked"] if account else token,
                        "count": 0,
                        "status": f"error: {res.get('error')}",
                        "last_asset_clear_at": account["last_asset_clear_at"]
                        if account
                        else None,
                    }
                if detail:
                    online_details.append(detail)
            online_stats = {
                "count": total,
                "status": "ok" if selected_tokens else "no_token",
                "token": None,
                "last_asset_clear_at": None,
            }
            scope = "selected"
        elif scope == "all":
            total = 0
            tokens = list(dict.fromkeys([account["token"] for account in accounts]))
            raw_results = await ListService.fetch_assets_details(
                tokens,
                account_map,
            )
            for token, res in raw_results.items():
                if res.get("ok"):
                    data = res.get("data", {})
                    detail = data.get("detail")
                    total += data.get("count", 0)
                else:
                    account = account_map.get(token)
                    detail = {
                        "token": token,
                        "token_masked": account["token_masked"] if account else token,
                        "count": 0,
                        "status": f"error: {res.get('error')}",
                        "last_asset_clear_at": account["last_asset_clear_at"]
                        if account
                        else None,
                    }
                if detail:
                    online_details.append(detail)
            online_stats = {
                "count": total,
                "status": "ok" if accounts else "no_token",
                "token": None,
                "last_asset_clear_at": None,
            }
        else:
            token = selected_token
            if token:
                raw_results = await ListService.fetch_assets_details(
                    [token],
                    account_map,
                )
                res = raw_results.get(token, {})
                data = res.get("data", {})
                detail = data.get("detail") if res.get("ok") else None
                if detail:
                    online_stats = {
                        "count": data.get("count", 0),
                        "status": detail.get("status", "ok"),
                        "token": detail.get("token"),
                        "token_masked": detail.get("token_masked"),
                        "last_asset_clear_at": detail.get("last_asset_clear_at"),
                    }
                else:
                    match = next((a for a in accounts if a["token"] == token), None)
                    online_stats = {
                        "count": 0,
                        "status": f"error: {res.get('error')}",
                        "token": token,
                        "token_masked": match["token_masked"] if match else token,
                        "last_asset_clear_at": match["last_asset_clear_at"]
                        if match
                        else None,
                    }
            else:
                online_stats = {
                    "count": 0,
                    "status": "not_loaded",
                    "token": None,
                    "last_asset_clear_at": None,
                }

        response = {
            "local_image": image_stats,
            "local_video": video_stats,
            "online": online_stats,
            "online_accounts": accounts,
            "online_scope": scope or "none",
            "online_details": online_details,
        }
        return response
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@router.get("/cache/list", dependencies=[Depends(verify_app_key)])
async def list_local(
    cache_type: str = "image",
    type_: str = Query(default=None, alias="type"),
    page: int = 1,
    page_size: int = 1000,
):
    """列出本地缓存文件"""
    from app.services.grok.utils.cache import CacheService

    try:
        if type_:
            cache_type = type_
        cache_service = CacheService()
        result = cache_service.list_files(cache_type, page, page_size)
        return {"status": "success", **result}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@router.post("/cache/clear", dependencies=[Depends(verify_app_key)])
async def clear_local(data: dict):
    """清理本地缓存"""
    from app.services.grok.utils.cache import CacheService

    cache_type = data.get("type", "image")

    try:
        cache_service = CacheService()
        result = cache_service.clear(cache_type)
        return {"status": "success", "result": result}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@router.post("/cache/item/delete", dependencies=[Depends(verify_app_key)])
async def delete_local_item(data: dict):
    """删除单个本地缓存文件"""
    from app.services.grok.utils.cache import CacheService

    cache_type = data.get("type", "image")
    name = data.get("name")
    if not name:
        raise HTTPException(status_code=400, detail="Missing file name")
    try:
        cache_service = CacheService()
        result = cache_service.delete_file(cache_type, name)
        return {"status": "success", "result": result}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@router.post("/cache/online/clear", dependencies=[Depends(verify_app_key)])
async def clear_online(data: dict):
    """清理在线缓存"""
    try:
        mgr = await get_token_manager()
        tokens = data.get("tokens")

        if isinstance(tokens, list):
            token_list = [t.strip() for t in tokens if isinstance(t, str) and t.strip()]
            if not token_list:
                raise HTTPException(status_code=400, detail="No tokens provided")

            token_list = list(dict.fromkeys(token_list))

            results = {}
            raw_results = await DeleteService.clear_assets(
                token_list,
                mgr,
            )
            for token, res in raw_results.items():
                if res.get("ok"):
                    results[token] = res.get("data", {})
                else:
                    results[token] = {"status": "error", "error": res.get("error")}

            return {"status": "success", "results": results}

        token = data.get("token") or mgr.get_token()
        if not token:
            raise HTTPException(
                status_code=400, detail="No available token to perform cleanup"
            )

        raw_results = await DeleteService.clear_assets(
            [token],
            mgr,
        )
        res = raw_results.get(token, {})
        data = res.get("data", {})
        if res.get("ok") and data.get("status") == "success":
            return {"status": "success", "result": data.get("result")}
        return {"status": "error", "error": data.get("error") or res.get("error")}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@router.post("/cache/online/clear/async", dependencies=[Depends(verify_app_key)])
async def clear_online_async(data: dict):
    """清理在线缓存（异步批量 + SSE 进度）"""
    mgr = await get_token_manager()
    tokens = data.get("tokens")
    if not isinstance(tokens, list):
        raise HTTPException(status_code=400, detail="No tokens provided")

    token_list = [t.strip() for t in tokens if isinstance(t, str) and t.strip()]
    if not token_list:
        raise HTTPException(status_code=400, detail="No tokens provided")

    task = create_task(len(token_list))

    async def _run():
        try:
            async def _on_item(item: str, res: dict):
                ok = bool(res.get("data", {}).get("ok"))
                task.record(ok)

            raw_results = await DeleteService.clear_assets(
                token_list,
                mgr,
                include_ok=True,
                on_item=_on_item,
                should_cancel=lambda: task.cancelled,
            )

            if task.cancelled:
                task.finish_cancelled()
                return

            results = {}
            ok_count = 0
            fail_count = 0
            for token, res in raw_results.items():
                data = res.get("data", {})
                if data.get("ok"):
                    ok_count += 1
                    results[token] = {"status": "success", "result": data.get("result")}
                else:
                    fail_count += 1
                    results[token] = {"status": "error", "error": data.get("error")}

            result = {
                "status": "success",
                "summary": {
                    "total": len(token_list),
                    "ok": ok_count,
                    "fail": fail_count,
                },
                "results": results,
            }
            task.finish(result)
        except Exception as e:
            task.fail_task(str(e))
        finally:
            import asyncio
            asyncio.create_task(expire_task(task.id, 300))

    import asyncio
    asyncio.create_task(_run())

    return {
        "status": "success",
        "task_id": task.id,
        "total": len(token_list),
    }


@router.post("/cache/online/load/async", dependencies=[Depends(verify_app_key)])
async def load_cache_async(data: dict):
    """在线资产统计（异步批量 + SSE 进度）"""
    from app.services.grok.utils.cache import CacheService

    mgr = await get_token_manager()

    accounts = []
    for pool_name, pool in mgr.pools.items():
        for info in pool.list():
            raw_token = info.token[4:] if info.token.startswith("sso=") else info.token
            masked = (
                f"{raw_token[:8]}...{raw_token[-16:]}"
                if len(raw_token) > 24
                else raw_token
            )
            accounts.append(
                {
                    "token": raw_token,
                    "token_masked": masked,
                    "pool": pool_name,
                    "status": info.status,
                    "last_asset_clear_at": info.last_asset_clear_at,
                }
            )

    account_map = {a["token"]: a for a in accounts}

    tokens = data.get("tokens")
    scope = data.get("scope")
    selected_tokens: List[str] = []
    if isinstance(tokens, list):
        selected_tokens = [str(t).strip() for t in tokens if str(t).strip()]

    if not selected_tokens and scope == "all":
        selected_tokens = [account["token"] for account in accounts]
        scope = "all"
    elif selected_tokens:
        scope = "selected"
    else:
        raise HTTPException(status_code=400, detail="No tokens provided")

    task = create_task(len(selected_tokens))

    async def _run():
        try:
            cache_service = CacheService()
            image_stats = cache_service.get_stats("image")
            video_stats = cache_service.get_stats("video")

            async def _on_item(item: str, res: dict):
                ok = bool(res.get("data", {}).get("ok"))
                task.record(ok)

            raw_results = await ListService.fetch_assets_details(
                selected_tokens,
                account_map,
                include_ok=True,
                on_item=_on_item,
                should_cancel=lambda: task.cancelled,
            )

            if task.cancelled:
                task.finish_cancelled()
                return

            online_details = []
            total = 0
            for token, res in raw_results.items():
                data = res.get("data", {})
                detail = data.get("detail")
                if detail:
                    online_details.append(detail)
                total += data.get("count", 0)

            online_stats = {
                "count": total,
                "status": "ok" if selected_tokens else "no_token",
                "token": None,
                "last_asset_clear_at": None,
            }

            result = {
                "local_image": image_stats,
                "local_video": video_stats,
                "online": online_stats,
                "online_accounts": accounts,
                "online_scope": scope or "none",
                "online_details": online_details,
            }
            task.finish(result)
        except Exception as e:
            task.fail_task(str(e))
        finally:
            import asyncio
            asyncio.create_task(expire_task(task.id, 300))

    import asyncio
    asyncio.create_task(_run())

    return {
        "status": "success",
        "task_id": task.id,
        "total": len(selected_tokens),
    }