"""
[v1.25.4] Web Control — 聊天内嵌网页控制器

架构说明:
- 服务端代理: 通过 /api/web_control/proxy 获取并改写网页内容, 注入控制脚本
- 会话管理: 每个聊天会话可创建一个 web_control 会话, 维护 cookie/命令队列/结果 Future
- 双向通信: Agent → 命令队列 → 客户端轮询 → 执行 → POST 结果 → Agent 阻塞等待
- 前端面板: 基础JS框架 + 动态容器(iframe), 服务端下发控制脚本在容器内执行
- [v1.25.3] 人机交互: human_interact action 允许 Agent 暂停控制, 用户手动登录后继续
- [v1.25.4] 登录流程: login action 提供一站式登录编排 + 凭证管理

工具 web_control 支持的 action:
  open            — 打开 URL (发送 v2_web_control SSE 事件, 前端打开面板)
  navigate        — 在已打开的面板中导航到新 URL
  click           — 点击元素 (selector: CSS 选择器)
  fill            — 填写输入框 (selector + value)
  scroll          — 滚动页面 (direction: up/down/top/bottom, distance: px)
  evaluate        — 执行 JavaScript (script: JS 代码)
  get_content     — 获取页面内容 (what: text/html/url/title/cookies/links)
  set_cookies     — 设置 cookie (cookies: [{name, value, domain, path}])
  get_cookies     — 获取当前 cookie
  wait            — 等待 (time: 毫秒 或 selector: CSS选择器, timeout: 秒)
  screenshot      — 截图 (返回 base64 PNG)
  human_interact  — [v1.25.3] 切换为人机交互模式, 用户可手动操作页面 (登录/验证码等)
  save_cookies    — [v1.25.3] 保存当前会话 cookies 到持久化文件
  load_cookies    — [v1.25.3] 从持久化文件加载 cookies
  login           — [v1.25.4] 一站式登录流程: 导航到登录页 → 人机交互 → 自动保存凭证
  save_credentials — [v1.25.4] 保存登录凭证 (cookies + 元信息) 到凭证库
  list_credentials — [v1.25.4] 列出已保存的凭证
  delete_credentials — [v1.25.4] 删除指定凭证
  close           — 关闭 web_control 面板
"""

# [v1.25.4] 常用平台登录 URL 模板
LOGIN_URLS = {
    "qq": "https://qun.qq.com/",
    "qq_mail": "https://mail.qq.com/",
    "wechat_work": "https://work.weixin.qq.com/wework_admin/frame",
    "wechat_mp": "https://mp.weixin.qq.com/",
    "telegram": "https://web.telegram.org/",
    "telegram_web": "https://web.telegram.org/",
    "discord": "https://discord.com/login",
    "feishu": "https://open.feishu.cn/",
    "feishu_admin": "https://feishu.cn/admin",
    "lark": "https://open.larksuite.com/",
    "dingtalk": "https://login.dingtalk.com/",
    "github": "https://github.com/login",
    "google": "https://accounts.google.com/",
    "bilibili": "https://passport.bilibili.com/login",
    "taobao": "https://login.taobao.com/",
    "zhihu": "https://www.zhihu.com/signin",
    "weibo": "https://passport.weibo.com/sso/signin",
}

import asyncio
import json
import os
import re
import time
import uuid
import hashlib
from urllib.parse import urlparse, urljoin, quote as url_quote, unquote
from typing import Dict, List, Optional, Any
from .logger import get_logger
logger = get_logger("myagent.web_control")


# ────────────────────────────────────────────────────────────────────────
# 控制脚本 — 注入到每个被代理页面中, 监听 postMessage 命令并执行
# ────────────────────────────────────────────────────────────────────────

CONTROL_SCRIPT = """
(function() {
  'use strict';
  window.__webControlReady = true;
  window.__webControlSessionId = '{{SESSION_ID}}';

  // 监听来自父窗口的命令
  window.addEventListener('message', function(event) {
    if (event.data && event.data.__webControlCmd) {
      handleCommand(event.data);
    }
  });

  // 通知父窗口页面已就绪
  function notifyParent(type, data) {
    if (window.parent && window.parent !== window) {
      window.parent.postMessage({ __webControlEvent: true, type: type, data: data }, '*');
    }
  }

  // 页面加载完成后通知父窗口
  function onReady() {
    notifyParent('ready', {
      url: window.location.href,
      title: document.title,
      sid: window.__webControlSessionId
    });
  }

  if (document.readyState === 'complete') {
    onReady();
  } else {
    window.addEventListener('load', onReady);
  }

  // 命令处理
  async function handleCommand(cmd) {
    var cmdId = cmd.cmdId;
    try {
      var result = null;
      switch (cmd.action) {
        case 'click':
          result = doClick(cmd.params || {});
          break;
        case 'fill':
          result = doFill(cmd.params || {});
          break;
        case 'scroll':
          result = doScroll(cmd.params || {});
          break;
        case 'evaluate':
          result = doEvaluate(cmd.params || {});
          break;
        case 'get_content':
          result = doGetContent(cmd.params || {});
          break;
        case 'set_cookies':
          result = doSetCookies(cmd.params || {});
          break;
        case 'get_cookies':
          result = doGetCookies(cmd.params || {});
          break;
        case 'wait':
          result = await doWait(cmd.params || {});
          break;
        case 'screenshot':
          result = await doScreenshot(cmd.params || {});
          break;
        case 'human_interact':
          // [v1.25.3] 切换人机模式
          window.__webControlHumanMode = true;
          // 显示人机模式提示横幅
          showHumanModeBanner(cmd.params || {});
          result = { success: true, human_mode: true, message: '已切换到人机交互模式, 用户可自由操作页面' };
          break;
        case 'agent_mode':
          // [v1.25.3] 恢复 Agent 控制模式
          window.__webControlHumanMode = false;
          hideHumanModeBanner();
          result = { success: true, human_mode: false, message: '已恢复 Agent 控制模式' };
          break;
        default:
          result = { success: false, error: 'Unknown action: ' + cmd.action };
      }
      result.cmdId = cmdId;
      result.success = result.success !== false;
      notifyParent('result', result);
    } catch (e) {
      notifyParent('result', { cmdId: cmdId, success: false, error: e.message || String(e) });
    }
  }

  // ── 操作实现 ──

  function doClick(p) {
    var el = document.querySelector(p.selector);
    if (!el) return { success: false, error: 'Element not found: ' + p.selector };
    el.scrollIntoView({ block: 'center', behavior: 'instant' });
    el.click();
    return { success: true, tag: el.tagName, text: (el.textContent || '').substring(0, 200) };
  }

  function doFill(p) {
    var el = document.querySelector(p.selector);
    if (!el) return { success: false, error: 'Element not found: ' + p.selector };
    var tag = (el.tagName || '').toLowerCase();
    if (tag === 'input' || tag === 'textarea' || tag === 'select') {
      el.focus();
      el.value = p.value || '';
      el.dispatchEvent(new Event('input', { bubbles: true }));
      el.dispatchEvent(new Event('change', { bubbles: true }));
      return { success: true, tag: tag, filled: p.value };
    }
    // 尝试 contenteditable
    if (el.isContentEditable) {
      el.focus();
      el.textContent = p.value || '';
      el.dispatchEvent(new Event('input', { bubbles: true }));
      return { success: true, tag: 'contenteditable', filled: p.value };
    }
    return { success: false, error: 'Element is not fillable: ' + p.selector };
  }

  function doScroll(p) {
    var dir = p.direction || 'down';
    var dist = parseInt(p.distance) || 300;
    switch (dir) {
      case 'up':
        window.scrollBy(0, -dist);
        break;
      case 'down':
        window.scrollBy(0, dist);
        break;
      case 'top':
        window.scrollTo(0, 0);
        break;
      case 'bottom':
        window.scrollTo(0, document.body.scrollHeight);
        break;
      default:
        window.scrollBy(0, dist);
    }
    return {
      success: true,
      scrollX: window.scrollX || window.pageXOffset,
      scrollY: window.scrollY || window.pageYOffset,
      pageHeight: document.body.scrollHeight
    };
  }

  function doEvaluate(p) {
    var script = p.script || '';
    try {
      var result = (new Function('return (' + script + ')'))();
      var resultStr = typeof result === 'object' ? JSON.stringify(result) : String(result);
      if (resultStr && resultStr.length > 5000) resultStr = resultStr.substring(0, 5000) + '...(truncated)';
      return { success: true, result: resultStr, type: typeof result };
    } catch (e) {
      return { success: false, error: 'Script error: ' + e.message };
    }
  }

  function doGetContent(p) {
    var what = p.what || 'text';
    switch (what) {
      case 'text':
        return { success: true, content: (document.body.innerText || document.body.textContent || '').substring(0, 8000) };
      case 'html':
        return { success: true, content: (document.documentElement.outerHTML || '').substring(0, 15000) };
      case 'url':
        return { success: true, content: window.location.href };
      case 'title':
        return { success: true, content: document.title };
      case 'cookies':
        return { success: true, content: document.cookie };
      case 'links':
        var links = [];
        document.querySelectorAll('a[href]').forEach(function(a) {
          links.push({ text: a.textContent.trim().substring(0, 100), href: a.href });
        });
        return { success: true, content: JSON.stringify(links.slice(0, 100)) };
      case 'images':
        var imgs = [];
        document.querySelectorAll('img[src]').forEach(function(img) {
          imgs.push({ src: img.src, alt: (img.alt || '').substring(0, 100) });
        });
        return { success: true, content: JSON.stringify(imgs.slice(0, 50)) };
      case 'forms':
        var forms = [];
        document.querySelectorAll('form').forEach(function(f) {
          var inputs = [];
          f.querySelectorAll('input,textarea,select').forEach(function(el) {
            inputs.push({ tag: el.tagName, type: el.type, name: el.name, id: el.id, value: el.value });
          });
          forms.push({ action: f.action, method: f.method, inputs: inputs });
        });
        return { success: true, content: JSON.stringify(forms.slice(0, 20)) };
      case 'inputs':
        var allInputs = [];
        document.querySelectorAll('input,textarea,select').forEach(function(el) {
          allInputs.push({
            tag: el.tagName, type: el.type || 'textarea', name: el.name,
            id: el.id, placeholder: el.placeholder, value: el.value
          });
        });
        return { success: true, content: JSON.stringify(allInputs.slice(0, 50)) };
      default:
        return { success: true, content: (document.body.innerText || '').substring(0, 8000) };
    }
  }

  function doSetCookies(p) {
    var cookies = p.cookies || [];
    for (var i = 0; i < cookies.length; i++) {
      var c = cookies[i];
      var parts = [c.name + '=' + (c.value || '')];
      if (c.domain) parts.push('domain=' + c.domain);
      if (c.path) parts.push('path=' + c.path);
      if (c.maxAge) parts.push('max-age=' + c.maxAge);
      if (c.secure) parts.push('secure');
      if (c.sameSite) parts.push('samesite=' + c.sameSite);
      document.cookie = parts.join('; ');
    }
    return { success: true, set_count: cookies.length, current_cookies: document.cookie };
  }

  function doGetCookies(p) {
    return { success: true, cookies: document.cookie };
  }

  function doWait(p) {
    var waitTime = parseInt(p.time) || 1000;
    if (p.selector) {
      return new Promise(function(resolve) {
        var el = document.querySelector(p.selector);
        if (el) {
          resolve({ success: true, found: true, elapsed: 0 });
          return;
        }
        var start = Date.now();
        var maxWait = (parseInt(p.timeout) || 10) * 1000;
        var interval = setInterval(function() {
          if (document.querySelector(p.selector)) {
            clearInterval(interval);
            resolve({ success: true, found: true, elapsed: Date.now() - start });
          } else if (Date.now() - start > maxWait) {
            clearInterval(interval);
            resolve({ success: false, found: false, error: 'Timeout waiting for: ' + p.selector, elapsed: Date.now() - start });
          }
        }, 200);
      });
    }
    return new Promise(function(resolve) {
      setTimeout(function() {
        resolve({ success: true, waited_ms: waitTime });
      }, waitTime);
    });
  }

  // [v1.25.3] 人机模式横幅
  function showHumanModeBanner(p) {
    var banner = document.createElement('div');
    banner.id = '__wcHumanBanner';
    var promptText = (p && p.prompt) ? p.prompt : '人机交互模式 — 请在此页面完成登录或验证操作';
    var platformText = (p && p.platform) ? ' (' + p.platform + ')' : '';
    banner.style.cssText = 'position:fixed;top:0;left:0;right:0;z-index:999999;padding:10px 16px;' +
      'background:linear-gradient(135deg,#667eea 0%,#764ba2 100%);color:#fff;font-size:14px;font-weight:600;' +
      'display:flex;align-items:center;justify-content:space-between;box-shadow:0 2px 12px rgba(0,0,0,0.3);font-family:-apple-system,BlinkMacSystemFont,sans-serif;';
    banner.innerHTML = '<span>🎯 ' + promptText + platformText + '</span>';
    document.body.appendChild(banner);
    // 为页面内容添加顶部间距
    document.body.style.paddingTop = '44px';
    // [v1.25.4] 记录进入人机模式时的 URL, 用于检测登录成功
    window.__webControlLoginStartUrl = window.location.href;
    // [v1.25.4] 监听 URL 变化 (登录成功通常伴随跳转)
    window.__webControlLastNotifiedUrl = window.location.href;
    var _urlCheckInterval = setInterval(function() {
      if (!window.__webControlHumanMode) { clearInterval(_urlCheckInterval); return; }
      var currentUrl = window.location.href;
      if (currentUrl !== window.__webControlLastNotifiedUrl) {
        window.__webControlLastNotifiedUrl = currentUrl;
        // 通知父窗口 URL 发生了变化 (可能登录成功)
        notifyParent('url_change', { url: currentUrl, previous_url: window.__webControlLoginStartUrl });
      }
    }, 1000);
  }
  function hideHumanModeBanner() {
    var banner = document.getElementById('__wcHumanBanner');
    if (banner) banner.remove();
    document.body.style.paddingTop = '';
    window.__webControlLoginStartUrl = null;
  }

  function doScreenshot(p) {
    // html2canvas 未加载时返回提示
    if (typeof html2canvas === 'undefined') {
      return { success: false, error: 'html2canvas not loaded. Use get_content with what=screenshot_html instead.' };
    }
    return html2canvas(document.body, { useCORS: true, allowTaint: true }).then(function(canvas) {
      var dataUrl = canvas.toDataURL('image/png');
      return { success: true, dataUrl: dataUrl, width: canvas.width, height: canvas.height };
    });
  }

  // [v1.25.3] 人机交互模式标志
  window.__webControlHumanMode = false;

  // 拦截链接点击, 重定向到代理（人机模式下跳过拦截）
  document.addEventListener('click', function(e) {
    if (window.__webControlHumanMode) return; // 人机模式不拦截
    var link = e.target.closest('a');
    if (link && link.href) {
      // 允许 target=_blank 和特殊链接
      if (link.target === '_blank') return;
      if (link.href.startsWith('javascript:') || link.href.startsWith('#')) return;
      // 重写 href 到代理
      var proxyBase = window.__webControlProxyBase || '';
      if (proxyBase) {
        e.preventDefault();
        window.location.href = proxyBase + '?url=' + encodeURIComponent(link.href) + '&sid=' + (window.__webControlSessionId || '');
      }
    }
  }, true);

  // 拦截表单提交（人机模式下跳过拦截）
  document.addEventListener('submit', function(e) {
    if (window.__webControlHumanMode) return; // 人机模式不拦截
    var form = e.target;
    if (form.method && form.method.toLowerCase() === 'post') {
      // POST 表单暂不代理, 仅阻止默认行为
      e.preventDefault();
      window.__webControlLastForm = {
        action: form.action,
        method: form.method,
        inputs: Array.from(form.querySelectorAll('input,textarea,select')).map(function(el) {
          return { name: el.name, value: el.value, type: el.type };
        })
      };
    } else if (form.action) {
      var proxyBase = window.__webControlProxyBase || '';
      if (proxyBase) {
        e.preventDefault();
        // 收集表单数据
        var formData = new FormData(form);
        var queryString = new URLSearchParams(formData).toString();
        var targetUrl = form.action + (form.action.includes('?') ? '&' : '?') + queryString;
        window.location.href = proxyBase + '?url=' + encodeURIComponent(targetUrl) + '&sid=' + (window.__webControlSessionId || '');
      }
    }
  }, true);
})();
"""


class WebControlSession:
    """单个 web_control 会话 — 管理 cookie、命令队列、结果 Future"""

    def __init__(self, session_id: str):
        self.session_id = session_id
        self.created_at = time.time()
        self.last_activity = time.time()
        self.current_url = ""
        self.cookies: Dict[str, str] = {}  # "domain::name" -> value
        self.command_queue: asyncio.Queue = asyncio.Queue()
        self.pending_results: Dict[str, asyncio.Future] = {}
        self.is_panel_open = False
        self._closed = False
        # [v1.25.3] 人机交互模式
        self.human_mode = False
        self.human_event: Optional[asyncio.Event] = None  # Agent 等待用户完成操作
        self.human_result: Optional[Dict[str, Any]] = None

    def is_alive(self) -> bool:
        if self._closed:
            return False
        # 会话超时 30 分钟
        return (time.time() - self.last_activity) < 1800

    def touch(self):
        self.last_activity = time.time()

    def close(self):
        self._closed = True
        # 取消所有 pending futures
        for future in self.pending_results.values():
            if not future.done():
                future.set_result({"success": False, "error": "Session closed"})
        self.pending_results.clear()
        # 唤醒等待 human_event 的协程
        if self.human_event and not self.human_event.is_set():
            self.human_event.set()

    # [v1.25.3] Cookie 持久化
    def save_cookies_to_file(self, label: str = "") -> str:
        """将当前 cookies 保存到文件, 返回文件路径"""
        import json as _json
        from pathlib import Path
        cookie_dir = Path.home() / ".myagent" / "data" / "web_cookies"
        cookie_dir.mkdir(parents=True, exist_ok=True)
        if not label:
            # 从 current_url 提取域名作为 label
            parsed = urlparse(self.current_url)
            label = parsed.hostname or "default"
        label = re.sub(r'[^a-zA-Z0-9._-]', '_', label)
        filepath = cookie_dir / f"{label}.json"
        data = {
            "url": self.current_url,
            "saved_at": time.strftime("%Y-%m-%d %H:%M:%S"),
            "cookies": [
                {"key": k, "value": v} for k, v in self.cookies.items()
            ]
        }
        filepath.write_text(_json.dumps(data, ensure_ascii=False, indent=2), encoding='utf-8')
        logger.info(f"[WebControl] Cookies 已保存: {filepath} ({len(self.cookies)} 个)")
        return str(filepath)

    def load_cookies_from_file(self, label: str = "") -> int:
        """从文件加载 cookies, 返回加载的数量"""
        import json as _json
        from pathlib import Path
        cookie_dir = Path.home() / ".myagent" / "data" / "web_cookies"
        if not label:
            parsed = urlparse(self.current_url)
            label = parsed.hostname or "default"
        label = re.sub(r'[^a-zA-Z0-9._-]', '_', label)
        filepath = cookie_dir / f"{label}.json"
        if not filepath.exists():
            return 0
        try:
            data = _json.loads(filepath.read_text(encoding='utf-8'))
            count = 0
            for item in data.get("cookies", []):
                self.cookies[item["key"]] = item["value"]
                count += 1
            logger.info(f"[WebControl] Cookies 已加载: {filepath} ({count} 个)")
            return count
        except Exception as e:
            logger.error(f"[WebControl] 加载 cookies 失败: {e}")
            return 0

    # [v1.25.4] 凭证管理 — 保存/列出/删除登录凭证
    def save_credentials_to_file(self, platform: str, label: str = "", extra: Dict = None) -> str:
        """保存完整登录凭证 (cookies + 元信息) 到凭证库, 返回文件路径"""
        import json as _json
        from pathlib import Path
        cred_dir = Path.home() / ".myagent" / "data" / "credentials"
        cred_dir.mkdir(parents=True, exist_ok=True)
        # platform 作为文件名
        safe_platform = re.sub(r'[^a-zA-Z0-9._-]', '_', platform).lower()
        if not safe_platform:
            safe_platform = "default"
        filepath = cred_dir / f"{safe_platform}.json"
        data = {
            "platform": platform,
            "url": self.current_url,
            "saved_at": time.strftime("%Y-%m-%d %H:%M:%S"),
            "cookie_count": len(self.cookies),
            "cookies": [
                {"key": k, "value": v} for k, v in self.cookies.items()
            ],
        }
        if extra:
            data["extra"] = extra
        if label:
            data["label"] = label
        filepath.write_text(_json.dumps(data, ensure_ascii=False, indent=2), encoding='utf-8')
        logger.info(f"[WebControl] 凭证已保存: {filepath} (平台: {platform}, cookies: {len(self.cookies)})")
        return str(filepath)

    @staticmethod
    def list_credentials() -> List[Dict]:
        """列出所有已保存的凭证"""
        import json as _json
        from pathlib import Path
        cred_dir = Path.home() / ".myagent" / "data" / "credentials"
        if not cred_dir.exists():
            return []
        results = []
        for f in sorted(cred_dir.glob("*.json")):
            try:
                data = _json.loads(f.read_text(encoding='utf-8'))
                results.append({
                    "platform": data.get("platform", f.stem),
                    "label": data.get("label", ""),
                    "url": data.get("url", ""),
                    "saved_at": data.get("saved_at", ""),
                    "cookie_count": data.get("cookie_count", 0),
                    "file": str(f),
                })
            except Exception:
                pass
        return results

    @staticmethod
    def delete_credentials(platform: str) -> bool:
        """删除指定平台的凭证"""
        from pathlib import Path
        cred_dir = Path.home() / ".myagent" / "data" / "credentials"
        safe_platform = re.sub(r'[^a-zA-Z0-9._-]', '_', platform).lower()
        filepath = cred_dir / f"{safe_platform}.json"
        if filepath.exists():
            filepath.unlink()
            logger.info(f"[WebControl] 凭证已删除: {filepath} (平台: {platform})")
            return True
        return False


class WebControlManager:
    """Web Control 管理器 — 管理所有会话、代理请求"""

    def __init__(self):
        self.sessions: Dict[str, WebControlSession] = {}
        self._proxy_cache: Dict[str, tuple] = {}  # url_hash -> (content, content_type, timestamp)
        self._cache_ttl = 60  # 缓存 60 秒
        self._lock = asyncio.Lock()

    def create_session(self) -> WebControlSession:
        session_id = uuid.uuid4().hex[:10]
        session = WebControlSession(session_id)
        self.sessions[session_id] = session
        logger.info(f"[WebControl] 创建会话: {session_id}")
        return session

    def get_session(self, session_id: str) -> Optional[WebControlSession]:
        session = self.sessions.get(session_id)
        if session and session.is_alive():
            return session
        return None

    def close_session(self, session_id: str):
        session = self.sessions.pop(session_id, None)
        if session:
            session.close()
            logger.info(f"[WebControl] 关闭会话: {session_id}")

    def cleanup_expired(self):
        """清理过期会话"""
        expired = [sid for sid, s in self.sessions.items() if not s.is_alive()]
        for sid in expired:
            self.close_session(sid)
        if expired:
            logger.info(f"[WebControl] 清理过期会话: {expired}")

    async def queue_command(
        self, session_id: str, action: str, params: Dict, timeout: int = 30
    ) -> Dict[str, Any]:
        """
        向客户端下发命令并等待结果。
        Agent 工具调用线程阻塞在此, 直到客户端返回结果或超时。
        
        [v1.20.9] 修复: 如果面板尚未打开（刚执行了 open 但客户端还没加载完成），
        会等待最多 15 秒让面板打开，而不是立即报错。
        """
        session = self.get_session(session_id)
        if not session:
            return {"success": False, "error": f"Session {session_id} not found or expired"}
        
        # [v1.20.9] 等待面板打开（最多 15 秒）
        if not session.is_panel_open:
            logger.info(f"[WebControl] 等待面板打开 (session: {session_id}, action: {action})")
            waited = 0
            while waited < 15:
                await asyncio.sleep(1)
                waited += 1
                if session.is_panel_open:
                    logger.info(f"[WebControl] 面板已打开 (等待了 {waited}s)")
                    break
                # 检查会话是否已关闭
                if session._closed:
                    return {"success": False, "error": "Session closed while waiting for panel"}
            
            if not session.is_panel_open:
                return {"success": False, "error": "Panel is not open. Use 'open' action first. (waited 15s)"}

        cmd_id = uuid.uuid4().hex[:10]
        command = {
            "id": cmd_id,
            "action": action,
            "params": params,
            "timestamp": time.time(),
        }

        # 放入队列
        session.command_queue.put_nowait(command)
        session.touch()

        # 创建 Future 并等待结果
        loop = asyncio.get_event_loop()
        future = loop.create_future()
        session.pending_results[cmd_id] = future

        try:
            result = await asyncio.wait_for(future, timeout=timeout)
            return result
        except asyncio.TimeoutError:
            session.pending_results.pop(cmd_id, None)
            logger.warning(f"[WebControl] 命令超时: {action} ({cmd_id})")
            return {"success": False, "error": f"Command '{action}' timed out after {timeout}s waiting for client"}

    async def poll_commands(self, session_id: str) -> Dict:
        """客户端轮询 — 返回待执行命令"""
        session = self.get_session(session_id)
        if not session:
            return {"commands": [], "sid": session_id, "error": "Session not found"}

        session.is_panel_open = True
        session.touch()

        commands = []
        while not session.command_queue.empty():
            try:
                commands.append(session.command_queue.get_nowait())
            except asyncio.QueueEmpty:
                break

        return {"commands": commands, "sid": session_id}

    async def submit_result(self, session_id: str, cmd_id: str, result: Dict) -> bool:
        """客户端提交命令执行结果"""
        session = self.get_session(session_id)
        if not session:
            return False

        future = session.pending_results.pop(cmd_id, None)
        if future and not future.done():
            future.set_result(result)
            session.touch()
            return True
        return False

    # ────────────────────────────────────────────────────────────────────
    # URL 代理 — 获取并改写网页内容
    # ────────────────────────────────────────────────────────────────────

    async def proxy_request(self, url: str, session_id: str = "", referer: str = "") -> tuple:
        """
        代理 HTTP 请求。
        返回 (content_bytes, content_type, headers_dict)
        """
        # URL 安全检查
        parsed = urlparse(url)
        if parsed.scheme not in ('http', 'https'):
            raise ValueError(f"Unsupported URL scheme: {parsed.scheme}")

        # 禁止访问内网地址（防止 SSRF 攻击）
        hostname = parsed.hostname
        import ipaddress
        if hostname:
            # 解析 hostname 为 IP（处理 DNS Rebinding）
            try:
                import socket
                resolved_ips = socket.getaddrinfo(hostname, parsed.port or 443, socket.AF_UNSPEC, socket.SOCK_STREAM)
                for family, _, _, _, sockaddr in resolved_ips:
                    ip = sockaddr[0]
                    try:
                        ip_obj = ipaddress.ip_address(ip)
                        # 阻止所有内网地址
                        if ip_obj.is_private or ip_obj.is_loopback or ip_obj.is_link_local or ip_obj.is_reserved or ip_obj.is_multicast:
                            raise ValueError(f"SSRF blocked: {hostname} resolves to private/reserved IP {ip}")
                    except ValueError:
                        raise
            except socket.gaierror:
                pass
            # 字符串级别的快速检查（防止绕过）
            blocked = (
                hostname in ('localhost', '127.0.0.1', '0.0.0.0', '::1', '::') or
                hostname.endswith('.local') or
                hostname.endswith('.internal') or
                hostname.startswith('10.') or
                hostname.startswith('192.168.') or
                # 172.16.0.0 - 172.31.255.255
                (hostname.startswith('172.') and len(hostname.split('.')) >= 3 and 16 <= int(hostname.split('.')[1]) <= 31) or
                # 169.254.0.0/16 云元数据
                hostname.startswith('169.254.') or
                # 100.64.0.0/10 CGN
                hostname.startswith('100.') and len(hostname.split('.')) >= 2 and 64 <= int(hostname.split('.')[1]) <= 127
            )
            if blocked:
                raise ValueError(f"SSRF blocked: cannot access internal address {hostname}")

        # 检查缓存（仅对非 session 请求）
        cache_key = hashlib.md5(url.encode()).hexdigest()
        if session_id:
            cached = self._proxy_cache.get(cache_key)
            if cached and (time.time() - cached[2]) < self._cache_ttl:
                return cached[0], cached[1], {}

        import aiohttp

        # [v1.23.26] 兼容不同 aiohttp 版本的 redirect 参数
        # aiohttp >= 3.0 使用 follow_redirects, < 3.0 使用 allow_redirects
        _aio_ver = tuple(int(x) for x in aiohttp.__version__.split('.')[:2])
        _redirect_kw = 'follow_redirects' if _aio_ver >= (3, 0) else 'allow_redirects'

        headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
            'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8',
            'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
            'Accept-Encoding': 'identity',  # 不要压缩, 方便改写
        }

        # 注入 session cookie
        session = self.get_session(session_id) if session_id else None
        if session and session.cookies:
            domain_cookies = []
            for key, val in session.cookies.items():
                parts = key.split('::', 1)
                if len(parts) == 2:
                    cookie_domain, cookie_name = parts
                    if hostname and (hostname == cookie_domain or hostname.endswith('.' + cookie_domain)):
                        domain_cookies.append(f"{cookie_name}={val}")
            if domain_cookies:
                headers['Cookie'] = '; '.join(domain_cookies)

        if referer:
            headers['Referer'] = referer

        async with aiohttp.ClientSession() as http_session:
            try:
                async with http_session.get(
                    url,
                    headers=headers,
                    allow_redirects=True,
                    timeout=aiohttp.ClientTimeout(total=20),
                    ssl=False,
                ) as resp:
                    content = await resp.read()
                    content_type = resp.headers.get('Content-Type', 'application/octet-stream')
                    resp_headers = dict(resp.headers)
            except Exception as e:
                raise RuntimeError(f"Proxy fetch failed: {e}")

        # 更新缓存（仅缓存小文件）
        if len(content) < 500000:  # < 500KB
            self._proxy_cache[cache_key] = (content, content_type, time.time())

        return content, content_type, resp_headers

    def rewrite_html(self, html: str, original_url: str, session_id: str) -> str:
        """
        改写 HTML 内容:
        1. 注入控制脚本
        2. 设置 proxy base URL 让相对链接走代理
        3. 改写绝对 URL 的链接/脚本/图片等资源
        """
        parsed = urlparse(original_url)
        base_origin = f"{parsed.scheme}://{parsed.netloc}"

        # 1. 注入控制脚本（在 <head> 之后立即注入）
        control_js = CONTROL_SCRIPT.replace('{{SESSION_ID}}', session_id)
        proxy_base = f"/api/web_control/proxy?sid={session_id}&url="

        inject_tag = (
            f'<script>'
            f'window.__webControlProxyBase="{proxy_base}";'
            f'window.__webControlSessionId="{session_id}";'
            f'</script>\n'
            f'<script>{control_js}</script>'
        )

        # 尝试在 <head> 后注入
        if '<head>' in html.lower():
            html = re.sub(
                r'(?i)<head[^>]*>',
                lambda m: m.group(0) + '\n' + inject_tag,
                html,
                count=1
            )
        elif '<html' in html.lower():
            html = re.sub(
                r'(?i)<html[^>]*>',
                lambda m: m.group(0) + '<head>' + inject_tag + '</head>',
                html,
                count=1
            )
        else:
            html = inject_tag + html

        # 2. 改写 URL 属性 — 将同源的相对/绝对 URL 重定向到代理
        # 仅改写 HTML 属性中的 URL (不影响内联脚本)
        def rewrite_attr(match):
            attr_name = match.group(1).lower()
            eq = match.group(2)       # "=" (可能带空格)
            q_open = match.group(3)    # 开引号 " 或 '
            url_val = match.group(4)   # URL 值
            q_close = match.group(5)   # 闭引号 (同 q_open)

            # 跳过特殊协议
            if not url_val or url_val.startswith('data:') or url_val.startswith('blob:') or url_val.startswith('#') or url_val.startswith('javascript:') or url_val.startswith('mailto:') or url_val.startswith('tel:'):
                return match.group(0)

            # 已经是代理 URL — 跳过
            if '/api/web_control/proxy' in url_val:
                return match.group(0)

            # 解析 URL
            if url_val.startswith('//'):
                url_val = parsed.scheme + ':' + url_val
            elif url_val.startswith('/'):
                url_val = base_origin + url_val
            elif not url_val.startswith('http://') and not url_val.startswith('https://'):
                url_val = urljoin(original_url, url_val)

            # 非同源资源: CDN 图片/脚本/样式直接访问, 不走代理
            url_parsed = urlparse(url_val)
            if url_parsed.netloc and url_parsed.netloc != parsed.netloc:
                if attr_name in ('src', 'data-src', 'data-original'):
                    return match.group(0)
                if attr_name == 'href' and any(url_val.lower().endswith(ext) for ext in ('.css', '.woff', '.woff2', '.ttf', '.eot', '.svg', '.png', '.jpg', '.jpeg', '.gif', '.webp', '.ico')):
                    return match.group(0)
                # 其他外链 (导航链接) 走代理
                if attr_name == 'href':
                    return f'{attr_name}{eq}{q_open}{proxy_base}{url_quote(url_val)}{q_close}'

            # 同源资源走代理
            return f'{attr_name}{eq}{q_open}{proxy_base}{url_quote(url_val)}{q_close}'

        # 匹配常见 URL 属性
        # group1=属性名, group2="=", group3=开引号, group4=URL值, group5=闭引号
        url_pattern = re.compile(
            r'((?:src|href|action|data-src|data-original|poster|formaction|content|cite|background))'
            r'(\s*=\s*)'
            r'(["\'])'
            r'([^"\']*?)'
            r'(\3)',
            re.IGNORECASE
        )

        # 分块处理以避免影响内联脚本中的属性
        # 简单方法: 仅在 <script> 和 <style> 标签外进行替换
        parts = re.split(r'(<script[^>]*>.*?</script>|<style[^>]*>.*?</style>)', html, flags=re.IGNORECASE | re.DOTALL)
        for i in range(len(parts)):
            if not re.match(r'^<(script|style)', parts[i], re.IGNORECASE):
                parts[i] = url_pattern.sub(rewrite_attr, parts[i])

        html = ''.join(parts)

        # 3. 添加 CSP meta 标签允许同源脚本执行
        if '<meta' not in html[:2000]:
            csp_meta = (
                '<meta http-equiv="Content-Security-Policy" '
                f'content="default-src * \'unsafe-inline\' \'unsafe-eval\' data: blob:; '
                f'script-src * \'unsafe-inline\' \'unsafe-eval\' \'self\'; '
                f'style-src * \'unsafe-inline\' \'self\' data:; '
                f'img-src * data: blob: \'self\'; '
                f'frame-src * \'self\'; '
                f'connect-src * \'self\'">'
            )
            if '</head>' in html:
                html = html.replace('</head>', csp_meta + '\n</head>', 1)
            else:
                html = csp_meta + html

        return html

    def rewrite_css(self, css: str, original_url: str, session_id: str) -> str:
        """改写 CSS 中的 url() 引用"""
        proxy_base = f"/api/web_control/proxy?sid={session_id}&url="

        def rewrite_css_url(match):
            url_val = match.group(1).strip().strip("'\"")
            if url_val.startswith('data:') or url_val.startswith('blob:') or '/api/web_control/proxy' in url_val:
                return match.group(0)
            if url_val.startswith('//'):
                parsed = urlparse(original_url)
                url_val = parsed.scheme + ':' + url_val
            elif not url_val.startswith('http'):
                url_val = urljoin(original_url, url_val)
            return f'url("{proxy_base}{url_quote(url_val)}")'

        return re.sub(r'url\(\s*([^\)]+)\s*\)', rewrite_css_url, css, flags=re.IGNORECASE)


# ────────────────────────────────────────────────────────────────────────
# 全局单例
# ────────────────────────────────────────────────────────────────────────

_web_control_manager: Optional[WebControlManager] = None


def get_web_control_manager() -> WebControlManager:
    global _web_control_manager
    if _web_control_manager is None:
        _web_control_manager = WebControlManager()
    return _web_control_manager