96 lines
3.0 KiB
Python
96 lines
3.0 KiB
Python
"""
|
||
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 = "12345678",
|
||
) -> 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 = "12345678",
|
||
) -> 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)}
|