ThreeFish-AI/data-extractor

{
  "title": "h1",
  "content": {
    "selector": ".content p",
    "multiple": true,
    "attr": "text"
  },
  "links": {
    "selector": "a",
    "multiple": true,
    "attr": "href"
  }
}

{
  "#username": "用户名",
  "input[name=\"password\"]": "密码",
  "select[name=country]": "China",
  "input[value=male]": "click",
  "input[name=agree]": true
}

{
  "max_images": 50,
  "max_bytes_per_image": 2000000,
  "timeout_seconds": 10
}

{
  "format_tables": true,
  "detect_code_language": true,
  "format_quotes": true,
  "enhance_images": true,
  "optimize_links": true,
  "format_lists": true
}

{
  "filter_domains": ["example.com", "blog.example.com"],
  "exclude_domains": ["ads.com", "tracker.net"],
  "internal_only": false
}

# 克隆仓库
git clone https://github.com/ThreeFish-AI/data-extractor.git
cd data-extractor

# 快速设置（推荐）
./scripts/setup.sh

# 或手动安装
# 使用 uv 安装依赖
uv sync

# 安装包括开发依赖
uv sync --extra dev

# 服务器设置
DATA_EXTRACTOR_SERVER_NAME=data-extractor
# DATA_EXTRACTOR_SERVER_VERSION=auto  # 版本号自动从 pyproject.toml 读取，无需手动配置

# 并发和延迟设置
DATA_EXTRACTOR_CONCURRENT_REQUESTS=16
DATA_EXTRACTOR_DOWNLOAD_DELAY=1.0
DATA_EXTRACTOR_RANDOMIZE_DOWNLOAD_DELAY=true

# 浏览器设置
DATA_EXTRACTOR_ENABLE_JAVASCRIPT=false
DATA_EXTRACTOR_BROWSER_HEADLESS=true
DATA_EXTRACTOR_BROWSER_TIMEOUT=30

# 反检测设置
DATA_EXTRACTOR_USE_RANDOM_USER_AGENT=true
DATA_EXTRACTOR_USE_PROXY=false
DATA_EXTRACTOR_PROXY_URL=

# 重试设置
DATA_EXTRACTOR_MAX_RETRIES=3
DATA_EXTRACTOR_REQUEST_TIMEOUT=30

# 使用命令行
data-extractor

# 使用 uv 运行（推荐）
uv run data-extractor

# 或者使用Python
python -m extractor.server

# 使用 uv 运行 Python 模块
uv run python -m extractor.server

{
  "mcpServers": {
    "data-extractor": {
      "command": "data-extractor",
      "args": []
    }
  }
}

{
  "mcpServers": {
    "data-extractor": {
      "command": "uv",
      "args": ["run", "data-extractor"],
      "cwd": "/path/to/your/data-extractor"
    }
  }
}

{
  "mcpServers": {
    "data-extractor": {
      "command": "uv",
      "args": [
        "run",
        "--with",
        "git+https://github.com/ThreeFish-AI/data-extractor.git@v0.1.4",
        "data-extractor"
      ]
    }
  }
}

{
  "mcpServers": {
    "data-extractor": {
      "command": "uv",
      "args": ["run", "python", "-m", "extractor.server"],
      "cwd": "/path/to/your/data-extractor"
    }
  }
}

{
  "url": "https://example.com",
  "method": "auto",
  "extract_config": {
    "title": "h1",
    "content": {
      "selector": ".content p",
      "multiple": true,
      "attr": "text"
    }
  }
}

{
  "success": true,
  "url": "https://example.com",
  "method": "auto",
  "data": {
    "title": "网站标题",
    "content": ["段落1内容", "段落2内容"]
  },
  "metadata": {
    "title": "网站标题",
    "description": "网站描述"
  },
  "timestamp": "2025-09-17T13:45:00"
}

{
  "urls": ["https://example1.com", "https://example2.com"],
  "method": "simple",
  "extract_config": {
    "title": "h1",
    "links": "a"
  }
}

{
  "url": "https://protected-site.com",
  "method": "playwright",
  "scroll_page": true,
  "wait_for_element": ".dynamic-content"
}

{
  "url": "https://example.com/contact",
  "form_data": {
    "input[name='name']": "John Doe",
    "input[name='email']": "john@example.com",
    "textarea[name='message']": "Hello world"
  },
  "submit": true,
  "method": "selenium"
}

{
  "url": "https://example.com",
  "internal_only": true
}

{
  "url": "https://company.com",
  "data_type": "contact"
}

{
  "url": "https://example.com"
}

{
  "url": "https://example.com/article",
  "method": "auto",
  "extract_main_content": true,
  "include_metadata": true,
  "custom_options": {
    "heading_style": "ATX",
    "bullets": "-",
    "wrap": false
  },
  "formatting_options": {
    "format_tables": true,
    "detect_code_language": true,
    "enhance_images": true,
    "apply_typography": false
  },
  "embed_images": true,
  "embed_options": {
    "max_images": 10,
    "max_bytes_per_image": 1500000,
    "timeout_seconds": 8
  }
}

{
  "success": true,
  "data": {
    "url": "https://example.com/article",
    "markdown": "# Article Title\n\nThis is the article content...",
    "metadata": {
      "title": "Article Title",
      "meta_description": "Article description",
      "word_count": 500,
      "character_count": 3000,
      "domain": "example.com"
    }
  }
}

{
  "urls": [
    "https://example.com/article1",
    "https://example.com/article2",
    "https://example.com/article3"
  ],
  "method": "auto",
  "extract_main_content": true,
  "include_metadata": true
}

{
  "success": true,
  "data": {
    "results": [
      {
        "success": true,
        "url": "https://example.com/article1",
        "markdown": "# Article 1\n\nContent...",
        "metadata": {...}
      },
      {
        "success": true,
        "url": "https://example.com/article2",
        "markdown": "# Article 2\n\nContent...",
        "metadata": {...}
      }
    ],
    "summary": {
      "total": 3,
      "successful": 2,
      "failed": 1,
      "success_rate": 0.67
    }
  }
}

{
  "pdf_source": "https://example.com/document.pdf",
  "method": "auto",
  "include_metadata": true,
  "page_range": [0, 10],
  "output_format": "markdown"
}

{
  "success": true,
  "data": {
    "text": "原始提取的文本内容",
    "markdown": "# 文档标题\n\n转换后的 Markdown 内容...",
    "metadata": {
      "title": "文档标题",
      "author": "作者姓名",
      "total_pages": 50,
      "pages_processed": 10,
      "file_size_bytes": 1024000
    },
    "source": "https://example.com/document.pdf",
    "method_used": "pymupdf",
    "word_count": 2500,
    "character_count": 15000
  }
}

{
  "pdf_sources": [
    "https://example.com/doc1.pdf",
    "/local/path/doc2.pdf",
    "https://example.com/doc3.pdf"
  ],
  "method": "auto",
  "include_metadata": true,
  "output_format": "markdown"
}

{
  "success": true,
  "data": {
    "results": [
      {
        "success": true,
        "source": "https://example.com/doc1.pdf",
        "text": "原始文本内容",
        "markdown": "# 文档1标题\n\n内容...",
        "metadata": {...},
        "word_count": 1500
      },
      {
        "success": true,
        "source": "/local/path/doc2.pdf",
        "text": "原始文本内容",
        "markdown": "# 文档2标题\n\n内容...",
        "metadata": {...},
        "word_count": 2000
      }
    ],
    "summary": {
      "total_pdfs": 3,
      "successful": 2,
      "failed": 1,
      "total_pages_processed": 45,
      "total_words_extracted": 3500,
      "method_used": "auto",
      "output_format": "markdown"
    }
  }
}

{
  "title": "h1",
  "links": "a"
}

{
  "products": {
    "selector": ".product",
    "multiple": true,
    "attr": "text"
  },
  "prices": {
    "selector": ".price",
    "multiple": true,
    "attr": "data-price"
  },
  "description": {
    "selector": ".description",
    "multiple": false,
    "attr": "text"
  }
}

class WebScraper:
    """主控制器，协调各种抓取方法"""

    def __init__(self):
        self.scrapy_wrapper = ScrapyWrapper()      # Scrapy 框架封装
        self.selenium_scraper = SeleniumScraper()  # 浏览器自动化
        self.simple_scraper = SimpleScraper()      # HTTP 请求

    async def scrape_url(self, url: str, method: str = "auto",
                         extract_config: Optional[Dict] = None) -> Dict:
        """智能选择最适合的抓取方法"""

class AntiDetectionScraper:
    """反检测专用抓取器"""

    async def scrape_with_stealth(self, url: str, method: str = "selenium"):
        """使用反检测技术抓取"""
        # 支持 undetected-chromedriver 和 Playwright 双引擎
        # 自动注入隐身脚本，模拟人类行为

class FormHandler:
    """智能表单处理器"""

    async def fill_and_submit_form(self, url: str, form_data: Dict):
        """自动识别表单元素类型并填写"""
        # 支持 input/select/textarea/checkbox/radio 等所有元素
        # 智能等待和提交策略

from extractor.utils import rate_limiter, retry_manager, cache_manager

# 限流控制
await rate_limiter.wait()

# 智能重试
result = await retry_manager.retry_async(scrape_function)

# 缓存管理
cache_manager.set(url, result, ttl=3600)

class DataExtractorSettings(BaseSettings):
    """Pydantic 配置管理"""

    # 服务器配置
    server_name: str = "Data Extractor MCP Server"
    concurrent_requests: int = 16

    # 浏览器配置
    enable_javascript: bool = False
    browser_timeout: int = 30

    # 反检测配置
    use_random_user_agent: bool = False

    model_config = SettingsConfigDict(
        env_prefix="DATA_EXTRACTOR_",  # 环境变量前缀
        env_file=".env"
    )

from fastmcp import FastMCP

app = FastMCP(settings.server_name, version=settings.server_version)
web_scraper = WebScraper()
anti_detection_scraper = AntiDetectionScraper()

@app.tool()
async def scrape_webpage(url: str, method: str = "auto",
                        extract_config: Optional[Dict] = None) -> Dict:
    """MCP 工具装饰器，自动处理输入验证和错误处理"""

@app.tool()
async def scrape_webpage(url: str, method: str = "auto",
                        extract_config: Optional[Dict] = None,
                        wait_for_element: Optional[str] = None) -> Dict:
    """
    支持的数据提取配置:
    {
        "title": "h1",                          # 简单选择器
        "products": {                           # 高级配置
            "selector": ".product",
            "multiple": true,
            "attr": "text"
        },
        "links": {
            "selector": "a",
            "multiple": true,
            "attr": "href"
        }
    }
    """

@app.tool()
async def scrape_with_stealth(url: str, method: str = "selenium",
                             extract_config: Optional[Dict] = None) -> Dict:
    """
    反检测技术:
    - undetected-chromedriver: 绕过 Selenium 检测
    - Playwright stealth: 原生反检测支持
    - 随机 User-Agent: 降低识别风险
    - 人类行为模拟: 鼠标移动、页面滚动
    """

@app.tool()
async def fill_and_submit_form(url: str, form_data: Dict,
                              submit: bool = False) -> Dict:
    """
    智能表单处理:
    - 自动识别 input/select/textarea/checkbox 元素
    - 支持复杂表单验证和提交
    - 等待页面响应和重定向处理
    """

# 编辑 pyproject.toml
vim pyproject.toml
# 更新 version = "x.y.z"

# 编辑 CHANGELOG.md，添加新版本条目
vim CHANGELOG.md

git add pyproject.toml CHANGELOG.md
git commit -m "chore: bump version to v1.2.3"
git tag v1.2.3
git push origin main --tags

# 1. 创建功能分支
git checkout -b feature/new-feature

# 2. 开发和测试
uv sync --extra dev
uv run pytest

# 3. 代码质量检查
uv run ruff check extractor/
uv run ruff format extractor/
uv run mypy extractor/

# 4. 提交PR
git push origin feature/new-feature
# 创建PR，CI自动运行测试

from extractor.scraper import WebScraper
from extractor.advanced_features import AntiDetectionScraper, FormHandler

# 基础抓取
scraper = WebScraper()
result = await scraper.scrape_url("https://example.com", method="simple")

# 反检测抓取
stealth_scraper = AntiDetectionScraper()
result = await stealth_scraper.scrape_with_stealth("https://protected-site.com")

# 表单自动化
form_handler = FormHandler()
result = await form_handler.fill_and_submit_form(
    "https://example.com/contact",
    {"input[name='email']": "test@example.com"}
)

# 简单配置
extract_config = {
    "title": "h1",
    "content": ".article-content"
}

# 高级配置
extract_config = {
    "products": {
        "selector": ".product-item",
        "multiple": True,
        "attr": "text"
    },
    "prices": {
        "selector": ".price",
        "multiple": True,
        "attr": "data-price"
    },
    "images": {
        "selector": "img.product-image",
        "multiple": True,
        "attr": "src"
    }
}

result = await scraper.scrape_url(url, extract_config=extract_config)

from extractor.utils import (
    rate_limiter, retry_manager, cache_manager,
    metrics_collector, error_handler
)

# 集成完整功能的抓取流程
async def enterprise_scrape(url: str):
    # 检查缓存
    cached_result = cache_manager.get(url)
    if cached_result:
        return cached_result

    # 速率限制
    await rate_limiter.wait()

    # 重试机制
    try:
        result = await retry_manager.retry_async(
            scraper.scrape_url, url, method="auto"
        )

        # 记录指标
        metrics_collector.record_request("GET", True, 1500, "scraper")

        # 缓存结果
        cache_manager.set(url, result, ttl=3600)

        return result

    except Exception as e:
        error_handler.handle_error(e, "enterprise_scrape")
        raise

# 通过 MCP 协议调用工具
import asyncio
from extractor.server import (
    scrape_webpage, scrape_multiple_webpages,
    scrape_with_stealth, fill_and_submit_form
)

# 基础页面抓取
async def basic_scraping_example():
    result = await scrape_webpage(
        url="https://example.com",
        method="auto",
        extract_config={
            "title": "h1",
            "content": ".main-content"
        }
    )
    print(f"页面标题: {result['data']['extracted_data']['title']}")

# 批量抓取
async def batch_scraping_example():
    urls = [
        "https://site1.com",
        "https://site2.com",
        "https://site3.com"
    ]

    results = await scrape_multiple_webpages(
        urls=urls,
        method="simple",
        extract_config={"title": "h1"}
    )

    for result in results['data']:
        print(f"URL: {result['url']}, 标题: {result.get('title', 'N/A')}")

# 反检测抓取
async def stealth_scraping_example():
    result = await scrape_with_stealth(
        url="https://protected-website.com",
        method="playwright",
        extract_config={
            "content": ".protected-content",
            "data": "[data-value]"
        }
    )
    return result

# 表单自动化
async def form_automation_example():
    result = await fill_and_submit_form(
        url="https://example.com/contact",
        form_data={
            "input[name='name']": "John Doe",
            "input[name='email']": "john@example.com",
            "textarea[name='message']": "Hello from Data Extractor!"
        },
        submit=True,
        submit_button_selector="button[type='submit']"
    )
    return result

async def ecommerce_scraping():
    # 抓取产品列表
    products_result = await scrape_webpage(
        url="https://shop.example.com/products",
        extract_config={
            "products": {
                "selector": ".product-card",
                "multiple": True,
                "attr": "text"
            },
            "prices": {
                "selector": ".price",
                "multiple": True,
                "attr": "text"
            },
            "product_links": {
                "selector": ".product-card a",
                "multiple": True,
                "attr": "href"
            }
        }
    )

    # 批量抓取产品详情
    product_urls = products_result['data']['extracted_data']['product_links']
    details = await scrape_multiple_webpages(
        urls=product_urls[:10],  # 限制前10个产品
        extract_config={
            "description": ".product-description",
            "specifications": ".specs-table",
            "images": {
                "selector": ".product-images img",
                "multiple": True,
                "attr": "src"
            }
        }
    )

    return {
        "products_overview": products_result,
        "product_details": details
    }

async def news_monitoring_system():
    news_sites = [
        "https://news.ycombinator.com",
        "https://techcrunch.com",
        "https://arstechnica.com"
    ]

    # 批量抓取新闻标题
    news_results = await scrape_multiple_webpages(
        urls=news_sites,
        extract_config={
            "headlines": {
                "selector": "h1, h2, .headline",
                "multiple": True,
                "attr": "text"
            },
            "timestamps": {
                "selector": ".timestamp, time",
                "multiple": True,
                "attr": "text"
            }
        }
    )

    # 提取所有链接用于深度分析
    all_links = []
    for site in news_sites:
        links_result = await extract_links(
            url=site,
            internal_only=True
        )
        all_links.extend(links_result['data']['links'])

    return {
        "news_headlines": news_results,
        "discovered_links": all_links
    }

async def compliance_check_workflow(target_url: str):
    # 1. 检查 robots.txt
    robots_result = await check_robots_txt(target_url)

    if not robots_result['data']['can_crawl']:
        return {"error": "网站禁止爬取", "robots_txt": robots_result}

    # 2. 获取页面基础信息
    page_info = await get_page_info(target_url)

    # 3. 执行合规的数据抓取
    scrape_result = await scrape_webpage(
        url=target_url,
        method="simple",  # 使用最轻量的方法
        extract_config={
            "public_content": ".main-content, .article",
            "meta_info": "meta[name='description']"
        }
    )

    # 4. 检查服务器性能影响
    metrics = await get_server_metrics()

    return {
        "compliance_check": robots_result,
        "page_info": page_info,
        "extracted_data": scrape_result,
        "performance_metrics": metrics
    }

# 编辑 pyproject.toml 中的 version 字段
vim pyproject.toml

# 在 CHANGELOG.md 中记录变更内容
vim CHANGELOG.md

# 更新 README.md 中的"当前最新稳定版本"
vim README.md

git add pyproject.toml CHANGELOG.md README.md
git commit -m "chore(release): bump version to vX.Y.Z"
git tag -a vX.Y.Z -m "Release version X.Y.Z"
git push && git push --tags

# 使用 uv 构建包
uv build

# 发布到 PyPI（如需要）
uv publish

# 检查当前版本
python -c "import extractor; print(extractor.__version__)"

# 或使用 uv
uv run python -c "from extractor import __version__; print(__version__)"