--- name: email-bridge description: Email management skill for AI assistants with real-time notifications, smart categorization (7 categories), verification code extraction, and HTML content sanitization. Supports Gmail, QQ Mail, and NetEase. homepage: https://github.com/ryanchan720/email-bridge source: https://github.com/ryanchan720/email-bridge version: 0.6.3 --- # Email Bridge Skill Email management skill for OpenClaw. Provides real-time email monitoring with smart categorization and clean notifications for AI assistants. ## Features - **Real-time notifications**: IMAP IDLE (QQ/NetEase) + polling (Gmail) - **Smart categorization**: 7 categories with subject-only classification - **Verification code extraction**: Context-aware, low false positive rate - **HTML content sanitization**: Clean text from HTML emails, remove invisible chars - **Prompt injection protection**: Safe email content for AI processing - **Multi-provider support**: Gmail (API), QQ Mail (IMAP), NetEase (IMAP) ## Installation ```bash cd skills/email-bridge pip install -e . ``` ## Setup (Manual CLI Required) ⚠️ **Security Note**: Do NOT share authorization codes in chat. Configure accounts via CLI only. ```bash # Add account (prompts for authorization code securely) email-bridge accounts add your@qq.com -p qq # Sync emails email-bridge sync # Start daemon for real-time notifications email-bridge daemon start -d ``` ### Getting Authorization Codes **QQ Mail:** https://service.mail.qq.com/detail/0/75 (send SMS, get 16-char code) **NetEase (163/126):** Settings → POP3/SMTP/IMAP → Enable → Get code **Gmail:** Requires OAuth setup (see README.md) ## Capabilities - **Receive emails**: Sync and read emails from configured accounts - **Send emails**: Send emails via SMTP - **Real-time notifications**: Push to OpenClaw via `openclaw system event` - **Smart categorization**: 7 categories with keyword-based classification - **Verification code extraction**: Context-aware extraction with low false positives - **Link extraction**: Extract action links from emails - **HTML sanitization**: Clean text extraction with invisible char removal - **Prompt injection protection**: Sanitize email content for safe AI processing ## Email Categories Subject-only classification for fast, reliable categorization: | Category | Icon | Description | Example Keywords | |----------|------|-------------|------------------| | verification | 🔐 | Verification codes, activation | 验证码, OTP, activate, 绑定邮箱 | | security | ⚠️ | Security alerts, login warnings | 安全提醒, security alert, 密码修改 | | transactional | 📦 | Orders, payments, shipping | 订单确认, receipt, 发货通知 | | promotion | 🎁 | Marketing, promotions, rewards | 奖励, 优惠, promo, discount | | subscription | 📰 | Newsletters, digests | newsletter, 订阅, weekly digest | | spam_like | 🚫 | Suspected spam | 中奖, FREE, click here now | | normal | — | Regular email | (default) | ## Trigger Keywords **Chinese:** 邮箱、邮件、发邮件、查看邮件、验证码、QQ邮箱、Gmail **English:** email, mail, send email, check email, verification code ## Common Commands ```bash # List recent emails email-bridge messages list -n 10 # Get verification codes from recent emails email-bridge codes # Send email email-bridge send -a -t recipient@example.com -s "Subject" -b "Body" # Daemon management email-bridge daemon status email-bridge daemon stop ``` ## Configuration Configuration file: `~/.email-bridge/config.json` **Default configuration** (auto-generated, minimal): ```json { "daemon": { "poll_interval": 300, "notify_openclaw": true } } ``` **Full configuration with all options** (customize as needed): ```json { "daemon": { "poll_interval": 300, "notify_openclaw": true, "notification": { "include_body": false, "body_max_length": 500, "include_verification_codes": true, "include_links": false } } } ``` ### Notification Options | Option | Default | Description | |--------|---------|-------------| | `include_body` | `false` | Include email body preview in notifications | | `body_max_length` | `500` | Max characters for body preview | | `include_verification_codes` | `true` | Auto-extract and show verification codes | | `include_links` | `false` | Include action links (verify/reset) | ## Notifications When new emails arrive, the daemon sends formatted notifications: ``` 📧 新邮件: account@qq.com 1. 🔐 Google 您的验证码 ✨ 验证码: 123456 2. ⚠️ Microsoft 登录提醒 📝 We noticed a new sign-in... 3. 🎁 OKX 150 USDT 奖励等您拿 📝 亲爱的欧易用户，欧易诚邀您加入邀请好友计划... ``` ## HTML Content Processing HTML-only emails are processed through: 1. **Tag stripping**: Remove `', '', html, flags=re.DOTALL | re.IGNORECASE) # Remove script blocks html = re.sub(r']*>.*?', '', html, flags=re.DOTALL | re.IGNORECASE) # Remove HTML tags html = re.sub(r'<[^>]+>', ' ', html) # Decode HTML entities import html as html_module html = html_module.unescape(html) # Normalize whitespace body_text = re.sub(r'\s+', ' ', html).strip() # Optionally include body preview if self.notification_config.get('include_body') and body_text: max_len = self.notification_config.get('body_max_length', 500) info['body_preview'] = sanitize_for_notification(body_text, max_length=max_len) # Extract verification codes for all emails (not just verification category) if self.notification_config.get('include_verification_codes'): codes = extract_verification_codes(f"{msg.subject}\n\n{body_text or ''}") if codes: info['codes'] = [c.code for c in codes[:3]] # Max 3 codes # Optionally extract links (disabled by default for security) if self.notification_config.get('include_links') and body_text: from email_bridge.extraction import extract_action_links links = extract_action_links(body_text, msg.body_html) # Only include verify/reset links, not unsubscribe filtered_links = [l for l in links if l.link_type in ('verify', 'reset')][:2] if filtered_links: info['links'] = [{'type': l.link_type, 'domain': l.domain} for l in filtered_links] result.append(info) if len(result) >= count: break return result except Exception as e: self._log(f"Error getting message info: {e}") return [] def _notify_openclaw(self, account_email: str, new_count: int, account_id: str = None): """Send notification to OpenClaw main session via system event. Uses structured JSON format internally for safety, then formats as human-readable text for display. """ if not self.notify_openclaw: return try: import subprocess # Get message details if available details = [] if account_id: details = self._get_latest_message_info(account_id, min(new_count, 3)) # Skip notification if no new messages to report if not details: self._log(f"{account_email}: No new messages to notify (already notified or outside time limit)") return # Mark messages as notified if account_id not in self._notified_messages: self._notified_messages[account_id] = set() for info in details: self._notified_messages[account_id].add(info["message_id"]) # Build structured data (for internal processing) notification = { "type": "email_notification", "account": account_email, "count": new_count, "messages": details } # Format as human-readable text message = self._format_notification(notification) self._log(f"Sending notification to OpenClaw: {len(details)} messages") result = subprocess.run( ["openclaw", "system", "event", "--text", message, "--mode", "now"], capture_output=True, text=True, timeout=60 ) if result.returncode == 0: self._log(f"OpenClaw notification sent successfully") else: self._log(f"OpenClaw notification failed (code {result.returncode}): {result.stderr.strip() or 'No error'}") except subprocess.TimeoutExpired: self._log("OpenClaw notification timed out") except FileNotFoundError: self._log("OpenClaw CLI not found at PATH") except Exception as e: self._log(f"Error sending notification: {e}") def _format_notification(self, notification: dict) -> str: """Format notification data as human-readable text. Adapts format based on email category for better UX. Args: notification: Structured notification dict Returns: Human-readable text message """ lines = [f"📧 新邮件: {notification['account']}"] for i, msg in enumerate(notification['messages'], 1): category = msg.get('category', 'normal') sender = msg['sender'] subject = msg['subject'] # Category emoji and format if category == 'verification': # 验证码邮件：重点显示验证码 lines.append(f"\n{i}. 🔐 {sender}") lines.append(f" {subject}") if msg.get('codes'): codes_str = ', '.join(msg['codes']) lines.append(f" ✨ 验证码: {codes_str}") elif msg.get('body_preview'): # 没提取到验证码但显示正文 lines.append(f" 📝 {msg['body_preview'][:80]}...") elif category == 'security': # 安全邮件：突出警示 lines.append(f"\n{i}. ⚠️ {sender}") lines.append(f" {subject}") if msg.get('body_preview'): lines.append(f" 📝 {msg['body_preview'][:100]}...") elif category == 'transactional': # 交易邮件：订单/支付/发货 lines.append(f"\n{i}. 📦 {sender}") lines.append(f" {subject}") if msg.get('body_preview'): lines.append(f" 📝 {msg['body_preview'][:100]}...") elif category == 'promotion': # 推广邮件：营销/优惠/奖励 lines.append(f"\n{i}. 🎁 {sender}") lines.append(f" {subject}") if msg.get('body_preview'): lines.append(f" 📝 {msg['body_preview'][:80]}...") elif category == 'subscription': # 订阅邮件：简化显示 lines.append(f"\n{i}. 📰 {sender}") lines.append(f" {subject}") # 订阅邮件不显示正文，避免噪音 elif category == 'spam_like': # 垃圾邮件：标记但不展开 lines.append(f"\n{i}. 🚫 {sender}") lines.append(f" {subject} (疑似垃圾)") else: # 普通邮件：标准格式 lines.append(f"\n{i}. {sender}") lines.append(f" {subject}") if msg.get('body_preview'): lines.append(f" 📝 {msg['body_preview'][:100]}...") if msg.get('codes'): codes_str = ', '.join(msg['codes']) lines.append(f" 🔐 验证码: {codes_str}") # Links for any category if msg.get('links'): for link in msg['links'][:2]: lines.append(f" 🔗 {link['type']}: {link['domain']}") if notification['count'] > len(notification['messages']): lines.append(f"\n... 还有 {notification['count'] - len(notification['messages'])} 封") return '\n'.join(lines) def _poll_account(self, account, stop_event: threading.Event): """Poll an account for new messages (for Gmail).""" # Track message count to detect new messages last_count = None while not stop_event.is_set(): try: # Get current message count from database messages = self.service.list_recent_messages(account_id=account.id, limit=1000) current_count = len(messages) # First poll: just record the count, don't notify if last_count is None: last_count = current_count self._log(f"{account.email}: Initial count = {current_count}") elif current_count > last_count: # New messages detected new_count = current_count - last_count self._log(f"{account.email}: {new_count} new message(s) detected (was {last_count}, now {current_count})") last_count = current_count # Sync to get message details self.service.sync_account(account.id, limit=new_count) # Notify OpenClaw self._notify_openclaw(account.email, new_count, account.id) # Trigger callback if self.on_new_email: self.on_new_email({ "account_id": account.id, "email": account.email, "new_count": new_count, }) else: # No new messages, just sync to keep database updated self._log(f"{account.email}: No new messages (count: {current_count})") self.service.sync_account(account.id, limit=10) last_count = current_count except Exception as e: self._log(f"{account.email}: Poll error: {e}") # Wait for next poll stop_event.wait(timeout=self.poll_interval) def main(): """CLI entry point for daemon management.""" import argparse parser = argparse.ArgumentParser(description="Email Bridge Daemon") parser.add_argument( "action", choices=["start", "stop", "status"], help="Action to perform", ) parser.add_argument( "-d", "--daemon", action="store_true", help="Run in background (daemon mode)", ) parser.add_argument( "-i", "--interval", type=int, default=300, help="Polling interval in seconds (default: 300)", ) args = parser.parse_args() daemon = EmailDaemon(poll_interval=args.interval) if args.action == "start": daemon.start(background=args.daemon) elif args.action == "stop": daemon.stop() elif args.action == "status": status = daemon.status() print(json.dumps(status, indent=2)) if __name__ == "__main__": main() FILE:email_bridge/db.py """SQLite database layer for email bridge.""" import json import sqlite3 from datetime import datetime from pathlib import Path from typing import Optional from .models import Account, AccountStatus, EmailProvider, Message, EmailCategory DEFAULT_DB_PATH = Path.home() / ".email-bridge" / "email_bridge.db" class Database: """SQLite-backed storage for accounts and messages.""" def __init__(self, db_path: Optional[Path] = None): self.db_path = db_path or DEFAULT_DB_PATH self.db_path.parent.mkdir(parents=True, exist_ok=True) self._conn: Optional[sqlite3.Connection] = None @property def conn(self) -> sqlite3.Connection: if self._conn is None: self._conn = sqlite3.connect(str(self.db_path), check_same_thread=False) self._conn.row_factory = sqlite3.Row self._init_schema() return self._conn def _init_schema(self): """Initialize database schema.""" self.conn.executescript(""" CREATE TABLE IF NOT EXISTS accounts ( id TEXT PRIMARY KEY, email TEXT UNIQUE NOT NULL, provider TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'active', display_name TEXT, config TEXT DEFAULT '{}', created_at TEXT NOT NULL, updated_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS messages ( id TEXT PRIMARY KEY, account_id TEXT NOT NULL, message_id TEXT NOT NULL, subject TEXT NOT NULL, sender TEXT NOT NULL, sender_name TEXT, recipients TEXT DEFAULT '[]', received_at TEXT NOT NULL, is_read INTEGER DEFAULT 0, is_starred INTEGER DEFAULT 0, category TEXT DEFAULT 'normal', preview TEXT, body_text TEXT, body_html TEXT, synced_at TEXT NOT NULL, provider_data TEXT DEFAULT '{}', UNIQUE(account_id, message_id) ); CREATE INDEX IF NOT EXISTS idx_messages_account ON messages(account_id); CREATE INDEX IF NOT EXISTS idx_messages_received ON messages(received_at DESC); CREATE INDEX IF NOT EXISTS idx_messages_read ON messages(is_read); CREATE INDEX IF NOT EXISTS idx_messages_category ON messages(category); """) # Account operations def list_accounts(self, include_disabled: bool = False) -> list[Account]: """List all accounts.""" query = "SELECT * FROM accounts" if not include_disabled: query += " WHERE status = ?" rows = self.conn.execute(query, (AccountStatus.ACTIVE.value,)).fetchall() else: rows = self.conn.execute(query).fetchall() return [self._row_to_account(row) for row in rows] def get_account(self, account_id: str) -> Optional[Account]: """Get account by ID.""" row = self.conn.execute( "SELECT * FROM accounts WHERE id = ?", (account_id,) ).fetchone() return self._row_to_account(row) if row else None def add_account(self, account: Account) -> Account: """Add a new account.""" self.conn.execute( """INSERT INTO accounts (id, email, provider, status, display_name, config, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)""", ( account.id, account.email, account.provider.value, account.status.value, account.display_name, json.dumps(account.config), account.created_at.isoformat(), account.updated_at.isoformat() ) ) self.conn.commit() return account def update_account(self, account: Account) -> Account: """Update an existing account.""" account.updated_at = datetime.utcnow() self.conn.execute( """UPDATE accounts SET email = ?, provider = ?, status = ?, display_name = ?, config = ?, updated_at = ? WHERE id = ?""", ( account.email, account.provider.value, account.status.value, account.display_name, json.dumps(account.config), account.updated_at.isoformat(), account.id ) ) self.conn.commit() return account def delete_account(self, account_id: str) -> bool: """Delete an account and its messages.""" self.conn.execute("DELETE FROM messages WHERE account_id = ?", (account_id,)) self.conn.execute("DELETE FROM accounts WHERE id = ?", (account_id,)) self.conn.commit() return True # Message operations def list_messages( self, account_id: Optional[str] = None, unread_only: bool = False, category: Optional[EmailCategory] = None, limit: int = 50, offset: int = 0 ) -> list[Message]: """List messages with optional filters.""" query = "SELECT * FROM messages WHERE 1=1" params = [] if account_id: query += " AND account_id = ?" params.append(account_id) if unread_only: query += " AND is_read = 0" if category: query += " AND category = ?" params.append(category.value) query += " ORDER BY received_at DESC LIMIT ? OFFSET ?" params.extend([limit, offset]) rows = self.conn.execute(query, params).fetchall() return [self._row_to_message(row) for row in rows] def search_messages( self, keyword: Optional[str] = None, start_time: Optional[datetime] = None, end_time: Optional[datetime] = None, account_id: Optional[str] = None, limit: int = 50 ) -> list[Message]: """Search messages by keyword and/or time range.""" query = "SELECT * FROM messages WHERE 1=1" params = [] if keyword: query += " AND (subject LIKE ? OR sender LIKE ? OR preview LIKE ?)" kw = f"%{keyword}%" params.extend([kw, kw, kw]) if start_time: query += " AND received_at >= ?" params.append(start_time.isoformat()) if end_time: query += " AND received_at <= ?" params.append(end_time.isoformat()) if account_id: query += " AND account_id = ?" params.append(account_id) query += " ORDER BY received_at DESC LIMIT ?" params.append(limit) rows = self.conn.execute(query, params).fetchall() return [self._row_to_message(row) for row in rows] def get_message(self, message_id: str) -> Optional[Message]: """Get a single message by ID.""" row = self.conn.execute( "SELECT * FROM messages WHERE id = ?", (message_id,) ).fetchone() return self._row_to_message(row) if row else None def find_message(self, message_id: str) -> Optional[Message]: """Find a message by flexible ID lookup. Tries exact match first, then message_id suffix, then partial match. """ # Try exact match first msg = self.get_message(message_id) if msg: return msg # Try matching by message_id suffix (e.g., "mock-001" matches "abc:mock-001") row = self.conn.execute( "SELECT * FROM messages WHERE message_id = ?", (message_id,) ).fetchone() if row: return self._row_to_message(row) # Try partial match on id (ends with) row = self.conn.execute( "SELECT * FROM messages WHERE id LIKE ?", (f"%:{message_id}%",) ).fetchone() if row: return self._row_to_message(row) return None def add_message(self, message: Message) -> Message: """Add or update a message.""" self.conn.execute( """INSERT OR REPLACE INTO messages (id, account_id, message_id, subject, sender, sender_name, recipients, received_at, is_read, is_starred, category, preview, body_text, body_html, synced_at, provider_data) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", ( message.id, message.account_id, message.message_id, message.subject, message.sender, message.sender_name, json.dumps(message.recipients), message.received_at.isoformat(), int(message.is_read), int(message.is_starred), message.category.value, message.preview, message.body_text, message.body_html, message.synced_at.isoformat(), json.dumps(message.provider_data) ) ) self.conn.commit() return message def mark_read(self, message_id: str, is_read: bool = True) -> bool: """Mark a message as read/unread.""" self.conn.execute( "UPDATE messages SET is_read = ? WHERE id = ?", (int(is_read), message_id) ) self.conn.commit() return True def count_unread(self, account_id: Optional[str] = None) -> int: """Count unread messages.""" if account_id: row = self.conn.execute( "SELECT COUNT(*) FROM messages WHERE account_id = ? AND is_read = 0", (account_id,) ).fetchone() else: row = self.conn.execute( "SELECT COUNT(*) FROM messages WHERE is_read = 0" ).fetchone() return row[0] if row else 0 def _row_to_account(self, row: sqlite3.Row) -> Account: return Account( id=row["id"], email=row["email"], provider=EmailProvider(row["provider"]), status=AccountStatus(row["status"]), display_name=row["display_name"], config=json.loads(row["config"]), created_at=datetime.fromisoformat(row["created_at"]), updated_at=datetime.fromisoformat(row["updated_at"]), ) def _row_to_message(self, row: sqlite3.Row) -> Message: return Message( id=row["id"], account_id=row["account_id"], message_id=row["message_id"], subject=row["subject"], sender=row["sender"], sender_name=row["sender_name"], recipients=json.loads(row["recipients"]), received_at=datetime.fromisoformat(row["received_at"]), is_read=bool(row["is_read"]), is_starred=bool(row["is_starred"]), category=EmailCategory(row["category"]), preview=row["preview"], body_text=row["body_text"], body_html=row["body_html"], synced_at=datetime.fromisoformat(row["synced_at"]), provider_data=json.loads(row["provider_data"]), ) def close(self): if self._conn: self._conn.close() self._conn = None FILE:email_bridge/extraction.py """Extraction helpers for email content. Provides utilities to extract useful information from emails: - Verification codes (OTP, confirmation codes) - Action links (verify, confirm, reset password, unsubscribe) """ import re from dataclasses import dataclass from typing import Optional from urllib.parse import urlparse, unquote @dataclass class VerificationCode: """Extracted verification code.""" code: str context: Optional[str] = None # e.g., "Google", "GitHub", "Your code" @dataclass class ActionLink: """Extracted action link.""" url: str link_type: str # verify, confirm, reset, unsubscribe, action domain: Optional[str] = None text: Optional[str] = None # Link text if available # Patterns for verification codes VERIFICATION_PATTERNS = [ # Numeric codes (4-8 digits) (r'\b(\d{4,8})\b', 'numeric'), # Alphanumeric codes (common formats) (r'\b([A-Z0-9]{4,8})\b', 'alphanumeric'), # Codes with dashes (r'\b([A-Z0-9]{3,4}-[A-Z0-9]{3,4})\b', 'dashed'), # Chinese verification code patterns (r'验证码[：:\s]*(\d{4,8})', 'chinese_numeric'), (r'验证码[：:\s]*([A-Z0-9]{4,8})', 'chinese_alphanumeric'), ] # Context patterns that indicate verification codes VERIFICATION_CONTEXTS = [ r'verification\s*code', r'confirm(?:ation)?\s*code', r'security\s*code', r'OTP', r'one[- ]?time[- ]?password', r'PIN', r'authenticate', r'验证码', r'校验码', r'确认码', r'动态码', r'Your\s+(?:verification|security|confirmation)?\s*code\s+is', r'Enter\s+(?:this|the)\s+code', r'Use\s+(?:this|the)\s+code', r'code\s+(?:is|:)\s*[A-Z0-9]', ] # Link patterns for different action types LINK_PATTERNS = { 'verify': [ r'verify', r'confirm', r'activate', r'validate', r'认证', r'验证', ], 'reset': [ r'reset', r'change\s+password', r'new\s+password', r'forgot\s+password', r'重置', r'修改密码', ], 'unsubscribe': [ r'unsubscribe', r'opt[- ]?out', r'remove', r'stop\s+receiving', r'cancel\s+subscription', r'退订', r'取消订阅', ], 'action': [ r'click\s+here', r'view\s+(?:your|the)', r'respond', r'reply', r'accept', r'decline', r'join', r'start', r'continue', ], } def extract_verification_codes(text: str) -> list[VerificationCode]: """Extract verification codes from email text. IMPORTANT: This function ONLY extracts codes that appear near verification keywords. Standalone numbers are NOT extracted to avoid false positives (addresses, order numbers, tracking numbers, etc.). Args: text: Email body text to search Returns: List of VerificationCode objects found """ if not text: return [] codes = [] seen = set() # Only extract codes that are near verification keywords # This is the ONLY extraction method to avoid false positives text_lower = text.lower() for context_pattern in VERIFICATION_CONTEXTS: # Find context matches for match in re.finditer(context_pattern, text_lower, re.IGNORECASE): # Look for codes in surrounding text (100 chars before and after) start = max(0, match.start() - 100) end = min(len(text), match.end() + 100) context_text = text[start:end] # Try to find codes in this context for code_pattern, _ in VERIFICATION_PATTERNS: for code_match in re.finditer(code_pattern, context_text): code = code_match.group(1) if code not in seen and _is_likely_code_in_context(code, context_text): seen.add(code) # Extract context (service name if available) ctx = _extract_service_context(text, match.start()) codes.append(VerificationCode(code=code, context=ctx)) return codes def _is_likely_code(code: str) -> bool: """Check if a string looks like a verification code. Filters out things like years, prices, phone numbers, etc. """ # Filter out years (2000-2100) if len(code) == 4 and code.isdigit(): year = int(code) if 2000 <= year <= 2100: return False # Filter out common non-code patterns if code.lower() in ['http', 'https', 'html', 'www', 'utf8', 'json']: return False # Filter out pure letters (likely words) if code.isalpha(): return False # Filter out very long numeric sequences (likely phone/tracking numbers) if code.isdigit() and len(code) > 8: return False # Filter out dates in YYYYMMDD format if code.isdigit() and len(code) == 8: if code.startswith('20'): # Likely a date like 20260321 return False return True def _is_likely_code_in_context(code: str, context: str) -> bool: """Check if a code is likely a verification code given surrounding context. This is more strict than _is_likely_code and considers the context to filter out order numbers, invoice numbers, etc. """ if not _is_likely_code(code): return False # Check if the code appears after order/invoice/tracking markers context_lower = context.lower() false_positive_markers = [ 'order #', 'order no', 'order:', 'invoice', 'tracking', 'reference #', 'ref #', 'account #', 'transaction', 'zip', 'postal', 'phone', 'tel:', 'fax:', # Address patterns 'ste', 'st', 'ave', 'avenue', 'blvd', 'rd', 'road', 'drive', 'dr', 'lane', 'ln', 'court', 'ct', ] # Look for these markers within 30 chars before the code code_pos = context.find(code) if code_pos > 0: before = context[max(0, code_pos - 30):code_pos].lower() for marker in false_positive_markers: if marker in before: return False return True def _extract_service_context(text: str, position: int) -> Optional[str]: """Try to extract service/company name near a code.""" # Look backwards for common patterns before = text[max(0, position - 200):position] # Pattern: "Your Google verification code" or "GitHub code" patterns = [ r'(?:your\s+)?([A-Z][a-zA-Z]+)\s+(?:verification|security|confirmation)\s+code', r'([A-Z][a-zA-Z]+)\s+code\s+is', r'from\s+([A-Z][a-zA-Z]+)', ] for pattern in patterns: match = re.search(pattern, before, re.IGNORECASE) if match: return match.group(1) return None def extract_action_links(text: str, html: Optional[str] = None) -> list[ActionLink]: """Extract action links from email content. Args: text: Plain text email body html: Optional HTML email body for better link extraction Returns: List of ActionLink objects found """ links = [] seen_urls = set() # Extract from HTML if available (better link text extraction) if html: links.extend(_extract_links_from_html(html, seen_urls)) # Also extract from plain text links.extend(_extract_links_from_text(text, seen_urls)) # Filter and classify links action_links = [] for link in links: classified = _classify_link(link) if classified: action_links.append(classified) return action_links def _extract_links_from_html(html: str, seen_urls: set) -> list[ActionLink]: """Extract links from HTML content.""" links = [] # Match text pattern = r']*href=["\']([^"\']+)["\'][^>]*>([^<]+)' for match in re.finditer(pattern, html, re.IGNORECASE): url = match.group(1).strip() text = match.group(2).strip() # Clean up URL url = _clean_url(url) if url and url not in seen_urls: seen_urls.add(url) domain = _extract_domain(url) links.append(ActionLink( url=url, link_type='unknown', domain=domain, text=text[:100] if text else None )) return links def _extract_links_from_text(text: str, seen_urls: set) -> list[ActionLink]: """Extract links from plain text content.""" links = [] # Match URLs url_pattern = r'https?://[^\s<>"{}|\\^`\[\]]+' for match in re.finditer(url_pattern, text): url = match.group(0).strip() url = _clean_url(url) if url and url not in seen_urls: seen_urls.add(url) domain = _extract_domain(url) # Try to get context before the URL context_start = max(0, match.start() - 50) context = text[context_start:match.start()].strip() links.append(ActionLink( url=url, link_type='unknown', domain=domain, text=context if context else None )) return links def _clean_url(url: str) -> str: """Clean and normalize a URL.""" # Remove trailing punctuation url = url.rstrip('.,;:)>]') # Decode URL encoding for better pattern matching try: url = unquote(url) except Exception: pass return url def _extract_domain(url: str) -> Optional[str]: """Extract domain from URL.""" try: parsed = urlparse(url) return parsed.netloc except Exception: return None def _classify_link(link: ActionLink) -> Optional[ActionLink]: """Classify a link by type based on URL and text.""" # Combine URL and text for pattern matching check_text = f"{link.url} {link.text or ''}".lower() for link_type, patterns in LINK_PATTERNS.items(): for pattern in patterns: if re.search(pattern, check_text, re.IGNORECASE): return ActionLink( url=link.url, link_type=link_type, domain=link.domain, text=link.text ) # Not an action link return None def extract_from_email( subject: str, body_text: Optional[str] = None, body_html: Optional[str] = None ) -> dict: """Extract all useful information from an email. Args: subject: Email subject line body_text: Plain text body body_html: HTML body Returns: Dict with 'codes' and 'links' keys """ # Combine subject with body for code extraction full_text = f"{subject}\n\n{body_text or ''}" codes = extract_verification_codes(full_text) links = extract_action_links(body_text or '', body_html) return { 'codes': codes, 'links': links, } FILE:email_bridge/models.py """Data models for email bridge.""" from datetime import datetime from typing import Optional from enum import Enum from pydantic import BaseModel, Field class EmailCategory(str, Enum): """Email categories.""" VERIFICATION = "verification" SECURITY = "security" SUBSCRIPTION = "subscription" SPAM_LIKE = "spam_like" NORMAL = "normal" class AccountStatus(str, Enum): """Account status.""" ACTIVE = "active" DISABLED = "disabled" class EmailProvider(str, Enum): """Supported email providers.""" MOCK = "mock" GMAIL = "gmail" QQ = "qq" NETEASE = "netease" class Account(BaseModel): """Email account configuration.""" id: str email: str provider: EmailProvider status: AccountStatus = AccountStatus.ACTIVE display_name: Optional[str] = None created_at: datetime = Field(default_factory=datetime.utcnow) updated_at: datetime = Field(default_factory=datetime.utcnow) # Provider-specific config (e.g., tokens, credentials) stored encrypted later config: dict = Field(default_factory=dict) class Message(BaseModel): """Email message metadata.""" id: str account_id: str message_id: str # Original message ID from provider subject: str sender: str sender_name: Optional[str] = None recipients: list[str] = Field(default_factory=list) received_at: datetime is_read: bool = False is_starred: bool = False category: EmailCategory = EmailCategory.NORMAL # Optional cached content preview: Optional[str] = None # First ~200 chars body_text: Optional[str] = None body_html: Optional[str] = None # Metadata synced_at: datetime = Field(default_factory=datetime.utcnow) provider_data: dict = Field(default_factory=dict) # Provider-specific raw data FILE:email_bridge/sanitize.py """Content sanitization for email notifications. Prevents prompt injection attacks from email content. """ import re from typing import Optional # Dangerous patterns that could be used for prompt injection DANGEROUS_PATTERNS = [ # Instruction override attempts r'ignore\s+(all\s+)?previous\s+instructions?', r'forget\s+(all\s+)?(previous\s+)?instructions?', r'disregard\s+(all\s+)?instructions?', r'override\s+(all\s+)?instructions?', # Role manipulation r'you\s+are\s+now\s+', r'act\s+as\s+(a|an)\s+', r'pretend\s+(to\s+be|you\s+are)\s+', r'role[\s\-]*play\s+(as|that)\s+', # System prompt injection r'new\s+instructions?:?\s*', r'system\s*:\s*', r'assistant\s*:\s*', r'user\s*:\s*', # Special tokens r'<\|.*?\|>', r'\[SYSTEM\]', r'\[INST\]', r'<<.*?>>', # Escape sequences r'\\n\\n', r'\\r\\n', ] # Invisible Unicode characters to remove (zero-width spaces, etc.) INVISIBLE_CHARS_PATTERN = r'[\u200b-\u200f\u2028-\u202f\u205f-\u206f\ufeff\u00ad\u034f\u061c\u17b4-\u17b5\u180e\u200c\u200d]+' def remove_invisible_chars(text: str) -> str: """Remove invisible Unicode characters from text. These characters are often used in HTML emails for formatting but appear as garbage in plain text output. Includes: - Zero-width spaces (U+200B, U+200C, U+200D) - Zero-width non-joiners/joiners - Word joiners, function applications, etc. - BOM (U+FEFF) - Soft hyphen (U+00AD) - Various other invisible formatting chars """ if not text: return text return re.sub(INVISIBLE_CHARS_PATTERN, '', text) def sanitize_for_notification( text: str, max_length: int = 500, placeholder: str = "[REMOVED]", ) -> str: """Sanitize text content for safe notification. Removes potential prompt injection patterns, invisible characters, and limits length. Args: text: The text to sanitize max_length: Maximum allowed length (default 500) placeholder: String to replace dangerous patterns with Returns: Sanitized text safe for inclusion in notifications """ if not text: return "" result = text # Remove invisible Unicode characters first result = remove_invisible_chars(result) # Remove dangerous patterns (case-insensitive) for pattern in DANGEROUS_PATTERNS: result = re.sub(pattern, placeholder, result, flags=re.IGNORECASE) # Remove control characters except newlines and tabs result = re.sub(r'[\x00-\x08\x0b\x0c\x0e-\x1f\x7f-\x9f]', '', result) # Normalize whitespace result = re.sub(r'\s+', ' ', result) # Truncate to max length if len(result) > max_length: result = result[:max_length].rsplit(' ', 1)[0] + "..." return result.strip() def sanitize_sender(sender: str, max_length: int = 100) -> str: """Sanitize sender name/email for display. Args: sender: Sender name or email max_length: Maximum allowed length Returns: Sanitized sender string """ if not sender: return "Unknown" # Remove any special characters that could be problematic result = re.sub(r'[<>"\']', '', sender) # Remove invisible characters result = remove_invisible_chars(result) # Limit length if len(result) > max_length: result = result[:max_length] + "..." return result.strip() def sanitize_subject(subject: str, max_length: int = 100) -> str: """Sanitize email subject for display. Args: subject: Email subject max_length: Maximum allowed length Returns: Sanitized subject string """ if not subject: return "(No subject)" # Remove invisible characters result = remove_invisible_chars(subject) # Remove newlines that could break formatting result = result.replace('\n', ' ').replace('\r', ' ') # Limit length if len(result) > max_length: result = result[:max_length].rsplit(' ', 1)[0] + "..." return result.strip() FILE:email_bridge/service.py """Service layer for email bridge business logic.""" import uuid from datetime import datetime from typing import Optional, List from .adapters.base import FetchOptions from .adapters.mock import MockAdapter from .adapters.gmail import GmailAdapter, GmailCredentialsNotFoundError, GmailAuthError from .adapters.imap import QQMailAdapter, NetEaseMailAdapter, IMAPAdapterError, IMAPAuthError from .adapters.smtp import SMTPAdapter, SMTPAdapterError, SMTPAuthError, SMTPSendError from .categories import detect_category from .db import Database from .models import Account, AccountStatus, EmailCategory, EmailProvider, Message class EmailBridgeService: """Core service for email management. This service provides a clean API layer between the CLI/presentation and the data/adapters layer. """ def __init__(self, db: Optional[Database] = None): self.db = db or Database() self._adapters = { EmailProvider.MOCK: MockAdapter(), EmailProvider.GMAIL: GmailAdapter(), EmailProvider.QQ: QQMailAdapter(), EmailProvider.NETEASE: NetEaseMailAdapter(), } # Account management def list_accounts(self, include_disabled: bool = False) -> list[Account]: """List all configured accounts.""" return self.db.list_accounts(include_disabled=include_disabled) def get_account(self, account_id: str) -> Optional[Account]: """Get account by ID.""" return self.db.get_account(account_id) def add_account( self, email: str, provider: EmailProvider, display_name: Optional[str] = None, config: Optional[dict] = None ) -> Account: """Add a new email account.""" account = Account( id=str(uuid.uuid4())[:8], email=email, provider=provider, display_name=display_name, config=config or {}, ) return self.db.add_account(account) def update_account( self, account_id: str, display_name: Optional[str] = None, status: Optional[AccountStatus] = None, config: Optional[dict] = None ) -> Optional[Account]: """Update an existing account.""" account = self.db.get_account(account_id) if not account: return None if display_name is not None: account.display_name = display_name if status is not None: account.status = status if config is not None: account.config = config return self.db.update_account(account) def disable_account(self, account_id: str) -> bool: """Disable an account (soft delete).""" account = self.db.get_account(account_id) if not account: return False account.status = AccountStatus.DISABLED self.db.update_account(account) return True def delete_account(self, account_id: str) -> bool: """Permanently delete an account and its messages.""" return self.db.delete_account(account_id) # Message management def list_recent_messages( self, account_id: Optional[str] = None, limit: int = 20 ) -> list[Message]: """List recent messages across all or one account.""" return self.db.list_messages(account_id=account_id, limit=limit) def list_unread_messages( self, account_id: Optional[str] = None, limit: int = 20 ) -> list[Message]: """List unread messages.""" return self.db.list_messages( account_id=account_id, unread_only=True, limit=limit ) def search_messages( self, keyword: Optional[str] = None, start_time: Optional[datetime] = None, end_time: Optional[datetime] = None, account_id: Optional[str] = None, limit: int = 50 ) -> list[Message]: """Search messages by keyword and/or time range.""" return self.db.search_messages( keyword=keyword, start_time=start_time, end_time=end_time, account_id=account_id, limit=limit ) def get_message(self, message_id: str) -> Optional[Message]: """Get a single message with full details. Supports flexible ID lookup: full ID, message_id only, or partial. """ return self.db.find_message(message_id) def mark_read(self, message_id: str, is_read: bool = True) -> bool: """Mark a message as read or unread.""" # First find the message to get its full ID msg = self.db.find_message(message_id) if msg: return self.db.mark_read(msg.id, is_read) return False # Sync operations def sync_account( self, account_id: str, since: Optional[datetime] = None, limit: int = 100 ) -> int: """Sync messages from a provider into local storage. Returns the number of messages synced. """ account = self.db.get_account(account_id) if not account: raise ValueError(f"Account {account_id} not found") if account.status != AccountStatus.ACTIVE: raise ValueError(f"Account {account_id} is not active") adapter = self._adapters.get(account.provider) if not adapter: raise ValueError(f"No adapter for provider {account.provider}") options = FetchOptions(since=since, limit=limit) count = 0 for raw_msg in adapter.fetch_messages(account, options): # Detect category category = detect_category( raw_msg.subject, raw_msg.body_text or "" ) # Generate preview preview = None if raw_msg.body_text: preview = raw_msg.body_text[:200].strip() # Create message record message = Message( id=f"{account_id}:{raw_msg.message_id}", account_id=account_id, message_id=raw_msg.message_id, subject=raw_msg.subject, sender=raw_msg.sender, sender_name=raw_msg.sender_name, recipients=raw_msg.recipients, received_at=raw_msg.received_at, is_read=raw_msg.is_read, category=category, preview=preview, body_text=raw_msg.body_text, body_html=raw_msg.body_html, provider_data=raw_msg.raw_data, ) self.db.add_message(message) count += 1 return count def sync_all_accounts(self, since: Optional[datetime] = None) -> dict[str, int]: """Sync all active accounts. Returns a dict mapping account_id to sync count. """ results = {} for account in self.db.list_accounts(include_disabled=False): try: count = self.sync_account(account.id, since=since) results[account.id] = count except Exception as e: results[account.id] = f"Error: {e}" return results # Stats def get_stats(self) -> dict: """Get overall statistics.""" accounts = self.db.list_accounts(include_disabled=True) active_accounts = [a for a in accounts if a.status == AccountStatus.ACTIVE] unread_count = self.db.count_unread() return { "total_accounts": len(accounts), "active_accounts": len(active_accounts), "unread_messages": unread_count, } # Send email def send_email( self, account_id: str, to: List[str], subject: str, body_text: Optional[str] = None, body_html: Optional[str] = None, cc: Optional[List[str]] = None, bcc: Optional[List[str]] = None, ) -> bool: """Send an email from an account. Args: account_id: Account to send from to: List of recipient email addresses subject: Email subject body_text: Plain text body body_html: HTML body cc: List of CC recipients bcc: List of BCC recipients Returns: True if sent successfully Raises: ValueError: Account not found or inactive SMTPAuthError: Authentication failed SMTPSendError: Sending failed """ account = self.db.get_account(account_id) if not account: raise ValueError(f"Account {account_id} not found") if account.status != AccountStatus.ACTIVE: raise ValueError(f"Account {account_id} is not active") # Use SMTP adapter for all providers # (Gmail can also use API, but SMTP is simpler for now) smtp_adapter = SMTPAdapter() from_name = account.display_name return smtp_adapter.send_email( account=account, to=to, subject=subject, body_text=body_text, body_html=body_html, cc=cc, bcc=bcc, from_name=from_name, ) FILE:fixtures/sample_emails.json { "messages": [ { "message_id": "mock-001", "account_email": "demo@example.com", "subject": "Your verification code is 123456", "sender": "noreply@example.com", "sender_name": "Example Service", "recipients": ["demo@example.com"], "received_at": "2026-03-21T09:30:00", "is_read": false, "body_text": "Your verification code is 123456.\n\nThis code will expire in 5 minutes.\n\nIf you didn't request this code, please ignore this email.", "raw_data": {} }, { "message_id": "mock-002", "account_email": "demo@example.com", "subject": "Security Alert: New login to your account", "sender": "security@example.com", "sender_name": "Security Team", "recipients": ["demo@example.com"], "received_at": "2026-03-21T08:15:00", "is_read": false, "body_text": "We detected a new login to your account from:\n\nDevice: Chrome on Windows\nLocation: Beijing, China\nTime: March 21, 2026 at 8:15 AM\n\nIf this was you, no action is needed.\nIf you didn't log in, please secure your account immediately.", "raw_data": {} }, { "message_id": "mock-003", "account_email": "demo@example.com", "subject": "Weekly Newsletter - March Edition", "sender": "newsletter@techblog.com", "sender_name": "Tech Blog Weekly", "recipients": ["demo@example.com"], "received_at": "2026-03-20T14:00:00", "is_read": true, "body_text": "Welcome to the March edition of our newsletter!\n\nTop stories this week:\n- AI advances continue to reshape industries\n- New Python 3.14 released with performance improvements\n- Cloud computing trends for 2026\n\nTo unsubscribe, click here.", "raw_data": {} }, { "message_id": "mock-004", "account_email": "demo@example.com", "subject": "Re: Project Update Meeting", "sender": "colleague@company.com", "sender_name": "Zhang Wei", "recipients": ["demo@example.com"], "received_at": "2026-03-20T11:30:00", "is_read": true, "body_text": "Hi,\n\nThanks for the update. The meeting notes look good.\nLet's schedule a follow-up for next week.\n\nBest regards,\nZhang Wei", "raw_data": {} }, { "message_id": "mock-005", "account_email": "demo@example.com", "subject": "CONGRATULATIONS! You've WON $1,000,000!!!", "sender": "winner@lottery-scam.fake", "sender_name": "International Lottery", "recipients": ["demo@example.com"], "received_at": "2026-03-19T22:45:00", "is_read": false, "body_text": "CONGRATULATIONS!!!\n\nYou have been selected as the WINNER of our international lottery!\nClick here NOW to claim your $1,000,000 prize!\n\nAct now - limited time offer!!!", "raw_data": {} }, { "message_id": "mock-006", "account_email": "demo@example.com", "subject": "您的QQ邮箱账户安全提醒", "sender": "service@qq.com", "sender_name": "QQ邮箱安全中心", "recipients": ["demo@example.com"], "received_at": "2026-03-19T16:20:00", "is_read": true, "body_text": "尊敬的用户：\n\n我们检测到您的账户在新设备上登录。\n\n登录时间：2026年3月19日 16:20\n登录地点：上海\n\n如果这不是您本人的操作，请立即修改密码。\n\nQQ邮箱安全中心", "raw_data": {} }, { "message_id": "mock-007", "account_email": "demo@example.com", "subject": "Netflix 订阅确认", "sender": "info@netflix.com", "sender_name": "Netflix", "recipients": ["demo@example.com"], "received_at": "2026-03-18T10:00:00", "is_read": true, "body_text": "您的 Netflix 订阅已成功续期。\n\n订阅类型：高级版\n下次扣款日期：2026年4月18日\n\n如需退订，请访问账户设置。", "raw_data": {} }, { "message_id": "mock-008", "account_email": "demo@example.com", "subject": "GitHub Security: New personal access token created", "sender": "noreply@github.com", "sender_name": "GitHub", "recipients": ["demo@example.com"], "received_at": "2026-03-17T15:45:00", "is_read": false, "body_text": "A new personal access token (classic) was recently created for your account.\n\nToken name: CI Token\nScopes: repo, workflow\n\nIf you did not make this change, please contact GitHub Support.", "raw_data": {} }, { "message_id": "mock-009", "account_email": "demo@example.com", "subject": "Invoice #INV-2026-0342 from Acme Corp", "sender": "billing@acmecorp.com", "sender_name": "Acme Corp Billing", "recipients": ["demo@example.com"], "received_at": "2026-03-15T09:00:00", "is_read": true, "body_text": "Dear Customer,\n\nPlease find attached Invoice #INV-2026-0342 for $299.00.\n\nPayment is due within 30 days.\n\nThank you for your business!\n\nAcme Corp", "raw_data": {} }, { "message_id": "mock-010", "account_email": "demo@example.com", "subject": "Your order has shipped!", "sender": "shipping@shop.example", "sender_name": "Online Shop", "recipients": ["demo@example.com"], "received_at": "2026-03-14T13:20:00", "is_read": true, "body_text": "Great news! Your order #ORD-12345 has shipped.\n\nTracking number: 1Z999AA10123456784\nExpected delivery: March 17, 2026\n\nItems in this order:\n- Widget Pro x 1\n- Gadget Plus x 2\n\nThank you for shopping with us!", "raw_data": {} } ] } FILE:install.sh #!/bin/bash # Email Bridge Installation Script # Usage: ./install.sh set -e echo "📧 Email Bridge Installer" echo "========================" # Check Python version PYTHON_VERSION=$(python3 --version 2>/dev/null | awk '{print $2}' | cut -d. -f1,2) if [ -z "$PYTHON_VERSION" ]; then echo "❌ Python 3 not found. Please install Python 3.10+ first." exit 1 fi echo "✓ Python $PYTHON_VERSION detected" # Check if uv is available (preferred) if command -v uv &> /dev/null; then echo "✓ Using uv for installation" uv sync source .venv/bin/activate elif command -v pip &> /dev/null; then echo "✓ Using pip for installation" python3 -m venv .venv source .venv/bin/activate pip install -e . else echo "❌ Neither uv nor pip found. Please install one of them." exit 1 fi echo "" echo "✅ Installation complete!" echo "" echo "Quick Start:" echo "-----------" echo "" echo "1. Add your email account:" echo " # QQ Mail (with authorization code)" echo " email-bridge accounts add your@qq.com -p qq --config '{\"password\": \"YOUR_AUTH_CODE\"}'" echo "" echo "2. Sync emails:" echo " email-bridge sync" echo "" echo "3. Start daemon for real-time notifications:" echo " email-bridge daemon start -d" echo "" echo "For Gmail setup, see README.md → Gmail 配置指南" echo "" echo "Documentation: https://github.com/ryanchan720/email-bridge" FILE:openclaw.plugin.json { "id": "email-bridge", "name": "Email Bridge", "version": "0.5.0", "description": "Email management skill with real-time notifications, verification code extraction, and multi-provider support (Gmail, QQ Mail, NetEase)", "author": "Ryan Chan", "license": "MIT", "repository": "https://github.com/ryanchan720/email-bridge", "keywords": ["email", "gmail", "qq-mail", "imap", "smtp", "verification-code"], "runtime": "python", "entry": "email_bridge.cli:main", "install": { "command": "pip install -e .", "dependencies": [ "click>=8.0", "pydantic>=2.0", "google-api-python-client>=2.0", "google-auth>=2.0", "google-auth-oauthlib>=1.0", "imaplib2>=3.6" ] }, "triggers": { "keywords": { "zh": ["邮箱", "邮件", "电子邮件", "发邮件", "查看邮件", "验证码", "QQ邮箱", "Gmail", "163邮箱", "126邮箱"], "en": ["email", "mail", "send email", "check email", "verification code"] } }, "capabilities": [ "email:receive", "email:send", "email:sync", "email:extract_code", "email:daemon" ], "config": { "accounts": { "type": "array", "description": "Email accounts to configure", "items": { "email": "string", "provider": "string", "password": "string" } }, "daemon": { "poll_interval": { "type": "number", "default": 300, "description": "Polling interval for Gmail (seconds)" }, "notify": { "type": "boolean", "default": true, "description": "Send notifications for new emails" } } } } FILE:pyproject.toml [project] name = "email-bridge" version = "0.6.3" description = "Personal email middleware for AI assistants with real-time notifications, smart categorization, and verification code extraction (Gmail, QQ Mail, NetEase)" readme = "README.md" license = {text = "MIT"} requires-python = ">=3.10" authors = [ {name = "Ryan Chan", email = "ryanchan720@qq.com"} ] keywords = ["email", "gmail", "qq-mail", "imap", "smtp", "verification-code", "notification"] classifiers = [ "Development Status :: 4 - Beta", "Environment :: Console", "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 :: Communications :: Email", ] dependencies = [ "click>=8.0", "pydantic>=2.0", "google-api-python-client>=2.0", "google-auth>=2.0", "google-auth-oauthlib>=1.0", "imaplib2>=3.6", ] [project.optional-dependencies] dev = [ "pytest>=7.0", "pytest-cov>=4.0", ] [project.scripts] email-bridge = "email_bridge.cli:main" [project.urls] Homepage = "https://github.com/ryanchan720/email-bridge" Repository = "https://github.com/ryanchan720/email-bridge" Issues = "https://github.com/ryanchan720/email-bridge/issues" [build-system] requires = ["hatchling"] build-backend = "hatchling.build" [tool.hatch.build.targets.wheel] packages = ["email_bridge"] FILE:references/gmail-setup.md # Gmail 配置指南 > ⚠️ **注意**：Gmail API 配置流程较复杂，需要 Google Cloud 项目和 OAuth 授权。 > 如果你只需要基本的收发邮件功能，建议使用 QQ 邮箱或网易邮箱（授权码配置更简单）。 ## 前置条件 - 一个 Google 账号（Gmail 邮箱） - 能访问 Google Cloud Console（可能需要科学上网） --- ## Step 1：创建 Google Cloud 项目 1. 打开 [Google Cloud Console](https://console.cloud.google.com/) 2. 登录你的 Google 账号 3. 页面顶部点击项目选择器，点击 **"NEW PROJECT"** 4. 项目名称随便填（如 `email-bridge`），点击 **"CREATE"** 5. 创建完成后选中该项目 --- ## Step 2：启用 Gmail API 1. 打开 [Gmail API 页面](https://console.cloud.google.com/apis/library/gmail.googleapis.com) 2. 确保顶部显示的是刚创建的项目 3. 点击 **"ENABLE"** --- ## Step 3：配置 OAuth 同意屏幕 1. 打开 [OAuth 同意屏幕配置](https://console.cloud.google.com/apis/credentials/consent) 2. 用户类型选择 **"External"**，点击 **"CREATE"** ### 填写配置 | 字段 | 填写内容 | |------|----------| | App name | `Email Bridge` 或任意名称 | | User support email | 选择你的邮箱 | | Developer contact | 填你的邮箱 | | 其他字段 | 可留空 | 点击 **"SAVE AND CONTINUE"**。 ### Scopes（权限范围） 1. 点击 **"ADD OR REMOVE SCOPES"** 2. 搜索 `gmail.modify`，勾选 `https://www.googleapis.com/auth/gmail.modify` 3. 点击 **"UPDATE"**，然后 **"SAVE AND CONTINUE"** > 💡 `gmail.modify` 包含读取和修改权限，以后发邮件也能用。 ### Test Users（测试用户） 1. 点击 **"ADD USERS"** 2. 填写你的 Gmail 地址 3. 点击 **"ADD"**，然后 **"SAVE AND CONTINUE"** 最后检查无误，点击 **"BACK TO DASHBOARD"**。 --- ## Step 4：创建 OAuth 客户端凭证 1. 打开 [Credentials 页面](https://console.cloud.google.com/apis/credentials) 2. 点击 **"CREATE CREDENTIALS"** → **"OAuth client ID"** 3. Application type 选择 **"Desktop app"** 4. Name 随便填（如 `Email Bridge CLI`） 5. 点击 **"CREATE"** 6. 在弹出窗口点击 **"DOWNLOAD JSON"** 下载凭证文件 --- ## Step 5：安装凭证文件 ```bash # 创建目录 mkdir -p ~/.email-bridge/gmail # 将下载的凭证文件移动过去，重命名为 credentials.json mv ~/Downloads/client_secret_xxx.json ~/.email-bridge/gmail/credentials.json ``` --- ## Step 6：添加账户并授权 ```bash # 添加 Gmail 账户 email-bridge accounts add your@gmail.com --provider gmail --name "Personal Gmail" # 首次同步，会打开浏览器要求授权 email-bridge sync ``` 授权完成后，token 会自动保存在 `~/.email-bridge/gmail/token_*.json`，后续同步无需再次授权。 --- ## 配置选项 ```bash # 自定义同步范围：最近 3 天，最多 50 封 email-bridge accounts add user@gmail.com -p gmail \ --config '{"sync_days": 3, "sync_max_messages": 50}' ``` | 配置项 | 说明 | 默认值 | |--------|------|--------| | `sync_days` | 同步最近 N 天的邮件 | 7 | | `sync_max_messages` | 每次同步最大邮件数 | 100 | | `credentials_path` | 自定义凭证文件路径 | `~/.email-bridge/gmail/credentials.json` | | `token_path` | 自定义 token 存储路径 | 自动生成 | --- ## 发送邮件 Gmail SMTP 发送需要**应用专用密码**（与 OAuth token 不同）： 1. 开启 Google 账户的两步验证 2. 访问 [应用专用密码](https://myaccount.google.com/apppasswords) 3. 生成一个新密码，选择"邮件"和"其他设备" 4. 更新账户配置： ```bash email-bridge accounts update --config '{"password": "YOUR_APP_PASSWORD"}' ``` FILE:scripts/email-bridge.sh #!/bin/bash # Email Bridge CLI wrapper # This script ensures email-bridge is installed and forwards commands set -e SKILL_DIR="$(cd "$(dirname "$0")/.." && pwd)" VENV_DIR="$SKILL_DIR/.venv" # Check if installed if [ ! -d "$VENV_DIR" ]; then echo "📧 Email Bridge 未安装，正在安装..." cd "$SKILL_DIR" if command -v uv &> /dev/null; then uv sync else python3 -m venv .venv source .venv/bin/activate pip install -e . fi echo "✅ 安装完成" fi # Activate venv and run command source "$VENV_DIR/bin/activate" email-bridge "$@" FILE:scripts/email_bridge.py #!/usr/bin/env python3 """ Email Bridge skill installer and entry point. When first called, installs the package if needed, then forwards commands. """ import subprocess import sys from pathlib import Path SKILL_DIR = Path(__file__).parent.parent VENV_DIR = SKILL_DIR / ".venv" EMAIL_BRIDGE_CMD = "email-bridge" def is_installed() -> bool: """Check if email-bridge is available.""" try: subprocess.run( [EMAIL_BRIDGE_CMD, "--version"], capture_output=True, check=True ) return True except (subprocess.CalledProcessError, FileNotFoundError): return False def install_package(): """Install email-bridge package.""" print("📧 正在安装 Email Bridge...") # Check Python version if sys.version_info < (3, 10): print("❌ 需要 Python 3.10 或更高版本") sys.exit(1) # Install using pip subprocess.run( [sys.executable, "-m", "pip", "install", "-e", str(SKILL_DIR)], check=True ) print("✅ Email Bridge 安装完成") def main(): """Main entry point.""" # Check if installed if not is_installed(): install_package() # Forward command to email-bridge cmd = [EMAIL_BRIDGE_CMD] + sys.argv[1:] result = subprocess.run(cmd) sys.exit(result.returncode) if __name__ == "__main__": main() FILE:test_installation.py #!/usr/bin/env python3 """ Email Bridge Test Script Run this to verify your Email Bridge installation. """ import sys from pathlib import Path # Add parent to path sys.path.insert(0, str(Path(__file__).parent)) from email_bridge.service import EmailBridgeService from email_bridge.models import EmailProvider def test_database(): """Test database initialization.""" print("📦 Testing database...") service = EmailBridgeService() accounts = service.list_accounts() print(f" ✓ Database initialized ({len(accounts)} accounts)") return service def test_mock_adapter(): """Test mock adapter.""" print("\n🎭 Testing mock adapter...") service = EmailBridgeService() # Check if mock account exists accounts = service.list_accounts() mock_account = next((a for a in accounts if a.provider == EmailProvider.MOCK), None) if not mock_account: # Add mock account mock_account = service.add_account( email="demo@example.com", provider=EmailProvider.MOCK, display_name="Demo Account" ) print(f" ✓ Created mock account: {mock_account.email}") # Sync messages count = service.sync_account(mock_account.id) print(f" ✓ Synced {count} messages from mock adapter") # List messages messages = service.list_recent_messages(account_id=mock_account.id, limit=5) print(f" ✓ Retrieved {len(messages)} messages") return True def test_verification_extraction(): """Test verification code extraction.""" print("\n🔢 Testing verification code extraction...") from email_bridge.extraction import extract_from_email # Test cases test_cases = [ ("验证码: 123456", "您的验证码是 123456，5分钟内有效"), ("Your code is: 789012", "Your verification code: 789012"), ("验证码：ABC123", "短信验证码 ABC123"), ] for subject, body in test_cases: result = extract_from_email(subject=subject, body_text=body) if result['codes']: print(f" ✓ Extracted: {result['codes'][0].code} from '{subject}'") else: print(f" ✗ Failed to extract from '{subject}'") return False return True def test_accounts_command(): """Test accounts list command.""" print("\n👥 Testing accounts command...") service = EmailBridgeService() accounts = service.list_accounts() print(f" ✓ Found {len(accounts)} account(s):") for acc in accounts: print(f" - {acc.email} ({acc.provider.value})") return True def main(): print("=" * 50) print("📧 Email Bridge Test Suite") print("=" * 50) tests = [ ("Database", test_database), ("Mock Adapter", test_mock_adapter), ("Verification Extraction", test_verification_extraction), ("Accounts Command", test_accounts_command), ] results = [] for name, test_func in tests: try: result = test_func() results.append((name, result, None)) except Exception as e: results.append((name, False, str(e))) print("\n" + "=" * 50) print("📊 Test Results") print("=" * 50) passed = sum(1 for _, r, _ in results if r) total = len(results) for name, result, error in results: status = "✅ PASS" if result else "❌ FAIL" print(f" {status}: {name}") if error: print(f" Error: {error}") print(f"\n总计: {passed}/{total} 通过") if passed == total: print("\n🎉 所有测试通过！Email Bridge 工作正常。") return 0 else: print("\n⚠️ 部分测试失败，请检查上述错误。") return 1 if __name__ == "__main__": sys.exit(main()) FILE:tests/__init__.py FILE:tests/test_basic.py """Basic tests for email bridge.""" import tempfile from pathlib import Path import pytest from email_bridge.models import Account, EmailProvider, AccountStatus, Message, EmailCategory from email_bridge.db import Database from email_bridge.service import EmailBridgeService from email_bridge.categories import detect_category, EmailCategory as Cat class TestModels: def test_account_creation(self): account = Account( id="test1", email="test@example.com", provider=EmailProvider.MOCK, ) assert account.id == "test1" assert account.email == "test@example.com" assert account.status == AccountStatus.ACTIVE def test_message_creation(self): from datetime import datetime msg = Message( id="acc:msg1", account_id="acc", message_id="msg1", subject="Test Subject", sender="sender@example.com", recipients=["recv@example.com"], received_at=datetime.utcnow(), ) assert msg.subject == "Test Subject" assert msg.is_read is False class TestDatabase: def test_database_operations(self, tmp_path): db = Database(db_path=tmp_path / "test.db") # Add account account = Account( id="test1", email="test@example.com", provider=EmailProvider.MOCK, ) db.add_account(account) # List accounts accounts = db.list_accounts() assert len(accounts) == 1 assert accounts[0].email == "test@example.com" # Get account acc = db.get_account("test1") assert acc is not None assert acc.email == "test@example.com" db.close() def test_message_operations(self, tmp_path): from datetime import datetime db = Database(db_path=tmp_path / "test.db") # Add account first account = Account(id="acc1", email="test@example.com", provider=EmailProvider.MOCK) db.add_account(account) # Add message msg = Message( id="acc1:msg1", account_id="acc1", message_id="msg1", subject="Test Subject", sender="sender@example.com", recipients=["test@example.com"], received_at=datetime.utcnow(), category=EmailCategory.NORMAL, ) db.add_message(msg) # List messages msgs = db.list_messages() assert len(msgs) == 1 assert msgs[0].subject == "Test Subject" # Find message found = db.find_message("msg1") assert found is not None assert found.subject == "Test Subject" db.close() class TestCategoryDetection: def test_verification_category(self): cat = detect_category("Your verification code is 123456", "") assert cat == Cat.VERIFICATION def test_security_category(self): cat = detect_category("Security Alert: New login", "") assert cat == Cat.SECURITY def test_subscription_category(self): cat = detect_category("Weekly Newsletter", "Click to unsubscribe") assert cat == Cat.SUBSCRIPTION def test_spam_category(self): cat = detect_category("CONGRATULATIONS! You've WON!!!", "") assert cat == Cat.SPAM_LIKE def test_normal_category(self): cat = detect_category("Re: Project Update Meeting", "Hi, thanks for the update") assert cat == Cat.NORMAL class TestService: def test_add_account(self, tmp_path): from email_bridge.db import Database db = Database(db_path=tmp_path / "test.db") service = EmailBridgeService(db=db) account = service.add_account( email="test@example.com", provider=EmailProvider.MOCK, display_name="Test Account" ) assert account.email == "test@example.com" assert account.provider == EmailProvider.MOCK db.close() def test_stats(self, tmp_path): from email_bridge.db import Database db = Database(db_path=tmp_path / "test.db") service = EmailBridgeService(db=db) stats = service.get_stats() assert stats["total_accounts"] == 0 assert stats["unread_messages"] == 0 db.close() FILE:tests/test_extraction.py """Tests for extraction module.""" import pytest from email_bridge.extraction import ( extract_verification_codes, extract_action_links, extract_from_email, VerificationCode, ActionLink, ) class TestVerificationCodeExtraction: def test_numeric_code_with_context(self): text = "Your verification code is 123456. Enter this code to verify your account." codes = extract_verification_codes(text) assert len(codes) >= 1 assert any(c.code == "123456" for c in codes) def test_chinese_verification_code(self): text = "您的验证码是：888999。请在5分钟内输入。" codes = extract_verification_codes(text) assert len(codes) >= 1 assert any(c.code == "888999" for c in codes) def test_alphanumeric_code(self): text = "Your security code is: ABC123XY" codes = extract_verification_codes(text) # Should find the alphanumeric code assert len(codes) >= 1 found = any("ABC" in c.code.upper() or "XY" in c.code.upper() for c in codes) assert found def test_dashed_code(self): text = "Enter confirmation code AB12-CD34 to continue" codes = extract_verification_codes(text) assert len(codes) >= 1 assert any("-" in c.code for c in codes) def test_empty_text(self): codes = extract_verification_codes("") assert codes == [] codes = extract_verification_codes(None) assert codes == [] def test_filters_out_years(self): text = "In 2024 we launched our product. Your code is 1234." codes = extract_verification_codes(text) # Should find 1234 but not 2024 (it's a year) assert any(c.code == "1234" for c in codes) assert not any(c.code == "2024" for c in codes) def test_otp_keyword(self): text = "OTP: 998877" codes = extract_verification_codes(text) assert len(codes) >= 1 assert any(c.code == "998877" for c in codes) class TestVerificationCodeFalsePositives: """Tests for filtering out false positives.""" def test_filters_order_numbers(self): """Order/invoice numbers should not be treated as verification codes.""" text = "Your order #123456 has been shipped. Track your package." codes = extract_verification_codes(text) # Should not match 123456 as a verification code assert not any(c.code == "123456" for c in codes) def test_filters_invoice_numbers(self): """Invoice numbers should not be treated as verification codes.""" text = "Invoice INV-123456 is due on March 30th." codes = extract_verification_codes(text) # INV-123456 should not be extracted as a verification code assert not any("123456" in c.code for c in codes) def test_filters_phone_numbers(self): """Phone numbers should not be treated as verification codes.""" text = "Call us at 1234567890 for support." codes = extract_verification_codes(text) # 10-digit number should not be extracted assert not any(c.code == "1234567890" for c in codes) def test_filters_tracking_numbers(self): """Long tracking numbers should not be treated as verification codes.""" text = "Tracking number: 123456789012345" codes = extract_verification_codes(text) # 15-digit number is too long for a verification code assert not any("123456789012345" in c.code for c in codes) def test_filters_credit_card_like_numbers(self): """Credit card-like patterns should not be treated as verification codes.""" text = "Card ending in 123456789012" codes = extract_verification_codes(text) # 12-digit number is too long assert not any("123456789012" in c.code for c in codes) def test_real_verification_code_still_works(self): """Real verification codes should still be extracted.""" text = "Your verification code is 123456. Do not share it." codes = extract_verification_codes(text) assert any(c.code == "123456" for c in codes) def test_real_code_with_context_still_works(self): """Codes with verification context should still be extracted.""" text = "GitHub: Your security code is 789012" codes = extract_verification_codes(text) assert any(c.code == "789012" for c in codes) def test_filters_price_like_numbers(self): """Numbers that look like prices should not be codes.""" # Note: This is tricky - 1999 could be a code or a price # But without context, we should be conservative text = "Your total is 1999 dollars." codes = extract_verification_codes(text) # Without verification context, should not extract assert not any(c.code == "1999" for c in codes) def test_filters_plain_numbers_without_context(self): """Plain numbers without verification context should not be extracted.""" text = "The meeting is at 1234 Main Street, room 5678." codes = extract_verification_codes(text) # These are not verification codes assert not any(c.code == "1234" for c in codes) assert not any(c.code == "5678" for c in codes) class TestActionLinkExtraction: def test_verify_link_from_text(self): text = "Click here to verify your email: https://example.com/verify?token=abc123" links = extract_action_links(text) assert len(links) >= 1 verify_link = next((l for l in links if l.link_type == "verify"), None) assert verify_link is not None assert "example.com" in verify_link.url def test_reset_link_from_html(self): html = """

Click below to reset your password:

Reset Password """ links = extract_action_links("", html) reset_link = next((l for l in links if l.link_type == "reset"), None) assert reset_link is not None assert reset_link.text == "Reset Password" def test_unsubscribe_link(self): text = "To stop receiving emails, visit https://example.com/unsubscribe" links = extract_action_links(text) unsub_link = next((l for l in links if l.link_type == "unsubscribe"), None) assert unsub_link is not None def test_multiple_links(self): html = """ Verify Email Reset Password Visit Homepage """ links = extract_action_links("", html) # Should have verify and reset links (not homepage - it's not an action) action_types = [l.link_type for l in links] assert "verify" in action_types or "reset" in action_types def test_empty_content(self): links = extract_action_links("") # May have some links found, but no action links classified assert all(l.link_type == "unknown" for l in links) or len(links) == 0 class TestExtractFromEmail: def test_full_extraction(self): subject = "Your verification code" body_text = """ Your verification code is 123456. Click here to verify: https://example.com/verify """ body_html = """ Verify Now """ result = extract_from_email(subject, body_text, body_html) assert "codes" in result assert "links" in result assert len(result["codes"]) >= 1 assert len(result["links"]) >= 1 def test_chinese_email(self): subject = "您的验证码" body_text = "您的验证码是：666888，请在10分钟内使用。" result = extract_from_email(subject, body_text) assert len(result["codes"]) >= 1 assert any(c.code == "666888" for c in result["codes"]) class TestVerificationCodeDataclass: def test_code_creation(self): code = VerificationCode(code="123456", context="GitHub") assert code.code == "123456" assert code.context == "GitHub" def test_optional_context(self): code = VerificationCode(code="123456") assert code.context is None class TestActionLinkDataclass: def test_link_creation(self): link = ActionLink( url="https://example.com/verify", link_type="verify", domain="example.com", text="Verify Email" ) assert link.url == "https://example.com/verify" assert link.link_type == "verify" assert link.domain == "example.com" assert link.text == "Verify Email" FILE:tests/test_gmail.py """Tests for Gmail adapter.""" import pytest from datetime import datetime from pathlib import Path from unittest.mock import Mock, patch, MagicMock from email_bridge.models import Account, EmailProvider from email_bridge.adapters.gmail import ( GmailAdapter, GmailAdapterError, GmailCredentialsNotFoundError, GmailAuthError, ) from email_bridge.adapters.base import FetchOptions, RawMessage class TestGmailAdapter: def test_provider_property(self): adapter = GmailAdapter() assert adapter.provider == EmailProvider.GMAIL def test_default_credentials_dir(self): adapter = GmailAdapter() expected = Path.home() / ".email-bridge" / "gmail" assert adapter.credentials_dir == expected def test_custom_credentials_dir(self): custom_dir = Path("/tmp/test-gmail") adapter = GmailAdapter(credentials_dir=custom_dir) assert adapter.credentials_dir == custom_dir def test_get_credentials_path_from_config(self): adapter = GmailAdapter() account = Account( id="test1", email="test@gmail.com", provider=EmailProvider.GMAIL, config={"credentials_path": "/custom/creds.json"} ) path = adapter._get_credentials_path(account) assert path == Path("/custom/creds.json") def test_get_credentials_path_default(self): adapter = GmailAdapter() account = Account( id="test1", email="test@gmail.com", provider=EmailProvider.GMAIL, ) path = adapter._get_credentials_path(account) assert path == adapter.credentials_dir / "credentials.json" def test_get_token_path_from_config(self): adapter = GmailAdapter() account = Account( id="test1", email="test@gmail.com", provider=EmailProvider.GMAIL, config={"token_path": "/custom/token.json"} ) path = adapter._get_token_path(account) assert path == Path("/custom/token.json") def test_get_token_path_default(self): adapter = GmailAdapter() account = Account( id="test1", email="test.user@gmail.com", provider=EmailProvider.GMAIL, ) path = adapter._get_token_path(account) # Email should be sanitized for filename assert "test_user_at_gmail_com" in str(path) def test_credentials_not_found_error(self): adapter = GmailAdapter(credentials_dir=Path("/nonexistent")) account = Account( id="test1", email="test@gmail.com", provider=EmailProvider.GMAIL, ) with pytest.raises(GmailCredentialsNotFoundError): adapter._get_credentials(account) def test_parse_message(self): adapter = GmailAdapter() # Mock Gmail API message data msg_data = { "id": "msg123", "threadId": "thread123", "labelIds": ["INBOX"], "snippet": "This is a test...", "payload": { "headers": [ {"name": "Subject", "value": "Test Subject"}, {"name": "From", "value": "Sender "}, {"name": "To", "value": "recipient@example.com"}, {"name": "Date", "value": "Mon, 20 Mar 2026 10:00:00 +0000"}, ], "mimeType": "text/plain", "body": { "data": "VGhpcyBpcyB0ZXN0IGJvZHk=" # base64 encoded "This is test body" } } } raw_msg = adapter._parse_message(msg_data) assert raw_msg.message_id == "msg123" assert raw_msg.subject == "Test Subject" assert raw_msg.sender == "sender@example.com" assert raw_msg.sender_name == "Sender" assert raw_msg.is_read is True # No UNREAD label assert raw_msg.body_text == "This is test body" def test_parse_message_unread(self): adapter = GmailAdapter() msg_data = { "id": "msg123", "threadId": "thread123", "labelIds": ["INBOX", "UNREAD"], "payload": { "headers": [ {"name": "Subject", "value": "Test"}, {"name": "From", "value": "sender@example.com"}, {"name": "Date", "value": "Mon, 20 Mar 2026 10:00:00 +0000"}, ], } } raw_msg = adapter._parse_message(msg_data) assert raw_msg.is_read is False def test_parse_message_with_name_parsing(self): adapter = GmailAdapter() msg_data = { "id": "msg123", "threadId": "thread123", "payload": { "headers": [ {"name": "Subject", "value": "Test"}, {"name": "From", "value": '"John Doe" '}, {"name": "Date", "value": "Mon, 20 Mar 2026 10:00:00 +0000"}, ], } } raw_msg = adapter._parse_message(msg_data) assert raw_msg.sender_name == "John Doe" assert raw_msg.sender == "john@example.com" def test_sync_config_from_account(self): adapter = GmailAdapter() # Test that account config is used for sync parameters account = Account( id="test1", email="test@gmail.com", provider=EmailProvider.GMAIL, config={ "sync_days": 3, "sync_max_messages": 50 } ) # We can't test actual fetch without mock service, but we can verify # the config is accessible assert account.config.get("sync_days") == 3 assert account.config.get("sync_max_messages") == 50 class TestGmailAdapterErrors: def test_error_hierarchy(self): # Test that errors are properly inherited assert issubclass(GmailCredentialsNotFoundError, GmailAdapterError) assert issubclass(GmailAuthError, GmailAdapterError) assert issubclass(GmailAdapterError, Exception) def test_credentials_error_message(self): error = GmailCredentialsNotFoundError("Credentials not found") assert "Credentials not found" in str(error) def test_auth_error_message(self): error = GmailAuthError("Auth failed") assert "Auth failed" in str(error)

Email Bridge

Comments (0)