Nexus/server/utils/browser_ui_state.py

"""Embedded browser UI state — history, tabs, panel layout."""

from __future__ import annotations

from datetime import datetime, timezone
from typing import Any
from urllib.parse import urlparse

BROWSER_UI_CONTEXT = "embedded_browser"
MAX_URL_HISTORY = 128
MAX_VISIT_LOG = 512
MAX_TABS = 12


def _utcnow_iso() -> str:
    return datetime.now(timezone.utc).replace(tzinfo=None).isoformat(timespec="seconds")


def _is_safe_url(url: str) -> bool:
    try:
        parsed = urlparse(url.strip())
    except Exception:
        return False
    if parsed.scheme not in ("http", "https"):
        return False
    if parsed.username or parsed.password:
        return False
    return bool(parsed.netloc)


def _hostname(url: str) -> str:
    try:
        return urlparse(url).hostname or url
    except Exception:
        return url


def empty_browser_state() -> dict[str, Any]:
    return {
        "url_history": [],
        "visits": [],
        "tabs": [],
        "active_tab_id": None,
        "minimized": False,
        "rect": None,
        "minimized_rect": None,
    }


def _parse_rect(raw: dict | None, *, default_w: float, default_h: float) -> dict[str, float] | None:
    if raw is None or not isinstance(raw, dict):
        return None
    try:
        return {
            "x": float(raw.get("x", 0)),
            "y": float(raw.get("y", 0)),
            "w": float(raw.get("w", default_w)),
            "h": float(raw.get("h", default_h)),
        }
    except (TypeError, ValueError):
        return None


def parse_browser_state(raw: dict | None) -> dict[str, Any]:
    base = empty_browser_state()
    if not raw or not isinstance(raw, dict):
        return base

    url_history = [
        u for u in raw.get("url_history") or []
        if isinstance(u, str) and _is_safe_url(u)
    ][:MAX_URL_HISTORY]

    visits: list[dict[str, str]] = []
    for item in raw.get("visits") or []:
        if not isinstance(item, dict):
            continue
        url = item.get("url")
        if not isinstance(url, str) or not _is_safe_url(url):
            continue
        visits.append({
            "url": url,
            "title": str(item.get("title") or _hostname(url))[:200],
            "at": str(item.get("at") or _utcnow_iso()),
        })
    visits = visits[:MAX_VISIT_LOG]

    tabs: list[dict[str, Any]] = []
    for item in raw.get("tabs") or []:
        if not isinstance(item, dict):
            continue
        tab_id = item.get("id")
        url = item.get("url")
        if not isinstance(tab_id, str) or not tab_id.strip():
            continue
        if not isinstance(url, str):
            continue
        if url.strip() and not _is_safe_url(url):
            continue
        if url.strip():
            stack = [
                u for u in item.get("stack") or [url]
                if isinstance(u, str) and _is_safe_url(u)
            ] or [url]
            stack_index = item.get("stack_index", len(stack) - 1)
            if not isinstance(stack_index, int):
                stack_index = len(stack) - 1
            stack_index = max(0, min(stack_index, len(stack) - 1))
        else:
            stack = []
            stack_index = -1
        tabs.append({
            "id": tab_id.strip()[:64],
            "url": url.strip(),
            "title": str(item.get("title") or (_hostname(url) if url.strip() else "新标签"))[:200],
            "stack": stack[-64:],
            "stack_index": stack_index,
        })
    tabs = tabs[:MAX_TABS]

    active_tab_id = raw.get("active_tab_id")
    if not isinstance(active_tab_id, str) or not any(t["id"] == active_tab_id for t in tabs):
        active_tab_id = tabs[0]["id"] if tabs else None

    rect = _parse_rect(raw.get("rect"), default_w=960, default_h=640)
    minimized_rect = _parse_rect(raw.get("minimized_rect"), default_w=320, default_h=52)

    minimized = bool(raw.get("minimized"))

    return {
        "url_history": url_history,
        "visits": visits,
        "tabs": tabs,
        "active_tab_id": active_tab_id,
        "minimized": minimized,
        "rect": rect,
        "minimized_rect": minimized_rect,
    }


def merge_browser_state(current: dict[str, Any], incoming: dict[str, Any]) -> dict[str, Any]:
    """Merge client state; incoming wins for tabs/active/minimized/rect."""
    parsed = parse_browser_state({**current, **incoming})
    if incoming.get("url_history") is not None:
        parsed["url_history"] = parse_browser_state({"url_history": incoming["url_history"]})["url_history"]
    if incoming.get("visits") is not None:
        parsed["visits"] = parse_browser_state({"visits": incoming["visits"]})["visits"]
    return parsed


def record_visit(state: dict[str, Any], url: str, title: str | None = None) -> dict[str, Any]:
    if not _is_safe_url(url):
        return state
    next_state = parse_browser_state(state)
    hist = [url] + [u for u in next_state["url_history"] if u != url]
    next_state["url_history"] = hist[:MAX_URL_HISTORY]
    visit = {
        "url": url,
        "title": (title or _hostname(url))[:200],
        "at": _utcnow_iso(),
    }
    visits = [visit] + next_state["visits"]
    next_state["visits"] = visits[:MAX_VISIT_LOG]
    return next_state