feat: add admin social login bindings

routes/api_social.py

@@ -6,7 +6,7 @@ from datetime import timedelta
 import database
 from app_logger import get_logger
 from db.utils import get_cst_now, get_cst_now_str
-from flask import Blueprint, jsonify, request
+from flask import Blueprint, jsonify, request, session
 from flask_login import current_user, login_required, login_user
 from services.accounts_service import load_user_accounts
 from services.models import User
@@ -82,6 +82,18 @@ def _binding_row(provider: str, binding: dict | None) -> dict:
     }
 
 
+def _admin_binding_row(provider: str, binding: dict | None) -> dict:
+    return {
+        "provider": provider,
+        "provider_label": provider_label(provider),
+        "bound": bool(binding),
+        "nickname": (binding or {}).get("nickname") or "",
+        "avatar_url": (binding or {}).get("avatar_url") or "",
+        "last_login_at": (binding or {}).get("last_login_at"),
+        "created_at": (binding or {}).get("created_at"),
+    }
+
+
 @api_social_bp.route("/api/auth/social/config", methods=["GET"])
 def social_public_config():
     return jsonify(public_social_config(database.get_system_config()))
@@ -135,6 +147,10 @@ def social_callback():
         return _social_error(error)
 
     binding = database.find_social_login_binding(profile.provider, profile.social_uid)
+    admin_binding = database.find_admin_social_login_binding_by_identity(profile.provider, profile.social_uid)
+    if admin_binding:
+        return jsonify({"error": "该第三方账号已绑定管理员账号"}), 409
+
     if binding:
         if mode == "bind":
             current_id = int(getattr(current_user, "id", 0) or 0)
@@ -204,6 +220,9 @@ def bind_social_account():
     existing_identity = database.find_social_login_binding(provider, social_uid)
     if existing_identity and int(existing_identity.get("user_id") or 0) != int(current_user.id):
         return jsonify({"error": "该第三方账号已绑定其他用户"}), 409
+    existing_admin_identity = database.find_admin_social_login_binding_by_identity(provider, social_uid)
+    if existing_admin_identity:
+        return jsonify({"error": "该第三方账号已绑定管理员账号"}), 409
 
     existing_provider = database.find_user_social_login_binding(int(current_user.id), provider)
     if existing_provider and str(existing_provider.get("social_uid") or "") != social_uid:
@@ -242,6 +261,138 @@ def admin_social_config():
     return protected()
 
 
+@api_social_bp.route("/yuyx/api/admin/social-bindings", methods=["GET"])
+def list_admin_social_bindings():
+    from routes.decorators import admin_required
+
+    @admin_required
+    def _inner():
+        cfg = database.get_system_config()
+        providers = parse_providers(cfg.get("social_login_providers")) or list(PROVIDER_LABELS.keys())
+        admin_id = int(session.get("admin_id") or 0)
+        existing = {
+            item["provider"]: item
+            for item in database.list_admin_social_login_bindings(admin_id)
+        }
+        public_cfg = public_social_config(cfg)
+        return jsonify(
+            {
+                "enabled": bool(public_cfg.get("enabled")),
+                "providers": providers,
+                "items": [_admin_binding_row(provider, existing.get(provider)) for provider in providers],
+            }
+        )
+
+    return _inner()
+
+
+@api_social_bp.route("/yuyx/api/admin/social-login-url", methods=["POST"])
+def admin_social_login_url():
+    from routes.decorators import admin_required
+
+    @admin_required
+    def _inner():
+        data = _get_json_payload()
+        provider = str(data.get("provider") or "").strip().lower()
+        redirect_uri = str(data.get("redirect_uri") or "").strip()
+        try:
+            result = fetch_social_login_url(
+                database.get_system_config(),
+                provider=provider,
+                mode="bind",
+                redirect_uri=redirect_uri,
+                allowed_hosts=_allowed_redirect_hosts(),
+            )
+        except SocialLoginError as error:
+            logger.warning(f"[admin/social/login-url] provider={provider or '-'} failed: {error.message}")
+            return _social_error(error)
+        return jsonify(result)
+
+    return _inner()
+
+
+@api_social_bp.route("/yuyx/api/admin/social-poll", methods=["POST"])
+def admin_social_poll():
+    from routes.decorators import admin_required
+
+    @admin_required
+    def _inner():
+        data = _get_json_payload()
+        provider = str(data.get("provider") or "").strip().lower()
+        state = str(data.get("state") or "").strip()
+        try:
+            result = poll_social_scan(database.get_system_config(), provider=provider, state=state)
+        except SocialLoginError as error:
+            logger.warning(f"[admin/social/poll] provider={provider or '-'} failed: {error.message}")
+            return _social_error(error)
+        return jsonify(result)
+
+    return _inner()
+
+
+@api_social_bp.route("/yuyx/api/admin/social-bindings/<provider>/callback", methods=["POST"])
+def bind_admin_social_callback(provider):
+    from routes.decorators import admin_required
+
+    @admin_required
+    def _inner():
+        data = _get_json_payload()
+        provider_value = str(data.get("provider") or provider or data.get("type") or "").strip().lower()
+        code = str(data.get("code") or "").strip()
+        admin_id = int(session.get("admin_id") or 0)
+
+        try:
+            profile = fetch_space_profile(database.get_system_config(), provider=provider_value, code=code)
+        except SocialLoginError as error:
+            logger.warning(f"[admin/social/callback] provider={provider_value or '-'} failed: {error.message}")
+            return _social_error(error)
+
+        user_identity = database.find_social_login_binding(profile.provider, profile.social_uid)
+        if user_identity:
+            return jsonify({"error": "该第三方账号已绑定普通用户"}), 409
+
+        existing_identity = database.find_admin_social_login_binding_by_identity(profile.provider, profile.social_uid)
+        if existing_identity and int(existing_identity.get("admin_id") or 0) != admin_id:
+            return jsonify({"error": "该第三方账号已绑定其他管理员"}), 409
+
+        existing_provider = database.find_admin_social_login_binding(admin_id, profile.provider)
+        if existing_provider and str(existing_provider.get("social_uid") or "") != profile.social_uid:
+            return jsonify({"error": f"当前管理员已绑定{provider_label(profile.provider)}"}), 409
+
+        binding = database.upsert_admin_social_login_binding(
+            admin_id=admin_id,
+            provider=profile.provider,
+            social_uid=profile.social_uid,
+            nickname=profile.nickname,
+            avatar_url=profile.avatar_url,
+        )
+        if not binding:
+            return jsonify({"error": "该第三方账号已绑定其他管理员"}), 409
+
+        logger.info(f"[admin/social/bind] admin_id={admin_id} provider={profile.provider}")
+        return jsonify({"success": True, "item": _admin_binding_row(profile.provider, binding)})
+
+    return _inner()
+
+
+@api_social_bp.route("/yuyx/api/admin/social-bindings/<provider>", methods=["DELETE"])
+def unbind_admin_social_account(provider):
+    from routes.decorators import admin_required
+
+    @admin_required
+    def _inner():
+        provider_value = str(provider or "").strip().lower()
+        if provider_value not in PROVIDER_LABELS:
+            return jsonify({"error": "不支持的登录方式"}), 400
+        admin_id = int(session.get("admin_id") or 0)
+        if not database.delete_admin_social_login_binding(admin_id, provider_value):
+            return jsonify({"error": "绑定记录不存在"}), 404
+        logger.info(f"[admin/social/unbind] admin_id={admin_id} provider={provider_value}")
+        return jsonify({"success": True})
+
+    return _inner()
+
+
 @api_social_bp.route("/yuyx/api/social-login/test", methods=["POST"])
 def test_admin_social_config():
     from routes.decorators import admin_required