""" 云文档配置:各平台 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)}