在授权期限内刷新用户凭证

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("凭证已安全保存；请按绝对截止时间检查有效期，不输出凭证")

# 凭证文件必须来自本页新版初始交换；缺少绝对截止时间时不能猜测补值。
client_key = env("DY_CLIENT_KEY")
refresh_token = env("DY_REFRESH_TOKEN")
token_file = env("DY_TOKEN_FILE")
with pending_file(token_file) as (path, output):
    previous = load_record(path)
    if previous.get("client_key") != client_key or not previous.get("open_id"):
        raise SystemExit("原凭证文件缺少归属或属于其他应用；尚未发送请求")
    if previous.get("refresh_token") != refresh_token:
        raise SystemExit("环境刷新凭证与原文件不一致；尚未发送请求")
    original_deadline = positive_int(previous.get("refresh_expires_at"), "refresh_expires_at")
    started_at = int(time.time())
    if original_deadline <= started_at:
        raise SystemExit("原 refresh_token 已到期，请用户重新授权；尚未发送请求")
    body = urlencode({"client_key": client_key, "refresh_token": refresh_token,
                      "grant_type": "refresh_token"}).encode("utf-8")
    data = post("https://open.douyin.com/oauth/refresh_token/", body)
    received_at = int(time.time())
    if data["open_id"] != previous["open_id"]:
        raise SystemExit("刷新结果用户不一致，原文件未覆盖")
    # 普通 refresh_token 接口不延长 refresh_token 本身的有效期。
    # 即使响应给了较长相对时长，也绝不从现在重新计满原期限。
    refresh_deadline = original_deadline
    if data.get("refresh_expires_in") is not None:
        refresh_deadline = min(original_deadline, started_at +
            positive_int(data["refresh_expires_in"], "refresh_expires_in"))
    record = dict(previous)
    record.update(data)
    record.update(client_key=client_key, open_id=previous["open_id"], received_at=received_at,
        access_expires_at=started_at + positive_int(data.get("expires_in"), "expires_in"),
        refresh_expires_at=refresh_deadline)
    if not isinstance(record.get("refresh_token"), str) or not record["refresh_token"]:
        raise SystemExit("刷新结果缺少凭证，原文件未覆盖")
    if min(record["access_expires_at"], refresh_deadline) <= received_at:
        raise SystemExit("响应到达时期限已过，原文件未覆盖")
    commit(path, output, record)