add d1kt SSO integration: OAuth2 login via WebView + phone binding

sentence_api/main.py

@@ -5,6 +5,8 @@ import mimetypes
 import shutil
 import subprocess
 import uuid
+import urllib.request
+import urllib.error
 from pathlib import Path
 from typing import Any, BinaryIO, Dict, List, Optional
 import json
@@ -41,6 +43,9 @@ from .models import (
     SentenceBoundaryDocument,
     SentenceTextUpdate,
     UserDubShareSummary,
+    SsoExchangeRequest,
+    SsoExchangeResponse,
+    SsoBindPhoneRequest,
     UserPublic,
     UserResultSummary,
     UserAdminListResponse,
@@ -201,6 +206,68 @@ def create_app(
             video_repository.delete_session(token)
         return Response(status_code=204)
 
+    D1KT_USERINFO_URL = "https://d1kt.cn/api/oauth2/userinfo"
+
+    @application.post("/api/v1/auth/sso/exchange", response_model=SsoExchangeResponse)
+    def sso_exchange(payload: SsoExchangeRequest) -> SsoExchangeResponse:
+        # Call the 第一课堂 userinfo endpoint to validate the access_token.
+        req = urllib.request.Request(
+            D1KT_USERINFO_URL,
+            headers={"Authorization": f"Bearer {payload.access_token}"},
+        )
+        try:
+            with urllib.request.urlopen(req, timeout=10) as resp:
+                userinfo = json.loads(resp.read().decode("utf-8"))
+        except (urllib.error.URLError, urllib.error.HTTPError, json.JSONDecodeError) as exc:
+            raise HTTPException(
+                status_code=401,
+                detail=f"第一课堂 access_token validation failed: {exc}",
+            )
+
+        d1kt_username = userinfo.get("username") or userinfo.get("name") or ""
+        if not d1kt_username:
+            raise HTTPException(status_code=401, detail="第一课堂 userinfo did not contain a username.")
+
+        fullname = userinfo.get("fullname") or userinfo.get("lastname") or d1kt_username
+        email = userinfo.get("email") or ""
+
+        # Try to find an existing local user with the same username.
+        existing = video_repository.get_user_by_username(d1kt_username)
+
+        if existing is None:
+            # Create a new SSO-linked user.
+            user = video_repository.create_sso_user(
+                username=d1kt_username,
+                nickname=fullname,
+                email=email,
+                sso_provider="d1kt",
+            )
+            if user is None:
+                raise HTTPException(status_code=409, detail="Failed to create SSO user.")
+            is_new_user = True
+        else:
+            user = existing
+            is_new_user = False
+
+        phone_bound = bool(user.get("phone_bound"))
+        token = video_repository.create_session(user["id"])
+        return SsoExchangeResponse(
+            token=token,
+            user=UserPublic(**user),
+            phone_bound=phone_bound,
+            is_new_user=is_new_user,
+        )
+
+    @application.post("/api/v1/auth/sso/bind-phone", response_model=UserPublic)
+    def sso_bind_phone(
+        payload: SsoBindPhoneRequest,
+        user: AuthenticatedUser = Depends(current_user),
+    ) -> UserPublic:
+        updated_user = video_repository.bind_phone(user.id, payload.phone)
+        if updated_user is None:
+            raise HTTPException(status_code=404, detail="User was not found.")
+        return UserPublic(**updated_user)
+
     @application.get("/api/v1/courses", response_model=List[CourseSummary])
     def list_courses(user: AuthenticatedUser = Depends(current_user)) -> List[CourseSummary]:
         return [