108 lines
3.8 KiB
Python
108 lines
3.8 KiB
Python
#!/usr/bin/env python3
|
|
"""通过当前登录的抖音浏览器会话调用个人信息接口并保存 JSON。"""
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any, cast
|
|
|
|
from account_log import visible
|
|
from cdp_explicit import ( # pyright: ignore[reportMissingImports]
|
|
add_cdp_args,
|
|
configure,
|
|
evaluate_script,
|
|
)
|
|
|
|
API_PATH = "/aweme/v1/web/user/profile/self/?device_platform=webapp&aid=6383"
|
|
OUTPUT = Path(__file__).with_name("current_user.json")
|
|
|
|
|
|
BROWSER_SCRIPT = f"""(async () => {{
|
|
if (location.origin !== 'https://www.douyin.com') return JSON.stringify({{status: 0, body: '{{}}'}});
|
|
const response = await fetch({json.dumps(API_PATH)}, {{credentials: "include", signal: AbortSignal.timeout(15000)}});
|
|
return JSON.stringify({{status: response.status, body: await response.text()}});
|
|
}})()"""
|
|
|
|
|
|
def get_user_from_browser() -> dict:
|
|
"""在当前抖音页面内使用登录态调用 self profile API。"""
|
|
completed = evaluate_script(BROWSER_SCRIPT)
|
|
if completed.returncode:
|
|
raise RuntimeError("CDP 调用失败")
|
|
return parse_user_response(completed.stdout)
|
|
|
|
|
|
def parse_user_response(value) -> dict:
|
|
"""Shared self-profile parsing for CLI and asynchronous desktop enrollment."""
|
|
try:
|
|
response = json.loads(value) if isinstance(value, str) else value
|
|
payload = json.loads(response["body"])
|
|
if not isinstance(payload, dict):
|
|
raise TypeError("invalid profile payload")
|
|
except (IndexError, KeyError, TypeError, json.JSONDecodeError) as exc:
|
|
raise RuntimeError("无法解析接口响应") from exc
|
|
if (
|
|
response.get("status") != 200
|
|
or payload.get("status_code") != 0
|
|
or not payload.get("user")
|
|
):
|
|
raise RuntimeError("接口返回异常，请手动检查登录态")
|
|
user = payload["user"]
|
|
if not isinstance(user, dict):
|
|
raise RuntimeError("个人信息无效，请手动检查登录状态")
|
|
uid = user.get("uid")
|
|
if (
|
|
type(uid) not in (str, int)
|
|
or not str(uid).isascii()
|
|
or not str(uid).isdecimal()
|
|
or str(uid).startswith("0")
|
|
or len(str(uid)) > 20
|
|
):
|
|
raise RuntimeError("尚未登录，请在打开的浏览器中手动登录")
|
|
return {**user, "uid": str(uid)}
|
|
|
|
|
|
def business_user(user: dict) -> dict:
|
|
"""Return full platform business data while retaining credential protection."""
|
|
result = cast(dict[str, Any], visible(user))
|
|
avatar = result.get("avatar_300x300") or result.get("avatar_thumb") or {}
|
|
avatars = avatar.get("url_list") if isinstance(avatar, dict) else []
|
|
result["douyin_id"] = result.get("short_id") or result.get("unique_id") or ""
|
|
result["avatar_url"] = avatars[0] if avatars else result.get("avatar_url") or ""
|
|
return result
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--output", type=Path, default=Path("current_user.json"))
|
|
add_cdp_args(parser)
|
|
args = parser.parse_args()
|
|
configure(args)
|
|
try:
|
|
user = business_user(get_user_from_browser())
|
|
args.output.write_text(
|
|
json.dumps(
|
|
{"fetched_at": datetime.now().astimezone().isoformat(), "user": user},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
+ "\n",
|
|
encoding="utf-8",
|
|
)
|
|
except (OSError, RuntimeError) as exc:
|
|
print(f"获取失败：{exc}", file=sys.stderr)
|
|
print(
|
|
"请核对显式 CDP 地址和期望 UID；登录失效请手动登录。",
|
|
file=sys.stderr,
|
|
)
|
|
return 1
|
|
print(f"已保存：{args.output}")
|
|
print(json.dumps(user, ensure_ascii=False, indent=2))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|