""" core/site_registry.py - 统一网站注册系统 ========================================= 管理所有可登录网站/服务的配置，支持动态增删。 架构: - 内置 20+ 网站配置（BUILTIN_SITES），开箱即用 - 自定义网站：~/.myagent/sites/*.json（运行时动态管理） - 被 browser_profile.py 引用，作为 Profile 预置配置的单一数据源 - 被 SKILL.md / CLI / Admin API 引用，展示可用网站列表 分类: - email: 邮箱服务 (Gmail, QQ邮箱, 139邮箱) - social: 社交媒体 (X.com, 微博, 微信) - cloud: 云服务/网盘 (百度网盘) - ai_chat: AI 聊天平台 (Gemini, DeepSeek, 智谱清言, 豆包, Kimi, 腾讯元宝, 通义千问, z.ai) - dev: 开发工具 (GitHub) - content: 内容平台 (抖音, 哔哩哔哩) - custom: 用户自定义 用法: from core.site_registry import get_site_registry reg = get_site_registry() sites = reg.list_sites() # 所有网站 site = reg.get_site("gmail") # 单个网站配置 reg.add_site(config_dict) # 添加自定义网站 reg.remove_site("my_site") # 删除自定义网站 """ from __future__ import annotations import json import os import time from pathlib import Path from typing import Any, Dict, List, Optional from core.logger import get_logger logger = get_logger("myagent.core.site_registry") # ══════════════════════════════════════════════════════════════ # 内置网站配置（20 个） # ══════════════════════════════════════════════════════════════ BUILTIN_SITES: Dict[str, Dict[str, Any]] = { # ── 邮箱 ── "gmail": { "display_name": "Gmail", "category": "email", "login_url": "https://accounts.google.com/signin", "detect_url": "https://mail.google.com/mail/u/0/", "home_url": "https://mail.google.com/mail/u/0/#inbox", "urls": { "inbox": "https://mail.google.com/mail/u/0/#inbox", "sent": "https://mail.google.com/mail/u/0/#sent", "drafts": "https://mail.google.com/mail/u/0/#drafts", "spam": "https://mail.google.com/mail/u/0/#spam", "search": "https://mail.google.com/mail/u/0/#search/{query}", }, "tips": [ "Google 登录安全严格，首次登录大概率需要验证码/手机验证", "建议使用 --no-headless 有头模式，方便 VNC 观察", "Gmail 页面动态渲染多，导航后需 stealth-wait 等待加载", "Cookie 失效后需重新走登录流程", ], }, "qq_mail": { "display_name": "QQ邮箱", "category": "email", "login_url": "https://mail.qq.com/", "detect_url": "https://mail.qq.com/cgi-bin/frame_html", "home_url": "https://mail.qq.com/cgi-bin/frame_html", "urls": { "inbox": "https://mail.qq.com/cgi-bin/frame_html?sid=undefined", "compose": "https://mail.qq.com/cgi-bin/qm_share?t=qm_mailme&email=", "contacts": "https://mail.qq.com/cgi-bin/addr_list", "search": "https://mail.qq.com/cgi-bin/search?query={query}", }, "tips": [ "支持 QQ 扫码登录和账号密码登录", "页面结构复杂，建议多用 stealth-screenshot 确认状态", "QQ邮箱可能有安全限制，频繁操作可能触发验证", ], }, "mail139": { "display_name": "139邮箱", "category": "email", "login_url": "https://mail.10086.cn/", "detect_url": "https://mail.10086.cn/", "home_url": "https://mail.10086.cn/", "urls": { "inbox": "https://mail.10086.cn/", "compose": "https://mail.10086.cn/RmC/mobileForm", "search": "https://mail.10086.cn/normal/search.action", }, "tips": [ "中国移动邮箱，支持手机号登录", "首次使用需在浏览器中手动完成登录", ], }, # ── 社交媒体 ── "x_com": { "display_name": "X.com (Twitter)", "category": "social", "login_url": "https://x.com/i/flow/login", "detect_url": "https://x.com/home", "home_url": "https://x.com/home", "urls": { "home": "https://x.com/home", "explore": "https://x.com/explore", "notifications": "https://x.com/notifications", "messages": "https://x.com/messages", "search": "https://x.com/search?q={query}", "profile": "https://x.com/{username}", }, "tips": [ "X.com 对自动化检测较严格，DrissionPage 反检测模式效果好", "发推文时注意字符限制", "可能遇到手机验证，使用 stealth-wait-manual 等待", ], }, "weibo": { "display_name": "微博", "category": "social", "login_url": "https://passport.weibo.com/sso/signin", "detect_url": "https://weibo.com/", "home_url": "https://weibo.com/", "urls": { "home": "https://weibo.com/", "explore": "https://s.weibo.com/top/summary", "search": "https://s.weibo.com/weibo?q={query}", "messages": "https://message.weibo.com/", "profile": "https://weibo.com/{username}", }, "tips": [ "支持微博扫码登录和账号密码登录", "微博反爬严格，操作间隔建议 > 3 秒", ], }, "wechat_web": { "display_name": "微信网页版", "category": "social", "login_url": "https://wx.qq.com/", "detect_url": "https://wx.qq.com/", "home_url": "https://wx.qq.com/", "urls": { "home": "https://wx.qq.com/", }, "tips": [ "微信网页版仅支持扫码登录，无账号密码入口", "需使用 stealth-wait-manual 等待扫码完成", "注意：微信网页版功能有限，部分账号已被禁止网页端登录", ], }, "wechat_mp": { "display_name": "微信公众号", "category": "social", "login_url": "https://mp.weixin.qq.com/", "detect_url": "https://mp.weixin.qq.com/cgi-bin/home?t=home/index", "home_url": "https://mp.weixin.qq.com/cgi-bin/home?t=home/index", "urls": { "home": "https://mp.weixin.qq.com/cgi-bin/home?t=home/index", "articles": "https://mp.weixin.qq.com/cgi-bin/appmsg?t=media/appmsg_edit_v2&action=edit&isNew=1&type=77&lang=zh_CN&token=", "fans": "https://mp.weixin.qq.com/cgi-bin/contactmanage?t=user/index&lang=zh_CN&token=", }, "tips": [ "仅支持管理员扫码登录", "登录后 Token 有效期较长，Profile 持久化效果好", ], }, # ── 云服务 / 网盘 ── "baidu_netdisk": { "display_name": "百度网盘", "category": "cloud", "login_url": "https://pan.baidu.com/", "detect_url": "https://pan.baidu.com/disk/main", "home_url": "https://pan.baidu.com/disk/main", "urls": { "home": "https://pan.baidu.com/disk/main", "share": "https://pan.baidu.com/disk/main#/share", "recycle": "https://pan.baidu.com/disk/main#/recycle", "search": "https://pan.baidu.com/disk/main#/search?key={query}", "apps": "https://pan.baidu.com/disk/main#/apps", }, "tips": [ "百度网盘对未登录用户限制下载速度", "首次登录需手机短信验证", "大文件操作可能需要较长时间，注意超时设置", ], }, # ── AI 聊天平台 ── "gemini": { "display_name": "Google Gemini", "category": "ai_chat", "login_url": "https://gemini.google.com/", "detect_url": "https://gemini.google.com/app", "home_url": "https://gemini.google.com/app", "urls": { "home": "https://gemini.google.com/app", "history": "https://gemini.google.com/app/activity", }, "tips": [ "需要 Google 账号登录", "对话界面动态渲染，需等待加载完成", "可配合 stealth-fill 在输入框中填写问题", ], }, "deepseek": { "display_name": "DeepSeek", "category": "ai_chat", "login_url": "https://chat.deepseek.com/", "detect_url": "https://chat.deepseek.com/", "home_url": "https://chat.deepseek.com/", "urls": { "home": "https://chat.deepseek.com/", "history": "https://chat.deepseek.com/list", }, "tips": [ "支持手机号登录和第三方账号登录", "对话输入框位于页面底部", ], }, "z_ai": { "display_name": "z.ai", "category": "ai_chat", "login_url": "https://z.ai/", "detect_url": "https://z.ai/", "home_url": "https://z.ai/", "urls": { "home": "https://z.ai/", }, "tips": [ "z.ai 智谱平台", ], }, "zhipu": { "display_name": "智谱清言", "category": "ai_chat", "login_url": "https://chatglm.cn/", "detect_url": "https://chatglm.cn/main/alltoolsdetail", "home_url": "https://chatglm.cn/main/alltoolsdetail", "urls": { "home": "https://chatglm.cn/main/alltoolsdetail", "history": "https://chatglm.cn/main/detail", }, "tips": [ "智谱 AI 官方对话平台", "支持手机号和微信扫码登录", ], }, "doubao": { "display_name": "豆包", "category": "ai_chat", "login_url": "https://www.doubao.com/", "detect_url": "https://www.doubao.com/chat/", "home_url": "https://www.doubao.com/chat/", "urls": { "home": "https://www.doubao.com/chat/", }, "tips": [ "字节跳动旗下 AI 助手", "支持手机号和抖音账号登录", ], }, "kimi": { "display_name": "Kimi", "category": "ai_chat", "login_url": "https://kimi.moonshot.cn/", "detect_url": "https://kimi.moonshot.cn/chat/", "home_url": "https://kimi.moonshot.cn/chat/", "urls": { "home": "https://kimi.moonshot.cn/chat/", "history": "https://kimi.moonshot.cn/history", }, "tips": [ "月之暗面旗下 AI 助手", "支持手机号登录", "支持长文本输入和文件上传", ], }, "tencent_yuanbao": { "display_name": "腾讯元宝", "category": "ai_chat", "login_url": "https://yuanbao.tencent.com/", "detect_url": "https://yuanbao.tencent.com/chat", "home_url": "https://yuanbao.tencent.com/chat", "urls": { "home": "https://yuanbao.tencent.com/chat", }, "tips": [ "腾讯混元大模型对话平台", "支持微信扫码和 QQ 登录", ], }, "qianwen": { "display_name": "通义千问", "category": "ai_chat", "login_url": "https://tongyi.aliyun.com/qianwen/", "detect_url": "https://tongyi.aliyun.com/qianwen/", "home_url": "https://tongyi.aliyun.com/qianwen/", "urls": { "home": "https://tongyi.aliyun.com/qianwen/", "history": "https://tongyi.aliyun.com/qianwen/#/chat-history", }, "tips": [ "阿里云旗下 AI 助手", "支持阿里云账号和支付宝登录", ], }, # ── 开发工具 ── "github": { "display_name": "GitHub", "category": "dev", "login_url": "https://github.com/login", "detect_url": "https://github.com/", "home_url": "https://github.com/", "urls": { "home": "https://github.com/", "notifications": "https://github.com/notifications", "pulls": "https://github.com/pulls", "issues": "https://github.com/issues", "search": "https://github.com/search?q={query}", }, "tips": [ "支持账号密码和 2FA 登录", "2FA 场景使用 stealth-wait-manual 等待", ], }, # ── 内容平台 ── "douyin": { "display_name": "抖音", "category": "content", "login_url": "https://www.douyin.com/", "detect_url": "https://www.douyin.com/", "home_url": "https://www.douyin.com/", "urls": { "home": "https://www.douyin.com/", "search": "https://www.douyin.com/search/{query}", "favorites": "https://www.douyin.com/favorite/", }, "tips": [ "支持抖音 App 扫码登录和手机号登录", "抖音页面大量 JS 渲染，需耐心等待加载", ], }, "bilibili": { "display_name": "哔哩哔哩", "category": "content", "login_url": "https://passport.bilibili.com/login", "detect_url": "https://www.bilibili.com/", "home_url": "https://www.bilibili.com/", "urls": { "home": "https://www.bilibili.com/", "search": "https://search.bilibili.com/all?keyword={query}", "favorites": "https://space.bilibili.com/{mid}/favlist", "history": "https://www.bilibili.com/account/history", }, "tips": [ "B站支持密码登录和第三方扫码", "视频播放页面信息丰富，stealth-content 返回量大", ], }, # ── Google 账号（通用，共享 Profile） ── "google": { "display_name": "Google 账号", "category": "dev", "login_url": "https://accounts.google.com/", "detect_url": "https://myaccount.google.com/", "home_url": "https://myaccount.google.com/", "urls": { "account": "https://myaccount.google.com/", "security": "https://myaccount.google.com/security", "privacy": "https://myaccount.google.com/privacy", }, "tips": [ "Google 账号是 Gmail/Gemini 等服务的统一登录入口", "Profile 与 Gmail 共享，无需单独登录", ], }, } # ══════════════════════════════════════════════════════════════ # 分类元数据 # ══════════════════════════════════════════════════════════════ SITE_CATEGORIES: Dict[str, Dict[str, str]] = { "email": {"name": "邮箱服务", "icon": "📧"}, "social": {"name": "社交媒体", "icon": "💬"}, "cloud": {"name": "云服务", "icon": "☁️"}, "ai_chat": {"name": "AI 对话", "icon": "🤖"}, "dev": {"name": "开发工具", "icon": "🛠️"}, "content": {"name": "内容平台", "icon": "🎬"}, "custom": {"name": "自定义", "icon": "⚙️"}, } # ══════════════════════════════════════════════════════════════ # SiteRegistry 类 # ══════════════════════════════════════════════════════════════ class SiteRegistry: """ 统一网站注册表。 加载顺序: 1. BUILTIN_SITES（内置，代码随包分发） 2. ~/.myagent/sites/*.json（用户自定义，可覆盖内置配置） 用法: reg = SiteRegistry() # 列出所有网站 sites = reg.list_sites() # 获取单个网站 site = reg.get_site("gmail") # 添加自定义网站 reg.add_site({ "name": "my_site", "display_name": "我的网站", "category": "custom", "login_url": "https://example.com/login", "detect_url": "https://example.com/dashboard", "home_url": "https://example.com/dashboard", }) # 删除自定义网站 reg.remove_site("my_site") """ def __init__(self, base_dir: Optional[str] = None): """ Args: base_dir: 自定义网站配置目录，默认 ~/.myagent/sites/ """ if base_dir: self.base_dir = Path(base_dir) else: # 遵循项目配置的数据目录 config_path = Path.home() / ".myagent" / "config.json" data_base = Path.home() / ".myagent" if config_path.exists(): try: cfg = json.loads(config_path.read_text(encoding="utf-8")) custom = cfg.get("data_dir", "") if custom: expanded = os.path.expanduser(custom) if not os.path.isabs(expanded): expanded = str(Path.home() / expanded) data_base = Path(expanded) except Exception: pass self.base_dir = data_base / "sites" self.base_dir.mkdir(parents=True, exist_ok=True) self._cache: Dict[str, Dict[str, Any]] = {} self._dirty = True # 是否需要重新加载 logger.debug(f"SiteRegistry 初始化: {self.base_dir}") def _load_custom_sites(self) -> Dict[str, Dict[str, Any]]: """从 ~/.myagent/sites/*.json 加载自定义网站配置""" custom = {} if not self.base_dir.exists(): return custom for f in sorted(self.base_dir.glob("*.json")): try: data = json.loads(f.read_text(encoding="utf-8")) name = data.get("name", f.stem) custom[name] = data except Exception as e: logger.warning(f"加载网站配置失败 ({f.name}): {e}") return custom def _ensure_loaded(self) -> None: """确保缓存是最新的""" if self._dirty: self._cache = {} # 先加载内置 self._cache.update(BUILTIN_SITES) # 再加载自定义（覆盖同名内置） self._cache.update(self._load_custom_sites()) self._dirty = False def list_sites( self, category: str = "", include_builtin: bool = True, include_custom: bool = True, ) -> List[Dict[str, Any]]: """ 列出所有网站。 Args: category: 按分类筛选（空=全部） include_builtin: 是否包含内置网站 include_custom: 是否包含自定义网站 Returns: 网站信息列表 """ self._ensure_loaded() custom_names = set(self._load_custom_sites().keys()) results = [] for name, site in sorted(self._cache.items()): is_builtin = name in BUILTIN_SITES is_custom = name in custom_names if not include_builtin and is_builtin and not is_custom: continue if not include_custom and is_custom and not is_builtin: continue if category and site.get("category") != category: continue results.append({ "name": name, "display_name": site.get("display_name", name), "category": site.get("category", "custom"), "login_url": site.get("login_url", ""), "home_url": site.get("home_url", ""), "is_builtin": is_builtin, "is_custom": is_custom, "has_profile": self._check_profile_exists(name), }) return results def get_site(self, name: str) -> Optional[Dict[str, Any]]: """ 获取网站配置。 Args: name: 网站名称 Returns: 网站配置字典（不存在返回 None） """ self._ensure_loaded() site = self._cache.get(name) if site: result = dict(site) result["name"] = name result["is_builtin"] = name in BUILTIN_SITES return result return None def get_categories(self) -> Dict[str, Dict[str, Any]]: """ 获取所有分类及其网站数量。 Returns: {category_name: {name, icon, count}} """ self._ensure_loaded() cats = {} for cat_info in SITE_CATEGORIES.values(): cats[cat_info["name"]] = {**cat_info, "count": 0} for site in self._cache.values(): cat = site.get("category", "custom") cat_name = SITE_CATEGORIES.get(cat, {}).get("name", "自定义") if cat_name in cats: cats[cat_name]["count"] += 1 return cats def get_all_site_names(self) -> List[str]: """获取所有网站名称列表""" self._ensure_loaded() return sorted(self._cache.keys()) def get_sites_by_category(self, category: str) -> List[Dict[str, Any]]: """获取指定分类下的所有网站""" return self.list_sites(category=category) def add_site(self, config: Dict[str, Any]) -> Dict[str, Any]: """ 添加自定义网站。 Args: config: 网站配置字典，必须包含 name 字段 Returns: 创建后的网站信息 Raises: ValueError: 配置无效或名称冲突 """ name = config.get("name", "").strip().lower().replace(" ", "_") if not name: raise ValueError("网站名称不能为空") if not all(c.isalnum() or c in "_-" for c in name): raise ValueError(f"网站名称只能包含字母、数字、下划线和连字符: {name}") # 确保基本字段存在 config["name"] = name config.setdefault("display_name", name) config.setdefault("category", "custom") config.setdefault("login_url", "") config.setdefault("detect_url", "") config.setdefault("home_url", "") config.setdefault("urls", {}) config.setdefault("tips", []) config["created_at"] = time.time() # 保存到文件 config_file = self.base_dir / f"{name}.json" config_file.write_text( json.dumps(config, ensure_ascii=False, indent=2), encoding="utf-8", ) self._dirty = True logger.info(f"已添加自定义网站: {name}") site = self.get_site(name) return site or config def update_site(self, name: str, updates: Dict[str, Any]) -> Optional[Dict[str, Any]]: """ 更新网站配置。 Args: name: 网站名称 updates: 要更新的字段 Returns: 更新后的网站信息 """ self._ensure_loaded() if name not in self._cache: return None # 不允许修改内置网站的 name 和 is_builtin updates.pop("name", None) updates.pop("is_builtin", None) updates["updated_at"] = time.time() # 如果是内置网站，自定义配置保存为覆盖 config_file = self.base_dir / f"{name}.json" existing = self._cache[name] merged = {**existing, **updates} config_file.write_text( json.dumps(merged, ensure_ascii=False, indent=2), encoding="utf-8", ) self._dirty = True logger.info(f"已更新网站配置: {name}") return self.get_site(name) def remove_site(self, name: str) -> bool: """ 删除自定义网站。 注意：内置网站不能删除，但可以隐藏。 Args: name: 网站名称 Returns: 是否成功删除 """ if name in BUILTIN_SITES: logger.warning(f"不能删除内置网站: {name}（可使用 update_site 禁用）") return False config_file = self.base_dir / f"{name}.json" if config_file.exists(): config_file.unlink() self._dirty = True logger.info(f"已删除自定义网站: {name}") return True return False def init_profiles(self, names: Optional[List[str]] = None) -> List[str]: """ 批量初始化网站的浏览器 Profile（仅创建目录结构）。 Args: names: 网站名称列表（空=全部内置网站） Returns: 已初始化的网站名称列表 """ from core.browser_profile import get_browser_profile_manager mgr = get_browser_profile_manager() initialized = [] site_names = names or list(BUILTIN_SITES.keys()) for name in site_names: site = self.get_site(name) if not site: continue try: profile = mgr.get_profile(name) if not profile.is_initialized(): profile.ensure_dirs() # 保存站点配置到 Profile if site.get("display_name"): profile.update_config( display_name=site["display_name"], login_url=site.get("login_url", ""), detect_url=site.get("detect_url", ""), ) initialized.append(name) logger.info(f"已初始化 Profile: {name}") except Exception as e: logger.error(f"初始化 Profile 失败 ({name}): {e}") return initialized def to_summary(self) -> Dict[str, Any]: """ 获取注册表摘要信息。 Returns: {total, builtin_count, custom_count, categories, sites} """ self._ensure_loaded() custom_names = set(self._load_custom_sites().keys()) builtin_count = sum(1 for n in self._cache if n in BUILTIN_SITES and n not in custom_names) custom_count = sum(1 for n in self._cache if n in custom_names and n not in BUILTIN_SITES) return { "total": len(self._cache), "builtin_count": builtin_count, "custom_count": custom_count, "categories": self.get_categories(), "sites": self.list_sites(), } def _check_profile_exists(self, name: str) -> bool: """检查网站是否有已创建的 Profile""" try: from core.browser_profile import get_browser_profile_manager mgr = get_browser_profile_manager() profile = mgr.get_profile(name) return profile.is_initialized() except Exception: return False # ══════════════════════════════════════════════════════════════ # 全局单例 # ══════════════════════════════════════════════════════════════ _registry: Optional[SiteRegistry] = None def get_site_registry() -> SiteRegistry: """获取全局 SiteRegistry 单例""" global _registry if _registry is None: _registry = SiteRegistry() return _registry