"""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()