api/system.py

from __future__ import annotations

import json
from typing import Any
from urllib.parse import quote, urlparse

import httpx
from fastapi import APIRouter, Header, HTTPException, Request
from fastapi.concurrency import run_in_threadpool
from fastapi.responses import Response, StreamingResponse
from pydantic import BaseModel, ConfigDict

from api.support import require_admin, require_identity, resolve_image_base_url
from services.backup_service import BackupError, backup_service
from services.config import config
from services.auth_service import auth_service
from services.image_service import delete_images, download_images_zip, get_image_download_response, get_image_response, get_thumbnail_response, list_images
from services.image_storage_service import ImageStorageError, image_storage_service
from services.image_tags_service import delete_tag, get_all_tags, set_tags
from services.log_service import log_service
from services.proxy_service import test_proxy


class SettingsUpdateRequest(BaseModel):
    model_config = ConfigDict(extra="allow")


class WechatLoginRequest(BaseModel):
    code: str = ""
    invite_code: str = ""


class NormalUserRegisterRequest(BaseModel):
    name: str = ""
    invite_code: str = ""


class ProxyTestRequest(BaseModel):
    url: str = ""


class ImageDeleteRequest(BaseModel):
    paths: list[str] = []
    start_date: str = ""
    end_date: str = ""
    all_matching: bool = False

class ImageDownloadRequest(BaseModel):
    paths: list[str]

class ImageTagsRequest(BaseModel):
    path: str
    tags: list[str]

class LogDeleteRequest(BaseModel):
    ids: list[str] = []
class BackupDeleteRequest(BaseModel):
    key: str = ""


class UserStorageImportRequest(BaseModel):
    payload: Any
    dry_run: bool = True


def _parse_json_object(value: object) -> dict[str, Any]:
    if isinstance(value, dict):
        return dict(value)
    if isinstance(value, str):
        try:
            parsed = json.loads(value)
        except json.JSONDecodeError:
            return {}
        return dict(parsed) if isinstance(parsed, dict) else {}
    return {}


def _parse_json_list(value: object) -> list[dict[str, Any]]:
    if isinstance(value, list):
        return [dict(item) for item in value if isinstance(item, dict)]
    if isinstance(value, str):
        try:
            parsed = json.loads(value)
        except json.JSONDecodeError:
            return []
        return [dict(item) for item in parsed if isinstance(item, dict)] if isinstance(parsed, list) else []
    return []


def _normalize_auth_key_import(payload: dict[str, Any]) -> list[dict[str, Any]]:
    raw_auth_keys = payload.get("auth_keys")
    rows = raw_auth_keys if isinstance(raw_auth_keys, list) else []
    items: list[dict[str, Any]] = []
    for row in rows:
        if not isinstance(row, dict):
            continue
        item = _parse_json_object(row.get("data")) if "data" in row else dict(row)
        item_id = str(item.get("id") or row.get("key_id") or "").strip()
        if item_id:
            item["id"] = item_id
            items.append(item)
    return items


def _normalize_shop_state_import(payload: dict[str, Any]) -> tuple[dict[str, Any], dict[str, dict[str, Any]]]:
    state: dict[str, Any] = {}
    named_states: dict[str, dict[str, Any]] = {}
    raw_shop_state = payload.get("shop_state")

    if isinstance(raw_shop_state, dict):
        state = dict(raw_shop_state)
    elif isinstance(raw_shop_state, list):
        for row in raw_shop_state:
            if not isinstance(row, dict):
                continue
            key = str(row.get("key") or "default").strip() or "default"
            item_state = _parse_json_object(row.get("data")) if "data" in row else dict(row)
            if key == "default":
                state = item_state
            else:
                named_states[key] = item_state

    imported_codes = _parse_json_list(payload.get("redeem_codes"))
    if imported_codes:
        existing_codes = state.get("codes") if isinstance(state.get("codes"), list) else []
        merged: dict[str, dict[str, Any]] = {}
        for item in [*existing_codes, *imported_codes]:
            if not isinstance(item, dict):
                continue
            data = _parse_json_object(item.get("data")) if "data" in item else dict(item)
            data.setdefault("id", item.get("code_id") or item.get("id"))
            data.setdefault("code_hash", item.get("code_hash"))
            data.setdefault("status", item.get("status"))
            if item.get("batch_id") and not data.get("batch_id"):
                data["batch_id"] = item.get("batch_id")
            if item.get("redeemed_by") and not data.get("redeemed_by"):
                data["redeemed_by"] = item.get("redeemed_by")
            identity = str(data.get("code_hash") or data.get("id") or "").strip()
            if identity:
                merged[identity] = data
        state["codes"] = list(merged.values())

    if not isinstance(state.get("ledger"), list):
        state["ledger"] = []
    if not isinstance(state.get("codes"), list):
        state["codes"] = []
    return state, named_states


def _normalize_user_storage_import(payload: object) -> tuple[list[dict[str, Any]], dict[str, Any], dict[str, dict[str, Any]]]:
    parsed = _parse_json_object(payload)
    if "full_backup" in parsed:
        parsed = _parse_json_object(parsed.get("full_backup"))
    auth_keys = _normalize_auth_key_import(parsed)
    shop_state, named_states = _normalize_shop_state_import(parsed)
    return auth_keys, shop_state, named_states


def create_router(app_version: str) -> APIRouter:
    router = APIRouter()

    @router.post("/auth/login")
    async def login(authorization: str | None = Header(default=None)):
        identity = require_identity(authorization)
        return {
            "ok": True,
            "version": app_version,
            "role": identity.get("role"),
            "subject_id": identity.get("id"),
            "name": identity.get("name"),
            "account_pool_enabled": bool(identity.get("role") == "admin" or identity.get("account_pool_enabled")),
            "image_memory_button_enabled": config.image_memory_button_enabled,
            "login_session_duration_hours": config.login_session_duration_hours_for_role(identity.get("role")),
        }

    @router.get("/api/auth/wechat/status")
    async def wechat_login_status():
        return {
            "enabled": config.wechat_login_enabled,
            "auto_register_enabled": config.wechat_auto_register_enabled,
            "login_session_duration_hours": config.normal_user_login_session_duration_hours,
        }

    @router.get("/api/branding")
    async def get_branding():
        return {"branding": config.get_branding()}

    @router.post("/api/auth/wechat/login")
    async def wechat_login(body: WechatLoginRequest):
        if not config.wechat_login_enabled:
            raise HTTPException(status_code=403, detail={"error": "微信验证码登录已关闭，请使用登录密钥。"})
        code = str(body.code or "").strip()
        if not code:
            raise HTTPException(status_code=400, detail={"error": "请输入微信验证码"})

        try:
            async with httpx.AsyncClient(timeout=15.0) as client:
                response = await client.post("https://wx.z-l.top/api/auth/verify", json={"code": code})
        except httpx.HTTPError as exc:
            raise HTTPException(status_code=502, detail={"error": "微信登录服务暂时不可用，请稍后再试"}) from exc

        try:
            payload = response.json()
        except ValueError as exc:
            raise HTTPException(status_code=502, detail={"error": "微信登录服务返回异常"}) from exc

        if not response.is_success or not payload.get("success"):
            message = str(payload.get("message") or "微信验证码无效或已过期")
            raise HTTPException(status_code=401, detail={"error": message})

        user = payload.get("user") if isinstance(payload.get("user"), dict) else {}
        openid = str(user.get("openid") or "").strip()
        if not openid:
            raise HTTPException(status_code=502, detail={"error": "微信登录服务未返回 openid"})

        try:
            item, raw_key, created = auth_service.get_or_create_wechat_normal_user(
                openid=openid,
                nickname=str(user.get("nickname") or "").strip(),
                avatar=str(user.get("avatar") or "").strip(),
                allow_create=config.wechat_auto_register_enabled,
            )
            if str(body.invite_code or "").strip():
                auth_service.attach_invite(str(item.get("id") or ""), body.invite_code)
                item = auth_service.get_key(str(item.get("id") or ""), role="normal") or item
        except ValueError as exc:
            raise HTTPException(status_code=403, detail={"error": str(exc)}) from exc
        except Exception as exc:
            print(f"[wechat-login] failed to save user: {exc}")
            raise HTTPException(status_code=503, detail={"error": f"微信登录已验证，但保存用户失败：{exc}"}) from exc
        return {
            "ok": True,
            "version": app_version,
            "role": "normal",
            "subject_id": item.get("id"),
            "name": item.get("name"),
            "account_pool_enabled": bool(item.get("account_pool_enabled")),
            "image_memory_button_enabled": config.image_memory_button_enabled,
            "key": raw_key,
            "created": created,
            "login_session_duration_hours": config.normal_user_login_session_duration_hours,
            "credits": auth_service.credit_summary(item),
            "wechat": {
                "openid": openid,
                "nickname": item.get("wechat_nickname") or user.get("nickname") or "",
                "avatar": item.get("wechat_avatar") or user.get("avatar") or "",
            },
        }

    @router.get("/version")
    async def get_version():
        return {"version": app_version}

    @router.get("/api/settings")
    async def get_settings(authorization: str | None = Header(default=None)):
        require_admin(authorization)
        return {"config": config.get()}

    @router.get("/api/announcements")
    async def get_announcements(authorization: str | None = Header(default=None)):
        identity = require_identity(authorization)
        return {"items": config.get_announcements(role=str(identity.get("role") or ""))}

    @router.post("/api/settings")
    async def save_settings(body: SettingsUpdateRequest, authorization: str | None = Header(default=None)):
        require_admin(authorization)
        try:
            return {"config": config.update(body.model_dump(mode="python"))}
        except ValueError as exc:
            raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc

    @router.get("/api/images")
    async def get_images(request: Request, start_date: str = "", end_date: str = "", limit: int = 500, authorization: str | None = Header(default=None)):
        require_admin(authorization)
        return list_images(resolve_image_base_url(request), start_date=start_date.strip(), end_date=end_date.strip(), limit=limit)

    @router.get("/images/{image_path:path}", include_in_schema=False)
    async def get_image(image_path: str):
        return get_image_response(image_path)

    @router.get("/api/images/proxy")
    async def proxy_image(url: str, authorization: str | None = Header(default=None)):
        require_identity(authorization)
        parsed = urlparse(str(url or "").strip())
        if parsed.scheme not in {"http", "https"} or not parsed.netloc:
            raise HTTPException(status_code=400, detail={"error": "invalid image url"})
        try:
            async with httpx.AsyncClient(follow_redirects=True, timeout=180) as client:
                response = await client.get(url, headers={"Accept": "image/*,*/*"})
        except httpx.HTTPError as exc:
            raise HTTPException(status_code=502, detail={"error": f"fetch image failed: {exc.__class__.__name__}"}) from exc
        if response.status_code >= 400:
            raise HTTPException(status_code=response.status_code, detail={"error": f"fetch image failed: HTTP {response.status_code}"})
        content_type = response.headers.get("content-type") or "image/png"
        if not content_type.lower().startswith("image/"):
            content_type = "image/png"
        return Response(content=response.content, media_type=content_type)

    @router.get("/image-thumbnails/{image_path:path}", include_in_schema=False)
    async def get_image_thumbnail(image_path: str):
        return get_thumbnail_response(image_path)

    @router.post("/api/images/delete")
    async def delete_images_endpoint(body: ImageDeleteRequest, authorization: str | None = Header(default=None)):
        require_admin(authorization)
        return delete_images(body.paths, start_date=body.start_date.strip(), end_date=body.end_date.strip(), all_matching=body.all_matching)

    @router.post("/api/images/download")
    async def download_images_endpoint(body: ImageDownloadRequest, authorization: str | None = Header(default=None)):
        require_admin(authorization)
        buf = download_images_zip(body.paths)
        return StreamingResponse(
            buf,
            media_type="application/zip",
            headers={"Content-Disposition": 'attachment; filename="images.zip"'},
        )

    @router.get("/api/images/download/{image_path:path}")
    async def download_single_image_endpoint(image_path: str, authorization: str | None = Header(default=None)):
        require_admin(authorization)
        return get_image_download_response(image_path)

    @router.get("/api/logs")
    async def get_logs(type: str = "", start_date: str = "", end_date: str = "", limit: int = 200, authorization: str | None = Header(default=None)):
        require_admin(authorization)
        return {"items": log_service.list(type=type.strip(), start_date=start_date.strip(), end_date=end_date.strip(), limit=limit)}

    @router.post("/api/logs/delete")
    async def delete_logs(body: LogDeleteRequest, authorization: str | None = Header(default=None)):
        require_admin(authorization)
        return log_service.delete(body.ids)

    @router.post("/api/proxy/test")
    async def test_proxy_endpoint(body: ProxyTestRequest, authorization: str | None = Header(default=None)):
        require_admin(authorization)
        candidate = (body.url or "").strip() or config.get_proxy_settings()
        if not candidate:
            raise HTTPException(status_code=400, detail={"error": "proxy url is required"})
        return {"result": await run_in_threadpool(test_proxy, candidate)}

    @router.get("/api/storage/info")
    async def get_storage_info(authorization: str | None = Header(default=None)):
        require_admin(authorization)
        storage = config.get_storage_backend()
        user_storage = config.get_user_storage_backend()
        account_pool_storage = config.get_account_pool_storage_backend()
        return {
            "backend": storage.get_backend_info(),
            "health": storage.health_check(),
            "user_backend": user_storage.get_backend_info(),
            "user_health": user_storage.health_check(),
            "account_pool_backend": account_pool_storage.get_backend_info(),
            "account_pool_health": account_pool_storage.health_check(),
        }

    @router.post("/api/admin/import-user-storage")
    async def import_user_storage(body: UserStorageImportRequest, authorization: str | None = Header(default=None)):
        require_admin(authorization)
        auth_keys, shop_state, named_states = _normalize_user_storage_import(body.payload)
        code_count = len(shop_state.get("codes") or []) if isinstance(shop_state, dict) else 0
        ledger_count = len(shop_state.get("ledger") or []) if isinstance(shop_state, dict) else 0
        if body.dry_run:
            return {
                "ok": True,
                "dry_run": True,
                "target": "local_primary_storage",
                "auth_keys": len(auth_keys),
                "codes": code_count,
                "ledger": ledger_count,
                "named_states": sorted(named_states.keys()),
            }

        storage = config.get_storage_backend()
        try:
            await run_in_threadpool(storage.save_auth_keys, auth_keys)
            await run_in_threadpool(storage.save_shop_state, shop_state)
            save_named_state = getattr(storage, "save_named_state", None)
            if callable(save_named_state):
                for key, state in named_states.items():
                    await run_in_threadpool(save_named_state, key, state)
        except Exception as exc:
            raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc
        return {
            "ok": True,
            "dry_run": False,
            "target": "local_primary_storage",
            "auth_keys": len(auth_keys),
            "codes": code_count,
            "ledger": ledger_count,
            "named_states": sorted(named_states.keys()),
        }

    @router.post("/auth/register")
    async def register_normal_user(body: NormalUserRegisterRequest):
        try:
            item, raw_key = auth_service.create_invited_normal_user(name=body.name, invite_code=body.invite_code)
        except ValueError as exc:
            raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc
        return {
            "ok": True,
            "key": raw_key,
            "role": "normal",
            "subject_id": item.get("id"),
            "name": item.get("name"),
            "account_pool_enabled": False,
            "image_memory_button_enabled": config.image_memory_button_enabled,
            "login_session_duration_hours": config.normal_user_login_session_duration_hours,
            "invite_code": item.get("invite_code"),
        }

    @router.post("/api/backup/test")
    async def test_backup_connection(authorization: str | None = Header(default=None)):
        require_admin(authorization)
        try:
            return {"result": await run_in_threadpool(backup_service.test_connection)}
        except BackupError as exc:
            raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc

    @router.post("/api/image-storage/test")
    async def test_image_storage_endpoint(authorization: str | None = Header(default=None)):
        require_admin(authorization)
        return {"result": await run_in_threadpool(image_storage_service.test_webdav)}

    @router.post("/api/image-storage/sync")
    async def sync_image_storage_endpoint(authorization: str | None = Header(default=None)):
        require_admin(authorization)
        try:
            return {"result": await run_in_threadpool(image_storage_service.sync_all)}
        except ImageStorageError as exc:
            raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc

    @router.get("/api/backups")
    async def get_backups(authorization: str | None = Header(default=None)):
        require_admin(authorization)
        try:
            return {
                "items": await run_in_threadpool(backup_service.list_backups),
                "state": backup_service.get_status(),
                "settings": backup_service.get_settings(),
            }
        except BackupError as exc:
            raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc

    @router.post("/api/backups/run")
    async def run_backup_endpoint(authorization: str | None = Header(default=None)):
        require_admin(authorization)
        try:
            return {"result": await run_in_threadpool(backup_service.run_backup)}
        except BackupError as exc:
            raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc

    @router.post("/api/backups/delete")
    async def delete_backup_endpoint(body: BackupDeleteRequest, authorization: str | None = Header(default=None)):
        require_admin(authorization)
        try:
            await run_in_threadpool(backup_service.delete_backup, body.key)
            return {"ok": True}
        except BackupError as exc:
            raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc

    @router.get("/api/backups/detail")
    async def get_backup_detail(key: str = "", authorization: str | None = Header(default=None)):
        require_admin(authorization)
        try:
            return {"item": await run_in_threadpool(backup_service.get_backup_detail, key)}
        except BackupError as exc:
            raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc

    @router.get("/api/backups/download")
    async def download_backup_endpoint(key: str = "", authorization: str | None = Header(default=None)):
        require_admin(authorization)
        try:
            item = await run_in_threadpool(backup_service.download_backup, key)
        except BackupError as exc:
            raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc
        filename = str(item.get("name") or "backup.bin")
        quoted = quote(filename)
        headers = {
            "Content-Disposition": f"attachment; filename*=UTF-8''{quoted}",
            "Content-Length": str(int(item.get("size") or 0)),
        }
        return Response(
            content=bytes(item.get("payload") or b""),
            media_type=str(item.get("content_type") or "application/octet-stream"),
            headers=headers,
        )


    @router.get("/api/images/tags")
    async def list_image_tags(authorization: str | None = Header(default=None)):
        require_admin(authorization)
        return {"tags": get_all_tags()}

    @router.post("/api/images/tags")
    async def update_image_tags(body: ImageTagsRequest, authorization: str | None = Header(default=None)):
        require_admin(authorization)
        rel = body.path.strip().lstrip("/")
        if not rel:
            raise HTTPException(status_code=400, detail={"error": "path is required"})
        tags = set_tags(rel, body.tags)
        return {"ok": True, "tags": tags}

    @router.delete("/api/images/tags/{tag}")
    async def delete_image_tag(tag: str, authorization: str | None = Header(default=None)):
        require_admin(authorization)
        count = delete_tag(tag)
        return {"ok": True, "removed_from": count}

    return router