diff --git a/.env.example b/.env.example index 248671e..d23fb80 100644 --- a/.env.example +++ b/.env.example @@ -1,14 +1,15 @@ -# === Telegram (optional, for TG channel fetching) === -TG_API_ID= -TG_API_HASH= +# Agent Eyes Configuration +# Copy to .env and fill in your values +# Or use: agent-eyes setup -# === Groq Whisper (optional, for video/audio transcription) === -# Free API key: https://console.groq.com/keys -GROQ_API_KEY= +# Exa Search (free 1000/month) — https://exa.ai +# EXA_API_KEY=exa-your-key-here -# === Gemini (optional, for AI-powered analysis) === -GEMINI_API_KEY= +# GitHub Token (optional, for higher rate limits) — https://github.com/settings/tokens +# GITHUB_TOKEN=ghp_your_token_here -# === Output === -OUTPUT_DIR=./output -INBOX_FILE=./unified_inbox.json +# Reddit ISP Proxy (optional, for full Reddit access) +# REDDIT_PROXY=http://user:pass@ip:port + +# Groq Whisper (optional, for video transcription) — https://console.groq.com +# GROQ_API_KEY=gsk_your_key_here diff --git a/.gitignore b/.gitignore index 45357b9..d4e546a 100644 --- a/.gitignore +++ b/.gitignore @@ -1,32 +1,10 @@ -# Python __pycache__/ -*.py[cod] +*.pyc +*.pyo +.pytest_cache/ *.egg-info/ dist/ build/ -.venv/ -venv/ - -# Environment .env - -# Telethon session -*.session -tg_session* - -# Data -*.db +.agent-eyes/ *.log -data/ -unified_inbox.json -output/ - -# Audio/video temp files -/tmp/media_* - -# OS -.DS_Store - -# IDE -.vscode/ -.idea/ diff --git a/README.md b/README.md index 2e86eb2..c72a77a 100644 --- a/README.md +++ b/README.md @@ -2,24 +2,22 @@ **Give your AI Agent eyes to see the entire internet.** -[![Python 3.10+](https://img.shields.io/badge/python-3.10+-blue.svg)](https://www.python.org/downloads/) -[![License: MIT](https://img.shields.io/badge/License-MIT-green.svg)](LICENSE) +Agent Eyes is infrastructure for the Agent world — install once, and your AI Agent can search and read content across 10+ platforms. Zero to minimal configuration. -Agent Eyes is an open-source infrastructure tool that gives any AI Agent the ability to **search** and **read** the entire internet. One install, 10+ platforms, unified output. +> *Built on top of [x-reader](https://github.com/runesleo/x-reader) by [@runes_leo](https://x.com/runes_leo) — thank you for the amazing open-source work. 🙏* -> 🙏 Built on the shoulders of [x-reader](https://github.com/runesleo/x-reader) by [@runes_leo](https://x.com/runes_leo). Thank you for the foundation! +--- ## Why Agent Eyes? -Your AI Agent is blind. It can only see what you manually feed it. +AI Agents are powerful, but blind. They can't browse Reddit, search Twitter, or read WeChat articles. Agent Eyes fixes that. -Agent Eyes gives it **eyes** — the ability to: -- **Search** the web, Reddit, GitHub with a single command -- **Read** any URL from 10+ platforms (articles, videos, tweets, posts) -- **Transcribe** videos and podcasts to text +- **One install, 10+ platforms** — web pages, GitHub, Reddit, Twitter, YouTube, Bilibili, WeChat, XiaoHongShu, RSS, Telegram +- **Search + Read** — not just URL extraction, but actual search capability (find, then read) +- **Agent-native design** — CLI, MCP Server, Python API — works with any AI Agent platform +- **Mostly free** — 99% of features cost nothing. One free API key unlocks full search. -Without this, your Agent is a chatbot waiting for instructions. -With this, it can autonomously find and consume information — just like you do. +--- ## Quick Start @@ -27,134 +25,232 @@ With this, it can autonomously find and consume information — just like you do # Install pip install git+https://github.com/Panniantong/agent-eyes.git -# Search the web -agent-eyes search "AI agent framework 2026" +# Check what's available +agent-eyes doctor -# Read any URL -agent-eyes read https://reddit.com/r/LocalLLaMA/comments/xxx -agent-eyes read https://github.com/openai/codex -agent-eyes read https://mp.weixin.qq.com/s/xxx -agent-eyes read https://x.com/elonmusk/status/xxx +# Read any URL (works immediately, no config needed) +agent-eyes read "https://github.com/openai/gpt-4" +agent-eyes read "https://www.bilibili.com/video/BV1xx411c7mD" -# Your Agent now has eyes 👁️ +# Search GitHub (works immediately, no config needed) +agent-eyes search-github "LLM framework" + +# Unlock search (one free API key) +agent-eyes setup +agent-eyes search "AI agent infrastructure 2025" +agent-eyes search-reddit "best self-hosted LLM" --sub LocalLLaMA +agent-eyes search-twitter "OpenClaw agent" ``` -## Supported Platforms +--- -| Platform | Read URL | Search | Notes | -|----------|:--------:|:------:|-------| -| 🔍 Web (any) | ✅ | ✅ Exa | Semantic search across the entire web | -| 🟠 Reddit | ✅ | ✅ | Posts + comments. Proxy support via `REDDIT_PROXY` | -| 🐙 GitHub | ✅ | ✅ | Repos (README), Issues, PRs | -| 🐦 X / Twitter | ✅ | — | Tweets and threads | -| 💬 WeChat (微信公众号) | ✅ | — | Anti-scraping bypass via Playwright | -| 📕 Xiaohongshu (小红书) | ✅ | — | Session persistence for login-gated content | -| ▶️ YouTube | ✅ | — | Subtitles + Whisper transcription | -| 📺 Bilibili (B站) | ✅ | — | Official API | -| ✈️ Telegram | ✅ | — | Channel message sync | -| 📡 RSS | ✅ | — | Any RSS/Atom feed | -| 🎙️ Podcasts | ✅ | — | 小宇宙, Apple Podcasts (via Whisper) | +## Platform Support -## Three Layers +### ✅ Zero Config (works out of the box) -Use the layers you need: +| Platform | Read | Search | Notes | +|----------|:----:|:------:|-------| +| Web Pages | ✅ | — | Any URL via Jina Reader | +| GitHub | ✅ | ✅ | Repos, issues, PRs, code | +| Bilibili | ✅ | — | Videos with subtitles | +| YouTube | ✅ | — | Videos with subtitles (needs yt-dlp) | +| RSS | ✅ | — | Any RSS/Atom feed | +| Single Tweet | ✅ | — | Via Jina Reader | -| Layer | What | For | -|-------|------|-----| -| **CLI** | `agent-eyes read/search` | Quick command-line use | -| **MCP Server** | 7 tools for any AI Agent | OpenClaw, Claude Code, etc. | -| **Python Library** | `from agent_eyes import UniversalReader` | Custom integrations | +### 🔑 One Free API Key (30 seconds to set up) -### As MCP Server (recommended for Agents) +| Platform | Read | Search | Notes | +|----------|:----:|:------:|-------| +| Web (semantic) | — | ✅ | Exa API (1000 free/month) | +| Reddit | — | ✅ | Via Exa (site:reddit.com) | +| Twitter/X | — | ✅ | Via Exa (site:x.com) | + +### ⚙️ Optional Setup + +| Platform | Read | Search | What's Needed | +|----------|:----:|:------:|---------------| +| Reddit (full) | ✅ | — | ISP proxy (~$3-10/mo) | +| Twitter (advanced) | ✅ | ✅ | birdx + browser cookies (free) | +| WeChat Articles | ✅ | — | Playwright (free, auto-installed) | +| XiaoHongShu | ✅ | — | Playwright + one-time login (free) | +| Video Transcription | ✅ | — | Groq API key (free) | + +--- + +## Three Ways to Use + +### 1. Command Line (CLI) ```bash -# Start the server -python mcp_server.py - -# Or with SSE transport -python mcp_server.py --transport sse +agent-eyes read # Read any URL +agent-eyes search "query" # Search the web +agent-eyes search-reddit "query" # Search Reddit +agent-eyes search-github "query" # Search GitHub +agent-eyes search-twitter "query" # Search Twitter +agent-eyes setup # Interactive setup wizard +agent-eyes doctor # Check platform status ``` -MCP Tools exposed: +### 2. MCP Server (for Claude Code, Cursor, etc.) -| Tool | Description | -|------|-------------| -| `read_url(url)` | Read any URL → structured content | -| `read_batch(urls)` | Read multiple URLs concurrently | -| `search(query)` | Semantic web search (Exa) | -| `search_reddit(query, subreddit?)` | Search Reddit | -| `search_github(query)` | Search GitHub repos | -| `list_inbox()` | View previously fetched content | -| `detect_platform(url)` | Identify platform from URL | +```bash +pip install agent-eyes[mcp] +python -m agent_eyes.integrations.mcp_server +``` -### As Python Library +Exposes 8 tools: `read_url`, `read_batch`, `detect_platform`, `search`, `search_reddit`, `search_github`, `search_twitter`, `get_status` + +Add to your MCP config: + +```json +{ + "mcpServers": { + "agent-eyes": { + "command": "python", + "args": ["-m", "agent_eyes.integrations.mcp_server"] + } + } +} +``` + +### 3. Python Library ```python +from agent_eyes import AgentEyes import asyncio -from agent_eyes.reader import UniversalReader -async def main(): - reader = UniversalReader() - - # Read any URL - content = await reader.read("https://github.com/openai/codex") - print(content.title) - print(content.content[:500]) +eyes = AgentEyes() -asyncio.run(main()) +# Read +result = asyncio.run(eyes.read("https://github.com/openai/gpt-4")) +print(result["title"]) +print(result["content"]) + +# Search +results = asyncio.run(eyes.search("AI agent framework")) +for r in results: + print(f"{r['title']} — {r['url']}") + +# Search Reddit +results = asyncio.run(eyes.search_reddit("best LLM", subreddit="LocalLLaMA")) + +# Health check +print(eyes.doctor_report()) ``` -## Install - -```bash -# Basic install -pip install git+https://github.com/Panniantong/agent-eyes.git - -# With browser fallback (for WeChat/XHS anti-scraping) -pip install "agent-eyes[browser] @ git+https://github.com/Panniantong/agent-eyes.git" -playwright install chromium - -# With Telegram support -pip install "agent-eyes[telegram] @ git+https://github.com/Panniantong/agent-eyes.git" - -# Everything -pip install "agent-eyes[all] @ git+https://github.com/Panniantong/agent-eyes.git" -playwright install chromium -``` +--- ## Configuration -| Variable | Required | Description | -|----------|----------|-------------| -| `EXA_API_KEY` | For search | Free key from [exa.ai](https://exa.ai) | -| `REDDIT_PROXY` | For Reddit (if IP blocked) | `http://user:pass@host:port` | -| `GITHUB_TOKEN` | No (higher rate limits) | GitHub personal access token | -| `GROQ_API_KEY` | For Whisper | Free key from [groq.com](https://console.groq.com/keys) | -| `TG_API_ID` | Telegram only | From https://my.telegram.org | -| `TG_API_HASH` | Telegram only | From https://my.telegram.org | +### Interactive Setup -## What's New (vs x-reader) +```bash +agent-eyes setup +``` -Agent Eyes extends x-reader with: +Walks you through configuring each platform step by step. Only asks for what you want to set up. -- 🟠 **Reddit support** — Read posts + comments, search subreddits. Proxy support for blocked IPs. -- 🐙 **GitHub support** — Read repos (README), issues, PRs. Search repositories. -- 🔍 **Web search** — Semantic search across the entire web via Exa. -- 🎯 **Agent-first design** — MCP Server with 7 tools, ready to plug into any AI Agent. +### Agent-Readable Guides -## Philosophy +Each platform has a detailed setup guide in `agent_eyes/guides/`. These are designed for AI Agents to read and follow — the Agent handles the technical steps, and only asks the user for things that require human action (logging in, copying API keys, buying proxies). -This is **Agent infrastructure**. In the Web 4.0 era where AI Agents act on behalf of humans, the first capability they need is the ability to **see the world**. +| Guide | What It Configures | +|-------|-------------------| +| `setup-exa.md` | Exa search API key (free) | +| `setup-reddit.md` | Reddit ISP proxy | +| `setup-twitter.md` | Twitter birdx cookies | +| `setup-xiaohongshu.md` | XiaoHongShu login | +| `setup-wechat.md` | WeChat Playwright | +| `setup-groq.md` | Groq Whisper API key (free) | -Agent Eyes is the sensory layer — the eyes — that every Agent needs. +### Manual Config + +Config file: `~/.agent-eyes/config.yaml` + +```yaml +exa_api_key: "exa-..." +github_token: "ghp_..." +reddit_proxy: "http://user:pass@ip:port" +groq_api_key: "gsk_..." +``` + +Environment variables also work (uppercase): `EXA_API_KEY`, `GITHUB_TOKEN`, etc. + +--- + +## Health Check + +```bash +$ agent-eyes doctor + +👁️ Agent Eyes Status +======================================== + +✅ Ready (no setup needed): + ✅ Web Pages + ✅ GitHub + ✅ Bilibili + ✅ RSS + ✅ Tweet (single) + ✅ GitHub Search + +🔍 Search (need free Exa API key): + ⬜ Web Search + ⬜ Reddit Search + ⬜ Twitter Search + +🔧 Optional (advanced setup): + ⬜ Twitter Advanced — Install birdx for timeline/deep search + ⬜ Reddit Reader — Need proxy for full post reading + ⬜ WeChat — pip install agent-eyes[browser] + ⬜ XiaoHongShu — pip install agent-eyes[browser] + +Status: 7/13 platforms active +Run `agent-eyes setup` to unlock more! +``` + +--- + +## Architecture + +Agent Eyes is **not** a fork of x-reader. It's a layer built on top of it. + +``` +┌──────────────────────────────────────┐ +│ Agent Eyes │ +│ ┌──────────┐ ┌──────────────────┐ │ +│ │ Search │ │ Config + Doctor │ │ +│ │ Exa │ │ Setup Wizard │ │ +│ │ Reddit │ │ Health Check │ │ +│ │ GitHub │ │ │ │ +│ │ Twitter │ │ Agent Guides │ │ +│ └──────────┘ └──────────────────┘ │ +│ ┌──────────────────────────────────┐│ +│ │ Readers (based on x-reader) ││ +│ │ Web · GitHub · Reddit · Twitter ││ +│ │ YouTube · Bilibili · WeChat ││ +│ │ XHS · RSS · Telegram ││ +│ └──────────────────────────────────┘│ +│ ┌──────────────────────────────────┐│ +│ │ Integrations ││ +│ │ CLI · MCP Server · OpenClaw Skill│ +│ └──────────────────────────────────┘│ +└──────────────────────────────────────┘ +``` + +- **Readers**: Content extraction from URLs (internalized from x-reader, MIT license) +- **Search**: Semantic search across platforms (Agent Eyes original) +- **Config + Doctor**: Configuration management and health checks (Agent Eyes original) +- **Integrations**: CLI, MCP Server, OpenClaw Skill (Agent Eyes original) + +--- ## Credits -- [x-reader](https://github.com/runesleo/x-reader) by [@runes_leo](https://x.com/runes_leo) — the original universal content reader that inspired and powers the core of Agent Eyes -- [Jina Reader](https://jina.ai/reader/) — universal web content extraction -- [Exa](https://exa.ai) — semantic web search API -- [yt-dlp](https://github.com/yt-dlp/yt-dlp) — video/audio extraction +- **[x-reader](https://github.com/runesleo/x-reader)** by [@runes_leo](https://x.com/runes_leo) — the URL-to-content extraction engine that Agent Eyes is built upon. MIT License. +- **[Exa](https://exa.ai)** — semantic search API powering web/Reddit/Twitter search. +- **[birdx](https://github.com/runesleo/birdx)** — Twitter CLI tool for advanced Twitter features. ## License -MIT — use it, fork it, build on it. +MIT diff --git a/agent_eyes/__init__.py b/agent_eyes/__init__.py index 8ab0f3e..0ceaffc 100644 --- a/agent_eyes/__init__.py +++ b/agent_eyes/__init__.py @@ -1,3 +1,9 @@ -"""x-reader: Universal content reader for 7+ platforms.""" +# -*- coding: utf-8 -*- +"""Agent Eyes — Give your AI Agent eyes to see the entire internet.""" -__version__ = "0.1.0" +__version__ = "1.0.0" +__author__ = "Neo Reid" + +from agent_eyes.core import AgentEyes + +__all__ = ["AgentEyes"] diff --git a/agent_eyes/cli.py b/agent_eyes/cli.py index 3a64656..619f889 100644 --- a/agent_eyes/cli.py +++ b/agent_eyes/cli.py @@ -1,138 +1,247 @@ # -*- coding: utf-8 -*- """ -x-reader CLI — fetch content from any platform. +Agent Eyes CLI — command-line interface. Usage: - x-reader # Fetch a single URL - x-reader ... # Fetch multiple URLs - x-reader list # Show inbox contents - x-reader clear # Clear inbox + agent-eyes read + agent-eyes search + agent-eyes search-reddit [--sub ] + agent-eyes search-github [--lang ] + agent-eyes search-twitter + agent-eyes setup + agent-eyes doctor + agent-eyes version """ import sys import asyncio +import argparse import json -from pathlib import Path -from dotenv import load_dotenv -load_dotenv() - -from agent_eyes.reader import UniversalReader -from agent_eyes.schema import UnifiedInbox, SourceType - - -def get_inbox_path() -> str: - import os - return os.getenv("INBOX_FILE", "unified_inbox.json") - - -def cmd_fetch(urls: list[str]): - """Fetch one or more URLs.""" - inbox = UnifiedInbox(get_inbox_path()) - reader = UniversalReader(inbox=inbox) - - async def run(): - if len(urls) == 1: - item = await reader.read(urls[0]) - print(f"✅ [{item.source_type.value}] {item.title[:60]}") - print(f" {item.url}") - print(f" {item.content[:200]}...") - else: - items = await reader.read_batch(urls) - for item in items: - print(f"✅ [{item.source_type.value}] {item.title[:60]}") - print(f"\n📦 Fetched {len(items)}/{len(urls)} URLs") - - try: - asyncio.run(run()) - except KeyboardInterrupt: - print("\n⏹ Cancelled") - except Exception as e: - print(f"❌ {e}") - sys.exit(1) - - -def cmd_list(): - """Show inbox contents.""" - inbox = UnifiedInbox(get_inbox_path()) - if not inbox.items: - print("📦 Inbox is empty") - return - - print(f"📦 Inbox: {len(inbox.items)} items\n") - - emoji_map = { - SourceType.TELEGRAM: "📢", SourceType.RSS: "📰", - SourceType.BILIBILI: "🎬", SourceType.XIAOHONGSHU: "📕", - SourceType.TWITTER: "🐦", SourceType.WECHAT: "💬", - SourceType.YOUTUBE: "▶️", SourceType.MANUAL: "✏️", - } - - for i, item in enumerate(inbox.items[-20:], 1): - emoji = emoji_map.get(item.source_type, "📄") - print(f" {i:2d}. {emoji} [{item.source_type.value:8s}] {item.title[:50]}") - - -def cmd_clear(): - """Clear inbox.""" - path = Path(get_inbox_path()) - if path.exists(): - confirm = input("Clear inbox? (y/N) ") - if confirm.lower() == 'y': - path.write_text("[]") - print("✅ Inbox cleared") - else: - print("📦 Inbox is already empty") - - -def cmd_login(platform: str): - """Open browser for manual login to a platform.""" - from agent_eyes.login import login - login(platform) +from agent_eyes import __version__ def main(): - if len(sys.argv) < 2: - print(""" -📖 x-reader — Universal content reader + parser = argparse.ArgumentParser( + prog="agent-eyes", + description="👁️ Give your AI Agent eyes to see the entire internet", + ) + sub = parser.add_subparsers(dest="command", help="Available commands") -Usage: - x-reader Fetch content from any URL - x-reader Fetch multiple URLs - x-reader login Login to a platform (saves session for browser fallback) - x-reader list Show inbox contents - x-reader clear Clear inbox + # ── read ── + p_read = sub.add_parser("read", help="Read content from a URL") + p_read.add_argument("url", help="URL to read") + p_read.add_argument("--json", dest="as_json", action="store_true", help="Output as JSON") -Supported platforms: - WeChat, Telegram, X/Twitter, YouTube, - Bilibili, Xiaohongshu, RSS, and any web page + # ── search ── + p_search = sub.add_parser("search", help="Search the web (Exa)") + p_search.add_argument("query", nargs="+", help="Search query") + p_search.add_argument("-n", "--num", type=int, default=5, help="Number of results") -Examples: - x-reader https://mp.weixin.qq.com/s/abc123 - x-reader https://x.com/elonmusk/status/123456 - x-reader https://www.xiaohongshu.com/explore/abc123 - x-reader login xhs -""") - return + # ── search-reddit ── + p_sr = sub.add_parser("search-reddit", help="Search Reddit") + p_sr.add_argument("query", nargs="+", help="Search query") + p_sr.add_argument("--sub", help="Subreddit filter") + p_sr.add_argument("-n", "--num", type=int, default=10, help="Number of results") - cmd = sys.argv[1].lower() + # ── search-github ── + p_sg = sub.add_parser("search-github", help="Search GitHub") + p_sg.add_argument("query", nargs="+", help="Search query") + p_sg.add_argument("--lang", help="Language filter") + p_sg.add_argument("-n", "--num", type=int, default=5, help="Number of results") - if cmd == "login": - if len(sys.argv) < 3: - print("❌ Usage: x-reader login ") - print(" Supported: xhs, wechat") - sys.exit(1) - cmd_login(sys.argv[2]) - elif cmd == "list": - cmd_list() - elif cmd == "clear": - cmd_clear() - elif cmd.startswith("http") or cmd.startswith("www.") or "." in cmd: - urls = [arg for arg in sys.argv[1:] if arg.startswith(("http", "www.")) or "." in arg] - cmd_fetch(urls) + # ── search-twitter ── + p_st = sub.add_parser("search-twitter", help="Search Twitter") + p_st.add_argument("query", nargs="+", help="Search query") + p_st.add_argument("-n", "--num", type=int, default=10, help="Number of results") + + # ── setup ── + sub.add_parser("setup", help="Interactive configuration wizard") + + # ── doctor ── + sub.add_parser("doctor", help="Check platform availability") + + # ── version ── + sub.add_parser("version", help="Show version") + + args = parser.parse_args() + + if not args.command: + parser.print_help() + sys.exit(0) + + if args.command == "version": + print(f"Agent Eyes v{__version__}") + sys.exit(0) + + if args.command == "doctor": + _cmd_doctor() + elif args.command == "setup": + _cmd_setup() + elif args.command == "read": + asyncio.run(_cmd_read(args)) + elif args.command.startswith("search"): + asyncio.run(_cmd_search(args)) + + +# ── Command handlers ──────────────────────────────── + + +def _cmd_doctor(): + from agent_eyes.config import Config + from agent_eyes.doctor import check_all, format_report + config = Config() + results = check_all(config) + print(format_report(results)) + + +def _cmd_setup(): + from agent_eyes.config import Config + + config = Config() + print() + print("👁️ Agent Eyes Setup") + print("=" * 40) + print() + + # Step 1: Exa + print("【推荐】全网搜索 — Exa Search API") + print(" 免费 1000 次/月，注册地址: https://exa.ai") + current = config.get("exa_api_key") + if current: + print(f" 当前状态: ✅ 已配置 ({current[:8]}...)") + change = input(" 要更换吗？[y/N]: ").strip().lower() + if change != "y": + print() + else: + key = input(" EXA_API_KEY: ").strip() + if key: + config.set("exa_api_key", key) + print(" ✅ 已更新！") + print() else: - print(f"❌ Unknown command: {cmd}") - print(" Run 'x-reader' with no args for help") + print(" 当前状态: ⬜ 未配置") + key = input(" EXA_API_KEY (回车跳过): ").strip() + if key: + config.set("exa_api_key", key) + print(" ✅ 全网搜索 + Reddit搜索 + Twitter搜索 已开启！") + else: + print(" ℹ️ 跳过。稍后可运行 agent-eyes setup 配置") + print() + + # Step 2: GitHub token + print("【可选】GitHub Token — 提高 API 限额") + print(" 无 token: 60 次/小时 | 有 token: 5000 次/小时") + print(" 获取: https://github.com/settings/tokens (无需任何权限)") + current = config.get("github_token") + if current: + print(f" 当前状态: ✅ 已配置") + else: + key = input(" GITHUB_TOKEN (回车跳过): ").strip() + if key: + config.set("github_token", key) + print(" ✅ GitHub API 已提升至 5000 次/小时！") + else: + print(" ℹ️ 跳过。公开 API 也能用") + print() + + # Step 3: Reddit proxy + print("【可选】Reddit 代理 — 完整阅读 Reddit 帖子+评论") + print(" Reddit 封锁很多 IP，需要 ISP 代理才能直接访问") + print(" 格式: http://用户名:密码@IP:端口") + current = config.get("reddit_proxy") + if current: + print(f" 当前状态: ✅ 已配置") + else: + proxy = input(" REDDIT_PROXY (回车跳过): ").strip() + if proxy: + config.set("reddit_proxy", proxy) + print(" ✅ Reddit 完整阅读已开启！") + else: + print(" ℹ️ 跳过。仍可通过搜索获取 Reddit 内容") + print() + + # Step 4: Groq (Whisper) + print("【可选】Groq API — 视频无字幕时的语音转文字") + print(" 免费额度，注册: https://console.groq.com") + current = config.get("groq_api_key") + if current: + print(f" 当前状态: ✅ 已配置") + else: + key = input(" GROQ_API_KEY (回车跳过): ").strip() + if key: + config.set("groq_api_key", key) + print(" ✅ 语音转文字已开启！") + else: + print(" ℹ️ 跳过") + print() + + # Summary + print("=" * 40) + print(f"✅ 配置已保存到 {config.config_path}") + print("运行 agent-eyes doctor 查看完整状态") + print() + + +async def _cmd_read(args): + from agent_eyes.core import AgentEyes + eyes = AgentEyes() + try: + result = await eyes.read(args.url) + if args.as_json: + print(json.dumps(result, ensure_ascii=False, indent=2)) + else: + print(f"\n📖 {result.get('title', 'Untitled')}") + print(f"🔗 {result.get('url', '')}") + if result.get("author"): + print(f"👤 {result['author']}") + print(f"\n{result.get('content', '')}") + except Exception as e: + print(f"❌ Error: {e}", file=sys.stderr) + sys.exit(1) + + +async def _cmd_search(args): + from agent_eyes.core import AgentEyes + eyes = AgentEyes() + query = " ".join(args.query) + num = args.num + + try: + if args.command == "search": + results = await eyes.search(query, num_results=num) + elif args.command == "search-reddit": + results = await eyes.search_reddit(query, subreddit=getattr(args, "sub", None), limit=num) + elif args.command == "search-github": + results = await eyes.search_github(query, language=getattr(args, "lang", None), limit=num) + elif args.command == "search-twitter": + results = await eyes.search_twitter(query, limit=num) + else: + print(f"Unknown command: {args.command}", file=sys.stderr) + sys.exit(1) + + if not results: + print("No results found.") + return + + for i, r in enumerate(results, 1): + title = r.get("title") or r.get("name") or r.get("text", "")[:60] + url = r.get("url", "") + snippet = r.get("snippet") or r.get("description") or r.get("text", "") + print(f"\n{i}. {title}") + print(f" 🔗 {url}") + if snippet: + print(f" {snippet[:200]}") + # Extra info for GitHub + if "stars" in r: + print(f" ⭐ {r['stars']} 🍴 {r.get('forks', 0)} 📝 {r.get('language', '')}") + + except ValueError as e: + print(f"⚠️ {e}", file=sys.stderr) + sys.exit(1) + except Exception as e: + print(f"❌ Error: {e}", file=sys.stderr) + sys.exit(1) if __name__ == "__main__": diff --git a/agent_eyes/config.py b/agent_eyes/config.py new file mode 100644 index 0000000..f048a62 --- /dev/null +++ b/agent_eyes/config.py @@ -0,0 +1,96 @@ +# -*- coding: utf-8 -*- +"""Configuration management for Agent Eyes. + +Stores settings in ~/.agent-eyes/config.yaml. +Auto-creates directory on first use. +""" + +import os +from pathlib import Path +from typing import Any, Optional + +import yaml + + +class Config: + """Manages Agent Eyes configuration.""" + + CONFIG_DIR = Path.home() / ".agent-eyes" + CONFIG_FILE = CONFIG_DIR / "config.yaml" + + # Feature → required config keys + FEATURE_REQUIREMENTS = { + "exa_search": ["exa_api_key"], + "reddit_proxy": ["reddit_proxy"], + "twitter_birdx": ["twitter_auth_token", "twitter_ct0"], + "groq_whisper": ["groq_api_key"], + "github_token": ["github_token"], + } + + def __init__(self, config_path: Optional[Path] = None): + self.config_path = Path(config_path) if config_path else self.CONFIG_FILE + self.config_dir = self.config_path.parent + self.data: dict = {} + self._ensure_dir() + self.load() + + def _ensure_dir(self): + """Create config directory if it doesn't exist.""" + self.config_dir.mkdir(parents=True, exist_ok=True) + + def load(self): + """Load config from YAML file.""" + if self.config_path.exists(): + with open(self.config_path, "r") as f: + self.data = yaml.safe_load(f) or {} + else: + self.data = {} + + def save(self): + """Save config to YAML file.""" + self._ensure_dir() + with open(self.config_path, "w") as f: + yaml.dump(self.data, f, default_flow_style=False, allow_unicode=True) + + def get(self, key: str, default: Any = None) -> Any: + """Get a config value. Also checks environment variables (uppercase).""" + # Config file first + if key in self.data: + return self.data[key] + # Then env var (uppercase) + env_val = os.environ.get(key.upper()) + if env_val: + return env_val + return default + + def set(self, key: str, value: Any): + """Set a config value and save.""" + self.data[key] = value + self.save() + + def delete(self, key: str): + """Delete a config key and save.""" + self.data.pop(key, None) + self.save() + + def is_configured(self, feature: str) -> bool: + """Check if a feature has all required config.""" + required = self.FEATURE_REQUIREMENTS.get(feature, []) + return all(self.get(k) for k in required) + + def get_configured_features(self) -> dict: + """Return status of all optional features.""" + return { + feature: self.is_configured(feature) + for feature in self.FEATURE_REQUIREMENTS + } + + def to_dict(self) -> dict: + """Return config as dict (masks sensitive values).""" + masked = {} + for k, v in self.data.items(): + if any(s in k.lower() for s in ("key", "token", "password", "proxy")): + masked[k] = f"{str(v)[:8]}..." if v else None + else: + masked[k] = v + return masked diff --git a/agent_eyes/core.py b/agent_eyes/core.py new file mode 100644 index 0000000..7293858 --- /dev/null +++ b/agent_eyes/core.py @@ -0,0 +1,133 @@ +# -*- coding: utf-8 -*- +""" +AgentEyes — the unified entry point. + +Usage: + from agent_eyes import AgentEyes + + eyes = AgentEyes() + content = await eyes.read("https://github.com/openai/gpt-4") + results = await eyes.search("AI agent framework") +""" + +import asyncio +from typing import Any, Dict, List, Optional + +from agent_eyes.config import Config +from agent_eyes.reader import UniversalReader + + +class AgentEyes: + """Give your AI Agent eyes to see the entire internet.""" + + def __init__(self, config: Optional[Config] = None): + self.config = config or Config() + self.reader = UniversalReader() + + # ── Reading ───────────────────────────────────────── + + async def read(self, url: str) -> Dict[str, Any]: + """ + Read content from any URL. Auto-detects platform. + + Supported: Web, GitHub, Reddit, Twitter, YouTube, + Bilibili, WeChat, XiaoHongShu, RSS, Telegram, etc. + + Returns: + Dict with title, content, url, author, etc. + """ + content = await self.reader.read(url) + return content.to_dict() + + async def read_batch(self, urls: List[str]) -> List[Dict[str, Any]]: + """Read multiple URLs concurrently.""" + contents = await self.reader.read_batch(urls) + return [c.to_dict() for c in contents] + + def detect_platform(self, url: str) -> str: + """Detect what platform a URL belongs to.""" + return self.reader._detect_platform(url) + + # ── Searching ─────────────────────────────────────── + + async def search(self, query: str, num_results: int = 5) -> List[Dict[str, Any]]: + """ + Semantic web search via Exa. Requires Exa API key. + + Args: + query: Search query + num_results: Number of results (max 10) + """ + from agent_eyes.search.exa import search_web + return await search_web(query, num_results=num_results, config=self.config) + + async def search_reddit( + self, + query: str, + subreddit: Optional[str] = None, + limit: int = 10, + ) -> List[Dict[str, Any]]: + """ + Search Reddit via Exa (bypasses IP blocks). + + Args: + query: Search query + subreddit: Optional subreddit filter + limit: Number of results + """ + from agent_eyes.search.reddit import search_reddit + return await search_reddit(query, subreddit=subreddit, num_results=limit, config=self.config) + + async def search_github( + self, + query: str, + language: Optional[str] = None, + limit: int = 5, + ) -> List[Dict[str, Any]]: + """ + Search GitHub repositories. + + Args: + query: Search query + language: Filter by language + limit: Number of results + """ + from agent_eyes.search.github import search_github + return await search_github(query, language=language, limit=limit, config=self.config) + + async def search_twitter( + self, + query: str, + limit: int = 10, + ) -> List[Dict[str, Any]]: + """ + Search Twitter. Uses birdx if available, else Exa. + + Args: + query: Search query + limit: Number of results + """ + from agent_eyes.search.twitter import search_twitter + return await search_twitter(query, limit=limit, config=self.config) + + # ── Health ────────────────────────────────────────── + + def doctor(self) -> Dict[str, dict]: + """Check all platform/feature availability.""" + from agent_eyes.doctor import check_all + return check_all(self.config) + + def doctor_report(self) -> str: + """Get formatted health report.""" + from agent_eyes.doctor import check_all, format_report + return format_report(check_all(self.config)) + + # ── Sync wrappers ─────────────────────────────────── + + def read_sync(self, url: str) -> Dict[str, Any]: + """Synchronous version of read().""" + return asyncio.run(self.read(url)) + + def search_sync(self, query: str, num_results: int = 5) -> List[Dict[str, Any]]: + """Synchronous version of search().""" + return asyncio.run(self.search(query, num_results)) diff --git a/agent_eyes/doctor.py b/agent_eyes/doctor.py new file mode 100644 index 0000000..08c8742 --- /dev/null +++ b/agent_eyes/doctor.py @@ -0,0 +1,226 @@ +# -*- coding: utf-8 -*- +"""Environment health checker for Agent Eyes. + +Checks all platforms, tools, and API keys. +Outputs a rich-formatted status report. +""" + +import shutil +import subprocess +from typing import Dict + +from agent_eyes.config import Config + + +STATUS_OK = "ok" +STATUS_WARN = "warn" +STATUS_OFF = "off" +STATUS_ERROR = "error" + + +def _check_command(cmd: str) -> bool: + """Check if a command exists on PATH.""" + return shutil.which(cmd) is not None + + +def _check_python_import(module: str) -> bool: + """Check if a Python module is importable.""" + try: + __import__(module) + return True + except ImportError: + return False + + +def check_all(config: Config) -> Dict[str, dict]: + """Check all features and return status dict.""" + results = {} + + # === Zero-config (always available) === + results["web"] = { + "status": STATUS_OK, + "name": "Web Pages", + "message": "Jina Reader (built-in)", + "tier": 0, + } + results["github_read"] = { + "status": STATUS_OK, + "name": "GitHub", + "message": "Public API (built-in)", + "tier": 0, + } + results["bilibili"] = { + "status": STATUS_OK, + "name": "Bilibili", + "message": "Public API (built-in)", + "tier": 0, + } + results["rss"] = { + "status": STATUS_OK, + "name": "RSS", + "message": "feedparser (built-in)", + "tier": 0, + } + results["tweet_read"] = { + "status": STATUS_OK, + "name": "Tweet (single)", + "message": "Jina Reader (built-in)", + "tier": 0, + } + + # YouTube — needs yt-dlp + if _check_command("yt-dlp"): + results["youtube"] = { + "status": STATUS_OK, + "name": "YouTube", + "message": "yt-dlp found", + "tier": 0, + } + else: + results["youtube"] = { + "status": STATUS_WARN, + "name": "YouTube", + "message": "Install yt-dlp: pip install yt-dlp", + "tier": 0, + } + + # === Needs API key (Tier 1) === + exa_key = config.get("exa_api_key") + if exa_key: + results["search_web"] = { + "status": STATUS_OK, + "name": "Web Search", + "message": "Exa API configured", + "tier": 1, + } + results["search_reddit"] = { + "status": STATUS_OK, + "name": "Reddit Search", + "message": "Via Exa (site:reddit.com)", + "tier": 1, + } + results["search_twitter"] = { + "status": STATUS_OK, + "name": "Twitter Search", + "message": "Via Exa (site:x.com)", + "tier": 1, + } + else: + for key in ("search_web", "search_reddit", "search_twitter"): + results[key] = { + "status": STATUS_OFF, + "name": {"search_web": "Web Search", "search_reddit": "Reddit Search", + "search_twitter": "Twitter Search"}[key], + "message": "Need Exa API key (free). Run: agent-eyes setup", + "tier": 1, + } + + # GitHub search (always works, token optional for higher limits) + github_token = config.get("github_token") + results["search_github"] = { + "status": STATUS_OK, + "name": "GitHub Search", + "message": f"API ({'authenticated' if github_token else 'public, 60 req/hr'})", + "tier": 0, + } + + # === Optional tools (Tier 2) === + + # Twitter advanced (birdx) + if _check_command("birdx"): + results["twitter_advanced"] = { + "status": STATUS_OK, + "name": "Twitter Advanced", + "message": "birdx found", + "tier": 2, + } + else: + results["twitter_advanced"] = { + "status": STATUS_OFF, + "name": "Twitter Advanced", + "message": "Install birdx for timeline/deep search", + "tier": 2, + } + + # Reddit full reader + reddit_proxy = config.get("reddit_proxy") + if reddit_proxy: + results["reddit_full"] = { + "status": STATUS_OK, + "name": "Reddit Reader", + "message": "Proxy configured", + "tier": 2, + } + else: + results["reddit_full"] = { + "status": STATUS_OFF, + "name": "Reddit Reader", + "message": "Need proxy for full post reading", + "tier": 2, + } + + # WeChat / XHS (playwright) + if _check_python_import("playwright"): + results["wechat"] = { + "status": STATUS_OK, + "name": "WeChat", + "message": "Playwright available", + "tier": 2, + } + results["xhs"] = { + "status": STATUS_OK, + "name": "XiaoHongShu", + "message": "Playwright available", + "tier": 2, + } + else: + for key in ("wechat", "xhs"): + results[key] = { + "status": STATUS_OFF, + "name": "WeChat" if key == "wechat" else "XiaoHongShu", + "message": "pip install agent-eyes[browser]", + "tier": 2, + } + + return results + + +def format_report(results: Dict[str, dict]) -> str: + """Format results as a readable text report.""" + lines = [] + lines.append("👁️ Agent Eyes Status") + lines.append("=" * 40) + + # Count stats + ok_count = sum(1 for r in results.values() if r["status"] == STATUS_OK) + total = len(results) + + # Group by tier + lines.append("") + lines.append("✅ Ready (no setup needed):") + for key, r in results.items(): + if r["tier"] == 0 and r["status"] == STATUS_OK: + lines.append(f" ✅ {r['name']}") + elif r["tier"] == 0 and r["status"] == STATUS_WARN: + lines.append(f" ⚠️ {r['name']} — {r['message']}") + + lines.append("") + lines.append("🔍 Search (need free Exa API key):") + for key, r in results.items(): + if r["tier"] == 1: + icon = "✅" if r["status"] == STATUS_OK else "⬜" + lines.append(f" {icon} {r['name']}") + + lines.append("") + lines.append("🔧 Optional (advanced setup):") + for key, r in results.items(): + if r["tier"] == 2: + icon = "✅" if r["status"] == STATUS_OK else "⬜" + lines.append(f" {icon} {r['name']} — {r['message']}") + + lines.append("") + lines.append(f"Status: {ok_count}/{total} platforms active") + if ok_count < total: + lines.append("Run `agent-eyes setup` to unlock more!") + + return "\n".join(lines) diff --git a/agent_eyes/fetchers/search.py b/agent_eyes/fetchers/search.py deleted file mode 100644 index 0d43290..0000000 --- a/agent_eyes/fetchers/search.py +++ /dev/null @@ -1,94 +0,0 @@ -# -*- coding: utf-8 -*- -"""Search fetcher — semantic web search via Exa API. - -Requires EXA_API_KEY env var. Get a free key at https://exa.ai -""" - -import os -import requests -from loguru import logger -from typing import Dict, Any, List, Optional - - -EXA_API_URL = "https://api.exa.ai/search" - - -async def search_web( - query: str, - num_results: int = 5, - search_type: str = "auto", -) -> List[Dict[str, Any]]: - """ - Search the web using Exa semantic search. - - Args: - query: Search query (supports site: prefix, e.g. "site:reddit.com AI agent") - num_results: Number of results to return (default 5, max 10) - search_type: "auto" (default) or "neural" or "keyword" - - Returns: - List of search results with title, url, snippet - """ - api_key = os.environ.get("EXA_API_KEY") - if not api_key: - raise ValueError( - "EXA_API_KEY not set. Get a free key at https://exa.ai\n" - "Then: export EXA_API_KEY=your_key_here" - ) - - logger.info(f"Exa search: {query} (n={num_results})") - - resp = requests.post( - EXA_API_URL, - headers={ - "Content-Type": "application/json", - "x-api-key": api_key, - }, - json={ - "query": query, - "numResults": min(num_results, 10), - "type": search_type, - "contents": { - "text": {"maxCharacters": 500}, - }, - }, - timeout=15, - ) - resp.raise_for_status() - data = resp.json() - - results = [] - for item in data.get("results", []): - results.append({ - "title": item.get("title", ""), - "url": item.get("url", ""), - "snippet": item.get("text", ""), - "published_date": item.get("publishedDate", ""), - "score": item.get("score", 0), - }) - - return results - - -async def search_reddit_via_exa( - query: str, - subreddit: Optional[str] = None, - num_results: int = 10, -) -> List[Dict[str, Any]]: - """ - Search Reddit content via Exa (bypasses Reddit IP blocks). - - Args: - query: Search query - subreddit: Optional subreddit to limit search (e.g. "LocalLLaMA") - num_results: Number of results - - Returns: - List of Reddit posts found - """ - if subreddit: - full_query = f"site:reddit.com/r/{subreddit} {query}" - else: - full_query = f"site:reddit.com {query}" - - return await search_web(full_query, num_results=num_results) diff --git a/agent_eyes/guides/setup-exa.md b/agent_eyes/guides/setup-exa.md new file mode 100644 index 0000000..e8a4678 --- /dev/null +++ b/agent_eyes/guides/setup-exa.md @@ -0,0 +1,54 @@ +# Exa Search 配置指南 + +## 功能说明 +Exa 是一个语义搜索引擎，免费提供 1000 次/月搜索。配置后解锁： +- 全网搜索 +- Reddit 搜索（通过 site:reddit.com） +- Twitter 搜索（通过 site:x.com） + +## Agent 可自动完成的步骤 + +1. 检查是否已配置： +```bash +agent-eyes doctor | grep "Web Search" +``` + +2. 如果用户提供了 key，写入配置： +```bash +# 方法1: 通过 setup 命令 +agent-eyes setup + +# 方法2: 直接写入配置文件 +python3 -c " +from agent_eyes.config import Config +c = Config() +c.set('exa_api_key', '用户提供的KEY') +print('✅ Exa API key saved') +" +``` + +3. 测试是否有效： +```bash +agent-eyes search "test query" -n 1 +``` + +## 需要用户手动做的步骤 + +请告诉用户： + +> 我需要一个 Exa API Key 来开启搜索功能。完全免费，每月 1000 次搜索。 +> +> 步骤： +> 1. 打开 https://exa.ai +> 2. 点击 "Get API Key" 或 "Sign Up" +> 3. 用 Google 账号或邮箱注册 +> 4. 注册后在 Dashboard 页面复制你的 API Key +> 5. 把 Key 发给我 +> +> 大概 30 秒就能搞定。 + +## Agent 收到 key 后的操作 + +1. 写入配置：`config.set("exa_api_key", key)` +2. 测试：运行一次搜索确认可用 +3. 反馈："✅ 全网搜索已开启！现在我可以帮你搜索全网、Reddit 和 Twitter 了。" diff --git a/agent_eyes/guides/setup-groq.md b/agent_eyes/guides/setup-groq.md new file mode 100644 index 0000000..8bcc922 --- /dev/null +++ b/agent_eyes/guides/setup-groq.md @@ -0,0 +1,47 @@ +# Groq Whisper 配置指南 + +## 功能说明 +当 YouTube/Bilibili 视频没有字幕时，用 Groq 的 Whisper API 进行语音转文字。Groq 提供免费额度。 + +## Agent 可自动完成的步骤 + +1. 检查是否已配置： +```bash +agent-eyes doctor | grep -i "groq\|whisper" +``` + +2. 如果用户提供了 key，写入配置： +```python +from agent_eyes.config import Config +c = Config() +c.set("groq_api_key", "用户提供的KEY") +``` + +3. 测试（可选）： +```bash +curl -s https://api.groq.com/openai/v1/models \ + -H "Authorization: Bearer 用户提供的KEY" \ + -o /dev/null -w "%{http_code}" +``` +返回 200 = 可用 + +## 需要用户手动做的步骤 + +请告诉用户： + +> 视频语音转文字需要一个 Groq API Key（免费）。 +> +> 步骤： +> 1. 打开 https://console.groq.com +> 2. 用 Google 账号或邮箱注册 +> 3. 点击左侧 "API Keys" +> 4. 点击 "Create API Key" +> 5. 复制生成的 Key，发给我 +> +> Groq 提供免费额度，日常使用完全够用。 + +## Agent 收到 key 后的操作 + +1. 写入配置：`config.set("groq_api_key", key)` +2. 测试 API 可用性 +3. 反馈："✅ 语音转文字已开启！现在遇到没有字幕的视频，我也能帮你提取内容了。" diff --git a/agent_eyes/guides/setup-reddit.md b/agent_eyes/guides/setup-reddit.md new file mode 100644 index 0000000..f4ab38a --- /dev/null +++ b/agent_eyes/guides/setup-reddit.md @@ -0,0 +1,60 @@ +# Reddit 代理配置指南 + +## 功能说明 +Reddit 封锁了大量服务器 IP，直接访问会返回 403。需要 ISP 代理（Residential/ISP Proxy）才能读取完整的 Reddit 帖子和评论。 + +**注意**：即使没有代理，也可以通过 Exa 搜索找到 Reddit 内容（只需要配 Exa API key）。代理只是用于读取完整帖子+评论。 + +## Agent 可自动完成的步骤 + +1. 检查当前状态： +```bash +agent-eyes doctor | grep "Reddit" +``` + +2. 如果用户提供了代理，测试连通性： +```bash +curl -s --proxy "用户提供的代理" \ + -H "User-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36" \ + "https://www.reddit.com/r/test.json?limit=1" \ + -o /dev/null -w "%{http_code}" +``` +返回 200 = 可用，403 = 代理被封，其他 = 配置错误 + +3. 写入配置： +```python +from agent_eyes.config import Config +c = Config() +c.set("reddit_proxy", "http://用户名:密码@IP:端口") +``` + +## 需要用户手动做的步骤 + +请告诉用户： + +> 要完整阅读 Reddit 帖子和评论，需要一个 ISP 代理（约 $3-10/月）。 +> +> 推荐代理服务商（选一个就行）： +> 1. **Smartproxy** (https://smartproxy.com) — ISP 代理，按流量计费 +> 2. **Bright Data** (https://brightdata.com) — 大厂，ISP 代理 +> 3. **IPRoyal** (https://iproyal.com) — 便宜，适合入门 +> 4. **ProxyEmpire** (https://proxyempire.io) — 有 Reddit 专用代理 +> +> 购买时选择： +> - 类型：**ISP Proxy**（不要选 Datacenter，会被封） +> - 地区：**美国** +> - 协议：**HTTP** +> +> 购买后你会得到一个代理地址，格式类似： +> `http://用户名:密码@IP地址:端口号` +> +> 把这个地址发给我就行。 +> +> ⚠️ 如果不想花钱，可以跳过。我仍然可以通过搜索引擎找到 Reddit 上的内容，只是不能读完整的帖子和评论。 + +## Agent 收到代理后的操作 + +1. 测试代理：用 curl 测试 reddit.com 是否返回 200 +2. 如果成功，写入配置：`config.set("reddit_proxy", proxy_url)` +3. 反馈："✅ Reddit 完整阅读已开启！现在我可以读取 Reddit 帖子和所有评论了。" +4. 如果失败，告诉用户："❌ 这个代理无法访问 Reddit，请检查代理是否有效，或换一个试试。" diff --git a/agent_eyes/guides/setup-twitter.md b/agent_eyes/guides/setup-twitter.md new file mode 100644 index 0000000..c71bfe3 --- /dev/null +++ b/agent_eyes/guides/setup-twitter.md @@ -0,0 +1,69 @@ +# Twitter 高级功能配置指南（birdx） + +## 功能说明 +基础 Twitter 功能（搜索+读单条推文）无需配置，开箱即用。 + +高级功能需要 birdx： +- 查看用户时间线 +- 深度搜索（更精确、更多结果） +- 读取完整线程（thread） +- 查看关注列表推文 + +birdx 是免费开源工具，但需要你的 Twitter 账号 cookie。 + +## Agent 可自动完成的步骤 + +1. 检查 birdx 是否安装： +```bash +which birdx && echo "installed" || echo "not installed" +``` + +2. 安装 birdx： +```bash +pip install birdx +``` + +3. 检查是否已配置 cookie： +```bash +birdx whoami 2>&1 +``` + +4. 如果用户提供了 cookie，配置 birdx： +```bash +# birdx 的 cookie 配置文件位置 +# 通常在 ~/.birdx/cookies.json 或通过环境变量 +export TWITTER_AUTH_TOKEN="用户提供的auth_token" +export TWITTER_CT0="用户提供的ct0" +``` + +5. 测试： +```bash +birdx search "test" -n 1 +``` + +## 需要用户手动做的步骤 + +请告诉用户： + +> Twitter 高级功能需要你的 Twitter 账号 cookie（完全免费）。 +> +> 步骤： +> 1. 用 Chrome 打开 https://x.com 并确保你已登录 +> 2. 按 **F12** 打开开发者工具（Mac 按 Cmd+Option+I） +> 3. 点击顶部的 **Application**（应用）标签 +> 4. 左侧找到 **Cookies** → **https://x.com** +> 5. 在列表中找到以下两个值，双击复制： +> - **auth_token** — 一串字母数字 +> - **ct0** — 一串字母数字 +> 6. 把这两个值发给我 +> +> ⚠️ 这些 cookie 让我能以你的身份读取推文（只读）。我不会发推、点赞或做任何操作。 +> ⚠️ cookie 大约 1-3 个月会过期，届时需要重新导出。 + +## Agent 收到 cookie 后的操作 + +1. 安装 birdx（如果没装）：`pip install birdx` +2. 配置 cookie：写入 birdx 配置 +3. 测试：`birdx whoami` 确认身份 +4. 反馈："✅ Twitter 高级功能已开启！你的账号是 @xxx。现在我可以查看时间线、读取线程了。" +5. 如果失败："❌ Cookie 无效或已过期，请重新导出。" diff --git a/agent_eyes/guides/setup-wechat.md b/agent_eyes/guides/setup-wechat.md new file mode 100644 index 0000000..a536826 --- /dev/null +++ b/agent_eyes/guides/setup-wechat.md @@ -0,0 +1,40 @@ +# 微信公众号配置指南 + +## 功能说明 +读取微信公众号文章。需要 Playwright 来处理微信的反爬机制。 + +## Agent 可自动完成的步骤 + +1. 检查 Playwright 是否安装： +```bash +python3 -c "import playwright; print('installed')" 2>&1 +``` + +2. 安装 Playwright + 浏览器： +```bash +pip install playwright +playwright install chromium +``` + +3. 安装完成后测试： +```bash +agent-eyes read "https://mp.weixin.qq.com/s/一个测试链接" +``` + +## 需要用户手动做的步骤 + +请告诉用户： + +> 微信公众号的配置很简单，只需要安装一个浏览器组件（约 150MB）。 +> +> 我来帮你安装，你不需要做任何事情。安装过程大约 1-2 分钟。 +> +> 安装好之后就可以直接读取微信公众号文章了，不需要登录。 + +## Agent 操作流程 + +1. 安装 Playwright：`pip install playwright` +2. 安装 Chromium：`playwright install chromium` +3. 测试：读一篇微信文章 +4. 反馈："✅ 微信公众号已配置！发给我任何公众号文章链接，我都能读取。" +5. 如果安装失败（空间不足等）："❌ 浏览器组件安装失败。可能是磁盘空间不足（需要约 150MB）。" diff --git a/agent_eyes/guides/setup-xiaohongshu.md b/agent_eyes/guides/setup-xiaohongshu.md new file mode 100644 index 0000000..d8f7a0e --- /dev/null +++ b/agent_eyes/guides/setup-xiaohongshu.md @@ -0,0 +1,42 @@ +# 小红书配置指南 + +## 功能说明 +读取小红书笔记内容。需要 Playwright（浏览器自动化）和一次性登录。 + +## Agent 可自动完成的步骤 + +1. 检查 Playwright 是否安装： +```bash +python3 -c "import playwright; print('installed')" 2>&1 +``` + +2. 安装 Playwright + 浏览器： +```bash +pip install playwright +playwright install chromium +``` + +3. 检查是否已有登录态： +```bash +# 检查 cookie 文件是否存在 +ls ~/.agent-eyes/xhs_cookies.json 2>/dev/null +``` + +## 需要用户手动做的步骤 + +请告诉用户： + +> 小红书需要登录一次（之后会记住你的登录状态）。 +> +> 我现在会打开一个浏览器窗口，显示小红书登录页面。你需要： +> 1. 用手机小红书 App 扫描屏幕上的二维码 +> 2. 在手机上确认登录 +> 3. 看到首页后告诉我"登录好了" +> +> 之后就不需要再登录了（除非 cookie 过期，大约 1-3 个月）。 + +## Agent 收到确认后的操作 + +1. 保存浏览器 cookie 到 `~/.agent-eyes/xhs_cookies.json` +2. 测试：读取一条小红书笔记 +3. 反馈："✅ 小红书已配置！现在我可以读取小红书笔记了。" diff --git a/agent_eyes/integrations/__init__.py b/agent_eyes/integrations/__init__.py new file mode 100644 index 0000000..40a96af --- /dev/null +++ b/agent_eyes/integrations/__init__.py @@ -0,0 +1 @@ +# -*- coding: utf-8 -*- diff --git a/agent_eyes/integrations/mcp_server.py b/agent_eyes/integrations/mcp_server.py new file mode 100644 index 0000000..7542255 --- /dev/null +++ b/agent_eyes/integrations/mcp_server.py @@ -0,0 +1,198 @@ +# -*- coding: utf-8 -*- +""" +Agent Eyes MCP Server — expose all capabilities as MCP tools. + +Run: python -m agent_eyes.integrations.mcp_server +Or: agent-eyes serve (after pip install) + +10 tools for any MCP-compatible AI Agent. +""" + +import asyncio +import json +import sys + +from agent_eyes.config import Config +from agent_eyes.core import AgentEyes + +try: + from mcp.server import Server + from mcp.server.stdio import stdio_server + from mcp.types import Tool, TextContent + HAS_MCP = True +except ImportError: + HAS_MCP = False + + +def create_server(): + """Create and configure the MCP server.""" + if not HAS_MCP: + print("MCP not installed. Install: pip install agent-eyes[mcp]", file=sys.stderr) + sys.exit(1) + + server = Server("agent-eyes") + config = Config() + eyes = AgentEyes(config) + + @server.list_tools() + async def list_tools(): + return [ + Tool( + name="read_url", + description="Read content from any URL. Supports: web pages, GitHub, Reddit, Twitter, YouTube, Bilibili, WeChat, XiaoHongShu, RSS, Telegram.", + inputSchema={ + "type": "object", + "properties": { + "url": {"type": "string", "description": "URL to read"}, + }, + "required": ["url"], + }, + ), + Tool( + name="read_batch", + description="Read multiple URLs concurrently.", + inputSchema={ + "type": "object", + "properties": { + "urls": {"type": "array", "items": {"type": "string"}, "description": "List of URLs"}, + }, + "required": ["urls"], + }, + ), + Tool( + name="detect_platform", + description="Detect what platform a URL belongs to (github, reddit, twitter, youtube, etc).", + inputSchema={ + "type": "object", + "properties": { + "url": {"type": "string", "description": "URL to detect"}, + }, + "required": ["url"], + }, + ), + Tool( + name="search", + description="Semantic web search using Exa. Find any information on the internet.", + inputSchema={ + "type": "object", + "properties": { + "query": {"type": "string", "description": "Search query"}, + "num_results": {"type": "integer", "description": "Number of results (1-10)", "default": 5}, + }, + "required": ["query"], + }, + ), + Tool( + name="search_reddit", + description="Search Reddit posts and discussions. Works even when Reddit blocks your IP.", + inputSchema={ + "type": "object", + "properties": { + "query": {"type": "string", "description": "Search query"}, + "subreddit": {"type": "string", "description": "Optional subreddit filter (e.g. 'LocalLLaMA')"}, + "limit": {"type": "integer", "description": "Number of results", "default": 10}, + }, + "required": ["query"], + }, + ), + Tool( + name="search_github", + description="Search GitHub repositories by topic, keyword, or technology.", + inputSchema={ + "type": "object", + "properties": { + "query": {"type": "string", "description": "Search query"}, + "language": {"type": "string", "description": "Filter by language (e.g. 'python')"}, + "limit": {"type": "integer", "description": "Number of results", "default": 5}, + }, + "required": ["query"], + }, + ), + Tool( + name="search_twitter", + description="Search Twitter/X posts. Uses birdx if available, otherwise Exa.", + inputSchema={ + "type": "object", + "properties": { + "query": {"type": "string", "description": "Search query"}, + "limit": {"type": "integer", "description": "Number of results", "default": 10}, + }, + "required": ["query"], + }, + ), + Tool( + name="get_status", + description="Get Agent Eyes status: which platforms are active, which need configuration.", + inputSchema={ + "type": "object", + "properties": {}, + }, + ), + ] + + @server.call_tool() + async def call_tool(name: str, arguments: dict): + try: + if name == "read_url": + result = await eyes.read(arguments["url"]) + return [TextContent(type="text", text=json.dumps(result, ensure_ascii=False, indent=2))] + + elif name == "read_batch": + results = await eyes.read_batch(arguments["urls"]) + return [TextContent(type="text", text=json.dumps(results, ensure_ascii=False, indent=2))] + + elif name == "detect_platform": + platform = eyes.detect_platform(arguments["url"]) + return [TextContent(type="text", text=f"Platform: {platform}")] + + elif name == "search": + results = await eyes.search( + arguments["query"], + num_results=arguments.get("num_results", 5), + ) + return [TextContent(type="text", text=json.dumps(results, ensure_ascii=False, indent=2))] + + elif name == "search_reddit": + results = await eyes.search_reddit( + arguments["query"], + subreddit=arguments.get("subreddit"), + limit=arguments.get("limit", 10), + ) + return [TextContent(type="text", text=json.dumps(results, ensure_ascii=False, indent=2))] + + elif name == "search_github": + results = await eyes.search_github( + arguments["query"], + language=arguments.get("language"), + limit=arguments.get("limit", 5), + ) + return [TextContent(type="text", text=json.dumps(results, ensure_ascii=False, indent=2))] + + elif name == "search_twitter": + results = await eyes.search_twitter( + arguments["query"], + limit=arguments.get("limit", 10), + ) + return [TextContent(type="text", text=json.dumps(results, ensure_ascii=False, indent=2))] + + elif name == "get_status": + report = eyes.doctor_report() + return [TextContent(type="text", text=report)] + + else: + return [TextContent(type="text", text=f"Unknown tool: {name}")] + + except Exception as e: + return [TextContent(type="text", text=f"Error: {str(e)}")] + + return server + + +async def main(): + server = create_server() + async with stdio_server() as (read_stream, write_stream): + await server.run(read_stream, write_stream, server.create_initialization_options()) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/agent_eyes/integrations/skill/SKILL.md b/agent_eyes/integrations/skill/SKILL.md new file mode 100644 index 0000000..20219bd --- /dev/null +++ b/agent_eyes/integrations/skill/SKILL.md @@ -0,0 +1,65 @@ +--- +name: agent-eyes +description: "Give your AI Agent eyes to see the entire internet. Search + read 10+ platforms with one install." +install: pip install git+https://github.com/Panniantong/agent-eyes.git +--- + +# Agent Eyes — 给 Agent 装上眼睛 👁️ + +安装后你就能搜索和阅读全网内容。 + +## 安装 + +```bash +pip install git+https://github.com/Panniantong/agent-eyes.git +``` + +## 装完直接能用的（零配置） + +- `agent-eyes read ` — 读取任意网页、GitHub、B站、YouTube、RSS +- `agent-eyes search-github "query"` — 搜索 GitHub 仓库 +- `agent-eyes doctor` — 查看所有平台状态 + +## 解锁更多功能 + +运行 `agent-eyes setup` 进行交互式配置，或者读取以下指南： + +| 功能 | 指南文件 | +|------|---------| +| 全网搜索 + Reddit搜索 + Twitter搜索 | `agent_eyes/guides/setup-exa.md` | +| Reddit 完整阅读 | `agent_eyes/guides/setup-reddit.md` | +| Twitter 高级（时间线/线程） | `agent_eyes/guides/setup-twitter.md` | +| 小红书 | `agent_eyes/guides/setup-xiaohongshu.md` | +| 微信公众号 | `agent_eyes/guides/setup-wechat.md` | +| 视频语音转文字 | `agent_eyes/guides/setup-groq.md` | + +每个指南里有详细的步骤说明，你（Agent）可以照着做，只有需要人类操作的部分（登录、复制 key）才需要问用户。 + +## MCP Server + +如果你的 Agent 平台支持 MCP： + +```bash +pip install agent-eyes[mcp] +python -m agent_eyes.integrations.mcp_server +``` + +提供 8 个工具：read_url, read_batch, detect_platform, search, search_reddit, search_github, search_twitter, get_status + +## Python API + +```python +from agent_eyes import AgentEyes +import asyncio + +eyes = AgentEyes() + +# 读取 +result = asyncio.run(eyes.read("https://github.com/openai/gpt-4")) + +# 搜索 +results = asyncio.run(eyes.search("AI agent framework")) + +# 健康检查 +print(eyes.doctor_report()) +``` diff --git a/agent_eyes/login.py b/agent_eyes/login.py deleted file mode 100644 index 1d1d877..0000000 --- a/agent_eyes/login.py +++ /dev/null @@ -1,91 +0,0 @@ -# -*- coding: utf-8 -*- -""" -Login manager — opens a visible browser for manual login, saves session. - -Usage: - x-reader login xhs # Login to Xiaohongshu - x-reader login wechat # Login to WeChat (if needed) - -Sessions are saved as Playwright storage_state JSON files. -""" - -from pathlib import Path -from loguru import logger - -SESSION_DIR = Path.home() / ".x-reader" / "sessions" - -PLATFORM_URLS = { - "xhs": "https://www.xiaohongshu.com/explore", - "xiaohongshu": "https://www.xiaohongshu.com/explore", - "wechat": "https://mp.weixin.qq.com", - "twitter": "https://x.com/login", - "x": "https://x.com/login", -} - - -def login(platform: str) -> None: - """ - Open a visible browser for the user to log in manually. - After login, saves cookies/localStorage to a session file. - - Args: - platform: Platform key (e.g. 'xhs', 'wechat') - """ - try: - from playwright.sync_api import sync_playwright - except ImportError: - print( - "❌ Playwright is not installed. Run:\n" - ' pip install "x-reader[browser]"\n' - " playwright install chromium" - ) - return - - platform = platform.lower() - login_url = PLATFORM_URLS.get(platform) - if not login_url: - supported = ", ".join(sorted(PLATFORM_URLS.keys())) - print(f"❌ Unknown platform: {platform}") - print(f" Supported: {supported}") - return - - SESSION_DIR.mkdir(parents=True, exist_ok=True) - session_path = SESSION_DIR / f"{platform}.json" - # Normalize alias to canonical name - if platform in ("xhs", "xiaohongshu"): - canonical = "xhs" - elif platform in ("twitter", "x"): - canonical = "twitter" - else: - canonical = platform - session_path = SESSION_DIR / f"{canonical}.json" - - print(f"🌐 Opening {platform} login page: {login_url}") - print(" Please log in manually in the browser window.") - print(" When done, close the browser or press Ctrl+C.\n") - - with sync_playwright() as p: - browser = p.chromium.launch(headless=False) - context = browser.new_context( - user_agent="Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " - "AppleWebKit/537.36 (KHTML, like Gecko) " - "Chrome/120.0.0.0 Safari/537.36", - ) - page = context.new_page() - page.goto(login_url) - - try: - # Wait for user to log in — blocks until browser is closed - page.wait_for_event("close", timeout=300_000) # 5 min max - except KeyboardInterrupt: - pass - except Exception: - pass # Browser closed by user - - # Save session regardless of how we got here - context.storage_state(path=str(session_path)) - logger.info(f"Session saved: {session_path}") - print(f"\n✅ Session saved to {session_path}") - - context.close() - browser.close() diff --git a/agent_eyes/reader.py b/agent_eyes/reader.py index 430790d..813a432 100644 --- a/agent_eyes/reader.py +++ b/agent_eyes/reader.py @@ -15,7 +15,7 @@ from agent_eyes.schema import ( from_bilibili, from_twitter, from_wechat, from_xiaohongshu, from_youtube, from_rss, from_telegram, ) -from agent_eyes.fetchers.jina import fetch_via_jina +from agent_eyes.readers.jina import fetch_via_jina class UniversalReader: @@ -91,39 +91,39 @@ class UniversalReader: """Dispatch to platform-specific fetcher.""" if platform == "bilibili": - from agent_eyes.fetchers.bilibili import fetch_bilibili + from agent_eyes.readers.bilibili import fetch_bilibili data = await fetch_bilibili(url) return from_bilibili(data) if platform == "twitter": - from agent_eyes.fetchers.twitter import fetch_twitter + from agent_eyes.readers.twitter import fetch_twitter data = await fetch_twitter(url) return from_twitter(data) if platform == "wechat": - from agent_eyes.fetchers.wechat import fetch_wechat + from agent_eyes.readers.wechat import fetch_wechat data = await fetch_wechat(url) return from_wechat(data) if platform == "xhs": - from agent_eyes.fetchers.xhs import fetch_xhs + from agent_eyes.readers.xhs import fetch_xhs data = await fetch_xhs(url) return from_xiaohongshu(data) if platform == "youtube": - from agent_eyes.fetchers.youtube import fetch_youtube + from agent_eyes.readers.youtube import fetch_youtube data = await fetch_youtube(url) return from_youtube(data) if platform == "rss": - from agent_eyes.fetchers.rss import fetch_rss + from agent_eyes.readers.rss import fetch_rss articles = await fetch_rss(url, limit=1) if articles: return from_rss(articles[0]) raise ValueError(f"No articles found in RSS feed: {url}") if platform == "reddit": - from agent_eyes.fetchers.reddit import fetch_reddit + from agent_eyes.readers.reddit import fetch_reddit data = await fetch_reddit(url) return UnifiedContent( source_type=SourceType.REDDIT, @@ -137,7 +137,7 @@ class UniversalReader: ) if platform == "github": - from agent_eyes.fetchers.github import fetch_github + from agent_eyes.readers.github import fetch_github data = await fetch_github(url) return UnifiedContent( source_type=SourceType.GITHUB, @@ -151,7 +151,7 @@ class UniversalReader: ) if platform == "telegram": - from agent_eyes.fetchers.telegram import fetch_telegram + from agent_eyes.readers.telegram import fetch_telegram # Extract channel username from t.me URL path = urlparse(url).path.strip("/").split("/")[0] channel = path if path else url diff --git a/agent_eyes/fetchers/__init__.py b/agent_eyes/readers/__init__.py similarity index 100% rename from agent_eyes/fetchers/__init__.py rename to agent_eyes/readers/__init__.py diff --git a/agent_eyes/fetchers/bilibili.py b/agent_eyes/readers/bilibili.py similarity index 100% rename from agent_eyes/fetchers/bilibili.py rename to agent_eyes/readers/bilibili.py diff --git a/agent_eyes/fetchers/browser.py b/agent_eyes/readers/browser.py similarity index 100% rename from agent_eyes/fetchers/browser.py rename to agent_eyes/readers/browser.py diff --git a/agent_eyes/fetchers/github.py b/agent_eyes/readers/github.py similarity index 100% rename from agent_eyes/fetchers/github.py rename to agent_eyes/readers/github.py diff --git a/agent_eyes/fetchers/jina.py b/agent_eyes/readers/jina.py similarity index 100% rename from agent_eyes/fetchers/jina.py rename to agent_eyes/readers/jina.py diff --git a/agent_eyes/fetchers/reddit.py b/agent_eyes/readers/reddit.py similarity index 100% rename from agent_eyes/fetchers/reddit.py rename to agent_eyes/readers/reddit.py diff --git a/agent_eyes/fetchers/rss.py b/agent_eyes/readers/rss.py similarity index 100% rename from agent_eyes/fetchers/rss.py rename to agent_eyes/readers/rss.py diff --git a/agent_eyes/fetchers/telegram.py b/agent_eyes/readers/telegram.py similarity index 100% rename from agent_eyes/fetchers/telegram.py rename to agent_eyes/readers/telegram.py diff --git a/agent_eyes/fetchers/twitter.py b/agent_eyes/readers/twitter.py similarity index 100% rename from agent_eyes/fetchers/twitter.py rename to agent_eyes/readers/twitter.py diff --git a/agent_eyes/fetchers/wechat.py b/agent_eyes/readers/wechat.py similarity index 100% rename from agent_eyes/fetchers/wechat.py rename to agent_eyes/readers/wechat.py diff --git a/agent_eyes/fetchers/xhs.py b/agent_eyes/readers/xhs.py similarity index 100% rename from agent_eyes/fetchers/xhs.py rename to agent_eyes/readers/xhs.py diff --git a/agent_eyes/fetchers/youtube.py b/agent_eyes/readers/youtube.py similarity index 100% rename from agent_eyes/fetchers/youtube.py rename to agent_eyes/readers/youtube.py diff --git a/agent_eyes/search/__init__.py b/agent_eyes/search/__init__.py new file mode 100644 index 0000000..c7ccd8a --- /dev/null +++ b/agent_eyes/search/__init__.py @@ -0,0 +1,16 @@ +# -*- coding: utf-8 -*- +"""Search module init.""" + +from agent_eyes.search.exa import search_web +from agent_eyes.search.reddit import search_reddit +from agent_eyes.search.github import search_github, search_github_issues +from agent_eyes.search.twitter import search_twitter, get_user_tweets + +__all__ = [ + "search_web", + "search_reddit", + "search_github", + "search_github_issues", + "search_twitter", + "get_user_tweets", +] diff --git a/agent_eyes/search/exa.py b/agent_eyes/search/exa.py new file mode 100644 index 0000000..e41b7e2 --- /dev/null +++ b/agent_eyes/search/exa.py @@ -0,0 +1,79 @@ +# -*- coding: utf-8 -*- +"""Exa semantic web search. + +Get a free API key at https://exa.ai (1000 searches/month free). +""" + +import os +import requests +from loguru import logger +from typing import Any, Dict, List, Optional + + +EXA_API_URL = "https://api.exa.ai/search" + + +def _get_api_key(config=None) -> str: + """Get Exa API key from config or env.""" + if config: + key = config.get("exa_api_key") + if key: + return key + key = os.environ.get("EXA_API_KEY") + if key: + return key + raise ValueError( + "Exa API key not configured.\n" + "Get a free key at https://exa.ai (1000 searches/month free)\n" + "Then run: agent-eyes setup" + ) + + +async def search_web( + query: str, + num_results: int = 5, + search_type: str = "auto", + config=None, +) -> List[Dict[str, Any]]: + """ + Semantic web search via Exa. + + Args: + query: Search query (supports site: prefix) + num_results: Number of results (default 5, max 10) + search_type: "auto" / "neural" / "keyword" + config: Optional Config instance + + Returns: + List of {title, url, snippet, published_date, score} + """ + api_key = _get_api_key(config) + logger.info(f"Exa search: {query} (n={num_results})") + + resp = requests.post( + EXA_API_URL, + headers={ + "Content-Type": "application/json", + "x-api-key": api_key, + }, + json={ + "query": query, + "numResults": min(num_results, 10), + "type": search_type, + "contents": {"text": {"maxCharacters": 500}}, + }, + timeout=15, + ) + resp.raise_for_status() + data = resp.json() + + results = [] + for item in data.get("results", []): + results.append({ + "title": item.get("title", ""), + "url": item.get("url", ""), + "snippet": item.get("text", ""), + "published_date": item.get("publishedDate", ""), + "score": item.get("score", 0), + }) + return results diff --git a/agent_eyes/search/github.py b/agent_eyes/search/github.py new file mode 100644 index 0000000..2ed8204 --- /dev/null +++ b/agent_eyes/search/github.py @@ -0,0 +1,118 @@ +# -*- coding: utf-8 -*- +"""GitHub search via public API (no key required).""" + +import os +import requests +from loguru import logger +from typing import Any, Dict, List, Optional + + +GITHUB_API = "https://api.github.com" + + +def _get_headers(config=None) -> dict: + """Get GitHub API headers with optional auth token.""" + headers = {"Accept": "application/vnd.github+json"} + token = None + if config: + token = config.get("github_token") + if not token: + token = os.environ.get("GITHUB_TOKEN") + if token: + headers["Authorization"] = f"Bearer {token}" + return headers + + +async def search_github( + query: str, + language: Optional[str] = None, + sort: str = "stars", + limit: int = 5, + config=None, +) -> List[Dict[str, Any]]: + """ + Search GitHub repositories. + + Args: + query: Search query + language: Filter by language (e.g. "python") + sort: Sort by "stars" / "forks" / "updated" + limit: Number of results (max 30) + config: Optional Config instance + + Returns: + List of {name, url, description, stars, language, updated} + """ + q = query + if language: + q += f" language:{language}" + + logger.info(f"GitHub search: {q} (n={limit})") + + resp = requests.get( + f"{GITHUB_API}/search/repositories", + headers=_get_headers(config), + params={"q": q, "sort": sort, "per_page": min(limit, 30)}, + timeout=15, + ) + resp.raise_for_status() + data = resp.json() + + results = [] + for repo in data.get("items", []): + results.append({ + "name": repo.get("full_name", ""), + "url": repo.get("html_url", ""), + "description": repo.get("description", ""), + "stars": repo.get("stargazers_count", 0), + "forks": repo.get("forks_count", 0), + "language": repo.get("language", ""), + "updated": repo.get("updated_at", ""), + "topics": repo.get("topics", []), + }) + return results + + +async def search_github_issues( + query: str, + repo: Optional[str] = None, + state: str = "open", + limit: int = 5, + config=None, +) -> List[Dict[str, Any]]: + """ + Search GitHub issues and discussions. + + Args: + query: Search query + repo: Filter by repo (e.g. "owner/repo") + state: "open" / "closed" + limit: Number of results + config: Optional Config instance + """ + q = query + if repo: + q += f" repo:{repo}" + q += f" state:{state}" + + resp = requests.get( + f"{GITHUB_API}/search/issues", + headers=_get_headers(config), + params={"q": q, "sort": "reactions", "per_page": min(limit, 30)}, + timeout=15, + ) + resp.raise_for_status() + data = resp.json() + + results = [] + for issue in data.get("items", []): + results.append({ + "title": issue.get("title", ""), + "url": issue.get("html_url", ""), + "body": (issue.get("body", "") or "")[:500], + "state": issue.get("state", ""), + "comments": issue.get("comments", 0), + "reactions": issue.get("reactions", {}).get("total_count", 0), + "created": issue.get("created_at", ""), + }) + return results diff --git a/agent_eyes/search/reddit.py b/agent_eyes/search/reddit.py new file mode 100644 index 0000000..41e8a9f --- /dev/null +++ b/agent_eyes/search/reddit.py @@ -0,0 +1,30 @@ +# -*- coding: utf-8 -*- +"""Reddit search via Exa (bypasses Reddit IP blocks).""" + +from typing import Any, Dict, List, Optional +from agent_eyes.search.exa import search_web + + +async def search_reddit( + query: str, + subreddit: Optional[str] = None, + num_results: int = 10, + config=None, +) -> List[Dict[str, Any]]: + """ + Search Reddit content via Exa semantic search. + + Args: + query: Search query + subreddit: Optional subreddit (e.g. "LocalLLaMA") + num_results: Number of results + config: Optional Config instance + + Returns: + List of {title, url, snippet, published_date, score} + """ + if subreddit: + full_query = f"site:reddit.com/r/{subreddit} {query}" + else: + full_query = f"site:reddit.com {query}" + return await search_web(full_query, num_results=num_results, config=config) diff --git a/agent_eyes/search/twitter.py b/agent_eyes/search/twitter.py new file mode 100644 index 0000000..819acd7 --- /dev/null +++ b/agent_eyes/search/twitter.py @@ -0,0 +1,112 @@ +# -*- coding: utf-8 -*- +"""Twitter search — uses birdx if available, falls back to Exa.""" + +import json +import shutil +import subprocess +from loguru import logger +from typing import Any, Dict, List, Optional + + +async def search_twitter( + query: str, + limit: int = 10, + config=None, +) -> List[Dict[str, Any]]: + """ + Search Twitter/X content. + + Strategy: + 1. If birdx is installed → use it (full search, timeline, threads) + 2. Otherwise → use Exa with site:x.com (basic search) + + Args: + query: Search query + limit: Number of results + config: Optional Config instance + + Returns: + List of {author, text, url, likes, retweets, date} + """ + if shutil.which("birdx"): + return await _search_birdx(query, limit) + else: + return await _search_exa(query, limit, config) + + +async def _search_birdx(query: str, limit: int) -> List[Dict[str, Any]]: + """Search Twitter via birdx CLI.""" + logger.info(f"birdx search: {query} (n={limit})") + try: + result = subprocess.run( + ["birdx", "search", query, "-n", str(limit), "--json"], + capture_output=True, text=True, timeout=30, + ) + if result.returncode != 0: + # birdx might not support --json, try plain output + result = subprocess.run( + ["birdx", "search", query, "-n", str(limit)], + capture_output=True, text=True, timeout=30, + ) + return _parse_birdx_text(result.stdout) + + data = json.loads(result.stdout) + if isinstance(data, list): + return data + return data.get("tweets", data.get("results", [])) + except (subprocess.TimeoutExpired, json.JSONDecodeError, FileNotFoundError) as e: + logger.error(f"birdx search failed: {e}") + return [] + + +def _parse_birdx_text(text: str) -> List[Dict[str, Any]]: + """Parse birdx plain text output into structured data.""" + results = [] + current = {} + for line in text.strip().split("\n"): + line = line.strip() + if not line: + if current: + results.append(current) + current = {} + continue + if line.startswith("@"): + current["author"] = line.split()[0] if line else "" + elif line.startswith("http"): + current["url"] = line + else: + current["text"] = current.get("text", "") + " " + line + if current: + results.append(current) + return results + + +async def _search_exa(query: str, limit: int, config=None) -> List[Dict[str, Any]]: + """Search Twitter via Exa (site:x.com).""" + from agent_eyes.search.exa import search_web + return await search_web( + f"site:x.com {query}", + num_results=limit, + config=config, + ) + + +async def get_user_tweets( + username: str, + limit: int = 10, +) -> List[Dict[str, Any]]: + """Get recent tweets from a user (requires birdx).""" + if not shutil.which("birdx"): + raise RuntimeError( + "birdx not installed. Install: pip install birdx\n" + "Then configure cookies: agent-eyes setup" + ) + try: + result = subprocess.run( + ["birdx", "user-tweets", f"@{username.lstrip('@')}", "-n", str(limit)], + capture_output=True, text=True, timeout=30, + ) + return _parse_birdx_text(result.stdout) + except subprocess.TimeoutExpired: + logger.error("birdx timed out") + return [] diff --git a/mcp_server.py b/mcp_server.py deleted file mode 100644 index b7195da..0000000 --- a/mcp_server.py +++ /dev/null @@ -1,161 +0,0 @@ -# -*- coding: utf-8 -*- -""" -x-reader MCP Server — expose content reading as MCP tools. - -Usage: - python mcp_server.py # stdio transport (for Claude Code) - python mcp_server.py --transport sse # SSE transport (for web clients) - -Claude Code config (~/.claude/claude_desktop_config.json): - { - "mcpServers": { - "agent-eyes": { - "command": "python", - "args": ["/path/to/x-reader/mcp_server.py"] - } - } - } -""" - -import asyncio -from dotenv import load_dotenv -from mcp.server.fastmcp import FastMCP - -load_dotenv() - -from agent_eyes.reader import UniversalReader -from agent_eyes.schema import UnifiedInbox - -mcp = FastMCP( - "agent-eyes", - instructions="Give your AI Agent eyes to see the entire internet. Search, read, and extract content from any platform.", -) - -reader = UniversalReader(inbox=UnifiedInbox()) - - -@mcp.tool() -async def read_url(url: str) -> str: - """ - Read content from any URL and return structured result. - - Supports: YouTube, Bilibili, X/Twitter, WeChat, Xiaohongshu, - Telegram, RSS, and any generic web page. - - Returns JSON with: title, content, url, source_type, platform metadata. - """ - import json - - content = await reader.read(url) - result = content.to_dict() - # Keep it readable - return json.dumps(result, ensure_ascii=False, indent=2) - - -@mcp.tool() -async def read_batch(urls: list[str]) -> str: - """ - Read multiple URLs concurrently. Returns JSON array of results. - - Failed URLs are logged but don't block other results. - """ - import json - - contents = await reader.read_batch(urls) - results = [c.to_dict() for c in contents] - return json.dumps(results, ensure_ascii=False, indent=2) - - -@mcp.tool() -async def list_inbox() -> str: - """ - List all items in the content inbox. - - Returns JSON array of previously fetched content. - """ - import json - - items = [item.to_dict() for item in reader.inbox.items] - return json.dumps(items, ensure_ascii=False, indent=2) - - -@mcp.tool() -async def detect_platform(url: str) -> str: - """ - Detect which platform a URL belongs to. - - Returns the platform name: youtube, bilibili, twitter, wechat, - xhs, reddit, github, telegram, rss, or generic. - """ - return reader._detect_platform(url) - - -# ==================== Search Tools (NEW in Agent Eyes) ==================== - -@mcp.tool() -async def search(query: str, num_results: int = 5) -> str: - """ - Search the entire web using semantic search (powered by Exa). - - Great for finding articles, blog posts, discussions on any topic. - Supports site: prefix, e.g. "site:reddit.com AI agent" to limit to specific sites. - - Requires EXA_API_KEY env var. Get a free key at https://exa.ai - - Args: - query: Search query - num_results: Number of results (1-10, default 5) - """ - import json - from agent_eyes.fetchers.search import search_web - - results = await search_web(query, num_results=num_results) - return json.dumps(results, ensure_ascii=False, indent=2) - - -@mcp.tool() -async def search_reddit(query: str, subreddit: str = "", limit: int = 10) -> str: - """ - Search Reddit posts. Bypasses Reddit IP blocks via Exa. - - Args: - query: Search query - subreddit: Optional subreddit name (e.g. "LocalLLaMA"). Empty = all of Reddit. - limit: Number of results (default 10) - """ - import json - from agent_eyes.fetchers.search import search_reddit_via_exa - - sub = subreddit if subreddit else None - results = await search_reddit_via_exa(query, subreddit=sub, num_results=limit) - return json.dumps(results, ensure_ascii=False, indent=2) - - -@mcp.tool() -async def search_github(query: str, limit: int = 5) -> str: - """ - Search GitHub repositories by keyword. - - Returns repos sorted by stars. No API key needed for public repos. - - Args: - query: Search query (e.g. "LLM agent framework", "language:python RAG") - limit: Number of results (default 5) - """ - import json - from agent_eyes.fetchers.github import search_github as _search_gh - - results = await _search_gh(query, limit=limit) - return json.dumps(results, ensure_ascii=False, indent=2) - - -if __name__ == "__main__": - import sys - - transport = "stdio" - if "--transport" in sys.argv: - idx = sys.argv.index("--transport") - if idx + 1 < len(sys.argv): - transport = sys.argv[idx + 1] - - mcp.run(transport=transport) diff --git a/pyproject.toml b/pyproject.toml index 98d5f50..20e3cc9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,40 +1,52 @@ [project] name = "agent-eyes" version = "1.0.0" -description = "Give your AI Agent eyes to see the entire internet — search, read, and extract content from 10+ platforms" +description = "Give your AI Agent eyes to see the entire internet. Search + Read 10+ platforms." readme = "README.md" license = {text = "MIT"} requires-python = ">=3.10" -authors = [ - {name = "Neo Reid"}, - {name = "Leo (x-reader)", email = "runes.leo@gmail.com"}, +authors = [{name = "Neo Reid"}] +keywords = ["ai-agent", "web-reader", "search", "mcp", "agent-infrastructure", "llm-tools"] +classifiers = [ + "Development Status :: 4 - Beta", + "Intended Audience :: Developers", + "License :: OSI Approved :: MIT License", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Topic :: Internet :: WWW/HTTP", + "Topic :: Scientific/Engineering :: Artificial Intelligence", ] -keywords = ["ai-agent", "web-reader", "search", "reddit", "github", "content-extraction", "mcp", "agent-infrastructure"] + dependencies = [ "requests>=2.28", "feedparser>=6.0", "python-dotenv>=1.0", "loguru>=0.7", + "pyyaml>=6.0", + "rich>=13.0", ] +[project.optional-dependencies] +browser = ["playwright>=1.40"] +mcp = ["mcp[cli]>=1.0"] +all = ["playwright>=1.40", "mcp[cli]>=1.0"] + +[project.scripts] +agent-eyes = "agent_eyes.cli:main" + [project.urls] Homepage = "https://github.com/Panniantong/agent-eyes" Repository = "https://github.com/Panniantong/agent-eyes" Issues = "https://github.com/Panniantong/agent-eyes/issues" -[project.optional-dependencies] -telegram = ["telethon>=1.34"] -mcp = ["mcp[cli]>=1.0"] -browser = ["playwright>=1.40"] -all = ["telethon>=1.34", "mcp[cli]>=1.0", "playwright>=1.40"] - -[project.scripts] -agent-eyes = "agent_eyes.cli:main" -x-reader = "agent_eyes.cli:main" - [build-system] requires = ["hatchling"] build-backend = "hatchling.build" [tool.hatch.build.targets.wheel] packages = ["agent_eyes"] + +[tool.hatch.build.targets.wheel.force-include] +"agent_eyes/guides" = "agent_eyes/guides" diff --git a/skills/analyzer/skill.md b/skills/analyzer/skill.md deleted file mode 100644 index ed1953b..0000000 --- a/skills/analyzer/skill.md +++ /dev/null @@ -1,131 +0,0 @@ -# Content Analyzer Skill - -> Any content → structured analysis report with actionable insights - -## Trigger - -When user sends content (URL, text, or transcript) with analysis intent: -- `/analyze [URL]` -- "Analyze this article" -- "What are the key takeaways?" -- Auto-triggered after video/podcast transcription (from video skill) - -## Pipeline - -### Step 1: Get Content - -Choose tool based on input type: - -| Input | Tool | -|-------|------| -| Tweet URL | `fetch_tweet` or Jina Reader | -| Web URL | `WebFetch` or Jina Reader | -| Local file | Read file directly | -| Transcript from video skill | Use directly | - -### Step 2: Multi-Dimensional Analysis - -Scan content across these dimensions. Only output dimensions with actual content — skip empty ones. - -```markdown -## 📖 Summary - -[1-3 sentence core thesis] - -**Source**: [author/publisher] · [date] -**Type**: [tweet/article/video/podcast/report] - ---- - -## 💡 Key Insights - -### 🎯 Core Arguments -- **Thesis**: [Main argument or finding] -- **Evidence**: [Supporting data or reasoning] -- **Strength**: [How convincing? What's missing?] - -### 🤖 Tools & Methods -- **What**: [Tools, frameworks, or techniques mentioned] -- **How**: [How they're used or applied] -- **Relevance**: [Could you use this?] - -### ⚙️ Workflow Ideas -- **Optimization**: [Process improvements mentioned] -- **Automation**: [What could be automated] -- **Integration**: [How to fit into existing workflow] - -### 📊 Data & Numbers -- **Key metrics**: [Important numbers mentioned] -- **Trends**: [Patterns in the data] -- **Gaps**: [What data is missing] - -### ⚠️ Risks & Warnings -- **Pitfalls**: [Explicitly mentioned risks] -- **Blind spots**: [What the author might be missing] -- **Counter-arguments**: [Alternative perspectives] - -### 🔗 Resources -- **Tools/APIs**: [Mentioned tools or data sources] -- **People**: [Worth following or referencing] -- **Further reading**: [Related content] - -### 🧠 Mental Model Shifts -- **Before**: [Common assumption] -- **After**: [New understanding from this content] -- **Impact**: [How this changes decisions] - ---- - -## ✅ Action Items - -### Quick Wins (< 30 min) -- [ ] [Action 1] — Impact: ★★★★ | Effort: Easy -- [ ] [Action 2] — Impact: ★★★ | Effort: Easy - -### Deeper Work (1-3 hours) -- [ ] [Action 3] — Impact: ★★★ | Effort: Medium -- [ ] [Action 4] — Impact: ★★ | Effort: Medium - -### Exploration (needs validation) -- [ ] [Action 5] — Impact: ★★★ | Effort: Hard | Nature: Exploratory -``` - -### Step 3: Personalized Relevance (Customizable) - -Map insights to YOUR context. Edit the dimensions below to match your own projects, interests, and systems. - -```markdown -## 🔄 How This Applies to Me - -### My Projects -- **[Project A]**: [How this insight connects] -- **[Project B]**: [What I could apply] - -### My Knowledge Base -- **Update**: [Which notes/docs to update] -- **New entry**: [What to add to my knowledge system] - -### My Decision Log -- **Changed my mind about**: [what and why] -- **Confirmed my belief that**: [what] -``` - -> **Customization**: Edit the dimensions in Step 2 and Step 3 to match your own -> domain. A trader might add "Market Impact" and "Risk Assessment". A developer -> might add "Architecture Patterns" and "Tech Debt". Make it yours. - -## Output Modes - -| Mode | Trigger | Output | -|------|---------|--------| -| **Full** (default) | `/analyze [URL]` | All dimensions | -| **Sparse** | `/analyze [URL] --sparse` | Only hit dimensions, skip empty | -| **Brief** | `/analyze [URL] --brief` | Action items only | - -## Best Practices - -1. **Scan all dimensions, but don't force-fill** — skip empty dimensions cleanly -2. **Actions must be specific** — not "learn about X" but "read X docs chapter Y" -3. **Distinguish fact from opinion** — mark the author's claims vs verified facts -4. **Source everything** — tag where each insight comes from in the original content -5. **ROI awareness** — not every action is worth doing, assess effort vs impact diff --git a/skills/video/skill.md b/skills/video/skill.md deleted file mode 100644 index 9729802..0000000 --- a/skills/video/skill.md +++ /dev/null @@ -1,292 +0,0 @@ -# Video & Podcast Digest Skill - -> Send a video/podcast link → get full transcript + structured summary - -## Supported Platforms - -| Platform | Type | Subtitles | Whisper Transcription | -|----------|------|-----------|----------------------| -| YouTube | Video | ✅ | ✅ | -| Bilibili | Video | ✅ | ✅ | -| X/Twitter | Video | ❌ | ✅ | -| Xiaoyuzhou (小宇宙) | Podcast | ❌ | ✅ | -| Apple Podcasts | Podcast | ❌ | ✅ | -| Direct links (mp3/mp4/m3u8) | Any | ❌ | ✅ | - -## Trigger - -Auto-triggered when a media URL is detected: -- YouTube: `youtube.com`, `youtu.be` -- Bilibili: `bilibili.com`, `b23.tv` -- X/Twitter: `x.com`, `twitter.com` (tweets with video) -- Xiaoyuzhou: `xiaoyuzhoufm.com` -- Apple Podcasts: `podcasts.apple.com` -- Direct: `.mp3`, `.mp4`, `.m3u8`, `.m4a`, `.webm` - -## Pipeline - -### Step 0: Detect Media Type - -| URL Pattern | Type | Pipeline | -|-------------|------|----------| -| `xiaoyuzhoufm.com/episode/` | Podcast | → Step 1b (Xiaoyuzhou) | -| `podcasts.apple.com` | Podcast | → Step 1c (Apple) | -| `bilibili.com`, `b23.tv` | Video | → Step 1d (Bilibili API) | -| `.mp3`, `.m4a` direct link | Audio | → Step 2b (direct download) | -| Other | Video | → Step 1a (subtitle extraction) | - -### Step 1a: Video — Extract Subtitles - -```bash -# Clean up temp files -rm -f /tmp/media_sub*.vtt /tmp/media_audio.mp3 /tmp/media_transcript*.json /tmp/media_segment_*.mp3 2>/dev/null || true - -# YouTube (prefer English, fallback Chinese) -yt-dlp --skip-download --write-auto-sub --sub-lang "en,zh-Hans" -o "/tmp/media_sub" "VIDEO_URL" - -# Bilibili -yt-dlp --skip-download --write-auto-sub --sub-lang "zh-Hans,zh" -o "/tmp/media_sub" "VIDEO_URL" -``` - -Check for subtitles: -```bash -ls /tmp/media_sub*.vtt 2>/dev/null -``` -- **Has subtitles** → Read VTT content, skip to Step 3 -- **No subtitles** → Step 2a (download audio) - -### Step 1b: Xiaoyuzhou (小宇宙) — Extract Audio URL - -```bash -# Extract CDN direct link from __NEXT_DATA__ -# Xiaoyuzhou is a Next.js SPA, but initial HTML contains audio URL in __NEXT_DATA__ -AUDIO_URL=$(curl -sL -H "User-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36" \ - "EPISODE_URL" \ - | grep -oE 'https://media\.xyzcdn\.net/[^"]+\.(m4a|mp3)' \ - | head -1) - -echo "Audio URL: $AUDIO_URL" - -# Download audio -curl -L -o /tmp/media_audio.mp3 "$AUDIO_URL" -``` - -> If curl extraction is empty (rare), fallback: use Puppeteer/browser to get rendered page and extract. - -→ Step 2b (check size & transcribe) - -### Step 1c: Apple Podcasts — via yt-dlp - -```bash -yt-dlp -f "ba[ext=m4a]/ba/b" --extract-audio --audio-format mp3 --audio-quality 5 \ - -o "/tmp/media_audio.%(ext)s" "APPLE_PODCAST_URL" -``` - -→ Step 2b (check size & transcribe) - -### Step 1d: Bilibili — API Direct Audio Stream - -yt-dlp returns 412 for Bilibili even with cookies. Use Bilibili's API instead: - -```bash -# 1. Extract BV number from URL -BV="BV1xxxxx" # Replace with actual BV number - -# 2. Get video info (title, duration, CID) -curl -s "https://api.bilibili.com/x/web-interface/view?bvid=$BV" \ - -H "User-Agent: Mozilla/5.0" -H "Referer: https://www.bilibili.com/" \ - | python3 -c "import json,sys; d=json.load(sys.stdin)['data']; print(f\"Title: {d['title']}\nDuration: {d['duration']}s\nCID: {d['cid']}\")" - -# 3. Get audio stream URL -CID= -AUDIO_URL=$(curl -s "https://api.bilibili.com/x/player/playurl?bvid=$BV&cid=$CID&fnval=16&qn=64" \ - -H "User-Agent: Mozilla/5.0" -H "Referer: https://www.bilibili.com/" \ - | python3 -c "import json,sys; print(json.load(sys.stdin)['data']['dash']['audio'][0]['baseUrl'])") - -# 4. Download audio (Referer header required, otherwise 403) -curl -L -o /tmp/media_audio.m4s \ - -H "User-Agent: Mozilla/5.0" -H "Referer: https://www.bilibili.com/" "$AUDIO_URL" - -# 5. Convert to mp3 -ffmpeg -y -i /tmp/media_audio.m4s -acodec libmp3lame -q:a 5 /tmp/media_audio.mp3 -``` - -→ Step 2b (check size & transcribe) - -### Step 2a: Video — Download Audio (when no subtitles) - -```bash -# YouTube may need --cookies-from-browser chrome to bypass bot detection -yt-dlp --cookies-from-browser chrome -f "ba[ext=m4a]/ba/b" --extract-audio --audio-format mp3 --audio-quality 5 \ - -o "/tmp/media_audio.%(ext)s" "VIDEO_URL" -``` - -### Step 2b: Check Audio Size & Segment - -```bash -FILE_SIZE=$(stat -f%z /tmp/media_audio.* 2>/dev/null || stat -c%s /tmp/media_audio.* 2>/dev/null) -echo "File size: $FILE_SIZE bytes" -``` - -- **≤ 25MB (25000000)** → Step 2c (transcribe directly) -- **> 25MB** → Split first, then transcribe each segment - -**Splitting large audio (>25MB)**: -```bash -# Get total duration -DURATION=$(ffprobe -v error -show_entries format=duration -of csv=p=0 /tmp/media_audio.* | head -1) - -# Split into 10-minute segments (keeps each under 25MB) -SEGMENT_SEC=600 -SEGMENTS=$(python3 -c "import math; print(math.ceil(float('$DURATION')/$SEGMENT_SEC))") - -# Cut segments -for i in $(seq 0 $((SEGMENTS-1))); do - START=$((i * SEGMENT_SEC)) - ffmpeg -y -i /tmp/media_audio.* -ss $START -t $SEGMENT_SEC -acodec libmp3lame -q:a 5 \ - "/tmp/media_segment_${i}.mp3" 2>/dev/null -done -``` - -→ Call Step 2c for each segment **sequentially** (parallel triggers Groq 524 timeout), concatenate results - -### Step 2c: Whisper Transcription - -**Prerequisite**: `GROQ_API_KEY` environment variable - -```bash -# Check API key -if [ -z "$GROQ_API_KEY" ]; then - echo "❌ GROQ_API_KEY not set. Get one at: https://console.groq.com/keys" - exit 1 -fi - -# Transcribe single file (replace AUDIO_FILE with actual path) -curl -s -X POST "https://api.groq.com/openai/v1/audio/transcriptions" \ - -H "Authorization: Bearer $GROQ_API_KEY" \ - -H "Content-Type: multipart/form-data" \ - -F "file=@AUDIO_FILE" \ - -F "model=whisper-large-v3-turbo" \ - -F "response_format=verbose_json" \ - -F "language=zh" \ - > /tmp/media_transcript.json - -# Extract plain text -python3 -c "import json; print(json.load(open('/tmp/media_transcript.json'))['text'])" -``` - -**Whisper model options**: -| Model | Speed | Accuracy | Use Case | -|-------|-------|----------|----------| -| `whisper-large-v3-turbo` | 10x realtime | High | Default choice | -| `whisper-large-v3` | 5x realtime | Highest | Professional/noisy content | - -**Language parameter**: -- Chinese: `language=zh` -- English: `language=en` -- Auto-detect: omit language parameter - -### Step 3: Structured Summary - -Choose output format based on media type: - -**Video (≤20 min)**: -1. **Overview** (1-2 sentences) -2. **Key Points** (3-5 bullet points) -3. **Notable Quotes** (if any) -4. **Action Items** (if applicable) - -**Podcast (>20 min)**: -1. **Overview** (2-3 sentences: who discussed what) -2. **Chapter Summary** (segmented by topic, 2-3 sentences each) -3. **Key Points** (5-8 bullet points) -4. **Notable Quotes** -5. **Action Items** (if applicable) - -## Output Format - -### Video - -``` -## 📺 Video Digest - -**Title**: [Video Title] -**Duration**: [x minutes] -**Language**: [Chinese/English] - -### Overview -[1-2 sentence summary] - -### Key Points -1. [Point 1] -2. [Point 2] -... - -### Notable Quotes -> "xxx" — [timestamp] - -### Action Items -- [if applicable] -``` - -### Podcast - -``` -## 🎙️ Podcast Digest - -**Show**: [Podcast Name] -**Episode**: [Episode Title] -**Duration**: [x minutes] -**Guests**: [if any] - -### Overview -[2-3 sentences: who discussed what, core conclusions] - -### Chapter Summary -#### 1. [Topic] (~xx:xx-xx:xx) -[2-3 sentences of core content] - -#### 2. [Topic] (~xx:xx-xx:xx) -[2-3 sentences of core content] -... - -### Key Points -1. [Point 1] -2. [Point 2] -... - -### Notable Quotes -> "xxx" - -### Action Items -- [if applicable] -``` - -## Error Handling - -| Situation | Action | -|-----------|--------| -| No subtitles + no GROQ_API_KEY | Prompt user to set API key | -| No subtitles + has API key | Auto Whisper transcription | -| Xiaoyuzhou curl extraction empty | Use Puppeteer/browser to get rendered HTML | -| Audio >25MB | ffmpeg segment (10min/segment), transcribe sequentially | -| Podcast >2 hours | Warn user about duration, confirm before proceeding | -| Groq 524 timeout | Do NOT parallelize — transcribe sequentially, sleep 5-8s between segments | -| Groq 429 rate limit | 7200s/hour limit, wait for retry-after header, then retry | -| yt-dlp Bilibili 412 | Use Bilibili API instead (Step 1d) | -| yt-dlp YouTube bot detection | Add `--cookies-from-browser chrome` | -| Network timeout | Retry once | -| Spotify links | Inform user: not supported (DRM protected) | - -## Groq Whisper Limits - -- Max 25MB per request -- Free tier: 7200 seconds of audio/hour (rolling window), ~20 hours/day -- Supported formats: mp3, mp4, mpeg, mpga, m4a, wav, webm - -## Dependencies - -- `yt-dlp`: video download + subtitle extraction -- `ffmpeg`: audio conversion + segmentation -- `curl`: Xiaoyuzhou audio download, Bilibili API -- `GROQ_API_KEY`: Whisper transcription API (free at https://console.groq.com/keys) diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..c1b3d87 --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,29 @@ +# -*- coding: utf-8 -*- +"""Tests for Agent Eyes CLI.""" + +import pytest +from unittest.mock import patch +from agent_eyes.cli import main + + +class TestCLI: + def test_version(self, capsys): + with pytest.raises(SystemExit) as exc_info: + with patch("sys.argv", ["agent-eyes", "version"]): + main() + assert exc_info.value.code == 0 + captured = capsys.readouterr() + assert "Agent Eyes v" in captured.out + + def test_no_command_shows_help(self, capsys): + with pytest.raises(SystemExit) as exc_info: + with patch("sys.argv", ["agent-eyes"]): + main() + assert exc_info.value.code == 0 + + def test_doctor_runs(self, capsys): + with patch("sys.argv", ["agent-eyes", "doctor"]): + main() + captured = capsys.readouterr() + assert "Agent Eyes" in captured.out + assert "✅" in captured.out diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 0000000..133628c --- /dev/null +++ b/tests/test_config.py @@ -0,0 +1,80 @@ +# -*- coding: utf-8 -*- +"""Tests for Agent Eyes config module.""" + +import os +import tempfile +from pathlib import Path + +import pytest +import yaml + +from agent_eyes.config import Config + + +@pytest.fixture +def tmp_config(tmp_path): + """Create a Config with a temporary directory.""" + config_file = tmp_path / "config.yaml" + return Config(config_path=config_file) + + +class TestConfig: + def test_init_creates_dir(self, tmp_path): + config_file = tmp_path / "subdir" / "config.yaml" + config = Config(config_path=config_file) + assert config_file.parent.exists() + + def test_set_and_get(self, tmp_config): + tmp_config.set("test_key", "test_value") + assert tmp_config.get("test_key") == "test_value" + + def test_get_default(self, tmp_config): + assert tmp_config.get("nonexistent") is None + assert tmp_config.get("nonexistent", "default") == "default" + + def test_get_from_env(self, tmp_config, monkeypatch): + monkeypatch.setenv("TEST_ENV_KEY", "env_value") + assert tmp_config.get("test_env_key") == "env_value" + + def test_config_file_priority_over_env(self, tmp_config, monkeypatch): + monkeypatch.setenv("MY_KEY", "from_env") + tmp_config.set("my_key", "from_config") + assert tmp_config.get("my_key") == "from_config" + + def test_save_and_load(self, tmp_config): + tmp_config.set("key1", "value1") + tmp_config.set("key2", 42) + + # Create new config from same file + config2 = Config(config_path=tmp_config.config_path) + assert config2.get("key1") == "value1" + assert config2.get("key2") == 42 + + def test_delete(self, tmp_config): + tmp_config.set("to_delete", "value") + assert tmp_config.get("to_delete") == "value" + tmp_config.delete("to_delete") + assert tmp_config.get("to_delete") is None + + def test_is_configured(self, tmp_config): + assert not tmp_config.is_configured("exa_search") + tmp_config.set("exa_api_key", "test-key") + assert tmp_config.is_configured("exa_search") + + def test_is_configured_reddit(self, tmp_config): + assert not tmp_config.is_configured("reddit_proxy") + tmp_config.set("reddit_proxy", "http://user:pass@ip:port") + assert tmp_config.is_configured("reddit_proxy") + + def test_get_configured_features(self, tmp_config): + features = tmp_config.get_configured_features() + assert isinstance(features, dict) + assert "exa_search" in features + assert all(v is False for v in features.values()) + + def test_to_dict_masks_sensitive(self, tmp_config): + tmp_config.set("exa_api_key", "super-secret-key-12345") + tmp_config.set("normal_setting", "visible") + masked = tmp_config.to_dict() + assert masked["exa_api_key"] == "super-se..." + assert masked["normal_setting"] == "visible" diff --git a/tests/test_core.py b/tests/test_core.py new file mode 100644 index 0000000..99e4bff --- /dev/null +++ b/tests/test_core.py @@ -0,0 +1,41 @@ +# -*- coding: utf-8 -*- +"""Tests for AgentEyes core class.""" + +import pytest +from unittest.mock import patch, MagicMock, AsyncMock +from pathlib import Path + +from agent_eyes.config import Config +from agent_eyes.core import AgentEyes + + +@pytest.fixture +def eyes(tmp_path): + config = Config(config_path=tmp_path / "config.yaml") + return AgentEyes(config=config) + + +class TestAgentEyes: + def test_init(self, eyes): + assert eyes.config is not None + assert eyes.reader is not None + + def test_detect_platform(self, eyes): + assert eyes.detect_platform("https://github.com/openai/gpt-4") == "github" + assert eyes.detect_platform("https://reddit.com/r/test") == "reddit" + assert eyes.detect_platform("https://x.com/elonmusk/status/123") == "twitter" + assert eyes.detect_platform("https://youtube.com/watch?v=abc") == "youtube" + assert eyes.detect_platform("https://bilibili.com/video/BV1xx") == "bilibili" + assert eyes.detect_platform("https://mp.weixin.qq.com/s/abc") == "wechat" + assert eyes.detect_platform("https://example.com") == "generic" + + def test_doctor(self, eyes): + results = eyes.doctor() + assert isinstance(results, dict) + assert "web" in results + assert "github_read" in results + + def test_doctor_report(self, eyes): + report = eyes.doctor_report() + assert isinstance(report, str) + assert "Agent Eyes" in report diff --git a/tests/test_doctor.py b/tests/test_doctor.py new file mode 100644 index 0000000..8a1aa8b --- /dev/null +++ b/tests/test_doctor.py @@ -0,0 +1,67 @@ +# -*- coding: utf-8 -*- +"""Tests for Agent Eyes doctor module.""" + +import pytest +from unittest.mock import patch, MagicMock + +from agent_eyes.config import Config +from agent_eyes.doctor import check_all, format_report, STATUS_OK, STATUS_OFF + + +@pytest.fixture +def tmp_config(tmp_path): + return Config(config_path=tmp_path / "config.yaml") + + +class TestDoctor: + def test_zero_config_platforms_always_ok(self, tmp_config): + results = check_all(tmp_config) + assert results["web"]["status"] == STATUS_OK + assert results["github_read"]["status"] == STATUS_OK + assert results["bilibili"]["status"] == STATUS_OK + assert results["rss"]["status"] == STATUS_OK + assert results["tweet_read"]["status"] == STATUS_OK + + def test_search_off_without_exa_key(self, tmp_config): + results = check_all(tmp_config) + assert results["search_web"]["status"] == STATUS_OFF + assert results["search_reddit"]["status"] == STATUS_OFF + assert results["search_twitter"]["status"] == STATUS_OFF + + def test_search_on_with_exa_key(self, tmp_config): + tmp_config.set("exa_api_key", "test-key") + results = check_all(tmp_config) + assert results["search_web"]["status"] == STATUS_OK + assert results["search_reddit"]["status"] == STATUS_OK + assert results["search_twitter"]["status"] == STATUS_OK + + def test_github_search_always_on(self, tmp_config): + results = check_all(tmp_config) + assert results["search_github"]["status"] == STATUS_OK + + def test_reddit_full_off_without_proxy(self, tmp_config): + results = check_all(tmp_config) + assert results["reddit_full"]["status"] == STATUS_OFF + + def test_reddit_full_on_with_proxy(self, tmp_config): + tmp_config.set("reddit_proxy", "http://user:pass@ip:port") + results = check_all(tmp_config) + assert results["reddit_full"]["status"] == STATUS_OK + + @patch("agent_eyes.doctor._check_command") + def test_birdx_detection(self, mock_cmd, tmp_config): + mock_cmd.return_value = True + results = check_all(tmp_config) + assert results["twitter_advanced"]["status"] == STATUS_OK + + def test_format_report_is_string(self, tmp_config): + results = check_all(tmp_config) + report = format_report(results) + assert isinstance(report, str) + assert "Agent Eyes" in report + assert "✅" in report + + def test_format_report_shows_count(self, tmp_config): + results = check_all(tmp_config) + report = format_report(results) + assert "platforms active" in report diff --git a/tests/test_search.py b/tests/test_search.py new file mode 100644 index 0000000..7a74c89 --- /dev/null +++ b/tests/test_search.py @@ -0,0 +1,142 @@ +# -*- coding: utf-8 -*- +"""Tests for Agent Eyes search modules.""" + +import pytest +from unittest.mock import patch, MagicMock + +from agent_eyes.config import Config + + +@pytest.fixture +def tmp_config(tmp_path): + c = Config(config_path=tmp_path / "config.yaml") + c.set("exa_api_key", "test-key") + return c + + +class TestExaSearch: + @patch("agent_eyes.search.exa.requests.post") + @pytest.mark.asyncio + async def test_search_web(self, mock_post, tmp_config): + mock_resp = MagicMock() + mock_resp.json.return_value = { + "results": [ + { + "title": "Test Result", + "url": "https://example.com", + "text": "This is a test snippet", + "publishedDate": "2025-01-01", + "score": 0.95, + } + ] + } + mock_resp.raise_for_status = MagicMock() + mock_post.return_value = mock_resp + + from agent_eyes.search.exa import search_web + results = await search_web("test query", config=tmp_config) + + assert len(results) == 1 + assert results[0]["title"] == "Test Result" + assert results[0]["url"] == "https://example.com" + assert results[0]["snippet"] == "This is a test snippet" + + def test_search_web_requires_key(self): + from agent_eyes.search.exa import _get_api_key + with pytest.raises(ValueError, match="Exa API key"): + _get_api_key(Config(config_path="/tmp/nonexistent/config.yaml")) + + +class TestRedditSearch: + @patch("agent_eyes.search.exa.requests.post") + @pytest.mark.asyncio + async def test_search_reddit(self, mock_post, tmp_config): + mock_resp = MagicMock() + mock_resp.json.return_value = {"results": []} + mock_resp.raise_for_status = MagicMock() + mock_post.return_value = mock_resp + + from agent_eyes.search.reddit import search_reddit + results = await search_reddit("test", config=tmp_config) + + # Verify it searched site:reddit.com + call_args = mock_post.call_args + query = call_args[1]["json"]["query"] + assert "site:reddit.com" in query + + @patch("agent_eyes.search.exa.requests.post") + @pytest.mark.asyncio + async def test_search_reddit_with_sub(self, mock_post, tmp_config): + mock_resp = MagicMock() + mock_resp.json.return_value = {"results": []} + mock_resp.raise_for_status = MagicMock() + mock_post.return_value = mock_resp + + from agent_eyes.search.reddit import search_reddit + await search_reddit("test", subreddit="LocalLLaMA", config=tmp_config) + + call_args = mock_post.call_args + query = call_args[1]["json"]["query"] + assert "site:reddit.com/r/LocalLLaMA" in query + + +class TestGitHubSearch: + @patch("agent_eyes.search.github.requests.get") + @pytest.mark.asyncio + async def test_search_github(self, mock_get): + mock_resp = MagicMock() + mock_resp.json.return_value = { + "items": [ + { + "full_name": "owner/repo", + "html_url": "https://github.com/owner/repo", + "description": "A test repo", + "stargazers_count": 100, + "forks_count": 20, + "language": "Python", + "updated_at": "2025-01-01", + "topics": ["ai"], + } + ] + } + mock_resp.raise_for_status = MagicMock() + mock_get.return_value = mock_resp + + from agent_eyes.search.github import search_github + results = await search_github("test") + + assert len(results) == 1 + assert results[0]["name"] == "owner/repo" + assert results[0]["stars"] == 100 + + @patch("agent_eyes.search.github.requests.get") + @pytest.mark.asyncio + async def test_search_github_with_language(self, mock_get): + mock_resp = MagicMock() + mock_resp.json.return_value = {"items": []} + mock_resp.raise_for_status = MagicMock() + mock_get.return_value = mock_resp + + from agent_eyes.search.github import search_github + await search_github("test", language="python") + + call_args = mock_get.call_args + assert "language:python" in call_args[1]["params"]["q"] + + +class TestTwitterSearch: + @patch("agent_eyes.search.twitter.shutil.which", return_value=None) + @patch("agent_eyes.search.exa.requests.post") + @pytest.mark.asyncio + async def test_search_twitter_exa_fallback(self, mock_post, mock_which, tmp_config): + mock_resp = MagicMock() + mock_resp.json.return_value = {"results": []} + mock_resp.raise_for_status = MagicMock() + mock_post.return_value = mock_resp + + from agent_eyes.search.twitter import search_twitter + results = await search_twitter("test", config=tmp_config) + + call_args = mock_post.call_args + query = call_args[1]["json"]["query"] + assert "site:x.com" in query