416 lines
17 KiB
Python
416 lines
17 KiB
Python
"""Explicit async CDP session. No global page, no daemon, no UI business actions."""
|
||
|
||
import asyncio
|
||
import json
|
||
import uuid
|
||
from contextlib import suppress
|
||
from urllib.parse import urlsplit
|
||
|
||
from account_store import decode
|
||
from douyin_im import EXPRESSION
|
||
from follow_user import validate_uid
|
||
from get_current_user import BROWSER_SCRIPT, business_user, parse_user_response
|
||
from subscribe_notifications import (
|
||
INSTALL,
|
||
WAIT,
|
||
detail_request_script,
|
||
history_request_script,
|
||
works_request_script,
|
||
)
|
||
|
||
|
||
class SessionError(RuntimeError):
|
||
pass
|
||
|
||
|
||
FOLLOW_JS = r"""async p => {
|
||
let sent=false;
|
||
if(location.origin!=='https://www.douyin.com') return {status:'failed',code:'WRONG_ORIGIN'};
|
||
try {
|
||
const get=async path=>{const r=await fetch(path,{credentials:'include',signal:AbortSignal.timeout(15000)});
|
||
const v=await r.json(); if(!r.ok||v.status_code!==0) throw Error('READ_FAILED'); return v;};
|
||
const prefix='?device_platform=webapp&aid=6383&channel=channel_pc_web';
|
||
const self=await get('/aweme/v1/web/user/profile/self/'+prefix);
|
||
if(String(self.user?.uid)!==p.expected) throw Error('IDENTITY_MISMATCH');
|
||
if(p.target===p.expected) throw Error('SELF_TARGET');
|
||
const path='/aweme/v1/web/user/profile/other/'+prefix+'&user_id='+p.target;
|
||
const profile=(await get(path)).user;
|
||
if(String(profile?.uid)!==p.target) throw Error('TARGET_MISMATCH');
|
||
if(![0,1,2].includes(profile.follow_status)) throw Error('UNKNOWN_FOLLOW_STATE');
|
||
if(profile.follow_status!==0 || p.check) return {status:'succeeded',follow_status:profile.follow_status,action:p.check?'checked':'already_following'};
|
||
sent=true;
|
||
const r=await fetch('/aweme/v1/web/commit/follow/user/'+prefix,{method:'POST',credentials:'include',
|
||
headers:{'Content-Type':'application/x-www-form-urlencoded;charset=UTF-8'},
|
||
body:new URLSearchParams({user_id:p.target,type:'1'}),signal:AbortSignal.timeout(20000)});
|
||
const result=await r.json();
|
||
if(r.ok && typeof result.status_code==='number' && result.status_code!==0)
|
||
return {status:'failed',code:result.status_code};
|
||
if(!r.ok || result.status_code!==0 || ![1,2].includes(result.follow_status)) return {status:'unknown',code:'UNCONFIRMED'};
|
||
const verify=(await get(path)).user;
|
||
return {status:String(verify?.uid)===p.target && [1,2].includes(verify.follow_status)?'succeeded':'unknown',action:'followed'};
|
||
} catch(e) {
|
||
if(sent) {
|
||
try {
|
||
const get=async path=>{const r=await fetch(path,{credentials:'include',signal:AbortSignal.timeout(10000)});return await r.json();};
|
||
const q='?device_platform=webapp&aid=6383&channel=channel_pc_web';
|
||
const me=await get('/aweme/v1/web/user/profile/self/'+q);
|
||
if(me.status_code===0 && String(me.user?.uid)===p.expected) {
|
||
const other=await get('/aweme/v1/web/user/profile/other/'+q+'&user_id='+p.target);
|
||
if(other.status_code===0 && String(other.user?.uid)===p.target && [1,2].includes(other.user.follow_status))
|
||
return {status:'succeeded',action:'verified_after_timeout'};
|
||
}
|
||
} catch(_) {}
|
||
}
|
||
return {status:sent?'unknown':'failed',code:['READ_FAILED','IDENTITY_MISMATCH','SELF_TARGET','TARGET_MISMATCH','UNKNOWN_FOLLOW_STATE'].includes(e.message)?e.message:'REQUEST_FAILED'};
|
||
}
|
||
}"""
|
||
|
||
|
||
def validate_endpoint(endpoint):
|
||
try:
|
||
url = urlsplit(endpoint)
|
||
valid = (
|
||
url.scheme == "http"
|
||
and url.hostname in ("127.0.0.1", "localhost", "::1")
|
||
and 0 < url.port < 65536
|
||
)
|
||
except (ValueError, TypeError):
|
||
raise ValueError("CDP 地址或端口无效") from None
|
||
if (
|
||
not valid
|
||
or url.username
|
||
or url.password
|
||
or url.query
|
||
or url.fragment
|
||
or url.path not in ("", "/")
|
||
):
|
||
raise ValueError("CDP 必须显式指定本机 http://127.0.0.1:端口，不允许远程地址")
|
||
return endpoint
|
||
|
||
|
||
class Session:
|
||
def __init__(self, playwright, endpoint, expected_uid, page_url=None):
|
||
self.playwright = playwright
|
||
self.endpoint = validate_endpoint(endpoint)
|
||
self.uid = validate_uid(expected_uid) if expected_uid is not None else None
|
||
self.page_url = page_url
|
||
self.browser = None
|
||
self.page = None
|
||
self.key = "__douyin_accounts_" + uuid.uuid4().hex
|
||
|
||
async def connect(self, discover=False):
|
||
if not self.browser or not self.browser.is_connected():
|
||
try:
|
||
self.browser = await self.playwright.chromium.connect_over_cdp(
|
||
self.endpoint, timeout=15000
|
||
)
|
||
except Exception:
|
||
raise SessionError("浏览器未连接，请启动账号浏览器") from None
|
||
if len(self.browser.contexts) != 1:
|
||
raise SessionError("浏览器上下文不唯一，拒绝自动选择")
|
||
pages = []
|
||
for page in self.browser.contexts[0].pages:
|
||
url = urlsplit(page.url)
|
||
if (
|
||
url.scheme == "https"
|
||
and url.netloc == "www.douyin.com"
|
||
and (
|
||
not url.path.startswith("/user/")
|
||
or url.path.rstrip("/") == "/user/self"
|
||
)
|
||
and (self.page_url is None or page.url == self.page_url)
|
||
):
|
||
pages.append(page)
|
||
if len(pages) != 1:
|
||
raise SessionError(
|
||
"请只保留一个抖音首页/自身页标签；不会自动选择其他账号或他人主页"
|
||
)
|
||
self.page = pages[0]
|
||
if discover and self.uid is None:
|
||
return await self.profile()
|
||
return await self.identity()
|
||
|
||
async def evaluate(self, expression, arg=None, *, main_world=False):
|
||
if not self.page or self.page.is_closed():
|
||
raise SessionError("账号页面已关闭")
|
||
try:
|
||
async with asyncio.timeout(55):
|
||
return await self.page.evaluate(
|
||
expression, arg, isolated_context=not main_world
|
||
)
|
||
except Exception:
|
||
raise SessionError(
|
||
"页面调用失败：请检查登录、网络、页面刷新或 SDK 变化"
|
||
) from None
|
||
|
||
async def profile(self):
|
||
try:
|
||
user = business_user(
|
||
parse_user_response(
|
||
await self.evaluate(BROWSER_SCRIPT, main_world=True)
|
||
)
|
||
)
|
||
except (RuntimeError, ValueError, TypeError, AttributeError):
|
||
raise SessionError(
|
||
"请在打开的浏览器中手动登录，登录后会自动读取账号信息"
|
||
) from None
|
||
return {
|
||
**user,
|
||
"nickname": user.get("nickname") or "",
|
||
"avatar": user.get("avatar_url") or "",
|
||
}
|
||
|
||
def require_bound(self):
|
||
if self.uid is None:
|
||
raise SessionError("账号尚未完成登录绑定，禁止执行业务操作")
|
||
|
||
async def identity(self):
|
||
self.require_bound()
|
||
user = await self.profile()
|
||
if user.get("uid") != self.uid:
|
||
raise SessionError(
|
||
"身份未通过核验：请手动登录期望账号，不会自动登录或继续写入"
|
||
)
|
||
return user
|
||
|
||
async def json(self, expression, *, main_world=False):
|
||
value = await self.evaluate(expression, main_world=main_world)
|
||
return decode(value) if isinstance(value, str) else value
|
||
|
||
async def install(self):
|
||
await self.identity()
|
||
await self.uninstall()
|
||
result = await self.json(
|
||
INSTALL.replace("KEY", json.dumps(self.key)).replace(
|
||
"UID", json.dumps(self.uid)
|
||
),
|
||
main_world=True,
|
||
)
|
||
if not result.get("connected"):
|
||
raise SessionError("通知连接未就绪，整组已暂停")
|
||
|
||
async def uninstall(self):
|
||
if self.page and not self.page.is_closed():
|
||
with suppress(SessionError):
|
||
await self.evaluate(
|
||
"key => {window[key]?.dispose(); delete window[key];}",
|
||
self.key,
|
||
main_world=True,
|
||
)
|
||
|
||
async def wait(self):
|
||
return await self.json(
|
||
WAIT.replace("KEY", json.dumps(self.key)), main_world=True
|
||
)
|
||
|
||
async def details(self, ids):
|
||
await self.identity()
|
||
response = await self.json(detail_request_script(ids), main_world=True)
|
||
if response.get("status") != 200:
|
||
raise SessionError("通知详情 HTTP 失败，ID 已保存")
|
||
# Parse in Python, preserving 64-bit numeric IDs even when no *_str field is present.
|
||
try:
|
||
value = json.loads(response["body"])
|
||
except (KeyError, TypeError, ValueError):
|
||
raise SessionError("通知详情响应不是有效 JSON，ID 已保存") from None
|
||
if not isinstance(value, dict) or value.get("status_code") != 0:
|
||
code = value.get("status_code") if isinstance(value, dict) else None
|
||
code = code if type(code) is int else "未知"
|
||
raise SessionError(f"通知详情接口失败（业务码 {code}），ID 已保存")
|
||
if "notice_list_v2" not in value:
|
||
raise SessionError("通知详情响应缺少列表字段，ID 已保存")
|
||
# A successful null/empty/partial response means unavailable IDs, not a lost login/socket.
|
||
notices = value["notice_list_v2"] if value["notice_list_v2"] is not None else []
|
||
if not isinstance(notices, list):
|
||
raise SessionError("通知详情列表格式异常，ID 已保存")
|
||
if any(
|
||
not isinstance(n, dict) or str(n.get("user_id")) != self.uid
|
||
for n in notices
|
||
):
|
||
raise SessionError("通知详情身份不符")
|
||
if not {n.get("nid_str") or str(n.get("nid")) for n in notices}.issubset(ids):
|
||
raise SessionError("通知详情返回了未请求的 ID，已停止处理")
|
||
return notices
|
||
|
||
async def history_notices(self, stop_ids=None):
|
||
notices = {}
|
||
stop_ids = set(stop_ids or ())
|
||
min_time = max_time = 0
|
||
seen_cursors = set()
|
||
for _ in range(1000):
|
||
response = await self.json(
|
||
history_request_script(min_time, max_time), main_world=True
|
||
)
|
||
try:
|
||
status = int(response["status"])
|
||
payload = json.loads(response["body"])
|
||
except (KeyError, TypeError, ValueError, json.JSONDecodeError) as exc:
|
||
raise SessionError("历史通知响应格式无效") from exc
|
||
if status != 200 or payload.get("status_code") != 0:
|
||
raise SessionError(
|
||
f"历史通知请求失败：HTTP {status}，业务码 {payload.get('status_code')}"
|
||
)
|
||
rows = payload.get("notice_list_v2")
|
||
if rows is None:
|
||
rows = []
|
||
if not isinstance(rows, list) or any(
|
||
not isinstance(row, dict) for row in rows
|
||
):
|
||
raise SessionError("历史通知列表格式无效")
|
||
reached_cache = False
|
||
for row in rows:
|
||
if str(row.get("user_id")) != self.uid:
|
||
raise SessionError("历史通知所属身份不符")
|
||
nid = validate_uid(row.get("nid_str") or row.get("nid"))
|
||
reached_cache = reached_cache or nid in stop_ids
|
||
notices.setdefault(nid, row)
|
||
if reached_cache or not payload.get("has_more"):
|
||
return list(notices.values())
|
||
cursor = (payload.get("min_time"), payload.get("max_time"))
|
||
if cursor in seen_cursors or cursor == (min_time, max_time):
|
||
raise SessionError("历史通知分页游标未推进")
|
||
if any(type(value) not in (int, float) for value in cursor):
|
||
raise SessionError("历史通知分页游标格式无效")
|
||
seen_cursors.add(cursor)
|
||
min_time, max_time = cursor
|
||
raise SessionError("历史通知超过 50000 条，已停止以避免无限分页")
|
||
|
||
async def works(self, all_pages=False, stop_ids=None):
|
||
profile = await self.identity()
|
||
stop_ids = set(stop_ids or ())
|
||
sec_uid = profile.get("sec_uid") or profile.get("secUid")
|
||
if not isinstance(sec_uid, str) or not sec_uid:
|
||
raise SessionError("当前账号缺少作品列表身份参数")
|
||
works = {}
|
||
cursor = 0
|
||
seen_cursors = set()
|
||
pages = 1000 if all_pages or stop_ids else 1
|
||
for _ in range(pages):
|
||
response = await self.json(
|
||
works_request_script(sec_uid, cursor), main_world=True
|
||
)
|
||
try:
|
||
status = int(response["status"])
|
||
payload = json.loads(response["body"])
|
||
except (KeyError, TypeError, ValueError, json.JSONDecodeError) as exc:
|
||
raise SessionError("作品列表响应格式无效") from exc
|
||
if status != 200 or payload.get("status_code") != 0:
|
||
raise SessionError(
|
||
f"作品列表请求失败：HTTP {status}，业务码 {payload.get('status_code')}"
|
||
)
|
||
rows = payload.get("aweme_list")
|
||
if rows is None:
|
||
rows = []
|
||
if not isinstance(rows, list) or any(
|
||
not isinstance(row, dict) for row in rows
|
||
):
|
||
raise SessionError("作品列表格式无效")
|
||
reached_cache = False
|
||
for row in rows:
|
||
author = row.get("author") or {}
|
||
if str(author.get("uid")) != self.uid:
|
||
raise SessionError("作品列表所属身份不符")
|
||
ident = validate_uid(row.get("aweme_id") or row.get("awemeId"))
|
||
reached_cache = reached_cache or ident in stop_ids
|
||
cover = ((row.get("video") or {}).get("cover") or {}).get(
|
||
"url_list"
|
||
) or []
|
||
if not cover and row.get("images"):
|
||
cover = (row["images"][0] or {}).get("url_list") or []
|
||
works.setdefault(
|
||
ident,
|
||
{
|
||
"aweme_id": ident,
|
||
"desc": row.get("desc") or "",
|
||
"create_time": row.get("create_time") or "",
|
||
"statistics": row.get("statistics") or {},
|
||
"cover": cover[0] if cover else "",
|
||
"business": row,
|
||
},
|
||
)
|
||
if (
|
||
reached_cache
|
||
or (not all_pages and not stop_ids)
|
||
or not payload.get("has_more")
|
||
):
|
||
return list(works.values())
|
||
next_cursor = payload.get("max_cursor")
|
||
if type(next_cursor) not in (int, float) or next_cursor < 0:
|
||
raise SessionError("作品列表分页游标格式无效")
|
||
if next_cursor in seen_cursors or next_cursor == cursor:
|
||
raise SessionError("作品列表分页游标未推进")
|
||
seen_cursors.add(next_cursor)
|
||
cursor = next_cursor
|
||
raise SessionError("作品列表超过 18000 条，已停止以避免无限分页")
|
||
|
||
async def follow(self, target, check=False):
|
||
self.require_bound()
|
||
validate_uid(target)
|
||
return await self.evaluate(
|
||
FOLLOW_JS,
|
||
{"target": target, "expected": self.uid, "check": check},
|
||
main_world=True,
|
||
)
|
||
|
||
async def im(
|
||
self,
|
||
action,
|
||
target,
|
||
text="",
|
||
confirm=False,
|
||
limit=20,
|
||
cursor="9223372036854775807",
|
||
):
|
||
self.require_bound()
|
||
validate_uid(target)
|
||
if (
|
||
action not in ("send", "history")
|
||
or type(limit) is not int
|
||
or not 1 <= limit <= 100
|
||
):
|
||
raise ValueError("私信操作或历史条数无效")
|
||
if (
|
||
not isinstance(cursor, str)
|
||
or not cursor.isascii()
|
||
or not cursor.isdecimal()
|
||
or len(cursor) > 19
|
||
or (len(cursor) == 19 and cursor > "9223372036854775807")
|
||
):
|
||
raise ValueError("历史游标无效")
|
||
if action == "send" and (
|
||
not isinstance(text, str) or not text.strip() or len(text) > 1000
|
||
):
|
||
raise ValueError("消息必须为 1..1000 字")
|
||
params = {
|
||
"action": action,
|
||
"uid": target,
|
||
"text": text,
|
||
"confirm": confirm,
|
||
"limit": limit,
|
||
"cursor": cursor,
|
||
}
|
||
expression = EXPRESSION.replace(
|
||
"__PARAMS__", json.dumps(params, ensure_ascii=True)
|
||
)
|
||
expression = expression.replace(
|
||
"if (String(profile.user.uid) === p.uid)",
|
||
f"if (String(profile.user.uid) !== {json.dumps(self.uid)}) fail('LOGIN_REQUIRED');\n if (String(profile.user.uid) === p.uid)",
|
||
)
|
||
return await self.json(expression, main_world=True)
|
||
|
||
async def disconnect(self):
|
||
await self.uninstall()
|
||
# CDP Browser.close disconnects the driver; it does not terminate an externally owned browser.
|
||
if self.browser:
|
||
with suppress(Exception):
|
||
await self.browser.close()
|
||
self.browser = None
|
||
self.page = None
|
||
|
||
async def close_browser(self):
|
||
if self.browser and self.browser.is_connected():
|
||
client = await self.browser.new_browser_cdp_session()
|
||
with suppress(Exception):
|
||
await client.send("Browser.close")
|
||
await self.disconnect()
|