from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional, Set, Tuple # --- 可选依赖：curl_cffi（你指定要用） --- try:
    from curl_cffi import requests as curl_requests  # type: ignore except Exception:  # pragma: no cover
    curl_requests = None def _parse_iso_dt(s: Optional[str]) -> Optional[datetime]:
    if not s:
        return None
    s = s.replace("Z", "+00:00")
    try:
        dt = datetime.fromisoformat(s)
        # 保证 tz-aware if dt.tzinfo is None:
            dt = dt.replace(tzinfo=timezone.utc)
        return dt
    except ValueError:
        return None def _now_utc() -> datetime:
    return datetime.now(timezone.utc)


@dataclass(frozen=True) class Provider:
    id: str
    name: str type: str
    model: str
    group: str
    endpoint: str

    latest_status: Optional[str] = None
    latest_latency_ms: Optional[int] = None
    latest_ping_latency_ms: Optional[int] = None
    latest_checked_at: Optional[datetime] = None
    latest_message: Optional[str] = None @staticmethod def from_dict(d: Dict[str, Any]) -> "Provider":
        latest = d.get("latest") or {}
        return Provider(
            id=str(d.get("id", "")),
            name=str(d.get("name", "")),
            type=str(d.get("type", "")),
            model=str(d.get("model", "")),
            group=str(d.get("group", "")),
            endpoint=str(d.get("endpoint", "")),
            latest_status=(latest.get("status") if isinstance(latest, dict) else None),
            latest_latency_ms=(latest.get("latencyMs") if isinstance(latest, dict) else None),
            latest_ping_latency_ms=(latest.get("pingLatencyMs") if isinstance(latest, dict) else None),
            latest_checked_at=_parse_iso_dt(latest.get("checkedAt") if isinstance(latest, dict) else None),
            latest_message=(latest.get("message") if isinstance(latest, dict) else None),
        )


class LinuxDoStatusClient:
    """

    SDK-ish client for https://check.linux.do/api/v1/status



    - 数值/元数据：用属性 .total .operational .generatedAt ...

    - 列表/字典：用方法 getAllGroups() / getOfflineModels() ...

    """

    DEFAULT_URL = "https://check.linux.do/api/v1/status" # 你之前定义的概念我保留，并做了可调的集合
    DEGRADED_STATUSES: Set[str] = {"degraded"}
    # “离线”通常你会想把这些算进去；另外 error 不在 summary 里时会单独暴露 .error
    OFFLINE_STATUSES: Set[str] = {"failed", "validationfailed", "maintenance"}

    def __init__(self,

        url: str = DEFAULT_URL,

        timeout_s: float = 15.0,

        impersonate: str = "chrome",

        auto_refresh: bool = True,

        data: Optional[Dict[str, Any]] = None,):
        self._url = url
        self._timeout_s = timeout_s
        self._impersonate = impersonate

        self._raw: Dict[str, Any] = {}
        self._providers: List[Provider] = []
        self._providers_raw_by_id: Dict[str, Dict[str, Any]] = {}
        self._fetched_at: Optional[datetime] = None if data is not None:
            self._load(data)
        elif auto_refresh:
            self.refresh()

     # 网络 / 载入  def refresh(self) -> "LinuxDoStatusClient":
        if curl_requests is None:
            raise RuntimeError(
                "curl_cffi 未安装。请先 pip install curl-cffi\n" "或把 data=... 传进来离线使用。"
            )

        r = curl_requests.get(
            self._url,
            timeout=self._timeout_s,
            impersonate=self._impersonate,  # chrome impersonate
            headers={"accept": "application/json", "cache-control": "no-cache"},
        )
        r.raise_for_status()
        data = r.json()
        if not isinstance(data, dict):
            raise ValueError("API response must be a JSON object")

        self._load(data)
        return self def refreshIfNeeded(self) -> "LinuxDoStatusClient":
        """按 metadata.pollIntervalMs 判断是否过期；过期才 refresh。""" if self.isStale:
            return self.refresh()
        return self def _load(self, data: Dict[str, Any]) -> None:
        self._raw = data

        providers = data.get("providers", [])
        if not isinstance(providers, list):
            raise ValueError('response["providers"] must be a list')

        self._providers_raw_by_id = {
            str(p.get("id", "")): p for p in providers if isinstance(p, dict) and p.get("id")
        }
        self._providers = [Provider.from_dict(p) for p in providers if isinstance(p, dict)]
        self._fetched_at = _now_utc()

     # SDK 属性：summary / metadata  @property def total(self) -> int:
        v = (self._raw.get("summary") or {}).get("total")
        return v if isinstance(v, int) else len(self._providers)

@property def operational(self) -> int:
        v = (self._raw.get("summary") or {}).get("operational")
        if isinstance(v, int):
            return v
        return self.statusCounts.get("operational", 0)

@property def degraded(self) -> int:
        v = (self._raw.get("summary") or {}).get("degraded")
        if isinstance(v, int):
            return v
        return self.statusCounts.get("degraded", 0)

@property def failed(self) -> int:
        v = (self._raw.get("summary") or {}).get("failed")
        if isinstance(v, int):
            return v
        return self.statusCounts.get("failed", 0)

@property def validationFailed(self) -> int:
        v = (self._raw.get("summary") or {}).get("validationFailed")
        if isinstance(v, int):
            return v
        return self.statusCounts.get("validationfailed", 0)

@property def maintenance(self) -> int:
        v = (self._raw.get("summary") or {}).get("maintenance")
        if isinstance(v, int):
            return v
        return self.statusCounts.get("maintenance", 0)

@property def avgLatencyMs(self) -> Optional[int]:
        v = (self._raw.get("summary") or {}).get("avgLatencyMs")
        return v if isinstance(v, int) else None @property def generatedAt(self) -> Optional[datetime]:
        dt = (self._raw.get("metadata") or {}).get("generatedAt")
        return _parse_iso_dt(dt if isinstance(dt, str) else None)

@property def generatedAtString(self) -> Optional[str]:
        s = (self._raw.get("metadata") or {}).get("generatedAt")
        return s if isinstance(s, str) else None @property def pollIntervalMs(self) -> Optional[int]:
        v = (self._raw.get("metadata") or {}).get("pollIntervalMs")
        return v if isinstance(v, int) else None @property def pollIntervalLabel(self) -> Optional[str]:
        v = (self._raw.get("metadata") or {}).get("pollIntervalLabel")
        return v if isinstance(v, str) else None @property def fetchedAt(self) -> Optional[datetime]:
        """本地拉取时间（不是接口 generatedAt）。""" return self._fetched_at

@property def isStale(self) -> bool:
        """

        依据 generatedAt + pollIntervalMs 判断是否过期：

        - 没 metadata 就当 stale（更安全）

        """
        gen = self.generatedAt
        interval = self.pollIntervalMs
        if gen is None or interval is None:
            return True return (_now_utc() - gen).total_seconds() * 1000 > interval

     # 扩展：状态统计 & “error”这种 summary 没给的状态  @property def statusCounts(self) -> Dict[str, int]:
        """

        从 providers.latest.status 直接统计（比 summary 更“全”）

        例如：operational / degraded / error / failed / ...

        """
        counts: Dict[str, int] = {}
        for p in self._providers:
            k = (p.latest_status or "unknown").strip().lower()
            counts[k] = counts.get(k, 0) + 1 return counts

@property def error(self) -> int:
        """接口 summary 里不一定有 error，这里从 providers 计算。""" return self.statusCounts.get("error", 0)

@property def offline(self) -> int:
        """

        “离线”综合数：failed + validationFailed + maintenance + error

        （你也可以按自己口味改 OFFLINE_STATUSES / 规则）

        """ return self.failed + self.validationFailed + self.maintenance + self.error

     # SDK 方法：group/model/provider 查询  def providers(self) -> List[Provider]:
        return list(self._providers)

    def getProvider(self, key: str) -> Optional[Provider]:
        """按 id 或 name 精确匹配（大小写不敏感）。"""
        k = key.strip().lower()
        for p in self._providers:
            if p.id.lower() == k or p.name.lower() == k:
                return p
        return None def __getitem__(self, key: str) -> Provider:
        p = self.getProvider(key)
        if p is None:
            raise KeyError(f"provider not found: {key}")
        return p

    def getProviderRaw(self, provider_id: str) -> Optional[Dict[str, Any]]:
        """拿原始 provider dict（含 statistics / timeline）。""" return self._providers_raw_by_id.get(provider_id)

    def getAllGroups(self) -> List[str]:
        groups = {p.group for p in self._providers if p.group}
        return sorted(groups)

    def getAllTypes(self) -> List[str]:
        types_ = {p.type for p in self._providers if p.type}
        return sorted(types_)

    def getAllModels(self, by_group: bool = True) -> Dict[str, List[str]] | List[str]:
        if not by_group:
            models = {p.model for p in self._providers if p.model}
            return sorted(models)

        m: Dict[str, Set[str]] = {}
        for p in self._providers:
            if p.group and p.model:
                m.setdefault(p.group, set()).add(p.model)
        return {g: sorted(models) for g, models in sorted(m.items(), key=lambda x: x[0].lower())}

    def getModelsByGroup(self, group: str) -> List[str]:
        g = group.strip().lower()
        models = {p.model for p in self._providers if p.group.lower() == g and p.model}
        return sorted(models)

    def getProvidersByGroup(self, group: str) -> List[Provider]:
        g = group.strip().lower()
        return [p for p in self._providers if p.group.lower() == g]

    def getProvidersByModel(self, model: str) -> List[Provider]:
        m = model.strip().lower()
        return [p for p in self._providers if p.model.lower() == m]

    def getProvidersByStatus(self, status: str) -> List[Provider]:
        s = status.strip().lower()
        return [p for p in self._providers if (p.latest_status or "").lower() == s]

     # 你要的：离线/降级 models（保持原来的返回风格）  def getDegradedModels(self, by_group: bool = True) -> Dict[str, List[str]] | List[str]:
        return self._models_by_status(self.DEGRADED_STATUSES, by_group=by_group)

    def getOfflineModels(self, by_group: bool = True) -> Dict[str, List[str]] | List[str]:
        # offline = OFFLINE_STATUSES + {"error"}（更符合你想要的直觉）
        statuses = set(self.OFFLINE_STATUSES) | {"error"}
        return self._models_by_status(statuses, by_group=by_group)

    def _models_by_status(self, statuses: Set[str], by_group: bool) -> Dict[str, List[str]] | List[str]:
        statuses_l = {s.lower() for s in statuses}

        def hit(p: Provider) -> bool:
            return (p.latest_status or "").lower() in statuses_l

        if not by_group:
            models = {p.model for p in self._providers if p.model and hit(p)}
            return sorted(models)

        m: Dict[str, Set[str]] = {}
        for p in self._providers:
            if p.group and p.model and hit(p):
                m.setdefault(p.group, set()).add(p.model)

        return {g: sorted(models) for g, models in sorted(m.items(), key=lambda x: x[0].lower())}

     # 额外：快/慢、分组汇总（很 SDK 常用）  def getFastestProviders(self, n: int = 5) -> List[Provider]:
        def key(p: Provider) -> Tuple[int, int]:
            return (0 if p.latest_latency_ms is not None else 1, p.latest_latency_ms or 10**18)

        return sorted(self._providers, key=key)[: max(0, int(n))]

    def getSlowestProviders(self, n: int = 5) -> List[Provider]:
        def key(p: Provider) -> Tuple[int, int]:
            return (0 if p.latest_latency_ms is not None else 1, -(p.latest_latency_ms or -10**18))

        return sorted(self._providers, key=key)[: max(0, int(n))]

    def getGroupSummary(self) -> Dict[str, Dict[str, int]]:
        """

        返回：

        {

          "Packy": {"total": 10, "operational": 8, "degraded": 1, "error": 1, ...},

          ...

        }

        """
        out: Dict[str, Dict[str, int]] = {}
        for p in self._providers:
            g = p.group or "unknown"
            st = (p.latest_status or "unknown").lower()
            out.setdefault(g, {"total": 0})
            out[g]["total"] += 1
            out[g][st] = out[g].get(st, 0) + 1 # 排序输出（可读性更像 SDK） return {g: out[g] for g in sorted(out.keys(), key=lambda x: x.lower())}

     # 原始数据（调试用）  def raw(self) -> Dict[str, Any]:
        return dict(self._raw)

简单搓了一下
反正我集成到我的 Bot 里面了

至于为什么要用 curl_cffi? 因为监测站有 Cloudflare 盾

想象一下这样的场景：你正在和一个 AI 医疗助手聊天，它刚刚帮你记录了头痛的症状。第二天，你再次咨询时，它竟然完全忘记了你是谁，还要你重新介绍一遍病情…… 是不是很抓狂？这就是传统 AI 应用的 “金鱼记忆” 问题 —— 每次对话都是 “初次见面”，永远记不住历史信息。

今天，为大家介绍一款我最近开发的两款 AI 记忆存储产品 —— PowerMem + seekdb，一个让 AI 拥有 “超强记忆力” 的持久化记忆系统。

• PowerMem：https://github.com/oceanbase/powermem

• seekdb：https://github.com/oceanbase/seekdb

传统的 AI 对话系统每次会话都是 “失忆” 的。用户每次都需要重复说明自己的信息、偏好和历史，体验割裂、效率低下。开发者想要构建 “有记忆的 AI”，却面临数据持久化、智能提取、多模态支持、权限控制等诸多复杂问题，往往需要从零开始构建记忆系统，重复造轮子。

PowerMem 应运而生 —— 一款开源的 AI 记忆管理 SDK，致力于解决 80% 的 AI 记忆管理问题。我们相信，通过提供一套完整、易用、高性能的记忆管理解决方案，可以让开发者专注于业务创新，而不是重复造轮子。

不是简单的 "记事本"，而是 "最强大脑"

PowerMem 是什么？

PowerMem 建立在这样一个原则之上：AI 系统应该能够像人类一样随着时间积累知识和经验。这一理念驱动了 PowerMem 设计和实施的每个方面：

• 智能提取和保留：PowerMem 通过 LLM 模型进行记忆的提取，根据重要性和相关性确定哪些信息值得记住。

• 上下文理解：PowerMem 维护跨交互的上下文以实现有意义的个性化体验。

• 持续学习：PowerMem 使 AI 系统能够从每次交互中学习并随着时间的推移而改进。

• 自适应遗忘：像人类记忆一样，PowerMem 实现了自适应遗忘机制以防止信息过载。

PowerMem 的核心特性包括：

1. 开发者友好：轻量级接入方式，提供简洁的 Python SDK / MCP 支持，让开发者快速集成到现有项目中

2. 智能记忆管理：记忆的智能提取，通过 LLM 自动从对话中提取关键事实，智能检测重复、更新冲突信息并合并相关记忆，确保记忆库的准确性和一致性。举个例子：还记得上学时老师让你划重点吗？PowerMem 就是 AI 的学霸助手，不需要你手动标注，AI 自动帮你划重点。

    # 用户说了一堆话
   
   messages = [
   
   {"role": "user", "content": "Hi, my name is Alice. I'm a software engineer at Google."},
   
   {"role": "assistant", "content": "Nice to meet you, Alice!"},
   
   {"role": "user", "content": "I love Python programming and machine learning."}
   
   ]
   
   # PowerMem 自动提取关键事实
   
   memory.add(messages=messages, user_id="alice", infer=True)
   
   # 结果：自动提取出 4 条关键记忆 # 1. Name is Alice # 2. Is a software engineer at Google # 3. Loves Python programming # 4. Loves machine learning 

3. 艾宾浩斯遗忘曲线：基于认知科学的记忆遗忘规律，自动计算记忆保留率并实现时间衰减加权，优先返回最近且相关的记忆，让 AI 系统像人类一样自然 “遗忘” 过时信息。还记得那个著名的遗忘曲线吗？PowerMem 把它用在了 AI 记忆上，简单来说，就像人类大脑一样，重要的、最近的信息记得更牢。

   • 最近的信息：权重高，优先召回

   • 久远的信息：权重低，自然衰减

   • 过时信息：自动清理，保持记忆库新鲜

4. 多智能体支持：智能体共享 / 隔离记忆，为每个智能体提供独立的记忆空间，支持跨智能体记忆共享和协作，通过作用域控制实现灵活的权限管理。

5. 多模态支持：不仅记得文字，还看得懂图片。文本、图像、语音记忆：自动将图像和音频转换为文本描述并存储，支持多模态混合内容（文本 + 图像 + 音频）的检索，让 AI 系统理解更丰富的上下文信息。

    # 存储图片记忆
   
   memory.add(
   
   messages=[{"role": "user", "content": "这是我的X光片"}],
   
   images=["xray_image.jpg"],
   
   user_id="patient_001"
   
   )
   
   # 搜索时，文字 + 图片一起检索
   
   results = memory.search("X光片结果", user_id="patient_001")
   
   

6. 深度优化数据存储：支持子存储（Sub Stores），通过子存储实现数据的分区管理，支持自动路由查询，显著提升超大规模数据的查询性能和资源利用率。

7. 混合检索：融合向量检索、全文搜索和图检索的多路召回能力，通过 LLM 构建知识图谱并支持多跳图遍历，精准检索复杂的记忆关联关系。

seekdb 是什么？

OceanBase seekdb 是 OceanBase 打造的一款开发者友好的 AI 原生数据库产品，专注于为 AI 应用提供高效的混合搜索能力，支持向量、文本、结构化与半结构化数据的统一存储与检索，并通过内置 AI Functions 支持数据嵌入、重排与库内实时推理。 seekdb 在继承 OceanBase 原核心引擎高性能优势与 MySQL 全面兼容特性的基础上，通过深度优化数据搜索架构，为开发者提供更符合 AI 应用数据处理需求的解决方案。

PowerMem + seekdb (1 + 1>2 ) 的持久化记忆解决方案

PowerMem 的架构旨在模块化、可扩展，包括如下层：

• 核心记忆引擎 (core)：管理所有记忆操作，包括智能记忆处理器、分层记忆管理、全生命周期记忆管理模块

• 模型层：提供与流行 LLM 和嵌入模型的无缝集成

• 存储层：支持多种存储后端的灵活接口（特别地，我们在 seekdb /oceanbase 上做了深度适配，充分利用了 seekdb 的混搜能力）。

所以，PowerMem + seekdb 的组合不是简单的数据存储，而是一个真正智能的持久化记忆系统。

1 分钟快速上手：让 AI 秒变 "记忆大师"

第一步：安装

pip install powermem

第二步：使用

 from powermem import Memory, auto_config

# 自动从 .env 加载配置

memory = Memory(config=auto_config())

# 添加记忆（自动提取事实）

memory.add("用户喜欢喝咖啡", user_id="user123")

# 搜索记忆（智能检索）

results = memory.search("用户偏好", user_id="user123")

就这么简单！4 行代码，让你的 AI 拥有 "记忆力"！

结语：

还在为 AI 的 “金鱼记忆” 而烦恼吗？

还在为 Token 成本居高不下而头疼吗？

还在为检索准确率低而困扰吗？

是时候给 AI 装个 “外挂记忆” 了～

相关资源

• PowerMem GitHub 仓库

• seekdb GitHub 仓库

• PowerMem 官方网站

• seekdb 官方网站