92 lines
2.4 KiB
Python
92 lines
2.4 KiB
Python
|
"""
|
||
|
|
Web Fetch Tool
|
||
|
|
|
||
|
|
Web content fetching and screenshot tool.
|
||
|
|
"""
|
||
|
|
|
||
|
|
import asyncio
|
||
|
|
from typing import Dict, Any, Optional, List
|
||
|
|
from dataclasses import dataclass
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass
|
||
|
|
class FetchResult:
|
||
|
|
"""Fetch result container"""
|
||
|
|
|
||
|
|
url: str
|
||
|
|
title: Optional[str]
|
||
|
|
content: str
|
||
|
|
images: List[str]
|
||
|
|
links: List[str]
|
||
|
|
status: int
|
||
|
|
|
||
|
|
|
||
|
|
class WebFetch:
|
||
|
|
"""Web fetch tool"""
|
||
|
|
|
||
|
|
def __init__(self, config: dict):
|
||
|
|
self.timeout = config.get("timeout", 30)
|
||
|
|
self.user_agent = config.get("user_agent", "Mozilla/5.0 (compatible; Jarvis/1.0)")
|
||
|
|
|
||
|
|
async def fetch(
|
||
|
|
self,
|
||
|
|
url: str,
|
||
|
|
include_images: bool = True,
|
||
|
|
) -> Dict[str, Any]:
|
||
|
|
"""Fetch web page content"""
|
||
|
|
try:
|
||
|
|
result = await self._do_fetch(url, include_images)
|
||
|
|
return {
|
||
|
|
"status": "success",
|
||
|
|
"result": {
|
||
|
|
"url": result.url,
|
||
|
|
"title": result.title,
|
||
|
|
"content": result.content,
|
||
|
|
"images": result.images if include_images else [],
|
||
|
|
"links": result.links,
|
||
|
|
"status": result.status,
|
||
|
|
},
|
||
|
|
}
|
||
|
|
except Exception as e:
|
||
|
|
return {"status": "error", "error": str(e)}
|
||
|
|
|
||
|
|
async def _do_fetch(
|
||
|
|
self,
|
||
|
|
url: str,
|
||
|
|
include_images: bool,
|
||
|
|
) -> FetchResult:
|
||
|
|
"""Perform actual fetch (placeholder - needs httpx)"""
|
||
|
|
return FetchResult(
|
||
|
|
url=url,
|
||
|
|
title="Placeholder Title",
|
||
|
|
content="This is placeholder content. Configure httpx/beautifulsoup4 for real fetching.",
|
||
|
|
images=[],
|
||
|
|
links=[],
|
||
|
|
status=200,
|
||
|
|
)
|
||
|
|
|
||
|
|
async def screenshot(
|
||
|
|
self,
|
||
|
|
url: str,
|
||
|
|
) -> Dict[str, Any]:
|
||
|
|
"""Take screenshot of web page (placeholder)"""
|
||
|
|
return {
|
||
|
|
"status": "error",
|
||
|
|
"error": "Screenshot requires puppeteer or playwright integration",
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def create_web_fetch_executor(config: dict):
|
||
|
|
"""Create web fetch executor"""
|
||
|
|
fetcher = WebFetch(config)
|
||
|
|
|
||
|
|
async def execute(command: str, parameters: dict) -> dict:
|
||
|
|
if command == "fetch":
|
||
|
|
return await fetcher.fetch(**parameters)
|
||
|
|
elif command == "screenshot":
|
||
|
|
return await fetcher.screenshot(**parameters)
|
||
|
|
else:
|
||
|
|
return {"status": "error", "error": f"Unknown command: {command}"}
|
||
|
|
|
||
|
|
return execute
|