sdk/oapiSdk/test.py

from flask import Flask, request, jsonify
import os, base64, hashlib, time, requests, json

app = Flask(__name__)

APP_ID = None
APP_SECRET = None
REDIRECT_URI = None
code_verifier_store = {}
token_data = {}

# 生成 code_verifier / code_challenge
def generate_code_verifier():
    return base64.urlsafe_b64encode(os.urandom(32)).rstrip(b'=').decode('utf-8')

def generate_code_challenge(code_verifier):
    digest = hashlib.sha256(code_verifier.encode('utf-8')).digest()
    return base64.urlsafe_b64encode(digest).rstrip(b'=').decode('utf-8')

def save_token(app_id, token):
    with open(f"token_{app_id}.json", "w", encoding="utf-8") as f:
        json.dump(token, f)

def load_token(app_id):
    path = f"token_{app_id}.json"
    if os.path.exists(path):
        with open(path, "r", encoding="utf-8") as f:
            return json.load(f)
    return None

def ensure_token_valid():
    global token_data
    if not token_data:
        return False
    if time.time() >= token_data.get("expires_at", 0):
        # refresh token
        resp = requests.post(
            "https://open.feishu.cn/open-apis/authen/v2/oauth/token",
            json={
                "grant_type": "refresh_token",
                "client_id": APP_ID,
                "client_secret": APP_SECRET,
                "refresh_token": token_data["refresh_token"]
            }
        )
        data = resp.json()
        if data.get("code") == 0:
            token_data["access_token"] = data["access_token"]
            token_data["refresh_token"] = data["refresh_token"]
            token_data["expires_at"] = time.time() + data["expires_in"] - 60
            save_token(APP_ID, token_data)
            return True
        return False
    return True

# 1️⃣ 生成 PKCE 参数
@app.route("/pkce", methods=["GET"])
def get_pkce():
    global code_verifier_store
    code_verifier = generate_code_verifier()
    code_challenge = generate_code_challenge(code_verifier)
    # 保存，方便 callback 时使用
    code_verifier_store["verifier"] = code_verifier
    code_verifier_store["challenge"] = code_challenge
    return jsonify({
        "code_verifier": code_verifier,
        "code_challenge": code_challenge
    })

# 2️⃣ 回调换取 token
@app.route("/oauth/callback")
def oauth_callback():
    global token_data
    code = request.args.get("code")
    if not code:
        return jsonify({"error": "Missing code"}), 400

    if not all([APP_ID, APP_SECRET, REDIRECT_URI]):
        return jsonify({"error": "Missing config"}), 400

    verifier = code_verifier_store.get("verifier")
    if not verifier:
        return jsonify({"error": "Missing code_verifier"}), 400

    resp = requests.post(
        "https://open.feishu.cn/open-apis/authen/v2/oauth/token",
        json={
            "grant_type": "authorization_code",
            "client_id": APP_ID,
            "client_secret": APP_SECRET,
            "code": code,
            "redirect_uri": REDIRECT_URI,
            "code_verifier": verifier
        }
    )
    data = resp.json()
    print("[callback] token resp:", data)

    if data.get("code") == 0:
        token_data = {
            "access_token": data["access_token"],
            "refresh_token": data["refresh_token"],
            "expires_at": time.time() + data["expires_in"] - 60
        }
        save_token(APP_ID, token_data)
        return "授权成功，可以回到终端调用接口了。"
    else:
        return jsonify({"error": "Token获取失败", "raw": data}), 400

# 3️⃣ 设置配置参数
@app.route("/set_config", methods=["POST"])
def set_config():
    global APP_ID, APP_SECRET, REDIRECT_URI, token_data
    data = request.get_json(force=True)
    APP_ID = data.get("APP_ID")
    APP_SECRET = data.get("APP_SECRET")
    REDIRECT_URI = data.get("REDIRECT_URI")
    if not all([APP_ID, APP_SECRET, REDIRECT_URI]):
        return jsonify({"error": "缺少 APP_ID, APP_SECRET, REDIRECT_URI"}), 400

    token_data = load_token(APP_ID)
    return jsonify({"message": "配置成功", "token_data": token_data})

# 4️⃣ 获取日历和日程
@app.route("/calendar_events", methods=["POST"])
def calendar_events():
    global token_data
    if not ensure_token_valid():
        challenge = code_verifier_store.get("challenge")
        if not challenge:
            return jsonify({"error": "请先访问 /pkce 获取 PKCE 参数"}), 400
        auth_url = (
            f"https://accounts.feishu.cn/open-apis/authen/v1/authorize"
            f"?client_id={APP_ID}"
            f"&redirect_uri={REDIRECT_URI}"
            f"&scope=calendar:calendar.event:read%20calendar:calendar:readonly%20offline_access"
            f"&state=auth"
            f"&response_type=code"
            f"&code_challenge={challenge}"
            f"&code_challenge_method=S256"
        )
        return jsonify({"message": "需要授权", "auth_url": auth_url}), 401

    headers = {"Authorization": f"Bearer {token_data['access_token']}"}
    calendars_resp = requests.get("https://open.feishu.cn/open-apis/calendar/v4/calendars", headers=headers).json()
    if calendars_resp.get("code") != 0:
        return jsonify({"error": "获取日历列表失败", "raw": calendars_resp}), 500

    calendars = calendars_resp.get("data", {}).get("calendar_list", [])
    all_events = []
    for c in calendars:
        cid = c["calendar_id"]
        ev_resp = requests.get(
            f"https://open.feishu.cn/open-apis/calendar/v4/calendars/{cid}/events",
            headers=headers
        ).json()
        if ev_resp.get("code") == 0:
            all_events.extend(ev_resp.get("data", {}).get("items", []))

    return jsonify({"events": all_events})


if __name__ == "__main__":

    app.run(host="0.0.0.0", port=8888 ,debug=True)