s''

app/sdk/memos.py

@@ -0,0 +1,95 @@
+"""
+Memos SDK
+用户创建等操作，仅 nomadvip 使用
+"""
+import logging
+from typing import Optional
+
+import requests
+
+
+class MemosSDK:
+    """Memos 用户管理 SDK"""
+
+    def __init__(
+        self,
+        api_url: str,
+        token: str,
+        enabled: bool = True,
+    ):
+        self.api_url = api_url.rstrip("/")
+        self.token = token
+        self.enabled = enabled and bool(api_url and token)
+
+    def create_user(
+        self,
+        user_id: str,
+        password: str = "123456",
+    ) -> dict:
+        """
+        创建 Memos 用户
+        :param user_id: PocketBase 用户 ID，如 user1766043770158
+        :return: {"success": bool, "data"?: dict, "error"?: str}
+        """
+        if not self.enabled:
+            return {"success": False, "error": "Memos 未启用"}
+
+        try:
+            if user_id.startswith("user_"):
+                cleaned = "user" + user_id.split("user_", 1)[1]
+            elif "_" in user_id:
+                cleaned = user_id.replace("_", "")
+            else:
+                cleaned = user_id
+
+            username = cleaned
+            data = {
+                "username": username,
+                "password": password,
+                "role": "USER",
+                "state": "NORMAL",
+            }
+            headers = {
+                "Content-Type": "application/json",
+                "Authorization": f"Bearer {self.token}",
+            }
+            resp = requests.post(
+                self.api_url,
+                json=data,
+                headers=headers,
+                timeout=10,
+            )
+            if resp.status_code in (200, 201):
+                logging.info(f"memos 用户创建成功: {username}")
+                return {"success": True, "data": resp.json()}
+            else:
+                logging.error(f"memos 创建失败 - username: {username}, status: {resp.status_code}, response: {resp.text}")
+                return {"success": False, "error": resp.text}
+        except Exception as e:
+            logging.error(f"创建 memos 用户异常 - user_id: {user_id}, error: {e}")
+            return {"success": False, "error": str(e)}
+
+    def create_user_by_username(
+        self,
+        username: str,
+        password: str = "123456",
+    ) -> dict:
+        """按用户名直接创建（用于 /create_user 接口）"""
+        if not self.enabled:
+            return {"success": False, "error": "Memos 未启用"}
+        try:
+            data = {
+                "username": username,
+                "password": password,
+                "role": "USER",
+                "state": "NORMAL",
+            }
+            headers = {
+                "Content-Type": "application/json",
+                "Authorization": f"Bearer {self.token}",
+            }
+            resp = requests.post(self.api_url, json=data, headers=headers, timeout=10)
+            return resp.json()
+        except Exception as e:
+            logging.error(f"创建 memos 用户异常 - username: {username}, error: {e}")
+            return {"error": str(e)}