143 lines
5.7 KiB
Python
143 lines
5.7 KiB
Python
#!/usr/bin/env python3
|
||
"""直接传入数字 UID，通过接口关注；已关注则跳过，不打开目标主页。"""
|
||
|
||
import argparse
|
||
import json
|
||
import subprocess
|
||
import sys
|
||
from urllib.parse import urlencode
|
||
|
||
from cdp_explicit import add_cdp_args, configure, evaluate_script
|
||
from get_current_user import get_user_from_browser
|
||
|
||
PROFILE = "/aweme/v1/web/user/profile/other/"
|
||
FOLLOW = "/aweme/v1/web/commit/follow/user/"
|
||
PARAMS = {"device_platform": "webapp", "aid": 6383, "channel": "channel_pc_web"}
|
||
|
||
|
||
def validate_uid(uid: str) -> str:
|
||
if (
|
||
not isinstance(uid, str)
|
||
or not uid.isascii()
|
||
or not uid.isdecimal()
|
||
or uid.startswith("0")
|
||
):
|
||
raise ValueError("UID 必须为不带前导零的正整数数字字符串，不是主页链接或抖音号")
|
||
return uid
|
||
|
||
|
||
def request(path: str, params: dict, body: dict | None = None) -> dict:
|
||
url = path + "?" + urlencode({**PARAMS, **params})
|
||
expression = """new Promise((resolve, reject) => {
|
||
if (location.origin !== 'https://www.douyin.com')
|
||
return reject(new Error('请先选择已登录的抖音标签页'));
|
||
const xhr = new XMLHttpRequest();
|
||
xhr.open(METHOD, URL);
|
||
xhr.withCredentials = true;
|
||
xhr.timeout = 20000;
|
||
if (BODY !== null)
|
||
xhr.setRequestHeader('Content-Type', 'application/x-www-form-urlencoded; charset=UTF-8');
|
||
xhr.onload = () => resolve(JSON.stringify({status: xhr.status, body: xhr.responseText}));
|
||
xhr.onerror = () => reject(new Error('网络错误；如已发送关注，请先查询状态，勿自动重试'));
|
||
xhr.ontimeout = () => reject(new Error('请求超时；如已发送关注，请先查询状态，勿自动重试'));
|
||
xhr.send(BODY);
|
||
})""".replace("METHOD", json.dumps("POST" if body is not None else "GET"))
|
||
expression = expression.replace("URL", json.dumps(url)).replace(
|
||
"BODY", json.dumps(urlencode(body) if body is not None else None)
|
||
)
|
||
result = evaluate_script(expression)
|
||
try:
|
||
response = json.loads(result.stdout.strip().splitlines()[-1])
|
||
payload = json.loads(response["body"])
|
||
if not isinstance(payload, dict):
|
||
raise ValueError("响应不是对象")
|
||
except (ValueError, KeyError, IndexError, TypeError) as exc:
|
||
raise RuntimeError("无法解析接口响应，停止操作") from exc
|
||
if response["status"] != 200 or payload.get("status_code") != 0:
|
||
raise RuntimeError(
|
||
f"接口失败：HTTP {response['status']}，status_code={payload.get('status_code')}；"
|
||
"请检查登录态或风控提示，不会自动登录、重试或处理验证码"
|
||
)
|
||
return payload
|
||
|
||
|
||
def get_profile(uid: str) -> dict:
|
||
user = request(PROFILE, {"user_id": validate_uid(uid)}).get("user")
|
||
if not isinstance(user, dict) or user.get("uid") != uid:
|
||
raise RuntimeError("返回用户 UID 与目标不匹配，停止操作")
|
||
return user
|
||
|
||
|
||
def follow_user(uid: str, check_only: bool = False) -> dict:
|
||
uid = validate_uid(uid)
|
||
# 复用登录态检查；失败即退出，绝不尝试登录。
|
||
try:
|
||
current = get_user_from_browser()
|
||
except RuntimeError as exc:
|
||
raise RuntimeError("无法确认登录状态，请手动检查或登录后再运行") from exc
|
||
if not current.get("uid") or str(current["uid"]) == "0":
|
||
raise RuntimeError("登录状态已失效，请手动登录后再运行")
|
||
if uid == str(current["uid"]):
|
||
raise ValueError("不能关注自己")
|
||
user = get_profile(uid)
|
||
status = user.get("follow_status")
|
||
if type(status) is not int or status not in (0, 1, 2):
|
||
raise RuntimeError(f"未知关注状态 {status!r}，停止而不发送请求")
|
||
action = "checked" if check_only else "already_following"
|
||
if not check_only and status == 0:
|
||
result = request(FOLLOW, {}, {"user_id": user["uid"], "type": 1})
|
||
if type(result.get("follow_status")) is not int or result[
|
||
"follow_status"
|
||
] not in (1, 2):
|
||
raise RuntimeError("关注响应未确认成功；请使用 --check 查询，勿自动重试")
|
||
verified = get_profile(uid)
|
||
if verified["uid"] != user["uid"] or verified.get("follow_status") not in (
|
||
1,
|
||
2,
|
||
):
|
||
raise RuntimeError(
|
||
"关注请求已发出，但回查未确认；请使用 --check 查询，勿自动重试"
|
||
)
|
||
status = verified["follow_status"]
|
||
action = "followed"
|
||
return {
|
||
"action": action,
|
||
"uid": user["uid"],
|
||
"sec_uid": user.get("sec_uid"),
|
||
"nickname": user.get("nickname"),
|
||
"follow_status": status,
|
||
}
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
|
||
parser.add_argument("uid", help="目标用户数字 UID（通知 users[].uid），不是抖音号")
|
||
parser.add_argument("--check", action="store_true", help="只查询，不关注")
|
||
add_cdp_args(parser)
|
||
args = parser.parse_args()
|
||
configure(args)
|
||
try:
|
||
uid = validate_uid(args.uid)
|
||
print(json.dumps(follow_user(uid, args.check), ensure_ascii=False, indent=2))
|
||
except (
|
||
OSError,
|
||
ValueError,
|
||
RuntimeError,
|
||
KeyError,
|
||
TypeError,
|
||
subprocess.SubprocessError,
|
||
) as exc:
|
||
# 不回显浏览器 stderr：它可能包含带签名的请求 URL。
|
||
message = (
|
||
"浏览器连接或执行失败，请检查 CDP 与登录态"
|
||
if isinstance(exc, subprocess.SubprocessError)
|
||
else str(exc)
|
||
)
|
||
print(f"失败：{message}。不会自动登录或重试关注。", file=sys.stderr)
|
||
return 1
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|