admin/OpenClaw-DeepReeder
1
0
Fork 0
mirror of https://github.com/astonysh/OpenClaw-DeepReeder.git synced 2026-08-14 00:57:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

🦞 Initial release: OpenClaw DeepReeder v1.0.0

Browse Source 
Autonomous web content ingestion engine for AI agents.

Features:
- Generic article/blog parser (Trafilatura + BeautifulSoup fallback)
- Twitter/X parser (via Nitter instances)
- YouTube transcript parser
- Clean Markdown output with YAML frontmatter
- Automatic URL detection and routing
- Structured memory storage
This commit is contained in:
Tony Li
2026-02-16 13:58:54 +01:00
commit d59ae38920
16 changed files with 1700 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+41
View File
@@ -0,0 +1,41 @@
# Python
__pycache__/
*.py[cod]
*$py.class
*.so
*.egg-info/
*.egg
dist/
build/
*.whl
# Virtual environments
.venv/
venv/
env/
# IDE
.idea/
.vscode/
*.swp
*.swo
*~
# OS
.DS_Store
Thumbs.db
# Environment variables
.env
.env.local
# Logs
*.log
# Testing
.pytest_cache/
.coverage
htmlcov/
# Distribution
*.tar.gz
LICENSE
+21
View File
@@ -0,0 +1,21 @@
MIT License
Copyright (c) 2026 OpenClaw
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
README.md
+156
View File
@@ -0,0 +1,156 @@
# 🦞 OpenClaw DeepReeder
> **Autonomous web content ingestion engine for AI agents.**
DeepReeder intercepts URLs from user messages, scrapes content intelligently using specialized parsers, formats it into clean Markdown with YAML frontmatter, and saves it to the agent's long-term memory.
---
## ✨ Features
| Parser | Sources | Method |
|--------|---------|--------|
| 🌐 **Generic** | Blogs, articles, docs | [Trafilatura](https://trafilatura.readthedocs.io/) with BeautifulSoup fallback |
| 🐦 **Twitter / X** | Tweets & threads | Nitter instance proxying |
| 🎬 **YouTube** | Video transcripts | [youtube-transcript-api](https://github.com/jdepoix/youtube-transcript-api) |
### Output Format
Every piece of content is saved as a `.md` file with structured YAML frontmatter:
```yaml
---
title: "Article Title"
source_url: "https://example.com/article"
domain: "example.com"
parser: "generic"
ingested_at: "2026-02-16T12:00:00Z"
content_hash: "sha256:abc123..."
word_count: 1500
---
# Article Title
The clean, extracted content goes here...
```
---
## 📦 Installation
```bash
# Clone the repository
git clone https://github.com/astonysh/OpenClaw-DeepReeder.git
cd OpenClaw-DeepReeder
# Create a virtual environment
python3 -m venv .venv
source .venv/bin/activate
# Install dependencies
pip install -e .
```
---
## 🚀 Quick Start
```python
from deepreader_skill import run
# Process a single URL
result = run("Check out this article: https://example.com/blog/post")
print(result)
# Process multiple URLs at once
result = run("""
Here are some links:
https://example.com/article
https://youtube.com/watch?v=dQw4w9WgXcQ
https://x.com/user/status/123456
""")
print(result)
```
### Example Output
```
📚 DeepReader — Processed 3 URL(s):
✅ How to Build AI Agents
Source: https://example.com/article
Saved to: memory/inbox/20260216_120000_how-to-build-ai-agents.md
Content: 3200 characters
✅ Rick Astley - Never Gonna Give You Up
Source: https://youtube.com/watch?v=dQw4w9WgXcQ
Saved to: memory/inbox/20260216_120001_rick-astley-never-gonna.md
Content: 15000 characters
✅ @user's tweet
Source: https://x.com/user/status/123456
Saved to: memory/inbox/20260216_120002_user-tweet.md
Content: 280 characters
```
---
## 🏗️ Architecture
```
deepreader_skill/
├── __init__.py # Entry point — run() function
├── manifest.json # Skill metadata & trigger config
├── requirements.txt # Dependencies
├── core/
│ ├── router.py # URL → Parser routing logic
│ ├── storage.py # Markdown file generation & saving
│ └── utils.py # URL extraction & helper utilities
└── parsers/
├── base.py # Abstract base parser & ParseResult model
├── generic.py # Generic article/blog parser
├── twitter.py # Twitter/X parser (via Nitter)
└── youtube.py # YouTube transcript parser
```
---
## 🔧 Configuration
DeepReeder uses sensible defaults out of the box. Configuration can be customized via environment variables:
| Variable | Default | Description |
|----------|---------|-------------|
| `DEEPREEDER_MEMORY_PATH` | `../../memory/inbox/` | Where to save ingested content |
| `DEEPREEDER_LOG_LEVEL` | `INFO` | Logging verbosity |
---
## 🤝 Contributing
Contributions are welcome! Feel free to:
1. Fork the repository
2. Create a feature branch (`git checkout -b feature/amazing-parser`)
3. Commit your changes (`git commit -m 'Add amazing parser'`)
4. Push to the branch (`git push origin feature/amazing-parser`)
5. Open a Pull Request
---
## 📄 License
This project is licensed under the **MIT License** — see the [LICENSE](LICENSE) file for details.
---
## 🔗 Links
- **Repository**: [github.com/astonysh/OpenClaw-DeepReeder](https://github.com/astonysh/OpenClaw-DeepReeder)
- **Issues**: [github.com/astonysh/OpenClaw-DeepReeder/issues](https://github.com/astonysh/OpenClaw-DeepReeder/issues)
---
<p align="center">
Built with 🦞 by <a href="https://github.com/astonysh">OpenClaw</a>
</p>
deepreader_skill/__init__.py
+158
View File
@@ -0,0 +1,158 @@
"""
DeepReader Skill for OpenClaw
==============================
An autonomous web content ingestion engine that:
1. Extracts URLs from user messages
2. Routes each URL to the appropriate specialized parser
3. Saves clean Markdown with YAML frontmatter to the agent's memory
Usage (from OpenClaw)::
from deepreader_skill import run
response = run("Check out this article: https://example.com/blog/post")
# → Scrapes the article, saves to memory/inbox/, returns confirmation.
Supported URL types:
- **Generic** (blogs, articles, docs) → via trafilatura
- **Twitter / X** → via Nitter instances
- **YouTube** → via youtube_transcript_api
"""
from __future__ import annotations
import logging
from typing import Any
from .core.router import ParserRouter
from .core.storage import StorageManager
from .core.utils import extract_urls
__version__ = "1.0.0"
__all__ = ["run"]
logger = logging.getLogger("deepreader")
# Configure logging if not already configured
if not logger.handlers:
handler = logging.StreamHandler()
handler.setFormatter(
logging.Formatter(
"%(asctime)s [%(name)s] %(levelname)s: %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
)
logger.addHandler(handler)
logger.setLevel(logging.INFO)
# ---------------------------------------------------------------------------
# Singleton instances (lazy init)
# ---------------------------------------------------------------------------
_router: ParserRouter | None = None
_storage: StorageManager | None = None
def _get_router() -> ParserRouter:
global _router
if _router is None:
_router = ParserRouter()
return _router
def _get_storage() -> StorageManager:
global _storage
if _storage is None:
_storage = StorageManager()
return _storage
# ---------------------------------------------------------------------------
# Public API — OpenClaw Entry Point
# ---------------------------------------------------------------------------
def run(text: str, **kwargs: Any) -> str:
"""Main entry point for the DeepReader skill.
Called by OpenClaw when a user message potentially contains URLs.
Args:
text: The raw user message (may contain one or more URLs).
kwargs: Reserved for future OpenClaw context (e.g. user_id,
chat_id, config overrides).
Returns:
A human-readable status message summarizing what was processed.
On failure, returns a graceful error description — never raises.
"""
try:
# Step 1: Extract URLs from the message
urls = extract_urls(text)
if not urls:
return "🔍 No URL detected in your message."
router = _get_router()
storage = _get_storage()
results: list[str] = []
errors: list[str] = []
for url in urls:
logger.info("Processing URL: %s", url)
# Step 2: Route to the correct parser
parse_result = router.route(url)
if not parse_result.success:
error_msg = (
f"❌ Failed to read **{url}**\n"
f" Reason: {parse_result.error}"
)
errors.append(error_msg)
logger.warning("Parse failed for %s: %s", url, parse_result.error)
continue
# Step 3: Save to memory
try:
filepath = storage.save(parse_result)
success_msg = (
f"✅ **{parse_result.title or 'Untitled'}**\n"
f" Source: {url}\n"
f" Saved to: `{filepath}`\n"
f" Content: {len(parse_result.content)} characters"
)
results.append(success_msg)
logger.info("Successfully saved %s", filepath)
except OSError as exc:
error_msg = (
f"❌ Parsed **{url}** but failed to save.\n"
f" Error: {exc}"
)
errors.append(error_msg)
logger.error("Storage error for %s: %s", url, exc)
# Step 4: Build the response
response_parts: list[str] = []
if results:
response_parts.append(f"📚 **DeepReader** — Processed {len(results)} URL(s):\n")
response_parts.extend(results)
if errors:
if results:
response_parts.append("\n---\n")
response_parts.append(f"⚠️ {len(errors)} URL(s) had issues:\n")
response_parts.extend(errors)
return "\n\n".join(response_parts)
except Exception as exc: # noqa: BLE001
logger.exception("DeepReader encountered an unexpected error")
return (
f"🚨 DeepReader encountered an unexpected error: {exc}\n"
"The agent remains operational. Please try again or check the logs."
)
deepreader_skill/core/__init__.py
+1
View File
@@ -0,0 +1 @@
# DeepReader Skill - Core Module
deepreader_skill/core/router.py
+113
View File
@@ -0,0 +1,113 @@
"""
DeepReader Skill - Parser Router
==================================
Decides which parser to use based on URL domain/pattern analysis.
Follows the **Strategy Pattern** — parsers are registered and
selected dynamically.
"""
from __future__ import annotations
import logging
from typing import Sequence
from ..core.utils import is_twitter_url, is_youtube_url
from ..parsers.base import BaseParser, ParseResult
from ..parsers.generic import GenericParser
from ..parsers.twitter import TwitterParser
from ..parsers.youtube import YouTubeParser
logger = logging.getLogger("deepreader.router")
class ParserRouter:
"""Select and execute the appropriate parser for a given URL.
The router maintains an ordered list of specialized parsers. For
each URL, it iterates through the list and uses the first parser
whose :meth:`can_handle` returns ``True``. If no specialized
parser matches, the :class:`GenericParser` is used as a fallback.
Usage::
router = ParserRouter()
result = router.route("https://twitter.com/user/status/123")
# → uses TwitterParser
result = router.route("https://example.com/blog/post")
# → uses GenericParser
"""
def __init__(
self,
extra_parsers: Sequence[BaseParser] | None = None,
) -> None:
"""Initialize with default + optional extra parsers.
Args:
extra_parsers: Additional parser instances to register.
They are checked **before** the built-in
parsers, allowing user overrides.
"""
# Specialized parsers (checked in order)
self._parsers: list[BaseParser] = []
# Register user-supplied parsers first (highest priority)
if extra_parsers:
self._parsers.extend(extra_parsers)
# Built-in specialized parsers
self._parsers.extend([
TwitterParser(),
YouTubeParser(),
])
# Fallback parser (always matches)
self._fallback = GenericParser()
logger.info(
"ParserRouter initialized with %d specialized parsers + generic fallback",
len(self._parsers),
)
def route(self, url: str) -> ParseResult:
"""Determine the correct parser and execute it.
Args:
url: The URL to parse.
Returns:
A :class:`ParseResult` with the extracted content or an
error description.
"""
# Check specialized parsers first
for parser in self._parsers:
if parser.can_handle(url):
logger.info("Routing %s → %s parser", url, parser.name)
return parser.parse(url)
# Fallback to generic parser
logger.info("Routing %s → generic parser (fallback)", url)
return self._fallback.parse(url)
def register_parser(self, parser: BaseParser, priority: bool = False) -> None:
"""Register a new parser at runtime.
Args:
parser: The parser instance to add.
priority: If ``True``, insert at the beginning of the list
(highest priority). Otherwise append.
"""
if priority:
self._parsers.insert(0, parser)
else:
# Insert before the fallback position
self._parsers.append(parser)
logger.info("Registered new parser: %s (priority=%s)", parser.name, priority)
@property
def available_parsers(self) -> list[str]:
"""Return names of all registered parsers (including fallback)."""
names = [p.name for p in self._parsers]
names.append(self._fallback.name)
return names
deepreader_skill/core/storage.py
+179
View File
@@ -0,0 +1,179 @@
"""
DeepReader Skill - Storage Manager
====================================
Responsible for formatting extracted content into Markdown with
YAML frontmatter and persisting it to the agent's memory directory.
"""
from __future__ import annotations
import logging
import os
import uuid
from datetime import datetime, timezone
from pathlib import Path
from ..core.utils import build_filename, content_hash, generate_excerpt, get_domain, get_domain_tag
from ..parsers.base import ParseResult
logger = logging.getLogger("deepreader.storage")
class StorageManager:
"""Persist parsed content to the agent's long-term memory.
By default, files are saved to ``../../memory/inbox/`` relative to
the skill directory. This can be overridden via the constructor.
File naming convention::
YYYY-MM-DD_{sanitized_title}.md
"""
def __init__(self, memory_dir: str | Path | None = None) -> None:
"""Initialize the storage manager.
Args:
memory_dir: Absolute or relative path to the memory inbox
directory. Defaults to ``../../memory/inbox/``
relative to this file.
"""
if memory_dir is None:
# Default: ../../memory/inbox/ relative to the skill package
skill_root = Path(__file__).resolve().parent.parent
self._memory_dir = skill_root.parent.parent / "memory" / "inbox"
else:
self._memory_dir = Path(memory_dir)
logger.info("StorageManager target directory: %s", self._memory_dir)
def save(self, result: ParseResult) -> str:
"""Format and save a :class:`ParseResult` to the memory directory.
Args:
result: The successfully parsed content.
Returns:
The absolute path of the saved ``.md`` file.
Raises:
OSError: If the file cannot be written.
"""
# Ensure directory exists
self._memory_dir.mkdir(parents=True, exist_ok=True)
# Build filename
filename = build_filename(result.title, result.url)
filepath = self._memory_dir / filename
# Handle duplicate filenames
if filepath.exists():
stem = filepath.stem
suffix = filepath.suffix
counter = 1
while filepath.exists():
filepath = self._memory_dir / f"{stem}_{counter}{suffix}"
counter += 1
# Generate the Markdown content
markdown = self._format_markdown(result)
# Write to disk
filepath.write_text(markdown, encoding="utf-8")
logger.info("Saved content to %s (%d bytes)", filepath, len(markdown))
return str(filepath)
def _format_markdown(self, result: ParseResult) -> str:
"""Build the full Markdown document with YAML frontmatter.
Output format::
---
uuid: {uuid4}
source: {url}
date: {iso_date}
type: external_resource
tags: [imported, {domain_tag}]
title: "{title}"
author: "{author}"
content_hash: {sha256}
---
# {Title}
## Summary
{excerpt or blank}
## Content
{body text}
"""
doc_uuid = str(uuid.uuid4())
iso_date = datetime.now(timezone.utc).isoformat()
domain = get_domain(result.url)
domain_tag = get_domain_tag(result.url)
# Build tags list
tags = ["imported", domain_tag]
if result.tags:
tags.extend(result.tags)
# Deduplicate while preserving order
seen: set[str] = set()
unique_tags: list[str] = []
for tag in tags:
if tag not in seen:
seen.add(tag)
unique_tags.append(tag)
tags_str = ", ".join(unique_tags)
# Escape title for YAML (handle quotes)
safe_title = result.title.replace('"', '\\"') if result.title else ""
safe_author = result.author.replace('"', '\\"') if result.author else ""
# Build the excerpt / summary
excerpt = result.excerpt or generate_excerpt(result.content)
# Assemble the document
lines: list[str] = [
"---",
f"uuid: {doc_uuid}",
f'source: "{result.url}"',
f"date: {iso_date}",
"type: external_resource",
f"tags: [{tags_str}]",
]
if safe_title:
lines.append(f'title: "{safe_title}"')
if safe_author:
lines.append(f'author: "{safe_author}"')
c_hash = content_hash(result.content)
lines.append(f"content_hash: {c_hash}")
lines.append("---")
lines.append("")
# Heading
lines.append(f"# {result.title or 'Untitled'}")
lines.append("")
# Summary section
lines.append("## Summary")
lines.append("")
if excerpt:
lines.append(f"> {excerpt}")
else:
lines.append("*(To be filled by the Agent)*")
lines.append("")
# Content section
lines.append("## Content")
lines.append("")
lines.append(result.content)
lines.append("")
return "\n".join(lines)
@property
def memory_dir(self) -> Path:
"""Return the resolved memory directory path."""
return self._memory_dir
deepreader_skill/core/utils.py
+176
View File
@@ -0,0 +1,176 @@
"""
DeepReader Skill - Utility Functions
=====================================
Helper functions for URL extraction, text cleaning,
filename sanitization, and content hashing.
"""
from __future__ import annotations
import hashlib
import re
import unicodedata
from datetime import datetime, timezone
from urllib.parse import urlparse
import tldextract
# ---------------------------------------------------------------------------
# URL Extraction
# ---------------------------------------------------------------------------
# Robust URL regex – captures http(s) URLs in free-form text.
_URL_PATTERN = re.compile(
r"https?://" # scheme
r"(?:[a-zA-Z0-9\-._~:/?#\[\]@!$&'()*+,;=%])+", # rest of URI chars
re.IGNORECASE,
)
def extract_urls(text: str) -> list[str]:
"""Return a deduplicated, order-preserved list of URLs found in *text*."""
seen: set[str] = set()
urls: list[str] = []
for match in _URL_PATTERN.finditer(text):
url = match.group(0).rstrip(".,;:!?)") # strip trailing punctuation
if url not in seen:
seen.add(url)
urls.append(url)
return urls
# ---------------------------------------------------------------------------
# Domain Helpers
# ---------------------------------------------------------------------------
def get_domain(url: str) -> str:
"""Return the registered domain of *url* (e.g. ``'twitter.com'``)."""
extracted = tldextract.extract(url)
return f"{extracted.domain}.{extracted.suffix}".lower()
def get_domain_tag(url: str) -> str:
"""Return a clean, tag-safe domain label (e.g. ``'twitter_com'``)."""
return get_domain(url).replace(".", "_")
# ---------------------------------------------------------------------------
# Text Cleaning
# ---------------------------------------------------------------------------
_WHITESPACE_RUNS = re.compile(r"[ \t]+")
_BLANK_LINES = re.compile(r"\n{3,}")
def clean_text(text: str) -> str:
"""Normalize whitespace and remove excessive blank lines."""
text = _WHITESPACE_RUNS.sub(" ", text)
text = _BLANK_LINES.sub("\n\n", text)
return text.strip()
def generate_excerpt(text: str, max_length: int = 280) -> str:
"""Generate a short excerpt from the beginning of *text*."""
cleaned = clean_text(text)
if len(cleaned) <= max_length:
return cleaned
# Cut at last space before max_length to avoid mid-word breaks.
truncated = cleaned[:max_length].rsplit(" ", 1)[0]
return f"{truncated}…"
# ---------------------------------------------------------------------------
# Filename Helpers
# ---------------------------------------------------------------------------
_SLUG_UNSAFE = re.compile(r"[^\w\s-]", re.UNICODE)
_SLUG_SEPARATOR = re.compile(r"[-\s]+")
def sanitize_title(title: str, max_length: int = 80) -> str:
"""Convert *title* into a filesystem-safe slug.
Example::
>>> sanitize_title("Hello, World! An Article — 2024")
'hello-world-an-article-2024'
"""
# Normalize unicode, strip accents
value = unicodedata.normalize("NFKD", title)
value = value.encode("ascii", "ignore").decode("ascii")
value = _SLUG_UNSAFE.sub("", value).strip().lower()
value = _SLUG_SEPARATOR.sub("-", value)
return value[:max_length].rstrip("-")
def build_filename(title: str | None, url: str) -> str:
"""Build a ``YYYY-MM-DD_{slug}.md`` filename.
Falls back to a URL hash if *title* is empty.
"""
date_prefix = datetime.now(timezone.utc).strftime("%Y-%m-%d")
if title:
slug = sanitize_title(title)
else:
slug = hashlib.sha256(url.encode()).hexdigest()[:12]
return f"{date_prefix}_{slug}.md"
# ---------------------------------------------------------------------------
# Hashing
# ---------------------------------------------------------------------------
def content_hash(text: str) -> str:
"""Return a SHA-256 hex digest for deduplication checks."""
return hashlib.sha256(text.encode("utf-8")).hexdigest()
# ---------------------------------------------------------------------------
# URL Classification
# ---------------------------------------------------------------------------
_TWITTER_DOMAINS = {"twitter.com", "x.com", "mobile.twitter.com", "mobile.x.com"}
_YOUTUBE_DOMAINS = {"youtube.com", "youtu.be", "www.youtube.com", "m.youtube.com"}
def is_twitter_url(url: str) -> bool:
"""Return ``True`` if *url* points to Twitter / X."""
parsed = urlparse(url)
return parsed.hostname in _TWITTER_DOMAINS if parsed.hostname else False
def is_youtube_url(url: str) -> bool:
"""Return ``True`` if *url* points to YouTube."""
parsed = urlparse(url)
hostname = parsed.hostname or ""
return hostname in _YOUTUBE_DOMAINS
def extract_youtube_video_id(url: str) -> str | None:
"""Extract the video ID from a YouTube URL.
Supports:
- ``https://www.youtube.com/watch?v=VIDEO_ID``
- ``https://youtu.be/VIDEO_ID``
- ``https://www.youtube.com/embed/VIDEO_ID``
- ``https://www.youtube.com/shorts/VIDEO_ID``
"""
parsed = urlparse(url)
# youtu.be short links
if parsed.hostname and "youtu.be" in parsed.hostname:
return parsed.path.lstrip("/").split("/")[0] or None
# Standard /watch?v= links
from urllib.parse import parse_qs
qs = parse_qs(parsed.query)
if "v" in qs:
return qs["v"][0]
# /embed/ or /shorts/ paths
path_parts = parsed.path.strip("/").split("/")
if len(path_parts) >= 2 and path_parts[0] in ("embed", "shorts", "v"):
return path_parts[1]
return None
deepreader_skill/manifest.json
+30
View File
@@ -0,0 +1,30 @@
{
"name": "DeepReader",
"description": "Autonomous web content ingestion engine. Intercepts URLs from user messages, scrapes content intelligently using specialized parsers (generic articles, Twitter/X, YouTube), formats it into clean Markdown with YAML frontmatter metadata, and saves it to the agent's long-term memory directory.",
"version": "1.0.0",
"author": "OpenClaw",
"entry_point": "deepreader_skill",
"args": {
"text": {
"type": "string",
"description": "The user message that may contain one or more URLs to process.",
"required": true
}
},
"triggers": [
{
"type": "message",
"pattern": "https?://",
"description": "Triggers when a user message contains a URL."
}
],
"capabilities": [
"web_scraping",
"content_extraction",
"youtube_transcription",
"twitter_reading",
"markdown_generation"
],
"memory_path": "../../memory/inbox/",
"tags": ["reader", "scraper", "ingestion", "memory"]
}
deepreader_skill/parsers/__init__.py
+1
View File
@@ -0,0 +1 @@
# DeepReader Skill - Parsers Module
deepreader_skill/parsers/base.py
+95
View File
@@ -0,0 +1,95 @@
"""
DeepReader Skill - Abstract Base Parser
========================================
All content parsers must inherit from :class:`BaseParser` and implement
the :meth:`parse` method.
"""
from __future__ import annotations
import logging
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
logger = logging.getLogger("deepreader.parsers")
@dataclass
class ParseResult:
"""Structured output from a parser.
Attributes:
url: The original URL that was parsed.
title: Extracted page / post / video title.
content: The main body text in Markdown format.
author: Author name if available.
excerpt: A short summary / excerpt.
tags: Additional tags derived from the content.
success: Whether the parse operation succeeded.
error: Human-readable error message on failure.
"""
url: str
title: str = ""
content: str = ""
author: str = ""
excerpt: str = ""
tags: list[str] = field(default_factory=list)
success: bool = True
error: str = ""
@classmethod
def failure(cls, url: str, error: str) -> ParseResult:
"""Convenience constructor for a failed parse."""
return cls(url=url, success=False, error=error)
class BaseParser(ABC):
"""Abstract base class for all DeepReader content parsers.
Subclasses **must** implement :meth:`parse`. They may optionally
override :meth:`can_handle` for more granular URL matching beyond
what the router provides.
"""
# Friendly name for logging / debugging.
name: str = "base"
# Default request timeout in seconds.
timeout: int = 30
# Default User-Agent string for HTTP requests.
user_agent: str = (
"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"
)
def can_handle(self, url: str) -> bool: # noqa: ARG002
"""Return ``True`` if this parser should handle the given *url*.
The default implementation always returns ``True``. Override in
subclasses for domain-specific checks.
"""
return True
@abstractmethod
def parse(self, url: str) -> ParseResult:
"""Fetch and parse the content at *url*.
Returns a :class:`ParseResult` — even on failure (use
``ParseResult.failure(...)``).
"""
...
def _get_headers(self) -> dict[str, str]:
"""Return default HTTP headers."""
return {
"User-Agent": self.user_agent,
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Accept-Language": "en-US,en;q=0.9",
"Accept-Encoding": "gzip, deflate, br",
"DNT": "1",
"Connection": "keep-alive",
"Upgrade-Insecure-Requests": "1",
}
deepreader_skill/parsers/generic.py
+159
View File
@@ -0,0 +1,159 @@
"""
DeepReader Skill - Generic Web Parser
=======================================
Uses **trafilatura** as the primary extraction engine, with a
BeautifulSoup fallback for edge cases. Designed for blog posts,
news articles, documentation pages, and general web content.
"""
from __future__ import annotations
import logging
import requests
import trafilatura
from bs4 import BeautifulSoup
from .base import BaseParser, ParseResult
logger = logging.getLogger("deepreader.parsers.generic")
class GenericParser(BaseParser):
"""Extract main content from any web page using trafilatura."""
name = "generic"
def parse(self, url: str) -> ParseResult:
"""Fetch *url* and extract the article body via trafilatura."""
try:
html = self._fetch_html(url)
if not html:
return ParseResult.failure(url, "Failed to download page content.")
# ------------------------------------------------------------------
# Primary: trafilatura (best for article-style pages)
# ------------------------------------------------------------------
result = self._extract_with_trafilatura(url, html)
if result and result.success and result.content:
logger.info("Trafilatura extracted %d chars from %s", len(result.content), url)
return result
# ------------------------------------------------------------------
# Fallback: BeautifulSoup heuristic extraction
# ------------------------------------------------------------------
logger.info("Trafilatura returned empty, trying BeautifulSoup fallback for %s", url)
result = self._extract_with_beautifulsoup(url, html)
if result and result.success and result.content:
logger.info("BS4 extracted %d chars from %s", len(result.content), url)
return result
return ParseResult.failure(url, "Could not extract meaningful content from the page.")
except requests.RequestException as exc:
logger.error("HTTP error for %s: %s", url, exc)
return ParseResult.failure(url, f"HTTP request failed: {exc}")
except Exception as exc: # noqa: BLE001
logger.exception("Unexpected error parsing %s", url)
return ParseResult.failure(url, f"Unexpected error: {exc}")
# ------------------------------------------------------------------
# Internal helpers
# ------------------------------------------------------------------
def _fetch_html(self, url: str) -> str | None:
"""Download the raw HTML content of *url*."""
response = requests.get(
url,
headers=self._get_headers(),
timeout=self.timeout,
allow_redirects=True,
)
response.raise_for_status()
return response.text
def _extract_with_trafilatura(self, url: str, html: str) -> ParseResult | None:
"""Use trafilatura to extract structured content."""
# trafilatura.extract returns plain/markdown text
extracted: str | None = trafilatura.extract(
html,
url=url,
include_comments=False,
include_tables=True,
include_images=False,
include_links=True,
output_format="txt",
favor_precision=True,
)
if not extracted:
return None
# Extract metadata via the dedicated trafilatura metadata API
meta = trafilatura.metadata.extract_metadata(html, default_url=url)
title = ""
author = ""
if meta:
title = meta.title or ""
author = meta.author or ""
# If trafilatura didn't give us a title, fall back to <title> tag
if not title:
title = self._extract_title_from_html(html)
from ..core.utils import clean_text, generate_excerpt
content = clean_text(extracted)
excerpt = generate_excerpt(content)
return ParseResult(
url=url,
title=title,
content=content,
author=author,
excerpt=excerpt,
)
def _extract_with_beautifulsoup(self, url: str, html: str) -> ParseResult | None:
"""Fallback extraction using BeautifulSoup heuristics."""
soup = BeautifulSoup(html, "lxml")
# Remove noisy elements
for tag in soup.find_all(["script", "style", "nav", "footer", "header", "aside"]):
tag.decompose()
# Try to find the main article body
article = (
soup.find("article")
or soup.find("main")
or soup.find(attrs={"role": "main"})
or soup.find("div", class_=lambda c: c and "content" in c)
)
if not article:
article = soup.body
if not article:
return None
text = article.get_text(separator="\n", strip=True)
title = self._extract_title_from_html(html)
from ..core.utils import clean_text, generate_excerpt
content = clean_text(text)
if len(content) < 50:
return None
return ParseResult(
url=url,
title=title,
content=content,
excerpt=generate_excerpt(content),
)
@staticmethod
def _extract_title_from_html(html: str) -> str:
"""Extract the ``<title>`` tag from raw HTML."""
soup = BeautifulSoup(html, "lxml")
tag = soup.find("title")
return tag.get_text(strip=True) if tag else ""
deepreader_skill/parsers/twitter.py
+266
View File
@@ -0,0 +1,266 @@
"""
DeepReader Skill - Twitter / X Parser
=======================================
Strategy-pattern implementation for reading tweets:
1. **Primary**: Rotate through public Nitter instances to fetch tweet
content without any API keys.
2. **Fallback**: Gracefully degrade with informative guidance on how to
plug in a scraping service (ZenRows, ScrapingBee) or browser cookies.
Why Nitter?
-----------
Twitter's official API is paywalled and rate-limited. Nitter is an
open-source alternative frontend that serves tweets as plain HTML,
making extraction trivial. However, public instances may go down,
so we rotate through several and retry.
Extending with a paid scraping service
---------------------------------------
If all Nitter instances fail, you can integrate a proxy/rendering
service. See the ``_fallback_scrape`` method for detailed guidance
on where to plug in ZenRows or browser cookies.
"""
from __future__ import annotations
import logging
import random
import re
from urllib.parse import urlparse
import requests
from bs4 import BeautifulSoup
from .base import BaseParser, ParseResult
logger = logging.getLogger("deepreader.parsers.twitter")
# ---------------------------------------------------------------------------
# Known public Nitter instances (community-maintained)
# Update this list periodically – instances come and go.
# ---------------------------------------------------------------------------
NITTER_INSTANCES: list[str] = [
"https://nitter.privacydev.net",
"https://nitter.poast.org",
"https://nitter.woodland.cafe",
"https://nitter.1d4.us",
"https://nitter.kavin.rocks",
"https://nitter.unixfox.eu",
"https://nitter.d420.de",
"https://nitter.moomoo.me",
]
class TwitterParser(BaseParser):
"""Parse tweets from Twitter / X via Nitter relay instances."""
name = "twitter"
timeout = 20
# Maximum number of Nitter instances to try before giving up.
max_retries: int = 4
def can_handle(self, url: str) -> bool:
"""Return ``True`` for twitter.com / x.com URLs."""
from ..core.utils import is_twitter_url
return is_twitter_url(url)
def parse(self, url: str) -> ParseResult:
"""Attempt to read a tweet via Nitter, with graceful fallbacks."""
tweet_path = self._extract_tweet_path(url)
if not tweet_path:
return ParseResult.failure(
url,
"Could not extract a valid tweet path from this URL. "
"Expected format: https://twitter.com/user/status/123456",
)
# Shuffle instances to spread load and improve resilience.
instances = random.sample(
NITTER_INSTANCES,
min(self.max_retries, len(NITTER_INSTANCES)),
)
last_error = ""
for instance in instances:
nitter_url = f"{instance}/{tweet_path}"
logger.info("Trying Nitter instance: %s", nitter_url)
try:
result = self._parse_nitter(url, nitter_url)
if result.success:
return result
last_error = result.error
except requests.RequestException as exc:
last_error = str(exc)
logger.warning("Nitter instance %s failed: %s", instance, exc)
continue
except Exception as exc: # noqa: BLE001
last_error = str(exc)
logger.warning("Unexpected error with %s: %s", instance, exc)
continue
# ------------------------------------------------------------------
# All Nitter instances failed → fallback
# ------------------------------------------------------------------
return self._fallback_scrape(url, last_error)
# ------------------------------------------------------------------
# Nitter HTML Parsing
# ------------------------------------------------------------------
def _parse_nitter(self, original_url: str, nitter_url: str) -> ParseResult:
"""Fetch and parse a single Nitter page."""
resp = requests.get(
nitter_url,
headers=self._get_headers(),
timeout=self.timeout,
)
resp.raise_for_status()
soup = BeautifulSoup(resp.text, "lxml")
# --- Tweet body ---
tweet_div = soup.find("div", class_="tweet-content") or soup.find(
"div", class_="main-tweet"
)
if not tweet_div:
return ParseResult.failure(
original_url,
f"Nitter page loaded but no tweet content found at {nitter_url}",
)
tweet_text = tweet_div.get_text(separator="\n", strip=True)
# --- Author ---
author_tag = soup.find("a", class_="fullname") or soup.find(
"span", class_="username"
)
author = author_tag.get_text(strip=True) if author_tag else ""
# --- Timestamp ---
date_tag = soup.find("span", class_="tweet-date")
timestamp = ""
if date_tag:
a_tag = date_tag.find("a")
timestamp = a_tag.get("title", "") if a_tag else date_tag.get_text(strip=True)
# Build a nice title
title = f"Tweet by {author}" if author else "Tweet"
if timestamp:
title += f" ({timestamp})"
# Collect reply context if present
replies: list[str] = []
reply_divs = soup.find_all("div", class_="reply")
for rd in reply_divs[:5]: # limit to first 5 replies
reply_content = rd.find("div", class_="tweet-content")
if reply_content:
replies.append(reply_content.get_text(separator=" ", strip=True))
content_parts = [tweet_text]
if replies:
content_parts.append("\n\n---\n### Replies\n")
for i, reply in enumerate(replies, 1):
content_parts.append(f"**Reply {i}:** {reply}\n")
from ..core.utils import clean_text, generate_excerpt
full_content = clean_text("\n".join(content_parts))
return ParseResult(
url=original_url,
title=title,
content=full_content,
author=author,
excerpt=generate_excerpt(full_content),
tags=["twitter"],
)
# ------------------------------------------------------------------
# Fallback Strategy
# ------------------------------------------------------------------
def _fallback_scrape(self, url: str, last_error: str) -> ParseResult:
"""Produce a graceful degradation result with integration guidance.
.. rubric:: How to extend with a paid scraping service
**Option A – ZenRows / ScrapingBee:**
1. Sign up at https://www.zenrows.com/ or https://www.scrapingbee.com/
2. Obtain your API key.
3. Replace the body of this method with::
import requests
api_key = "YOUR_ZENROWS_API_KEY"
params = {
"url": url,
"apikey": api_key,
"js_render": "true",
"premium_proxy": "true",
}
resp = requests.get("https://api.zenrows.com/v1/", params=params)
html = resp.text
# Then parse 'html' with BeautifulSoup to extract tweet text.
**Option B – Browser Cookies:**
1. Export your Twitter session cookies (e.g. with the *EditThisCookie*
browser extension) as a Netscape-format ``cookies.txt`` file.
2. Place the file at ``deepreader_skill/twitter_cookies.txt``.
3. Modify ``_fetch_with_cookies()`` below to load and send them::
import http.cookiejar
jar = http.cookiejar.MozillaCookieJar("twitter_cookies.txt")
jar.load()
session = requests.Session()
session.cookies = jar
resp = session.get(url, headers=self._get_headers())
**Option C – Playwright / Selenium headless browser:**
For the most reliable extraction, you can use a headless browser.
This is heavier but handles JavaScript-rendered content::
from playwright.sync_api import sync_playwright
with sync_playwright() as p:
browser = p.chromium.launch(headless=True)
page = browser.new_page()
page.goto(url, wait_until="networkidle")
html = page.content()
browser.close()
# Then parse 'html' as above.
Returns a :class:`ParseResult` with ``success=False`` and the
guidance embedded in the error message.
"""
error_msg = (
f"⚠️ All Nitter instances failed for this tweet.\n"
f"Last error: {last_error}\n\n"
f"The tweet URL was: {url}\n\n"
f"💡 To improve Twitter support, consider:\n"
f" 1. Updating the NITTER_INSTANCES list in twitter.py\n"
f" 2. Integrating a paid scraping service (ZenRows/ScrapingBee)\n"
f" 3. Using browser cookies for authenticated access\n"
f" See the _fallback_scrape() docstring for detailed instructions."
)
logger.warning("Twitter fallback triggered for %s", url)
return ParseResult.failure(url, error_msg)
# ------------------------------------------------------------------
# URL Utilities
# ------------------------------------------------------------------
@staticmethod
def _extract_tweet_path(url: str) -> str | None:
"""Extract the tweet path (``user/status/id``) from a Twitter URL.
Returns ``None`` if the URL doesn't match the expected pattern.
"""
parsed = urlparse(url)
# Match patterns like /username/status/1234567890
match = re.match(r"^/([^/]+)/status/(\d+)", parsed.path)
if match:
return f"{match.group(1)}/status/{match.group(2)}"
return None
deepreader_skill/parsers/youtube.py
+234
View File
@@ -0,0 +1,234 @@
"""
DeepReader Skill - YouTube Parser
==================================
Fetches video metadata and transcripts/subtitles using the
``youtube_transcript_api`` library. No API key required for
publicly available transcripts.
"""
from __future__ import annotations
import logging
import requests
from bs4 import BeautifulSoup
from .base import BaseParser, ParseResult
logger = logging.getLogger("deepreader.parsers.youtube")
class YouTubeParser(BaseParser):
"""Extract transcripts and metadata from YouTube videos."""
name = "youtube"
timeout = 25
# Preferred language codes for transcripts (in priority order).
preferred_languages: list[str] = ["en", "zh-Hans", "zh-Hant", "zh", "ja", "ko", "de", "fr", "es"]
def can_handle(self, url: str) -> bool:
from ..core.utils import is_youtube_url
return is_youtube_url(url)
def parse(self, url: str) -> ParseResult:
"""Fetch the YouTube video transcript and metadata."""
from ..core.utils import extract_youtube_video_id
video_id = extract_youtube_video_id(url)
if not video_id:
return ParseResult.failure(
url,
"Could not extract a valid YouTube video ID from this URL.",
)
# Step 1: Get video metadata from the page
title, author, description = self._fetch_metadata(url)
# Step 2: Get the transcript
transcript_text, transcript_lang = self._fetch_transcript(video_id)
if not transcript_text:
# If no transcript available, still save what we have
content_parts = []
if description:
content_parts.append(f"**Video Description:**\n\n{description}")
content_parts.append(
"\n\n> ⚠️ No transcript/subtitles available for this video. "
"The video may not have captions enabled."
)
content = "\n".join(content_parts)
else:
content_parts = []
if transcript_lang:
content_parts.append(f"*Transcript language: {transcript_lang}*\n")
content_parts.append(transcript_text)
if description:
content_parts.append(f"\n\n---\n\n**Video Description:**\n\n{description}")
content = "\n".join(content_parts)
from ..core.utils import clean_text, generate_excerpt
content = clean_text(content)
return ParseResult(
url=url,
title=title or f"YouTube Video ({video_id})",
content=content,
author=author,
excerpt=generate_excerpt(content),
tags=["youtube", "video"],
)
# ------------------------------------------------------------------
# Transcript Extraction
# ------------------------------------------------------------------
def _fetch_transcript(self, video_id: str) -> tuple[str, str]:
"""Fetch the transcript for a YouTube video.
Returns a tuple of ``(transcript_text, language_code)``.
Returns ``("", "")`` if no transcript is available.
"""
try:
from youtube_transcript_api import YouTubeTranscriptApi
# Try to get transcript in preferred languages first
try:
transcript_list = YouTubeTranscriptApi.list_transcripts(video_id)
# Try to find a manually created transcript in preferred languages
transcript = None
lang_code = ""
for lang in self.preferred_languages:
try:
transcript = transcript_list.find_manually_created_transcript([lang])
lang_code = lang
break
except Exception: # noqa: BLE001
continue
# Fall back to auto-generated transcripts
if transcript is None:
for lang in self.preferred_languages:
try:
transcript = transcript_list.find_generated_transcript([lang])
lang_code = f"{lang} (auto-generated)"
break
except Exception: # noqa: BLE001
continue
# Last resort: get whatever is available
if transcript is None:
try:
available = list(transcript_list)
if available:
transcript = available[0]
lang_code = transcript.language_code
except Exception: # noqa: BLE001
pass
if transcript is not None:
entries = transcript.fetch()
lines = [entry["text"] for entry in entries]
return self._format_transcript(lines), lang_code
except Exception as exc: # noqa: BLE001
logger.warning("Transcript list API failed for %s: %s", video_id, exc)
# Direct fetch as ultimate fallback
try:
entries = YouTubeTranscriptApi.get_transcript(video_id)
lines = [entry["text"] for entry in entries]
return self._format_transcript(lines), "auto"
except Exception as inner_exc: # noqa: BLE001
logger.warning("Direct transcript fetch failed: %s", inner_exc)
except ImportError:
logger.error(
"youtube_transcript_api is not installed. "
"Run: pip install youtube-transcript-api"
)
return "", ""
@staticmethod
def _format_transcript(lines: list[str]) -> str:
"""Join transcript lines into clean paragraphs.
Groups lines into paragraphs of ~5 sentences for readability.
"""
if not lines:
return ""
paragraphs: list[str] = []
current_paragraph: list[str] = []
sentence_count = 0
for line in lines:
line = line.strip()
if not line:
continue
current_paragraph.append(line)
# Count sentence-ending punctuation
if any(line.endswith(p) for p in (".", "!", "?", "。", "！", "？")):
sentence_count += 1
if sentence_count >= 5:
paragraphs.append(" ".join(current_paragraph))
current_paragraph = []
sentence_count = 0
# Don't forget the last paragraph
if current_paragraph:
paragraphs.append(" ".join(current_paragraph))
return "\n\n".join(paragraphs)
# ------------------------------------------------------------------
# Metadata Extraction
# ------------------------------------------------------------------
def _fetch_metadata(self, url: str) -> tuple[str, str, str]:
"""Extract title, channel name, and description from the YouTube page.
Returns ``(title, author, description)``.
"""
try:
resp = requests.get(url, headers=self._get_headers(), timeout=self.timeout)
resp.raise_for_status()
soup = BeautifulSoup(resp.text, "lxml")
# Title: <meta property="og:title">
title = ""
og_title = soup.find("meta", property="og:title")
if og_title:
title = og_title.get("content", "")
if not title:
title_tag = soup.find("title")
title = title_tag.get_text(strip=True) if title_tag else ""
# Remove " - YouTube" suffix
if title.endswith(" - YouTube"):
title = title[:-10].strip()
# Author / channel
author = ""
link_author = soup.find("link", attrs={"itemprop": "name"})
if link_author:
author = link_author.get("content", "")
if not author:
og_author = soup.find("meta", property="og:video:tag")
if og_author:
author = og_author.get("content", "")
# Description
description = ""
og_desc = soup.find("meta", property="og:description")
if og_desc:
description = og_desc.get("content", "")
return title, author, description
except Exception as exc: # noqa: BLE001
logger.warning("Failed to fetch YouTube metadata for %s: %s", url, exc)
return "", "", ""
deepreader_skill/requirements.txt
+24
View File
@@ -0,0 +1,24 @@
# DeepReader Skill - Dependencies
# Web scraping & content extraction
trafilatura>=1.12.0
requests>=2.31.0
lxml>=5.1.0
lxml-html-clean>=0.4.0
# YouTube transcript extraction
youtube-transcript-api>=0.6.2
# Data validation & settings
pydantic>=2.5.0
pydantic-settings>=2.1.0
# URL parsing & utilities
tldextract>=5.1.0
# HTML parsing (fallback)
beautifulsoup4>=4.12.0
# Unique ID generation (stdlib, listed for clarity)
# uuid - built-in
# re - built-in
# hashlib - built-in
pyproject.toml
+46
View File
@@ -0,0 +1,46 @@
[build-system]
requires = ["setuptools>=68.0", "wheel"]
build-backend = "setuptools.backends._legacy:_Backend"
[project]
name = "openclaw-deepreeder"
version = "1.0.0"
description = "Autonomous web content ingestion engine for OpenClaw agents. Scrapes articles, Twitter/X posts, and YouTube transcripts into clean Markdown."
readme = "README.md"
license = {text = "MIT"}
requires-python = ">=3.10"
authors = [
{name = "OpenClaw", email = "astonysh@users.noreply.github.com"},
]
keywords = ["web-scraping", "content-extraction", "markdown", "ai-agent", "openclaw"]
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 :: Text Processing :: Markup :: Markdown",
]
dependencies = [
"trafilatura>=1.12.0",
"requests>=2.31.0",
"lxml>=5.1.0",
"lxml-html-clean>=0.4.0",
"youtube-transcript-api>=0.6.2",
"pydantic>=2.5.0",
"pydantic-settings>=2.1.0",
"tldextract>=5.1.0",
"beautifulsoup4>=4.12.0",
]
[project.urls]
Homepage = "https://github.com/astonysh/OpenClaw-DeepReeder"
Repository = "https://github.com/astonysh/OpenClaw-DeepReeder"
Issues = "https://github.com/astonysh/OpenClaw-DeepReeder/issues"
[tool.setuptools.packages.find]
include = ["deepreader_skill*"]