140 lines
4.8 KiB
Python
140 lines
4.8 KiB
Python
"""
|
||
云文档配置：各平台 API 凭证的存储与读取。
|
||
飞书 App ID/Secret、语雀 Token、腾讯文档 Client ID/Secret。
|
||
"""
|
||
import json
|
||
from pathlib import Path
|
||
from typing import Any, Dict
|
||
|
||
from fastapi import APIRouter, HTTPException
|
||
from pydantic import BaseModel, Field
|
||
|
||
router = APIRouter(prefix="/settings/cloud-doc-config", tags=["cloud-doc-config"])
|
||
|
||
CONFIG_PATH = Path("data/cloud_doc_credentials.json")
|
||
|
||
PLATFORMS = ("feishu", "yuque", "tencent")
|
||
|
||
|
||
class FeishuConfig(BaseModel):
|
||
app_id: str = Field("", description="飞书应用 App ID")
|
||
app_secret: str = Field("", description="飞书应用 App Secret")
|
||
|
||
|
||
class YuqueConfig(BaseModel):
|
||
token: str = Field("", description="语雀 Personal Access Token")
|
||
default_repo: str = Field("", description="默认知识库 namespace，如 my/repo")
|
||
|
||
|
||
class TencentConfig(BaseModel):
|
||
client_id: str = Field("", description="腾讯文档应用 Client ID")
|
||
client_secret: str = Field("", description="腾讯文档应用 Client Secret")
|
||
|
||
|
||
class FeishuConfigRead(BaseModel):
|
||
app_id: str = ""
|
||
app_secret_configured: bool = False
|
||
|
||
|
||
class YuqueConfigRead(BaseModel):
|
||
token_configured: bool = False
|
||
default_repo: str = ""
|
||
|
||
|
||
class TencentConfigRead(BaseModel):
|
||
client_id: str = ""
|
||
client_secret_configured: bool = False
|
||
|
||
|
||
class CloudDocConfigRead(BaseModel):
|
||
feishu: FeishuConfigRead
|
||
yuque: YuqueConfigRead
|
||
tencent: TencentConfigRead
|
||
|
||
|
||
def _load_config() -> Dict[str, Any]:
|
||
if not CONFIG_PATH.exists():
|
||
return {}
|
||
try:
|
||
data = json.loads(CONFIG_PATH.read_text(encoding="utf-8"))
|
||
return data if isinstance(data, dict) else {}
|
||
except Exception:
|
||
return {}
|
||
|
||
|
||
def _save_config(data: Dict[str, Any]) -> None:
|
||
CONFIG_PATH.parent.mkdir(parents=True, exist_ok=True)
|
||
CONFIG_PATH.write_text(
|
||
json.dumps(data, ensure_ascii=False, indent=2),
|
||
encoding="utf-8",
|
||
)
|
||
|
||
|
||
def _mask_secrets_for_read(raw: Dict[str, Any]) -> CloudDocConfigRead:
|
||
f = raw.get("feishu") or {}
|
||
y = raw.get("yuque") or {}
|
||
t = raw.get("tencent") or {}
|
||
return CloudDocConfigRead(
|
||
feishu=FeishuConfigRead(
|
||
app_id=f.get("app_id") or "",
|
||
app_secret_configured=bool((f.get("app_secret") or "").strip()),
|
||
),
|
||
yuque=YuqueConfigRead(
|
||
token_configured=bool((y.get("token") or "").strip()),
|
||
default_repo=(y.get("default_repo") or "").strip(),
|
||
),
|
||
tencent=TencentConfigRead(
|
||
client_id=t.get("client_id") or "",
|
||
client_secret_configured=bool((t.get("client_secret") or "").strip()),
|
||
),
|
||
)
|
||
|
||
|
||
@router.get("", response_model=CloudDocConfigRead)
|
||
async def get_cloud_doc_config():
|
||
"""获取云文档配置（凭证以是否已配置返回，不返回明文）。"""
|
||
raw = _load_config()
|
||
return _mask_secrets_for_read(raw)
|
||
|
||
|
||
@router.put("", response_model=CloudDocConfigRead)
|
||
async def update_cloud_doc_config(payload: Dict[str, Any]):
|
||
"""
|
||
更新云文档配置。传各平台字段，未传的保留原值。
|
||
例: { "feishu": { "app_id": "xxx", "app_secret": "yyy" }, "yuque": { "token": "zzz", "default_repo": "a/b" } }
|
||
"""
|
||
raw = _load_config()
|
||
for platform in PLATFORMS:
|
||
if platform not in payload or not isinstance(payload[platform], dict):
|
||
continue
|
||
p = payload[platform]
|
||
if platform == "feishu":
|
||
if "app_id" in p and p["app_id"] is not None:
|
||
raw.setdefault("feishu", {})["app_id"] = str(p["app_id"]).strip()
|
||
if "app_secret" in p and p["app_secret"] is not None:
|
||
raw.setdefault("feishu", {})["app_secret"] = str(p["app_secret"]).strip()
|
||
elif platform == "yuque":
|
||
if "token" in p and p["token"] is not None:
|
||
raw.setdefault("yuque", {})["token"] = str(p["token"]).strip()
|
||
if "default_repo" in p and p["default_repo"] is not None:
|
||
raw.setdefault("yuque", {})["default_repo"] = str(p["default_repo"]).strip()
|
||
elif platform == "tencent":
|
||
if "client_id" in p and p["client_id"] is not None:
|
||
raw.setdefault("tencent", {})["client_id"] = str(p["client_id"]).strip()
|
||
if "client_secret" in p and p["client_secret"] is not None:
|
||
raw.setdefault("tencent", {})["client_secret"] = str(p["client_secret"]).strip()
|
||
_save_config(raw)
|
||
return _mask_secrets_for_read(raw)
|
||
|
||
|
||
def get_credentials(platform: str) -> Dict[str, str]:
|
||
"""供 cloud_doc_service 使用：读取某平台明文凭证。"""
|
||
raw = _load_config()
|
||
return (raw.get(platform) or {}).copy()
|
||
|
||
|
||
def get_all_credentials() -> Dict[str, Dict[str, str]]:
|
||
"""供推送流程使用：读取全部平台凭证（明文）。"""
|
||
raw = _load_config()
|
||
return {k: dict(v) for k, v in raw.items() if isinstance(v, dict)}
|