网站扫码授权后换取用户凭证

import json
import os
import time
import tempfile
from pathlib import Path
from contextlib import contextmanager
from urllib.parse import urlencode
import urllib.error
import urllib.request

def env(name):
    value = os.environ.get(name)
    if not value or not value.strip() or "REPLACE_WITH" in value:
        raise SystemExit("请设置真实环境变量: " + name)
    return value

def positive_int(value, name):
    if isinstance(value, bool):
        raise SystemExit("无效期限字段: " + name)
    try:
        number = int(value)
    except (TypeError, ValueError):
        raise SystemExit("无效期限字段: " + name)
    if number <= 0:
        raise SystemExit("无效期限字段: " + name)
    return number

def load_record(path):
    try:
        record = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, ValueError):
        raise SystemExit("无法读取原凭证文件；请检查路径与安全存储格式")
    if not isinstance(record, dict):
        raise SystemExit("原凭证文件格式错误")
    return record

@contextmanager
def pending_file(value):
    path = Path(value).expanduser().absolute()
    if path.is_symlink() or (path.exists() and not path.is_file()):
        raise SystemExit("凭证目标必须是私有目录中的普通文件，不能是目录或符号链接")
    # 先创建 0600 临时文件并确认可写，再消费授权码或发送刷新请求。
    # 这能前置发现路径配置错误；不能保证请求后永不发生磁盘或系统故障。
    try:
        output = tempfile.NamedTemporaryFile("w", encoding="utf-8",
            dir=path.parent, prefix=".douyin-", delete=False)
    except OSError:
        raise SystemExit("无法在凭证目录创建文件；尚未发送请求")
    try:
        os.chmod(output.name, 0o600)
        output.flush()
        os.fsync(output.fileno())
        yield path, output
    finally:
        output.close()
        try:
            os.unlink(output.name)
        except FileNotFoundError:
            pass

def post(url, payload):
    request = urllib.request.Request(url, data=payload,
        headers={"Content-Type": "application/x-www-form-urlencoded"}, method="POST")
    try:
        with urllib.request.urlopen(request, timeout=30) as response:
            result = json.load(response)
    except urllib.error.HTTPError as error:
        raise SystemExit("HTTP " + str(error.code) + "；不输出可能含凭证的原文")
    except (urllib.error.URLError, TimeoutError):
        raise SystemExit("网络或超时错误；不要盲目重复消费一次性授权码")
    except (ValueError, UnicodeError):
        raise SystemExit("响应不是有效 JSON，原凭证文件未覆盖")
    if not isinstance(result, dict) or not isinstance(result.get("data"), dict):
        raise SystemExit("响应缺少 data，原凭证文件未覆盖")
    data = result["data"]
    if str(data.get("error_code")) != "0" or not data.get("access_token"):
        raise SystemExit("业务请求失败，业务码: " + str(data.get("error_code")))
    if not isinstance(data.get("open_id"), str) or not data["open_id"]:
        raise SystemExit("响应缺少用户归属，原凭证文件未覆盖")
    return data

def commit(path, output, record):
    json.dump(record, output, ensure_ascii=False)
    output.flush()
    os.fsync(output.fileno())
    output.close()
    os.replace(output.name, path)
    print("凭证已安全保存；请按绝对截止时间检查有效期，不输出凭证")

# 所有配置先读取和检查，缺失项不能在 code 已消费后才被发现。
client_key = env("DY_CLIENT_KEY")
client_secret = env("DY_CLIENT_SECRET")
auth_code = env("DY_AUTH_CODE")
token_file = env("DY_TOKEN_FILE")
with pending_file(token_file) as (path, output):
    previous = load_record(path) if path.exists() else None
    if previous and (previous.get("client_key") != client_key or not previous.get("open_id")):
        raise SystemExit("原凭证文件缺少归属或属于其他应用；改用该用户的独立文件")
    body = urlencode({"client_key": client_key, "client_secret": client_secret,
                      "code": auth_code, "grant_type": "authorization_code"}).encode("utf-8")
    started_at = int(time.time())
    data = post("https://open.douyin.com/oauth/access_token/", body)
    received_at = int(time.time())
    if previous and previous["open_id"] != data["open_id"]:
        raise SystemExit("返回用户与原文件不一致；没有覆盖原用户凭证")
    if not isinstance(data.get("refresh_token"), str) or not data["refresh_token"]:
        raise SystemExit("响应缺少刷新凭证，原文件未覆盖")
    # 以请求开始时刻加服务端时长，避免把网络耗时计入可用期限。
    record = dict(data, client_key=client_key, received_at=received_at,
        access_expires_at=started_at + positive_int(data.get("expires_in"), "expires_in"),
        refresh_expires_at=started_at + positive_int(data.get("refresh_expires_in"), "refresh_expires_in"))
    if min(record["access_expires_at"], record["refresh_expires_at"]) <= received_at:
        raise SystemExit("响应到达时期限已过，原文件未覆盖")
    commit(path, output, record)