import os import sys import json import time import base64 import hashlib import secrets import urllib.parse import webbrowser import requests import mimetypes from http.server import HTTPServer, BaseHTTPRequestHandler # ============================= # 基础配置 # ============================= CLIENT_KEY = os.getenv("TIKTOK_CLIENT_KEY", "").strip() CLIENT_SECRET = os.getenv("TIKTOK_CLIENT_SECRET", "").strip() REDIRECT_URI = os.getenv("TIKTOK_REDIRECT_URI", "").strip() TARGET = os.getenv("TIKTOK_TARGET", "").strip() # sandbox or empty AUTH_URL = "https://www.tiktok.com/v2/auth/authorize/" TOKEN_URL = "https://open.tiktokapis.com/v2/oauth/token/" TOKENS_FILE = "tiktok_tokens.json" if not CLIENT_KEY or not REDIRECT_URI: print("❌ 请先 export TIKTOK_CLIENT_KEY 和 TIKTOK_REDIRECT_URI") sys.exit(1) # ============================= # PKCE # ============================= def generate_pkce(): verifier = base64.urlsafe_b64encode(secrets.token_bytes(32)).decode().rstrip("=") challenge = base64.urlsafe_b64encode( hashlib.sha256(verifier.encode()).digest() ).decode().rstrip("=") return verifier, challenge # ============================= # OAuth 回调服务器 # ============================= class CallbackHandler(BaseHTTPRequestHandler): code = None error = None def do_GET(self): parsed = urllib.parse.urlparse(self.path) query = urllib.parse.parse_qs(parsed.query) if "code" in query: CallbackHandler.code = query["code"][0] self.send_response(200) self.end_headers() self.wfile.write(b"OAuth success. You may close this window.") elif "error" in query: CallbackHandler.error = query["error"][0] self.send_response(400) self.end_headers() self.wfile.write(b"OAuth error.") else: self.send_response(404) self.end_headers() def wait_for_code(port=8899, timeout=300): server = HTTPServer(("127.0.0.1", port), CallbackHandler) server.timeout = 1 start = time.time() while time.time() - start < timeout: server.handle_request() if CallbackHandler.code: return CallbackHandler.code if CallbackHandler.error: raise Exception(f"OAuth error: {CallbackHandler.error}") raise TimeoutError("⏳ 等待回调超时") # ============================= # OAuth 登录 # ============================= def oauth_login(): verifier, challenge = generate_pkce() params = { "client_key": CLIENT_KEY, "response_type": "code", "scope": os.getenv("TIKTOK_SCOPE","user.info.basic").strip(), "redirect_uri": REDIRECT_URI, "state": secrets.token_urlsafe(16), "code_challenge": challenge, "code_challenge_method": "S256", } if TARGET: params["target"] = TARGET auth_link = AUTH_URL + "?" + urllib.parse.urlencode(params) print("AUTH_URL=", auth_link) webbrowser.open(auth_link) code = wait_for_code() print("✅ 收到授权 code") data = { "client_key": CLIENT_KEY, "client_secret": CLIENT_SECRET, "code": code, "grant_type": "authorization_code", "redirect_uri": REDIRECT_URI, "code_verifier": verifier, } if TARGET: data["target"] = TARGET resp = requests.post(TOKEN_URL, data=data) result = resp.json() if "access_token" not in result: print("❌ 获取 token 失败:", result) sys.exit(1) with open(TOKENS_FILE, "w") as f: json.dump(result, f, indent=2) print("🎉 OAuth 成功，已保存 tiktok_tokens.json") # ============================= # 发布视频 # ============================= def post_video(video_path, caption): if not os.path.exists(TOKENS_FILE): print("❌ 请先运行 auth") return if not os.path.exists(video_path): print("❌ 视频文件不存在:", video_path) return with open(TOKENS_FILE) as f: tokens = json.load(f) access_token = tokens.get("access_token") headers = { "Authorization": f"Bearer {access_token}", "Content-Type": "application/json; charset=UTF-8", } # 查询 privacy_level creator_info_url = "https://open.tiktokapis.com/v2/post/publish/creator_info/query/" r = requests.post(creator_info_url, headers=headers, json={}) creator_res = r.json() if creator_res.get("error", {}).get("code") != "ok": print("❌ creator_info 查询失败:") print(json.dumps(creator_res, indent=2)) return options = creator_res.get("data", {}).get("privacy_level_options", []) if not options: print("❌ 未获取 privacy_level_options") return privacy_level = "SELF_ONLY" if "SELF_ONLY" in options else options[0] print("使用 privacy_level:", privacy_level) size = os.path.getsize(video_path) init_url = "https://open.tiktokapis.com/v2/post/publish/video/init/" body = { "post_info": { "title": caption, "privacy_level": privacy_level, "disable_comment": False, "disable_duet": False, "disable_stitch": False, }, "source_info": { "source": "FILE_UPLOAD", "video_size": size, "chunk_size": size, "total_chunk_count": 1, }, } r = requests.post(init_url, headers=headers, json=body) init_res = r.json() if init_res.get("error", {}).get("code") != "ok": print("❌ init 失败:") print(json.dumps(init_res, indent=2)) return upload_url = init_res["data"]["upload_url"] with open(video_path, "rb") as f: video_bytes = f.read() put_headers = { "Content-Type": mimetypes.guess_type(video_path)[0] or "video/mp4", "Content-Length": str(size), "Content-Range": f"bytes 0-{size-1}/{size}", } r = requests.put(upload_url, headers=put_headers, data=video_bytes) if r.status_code not in (200, 201, 204): print("❌ 视频上传失败:", r.status_code, r.text) return print("🎉 视频上传完成") # ============================= # CLI # ============================= if __name__ == "__main__": if len(sys.argv) < 2: print("用法:") print(" python tiktok_direct.py auth") print(" python tiktok_direct.py post video.mp4 \"caption\"") sys.exit(0) cmd = sys.argv[1] if cmd == "auth": oauth_login() elif cmd == "post": if len(sys.argv) < 4: print("用法: python tiktok_direct.py post video.mp4 \"caption\"") else: post_video(sys.argv[2], sys.argv[3])