# src/pages/builder/templates/agent-python/local_blob_store.py
"""LocalFileBlobStore — 文件系统持久化的 pages-blob 替身，仅用于本地 dev。

触发方式：设置环境变量 PAGES_BLOB_LOCAL_PERSIST=1。
数据路径：默认 .edgeone/agent-python/.blob-local/<store-name>/<url_quote(key)>.bin
       （可用 PAGES_BLOB_LOCAL_DIR 覆盖）

接口与 pages_blob.Store 鸭子兼容（只实现 BlobBackedStore 实际用到的 4 个方法：
get / set / delete / list）。不依赖 pages_blob 包，离线环境也能用。

不支持的能力：TTL 过期清理（由 BlobBackedStore 的 envelope 自己兜底）、
content-type、cacheControl、CDN 头——本地 dev 场景不需要。
"""
from __future__ import annotations

import asyncio
import os
import urllib.parse
from dataclasses import dataclass, field
from pathlib import Path
from typing import List, Optional


@dataclass
class _BlobInfo:
    """鸭子兼容 pages_blob.types.BlobInfo"""
    key: str
    etag: str = ""


@dataclass
class _ListResult:
    """鸭子兼容 pages_blob.types.ListResult"""
    blobs: List[_BlobInfo] = field(default_factory=list)
    directories: List[str] = field(default_factory=list)


def _safe_filename(key: str) -> str:
    """key → 文件名。用 URL encode 保留可读性，同时避免路径穿越。

    注意不用 hash：debug 时能直接 ls 看到原 key。
    """
    encoded = urllib.parse.quote(key, safe="")
    # 加后缀，方便 rm -rf 清理
    return encoded + ".bin"


def _unsafe_key(filename: str) -> str:
    """反向：文件名 → key"""
    if filename.endswith(".bin"):
        filename = filename[:-4]
    return urllib.parse.unquote(filename)


class LocalFileBlobStore:
    """文件持久化的 pages_blob.Store 替身。

    每个 store 对应 base_dir/<store_name>/ 目录，每个 key 对应一个文件。
    同目录下多 handler 并行写入同一个 key：使用 atomic write（先写 .tmp
    再 rename），确保读取端永远看到完整内容。
    """

    def __init__(self, store_name: str, base_dir: Path) -> None:
        self._store_name = store_name
        self._dir = base_dir / store_name
        self._dir.mkdir(parents=True, exist_ok=True)

    def _path_for(self, key: str) -> Path:
        return self._dir / _safe_filename(key)

    async def get(self, key: str, *, type: str = "text", **_kwargs) -> Optional[object]:
        path = self._path_for(key)
        if not path.exists():
            return None
        # 用线程池读，避免阻塞 event loop（小文件其实无所谓，保持形状对齐）
        body = await asyncio.to_thread(path.read_bytes)
        if type == "bytes":
            return body
        if type == "json":
            import json
            return json.loads(body.decode("utf-8"))
        # 默认 text
        return body.decode("utf-8")

    async def set(
        self,
        key: str,
        value,
        *,
        only_if_new: bool = False,
        **_kwargs,
    ) -> None:
        path = self._path_for(key)
        if only_if_new and path.exists():
            # 与线上行为对齐：412 → BlobBackedStore 不直接调这个，
            # 但独立调用方可能走到；直接抛出和 pages_blob 对齐的异常。
            raise _PreconditionFailedError()

        if isinstance(value, str):
            data = value.encode("utf-8")
        elif isinstance(value, (bytes, bytearray)):
            data = bytes(value)
        else:
            # dict / list 等——直接 json 化
            import json
            data = json.dumps(value, ensure_ascii=False).encode("utf-8")

        # Atomic write：rename 是 POSIX 原子操作
        tmp = path.with_suffix(path.suffix + ".tmp")
        await asyncio.to_thread(tmp.write_bytes, data)
        await asyncio.to_thread(os.replace, tmp, path)

    async def set_json(self, key: str, value, *, only_if_new: bool = False, **_kwargs) -> None:
        import json
        await self.set(key, json.dumps(value, ensure_ascii=False), only_if_new=only_if_new)

    async def delete(self, key: str) -> None:
        path = self._path_for(key)
        try:
            await asyncio.to_thread(path.unlink)
        except FileNotFoundError:
            # 对齐 pages_blob：delete 不存在的 key 是幂等操作
            pass

    async def list(
        self,
        *,
        prefix: Optional[str] = None,
        directories: bool = False,
        cursor: Optional[str] = None,
        paginate: bool = True,
        **_kwargs,
    ) -> _ListResult:
        entries = await asyncio.to_thread(lambda: list(self._dir.iterdir()))
        keys = []
        for entry in entries:
            if not entry.is_file() or not entry.name.endswith(".bin"):
                continue
            key = _unsafe_key(entry.name)
            if prefix and not key.startswith(prefix):
                continue
            keys.append(key)
        keys.sort()
        return _ListResult(blobs=[_BlobInfo(key=k) for k in keys])


class _PreconditionFailedError(Exception):
    """和 pages_blob.errors.PreconditionFailedError 鸭子对齐"""
    code = "PRECONDITION_FAILED"


def get_local_base_dir() -> Optional[Path]:
    """解析本地持久化根目录；未启用返回 None。

    - PAGES_BLOB_LOCAL_DIR 指定 → 用它
    - PAGES_BLOB_LOCAL_PERSIST=1 → 用 .edgeone/agent-python/.blob-local
    - 否则 → None
    """
    explicit = os.environ.get("PAGES_BLOB_LOCAL_DIR")
    if explicit:
        return Path(explicit).expanduser().resolve()

    flag = os.environ.get("PAGES_BLOB_LOCAL_PERSIST", "").strip().lower()
    if flag in ("1", "true", "yes", "on"):
        # adapter.py 同目录下的 .blob-local（main.py 位置就是 OUTPUT_DIR）
        main_py = os.environ.get("_AGENT_PYTHON_MAIN", "")
        if main_py:
            return Path(main_py).parent / ".blob-local"
        # Fallback: cwd 下
        return Path.cwd() / ".blob-local"

    return None