Crawl4ai Skill

name: Security Scan

on:
  push:
    branches: [ main, master ]
  pull_request:
    branches: [ main, master ]

jobs:
  bandit:
    name: Bandit Security Scan
    runs-on: ubuntu-latest
    
    steps:
    - name: Checkout code
      uses: actions/checkout@v4
      
    - name: Set up Python
      uses: actions/setup-python@v5
      with:
        python-version: '3.11'
        
    - name: Install bandit
      run: |
        pip install bandit[toml]
        
    - name: Run Bandit
      run: |
        bandit -r src/ -f json -o bandit-report.json || true
        bandit -r src/ -ll -f screen
        
    - name: Upload Bandit report
      uses: actions/upload-artifact@v4
      if: always()
      with:
        name: bandit-security-report
        path: bandit-report.json
        retention-days: 30

  dependency-check:
    name: Dependency Security Check
    runs-on: ubuntu-latest
    
    steps:
    - name: Checkout code
      uses: actions/checkout@v4
      
    - name: Set up Python
      uses: actions/setup-python@v5
      with:
        python-version: '3.11'
        
    - name: Install pip-audit
      run: pip install pip-audit
      
    - name: Run pip-audit
      run: pip-audit -r requirements.txt || true

# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
venv/
env/
*.egg-info/
dist/
build/

# Testing
.pytest_cache/
.coverage
htmlcov/

# IDE
.vscode/
.idea/
*.swp
*.swo

# Output
crawl_output/
search_crawl_output/
*.md
!README.md
!SKILL.md
!CONTRIBUTING.md
!CHANGELOG.md
!SECURITY.md

# Login sessions and browser data (sensitive)
sessions/
browser_data/
*.json
!package.json

#!/bin/bash
# 深度爬取示例

# 全站爬取
echo "=== 深度爬取 example.com ==="
crawl4ai-skill crawl-site https://example.com \
    --max-depth 2 \
    --max-pages 10 \
    --output-dir ./example_crawl

echo ""
echo "=== 查看爬取结果 ==="
ls -la ./example_crawl/

echo ""
echo "=== 查看统计信息 ==="
cat ./example_crawl/stats.json

echo ""
echo "完成!"

#!/bin/bash
# 搜索并爬取示例

# 搜索并爬取
echo "=== 搜索 'python web scraping' ==="
crawl4ai-skill search "python web scraping" --num-results 5

echo ""
echo "=== 搜索并爬取前 3 个结果 ==="
crawl4ai-skill search-and-crawl "python web scraping tutorials" \
    --num-results 5 \
    --crawl-top 3 \
    --output-dir ./search_results

echo ""
echo "=== 查看结果 ==="
ls -la ./search_results/

echo ""
echo "完成!"

#!/bin/bash
# 简单爬取示例

# 爬取单个页面
echo "=== 爬取 example.com ==="
crawl4ai-skill crawl https://example.com

echo ""
echo "=== 爬取并保存到文件 ==="
crawl4ai-skill crawl https://example.com -o example.md

echo ""
echo "=== 使用 markdown_with_citations 格式 ==="
crawl4ai-skill crawl https://example.com -f markdown_with_citations

echo ""
echo "完成!"

MIT License

Copyright (c) 2026 Lancelin

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.

{
  "name": "crawl4ai-skill",
  "version": "0.3.0",
  "description": "Smart web crawling and search skill for AI agents. Provides DuckDuckGo search and intelligent web scraping with LLM-optimized Markdown output.",
  "keywords": [
    "agent-skill",
    "ai-agent",
    "skills.sh",
    "openclaw",
    "web-scraping",
    "web-crawling",
    "llm",
    "markdown",
    "search",
    "crawl4ai"
  ],
  "repository": {
    "type": "git",
    "url": "https://github.com/lancelin111/crawl4ai-skill.git"
  },
  "homepage": "https://github.com/lancelin111/crawl4ai-skill",
  "bugs": {
    "url": "https://github.com/lancelin111/crawl4ai-skill/issues"
  },
  "author": "Lancelin",
  "license": "MIT",
  "files": [
    "skills"
  ],
  "scripts": {
    "postinstall": "echo '✅ crawl4ai-skill installed. For OpenClaw: clawhub install crawl4ai-skill'"
  }
}

[project]
name = "crawl4ai-skill"
version = "0.3.0"
description = "Smart web crawling and search for OpenClaw"
authors = [{name = "Lancelin"}]
readme = "README.md"
requires-python = ">=3.9"
license = {text = "MIT"}
keywords = ["crawl4ai", "openclaw", "web-scraping", "search"]

dependencies = [
    "crawl4ai>=0.8.0",
    "ddgs>=8.0.0",
    "click>=8.1.0",
    "pydantic>=2.0.0",
    "aiohttp>=3.9.0",
]

[project.optional-dependencies]
dev = [
    "pytest>=8.0.0",
    "pytest-asyncio>=0.23.0",
    "pytest-cov>=4.1.0",
    "black>=24.0.0",
    "ruff>=0.1.0",
]

[project.scripts]
crawl4ai-skill = "src.cli:cli"

[build-system]
requires = ["setuptools>=61.0"]
build-backend = "setuptools.build_meta"

[tool.setuptools.packages.find]
where = ["."]
include = ["src*"]

[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]

[tool.black]
line-length = 88
target-version = ['py39']

[tool.ruff]
line-length = 88
target-version = "py39"

crawl4ai>=0.8.0
ddgs>=8.0.0
playwright>=1.40.0
click>=8.1.0
pydantic>=2.0.0
defusedxml>=0.7.1
cryptography>=41.0.0
pytest>=8.0.0
pytest-asyncio>=0.23.0
pytest-cov>=4.1.0
black>=24.0.0
ruff>=0.1.0

"""Crawl4AI Skill - Smart web crawling and search for OpenClaw"""

__version__ = "0.1.0"

"""CLI 入口 - Click 命令行接口

提供 crawl4ai-skill 命令行工具。
"""

import asyncio
import json
import signal
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional

import click

from .search import DuckDuckGoSearcher, SearchError, RateLimitError
from .crawler import SmartCrawler, CrawlError, InvalidURLError, HTTPError
from .parser import ContentParser


# 全局标志，用于优雅退出
_interrupted = False


def handle_interrupt(signum, frame):
    """处理 Ctrl+C 中断"""
    global _interrupted
    _interrupted = True
    click.echo("\n⚠ 收到中断信号，正在优雅退出...", err=True)


# 注册信号处理
signal.signal(signal.SIGINT, handle_interrupt)


@click.group()
@click.version_option(version="0.3.0", prog_name="crawl4ai-skill")
def cli():
    """Crawl4AI Skill - 智能搜索与爬取工具

    为 OpenClaw 提供 DuckDuckGo 搜索和智能网页爬取能力。

    \b
    示例:
      crawl4ai-skill search "python web scraping"
      crawl4ai-skill crawl https://example.com
      crawl4ai-skill crawl-site https://docs.example.com
    """
    pass


@cli.command()
@click.argument("query")
@click.option("--num-results", "-n", default=10, help="搜索结果数量 (1-100)")
@click.option("--output", "-o", type=click.Path(), help="输出文件路径 (JSON 格式)")
def search(query: str, num_results: int, output: Optional[str]):
    """搜索网页

    使用 DuckDuckGo 搜索，无需 API key。

    \b
    示例:
      crawl4ai-skill search "python web scraping"
      crawl4ai-skill search "AI tutorials" --num-results 5
      crawl4ai-skill search "machine learning" -o results.json
    """
    try:
        searcher = DuckDuckGoSearcher()
        results = searcher.search(query, num_results)

        if not results:
            click.echo(f"⚠ 未找到相关结果: {query}", err=True)
            raise SystemExit(0)

        output_data = {
            "query": query,
            "num_results": len(results),
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "results": [r.to_dict() for r in results],
        }

        if output:
            try:
                with open(output, "w", encoding="utf-8") as f:
                    json.dump(output_data, f, indent=2, ensure_ascii=False)
                click.echo(f"✓ 搜索结果已保存到 {output}")
            except PermissionError:
                click.echo(f"✗ 无法写入文件: {output}，权限不足", err=True)
                raise SystemExit(1)
            except OSError as e:
                click.echo(f"✗ 无法写入文件: {output}，{e}", err=True)
                raise SystemExit(1)
        else:
            click.echo(json.dumps(output_data, indent=2, ensure_ascii=False))

    except RateLimitError as e:
        click.echo(f"✗ 搜索被限流，请稍后重试: {e}", err=True)
        raise SystemExit(1)
    except SearchError as e:
        click.echo(f"✗ 搜索失败: {e}", err=True)
        raise SystemExit(1)


@cli.command()
@click.argument("url")
@click.option(
    "--format",
    "-f",
    default="fit_markdown",
    type=click.Choice(["fit_markdown", "markdown_with_citations", "raw_markdown"]),
    help="输出格式",
)
@click.option("--output", "-o", type=click.Path(), help="输出文件路径")
@click.option("--wait-for", "-w", help="等待元素加载 (CSS selector)")
@click.option("--timeout", "-t", default=30, help="超时时间（秒）")
@click.option(
    "--wait-until",
    default="domcontentloaded",
    type=click.Choice(["domcontentloaded", "networkidle", "load", "commit"]),
    help="等待策略：domcontentloaded(默认), networkidle(动态页面推荐), load, commit",
)
@click.option(
    "--delay",
    "-d",
    default=0.1,
    type=float,
    help="返回前额外等待时间（秒），用于 JS 动态页面",
)
@click.option("--include-metadata", is_flag=True, help="在输出中包含元数据头")
def crawl(
    url: str,
    format: str,
    output: Optional[str],
    wait_for: Optional[str],
    timeout: int,
    wait_until: str,
    delay: float,
    include_metadata: bool,
):
    """爬取单个网页

    提取网页内容并转换为 Markdown 格式。

    \b
    示例:
      crawl4ai-skill crawl https://example.com
      crawl4ai-skill crawl https://example.com --format markdown_with_citations
      crawl4ai-skill crawl https://example.com -o page.md
      crawl4ai-skill crawl https://xueqiu.com/S/BIDU --wait-until networkidle --delay 2
    """
    try:
        crawler = SmartCrawler()
        result = asyncio.run(
            crawler.crawl_page(
                url,
                format=format,
                wait_for=wait_for,
                timeout=timeout,
                wait_until=wait_until,
                delay_before_return_html=delay,
            )
        )

        if result.status == "failed":
            click.echo(f"✗ 爬取失败: {result.error}", err=True)
            raise SystemExit(1)

        # 格式化输出
        parser = ContentParser()
        markdown = result.markdown

        if include_metadata:
            metadata = {
                "title": result.title,
                "url": result.url,
                "timestamp": result.crawled_at,
                "format": format,
            }
            markdown = parser.format_markdown(markdown, metadata)

        if output:
            try:
                with open(output, "w", encoding="utf-8") as f:
                    f.write(markdown)
                click.echo(f"✓ 页面已保存到 {output}")
                click.echo(f"  标题: {result.title}")
                click.echo(f"  链接数: {len(result.links)}")
            except PermissionError:
                click.echo(f"✗ 无法写入文件: {output}，权限不足", err=True)
                raise SystemExit(1)
            except OSError as e:
                click.echo(f"✗ 无法写入文件: {output}，{e}", err=True)
                raise SystemExit(1)
        else:
            click.echo(markdown)

    except InvalidURLError as e:
        click.echo(f"✗ 无效的 URL: {e}", err=True)
        raise SystemExit(1)
    except HTTPError as e:
        click.echo(f"✗ HTTP 错误: {e}", err=True)
        raise SystemExit(1)
    except CrawlError as e:
        click.echo(f"✗ 爬取失败: {e}", err=True)
        raise SystemExit(1)


@cli.command("crawl-site")
@click.argument("url")
@click.option("--max-depth", "-d", default=2, help="最大爬取深度 (1-10)")
@click.option("--max-pages", "-p", default=50, help="最大页面数量 (1-1000)")
@click.option("--include-external", is_flag=True, help="包含外部链接")
@click.option(
    "--format",
    "-f",
    default="fit_markdown",
    type=click.Choice(["fit_markdown", "markdown_with_citations", "raw_markdown"]),
    help="输出格式",
)
@click.option(
    "--output-dir",
    "-o",
    default="./crawl_output",
    type=click.Path(),
    help="输出目录",
)
@click.option(
    "--strategy",
    "-s",
    default="auto",
    type=click.Choice(["auto", "sitemap", "recursive"]),
    help="爬取策略",
)
def crawl_site(
    url: str,
    max_depth: int,
    max_pages: int,
    include_external: bool,
    format: str,
    output_dir: str,
    strategy: str,
):
    """爬取整个站点

    支持 sitemap、llms-full.txt 和递归爬取策略。

    \b
    示例:
      crawl4ai-skill crawl-site https://docs.example.com
      crawl4ai-skill crawl-site https://example.com/sitemap.xml --strategy sitemap
      crawl4ai-skill crawl-site https://example.com --max-depth 3 --max-pages 100
    """
    try:
        # 创建输出目录
        output_path = Path(output_dir)
        try:
            output_path.mkdir(parents=True, exist_ok=True)
            pages_path = output_path / "pages"
            pages_path.mkdir(exist_ok=True)
        except PermissionError:
            click.echo(f"✗ 无法创建目录: {output_dir}，权限不足", err=True)
            raise SystemExit(1)
        except OSError as e:
            click.echo(f"✗ 无法创建目录: {output_dir}，{e}", err=True)
            raise SystemExit(1)

        click.echo(f"开始爬取: {url}")
        click.echo(f"策略: {strategy}, 最大深度: {max_depth}, 最大页面: {max_pages}")

        crawler = SmartCrawler()
        results = asyncio.run(
            crawler.crawl_site(
                url,
                max_depth=max_depth,
                max_pages=max_pages,
                include_external=include_external,
                format=format,
                strategy=strategy,
            )
        )

        if not results:
            click.echo("⚠ 未爬取到任何页面", err=True)
            return

        # 保存结果
        parser = ContentParser()
        index_data = {
            "start_url": url,
            "crawled_at": datetime.now(timezone.utc).isoformat(),
            "strategy": strategy,
            "max_depth": max_depth,
            "max_pages": max_pages,
            "pages": [],
        }

        success_count = 0
        failed_count = 0

        for i, result in enumerate(results, 1):
            page_id = f"page_{i:03d}"
            page_file = f"pages/{page_id}.md"

            if result.status == "success":
                success_count += 1
                # 保存页面内容
                try:
                    with open(output_path / page_file, "w", encoding="utf-8") as f:
                        metadata = {
                            "title": result.title,
                            "url": result.url,
                            "timestamp": result.crawled_at,
                            "format": format,
                        }
                        content = parser.format_markdown(result.markdown, metadata)
                        f.write(content)
                except (PermissionError, OSError) as e:
                    click.echo(f"  ⚠ 无法保存文件 {page_file}: {e}", err=True)
                    failed_count += 1
                    success_count -= 1
            else:
                failed_count += 1

            index_data["pages"].append(
                {
                    "id": page_id,
                    "url": result.url,
                    "title": result.title,
                    "depth": result.depth,
                    "file": page_file if result.status == "success" else None,
                    "status": result.status,
                    "error": result.error,
                    "links_found": len(result.links),
                }
            )

            click.echo(
                f"  [{i}/{len(results)}] {result.status}: {result.url[:60]}..."
                if len(result.url) > 60
                else f"  [{i}/{len(results)}] {result.status}: {result.url}"
            )

        # 保存索引
        with open(output_path / "index.json", "w", encoding="utf-8") as f:
            json.dump(index_data, f, indent=2, ensure_ascii=False)

        # 保存统计信息
        stats = {
            "total_pages": len(results),
            "successful": success_count,
            "failed": failed_count,
            "total_links_found": sum(len(r.links) for r in results),
        }
        with open(output_path / "stats.json", "w", encoding="utf-8") as f:
            json.dump(stats, f, indent=2, ensure_ascii=False)

        click.echo()
        click.echo(f"✓ 爬取完成!")
        click.echo(f"  成功: {success_count}, 失败: {failed_count}")
        click.echo(f"  输出目录: {output_path}")

    except InvalidURLError as e:
        click.echo(f"✗ 无效的 URL: {e}", err=True)
        raise SystemExit(1)
    except HTTPError as e:
        click.echo(f"✗ HTTP 错误: {e}", err=True)
        raise SystemExit(1)
    except CrawlError as e:
        click.echo(f"✗ 爬取失败: {e}", err=True)
        raise SystemExit(1)


@cli.command("search-and-crawl")
@click.argument("query")
@click.option("--num-results", "-n", default=5, help="搜索结果数量")
@click.option("--crawl-top", "-c", default=3, help="爬取前 N 个结果")
@click.option(
    "--format",
    "-f",
    default="fit_markdown",
    type=click.Choice(["fit_markdown", "markdown_with_citations", "raw_markdown"]),
    help="输出格式",
)
@click.option(
    "--output-dir",
    "-o",
    default="./search_crawl_output",
    type=click.Path(),
    help="输出目录",
)
def search_and_crawl(
    query: str,
    num_results: int,
    crawl_top: int,
    format: str,
    output_dir: str,
):
    """搜索并爬取

    先搜索，再爬取前 N 个结果。

    \b
    示例:
      crawl4ai-skill search-and-crawl "python web scraping tutorials"
      crawl4ai-skill search-and-crawl "AI tutorials" --num-results 10 --crawl-top 5
    """
    global _interrupted

    # 参数验证: crawl_top 不能超过 num_results
    if crawl_top > num_results:
        click.echo(f"⚠ crawl-top ({crawl_top}) 大于 num-results ({num_results})，已自动调整为 {num_results}", err=True)
        crawl_top = num_results

    try:
        # 创建输出目录
        output_path = Path(output_dir)
        try:
            output_path.mkdir(parents=True, exist_ok=True)
        except PermissionError:
            click.echo(f"✗ 无法创建目录: {output_dir}，权限不足", err=True)
            raise SystemExit(1)
        except OSError as e:
            click.echo(f"✗ 无法创建目录: {output_dir}，{e}", err=True)
            raise SystemExit(1)

        # 1. 搜索
        click.echo(f"搜索: {query}")
        searcher = DuckDuckGoSearcher()
        search_results = searcher.search(query, num_results)

        if not search_results:
            click.echo(f"⚠ 未找到相关结果: {query}", err=True)
            # 保存空结果
            search_data = {
                "query": query,
                "timestamp": datetime.now(timezone.utc).isoformat(),
                "results": [],
            }
            with open(output_path / "search_results.json", "w", encoding="utf-8") as f:
                json.dump(search_data, f, indent=2, ensure_ascii=False)
            click.echo(f"✓ 空结果已保存到 {output_path}")
            return

        click.echo(f"  找到 {len(search_results)} 条结果")

        # 保存搜索结果
        search_data = {
            "query": query,
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "results": [r.to_dict() for r in search_results],
        }
        with open(output_path / "search_results.json", "w", encoding="utf-8") as f:
            json.dump(search_data, f, indent=2, ensure_ascii=False)

        # 2. 爬取前 N 个
        urls_to_crawl = [r.url for r in search_results[:crawl_top]]
        click.echo(f"\n爬取前 {len(urls_to_crawl)} 个结果:")

        crawler = SmartCrawler()
        parser = ContentParser()
        crawl_results = []

        for i, url in enumerate(urls_to_crawl, 1):
            # 检查是否被中断
            if _interrupted:
                click.echo("\n⚠ 用户中断，保存已完成的结果...", err=True)
                break

            click.echo(f"  [{i}/{len(urls_to_crawl)}] {url[:60]}...")
            try:
                result = asyncio.run(crawler.crawl_page(url, format=format))
                crawl_results.append(result)

                # 保存页面
                if result.status == "success":
                    filename = f"page_{i:02d}.md"
                    with open(output_path / filename, "w", encoding="utf-8") as f:
                        metadata = {
                            "title": result.title,
                            "url": result.url,
                            "timestamp": result.crawled_at,
                            "format": format,
                        }
                        content = parser.format_markdown(result.markdown, metadata)
                        f.write(content)
                    click.echo(f"    ✓ 已保存: {filename}")
                else:
                    click.echo(f"    ✗ 失败: {result.error}")

            except Exception as e:
                click.echo(f"    ✗ 错误: {e}")

        # 保存索引
        index_data = {
            "query": query,
            "crawled_at": datetime.now(timezone.utc).isoformat(),
            "search_results": len(search_results),
            "crawled_pages": len([r for r in crawl_results if r.status == "success"]),
            "interrupted": _interrupted,
            "pages": [
                {
                    "url": r.url,
                    "title": r.title,
                    "status": r.status,
                }
                for r in crawl_results
            ],
        }
        with open(output_path / "index.json", "w", encoding="utf-8") as f:
            json.dump(index_data, f, indent=2, ensure_ascii=False)

        click.echo(f"\n✓ 完成! 输出目录: {output_path}")

    except RateLimitError as e:
        click.echo(f"✗ 搜索被限流，请稍后重试: {e}", err=True)
        raise SystemExit(1)
    except SearchError as e:
        click.echo(f"✗ 搜索失败: {e}", err=True)
        raise SystemExit(1)
    except InvalidURLError as e:
        click.echo(f"✗ 无效的 URL: {e}", err=True)
        raise SystemExit(1)
    except CrawlError as e:
        click.echo(f"✗ 爬取失败: {e}", err=True)
        raise SystemExit(1)


if __name__ == "__main__":
    cli()

"""配置管理 - 使用 Pydantic"""

from typing import Literal, Optional
from pydantic import BaseModel, Field


class CrawlConfig(BaseModel):
    """爬取配置"""

    timeout: int = Field(default=30, ge=1, le=300, description="超时时间（秒）")
    max_depth: int = Field(default=2, ge=1, le=10, description="最大爬取深度")
    max_pages: int = Field(default=50, ge=1, le=1000, description="最大页面数量")
    include_external: bool = Field(default=False, description="是否包含外部链接")
    format: Literal["fit_markdown", "markdown_with_citations", "raw_markdown"] = Field(
        default="fit_markdown", description="输出格式"
    )
    wait_for: Optional[str] = Field(default=None, description="等待元素（CSS selector）")


class SearchConfig(BaseModel):
    """搜索配置"""

    num_results: int = Field(default=10, ge=1, le=100, description="搜索结果数量")
    timeout: int = Field(default=10, ge=1, le=60, description="超时时间（秒）")


class SiteCrawlConfig(CrawlConfig):
    """全站爬取配置"""

    strategy: Literal["auto", "sitemap", "recursive"] = Field(
        default="auto", description="爬取策略"
    )
    output_dir: str = Field(default="./crawl_output", description="输出目录")


class SearchAndCrawlConfig(BaseModel):
    """搜索并爬取配置"""

    num_results: int = Field(default=5, ge=1, le=100, description="搜索结果数量")
    crawl_top: int = Field(default=3, ge=1, le=20, description="爬取前 N 个结果")
    format: Literal["fit_markdown", "markdown_with_citations", "raw_markdown"] = Field(
        default="fit_markdown", description="输出格式"
    )
    output_dir: str = Field(default="./search_crawl_output", description="输出目录")

"""爬取模块 - 基于 crawl4ai 的网页爬取

提供单页爬取和全站爬取功能，支持多种 Markdown 输出格式。
"""

import asyncio
import random
try:
    import defusedxml.ElementTree as ET
except ImportError:
    # Fallback to standard library with manual sanitization
    import xml.etree.ElementTree as ET
    import warnings
    warnings.warn("defusedxml not installed, using xml.etree.ElementTree (less secure)")
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Optional, List, Set, Dict, Any
from urllib.parse import urljoin, urlparse
import logging
import re

logger = logging.getLogger(__name__)


def _validate_url_scheme(url: str) -> None:
    """验证 URL 只允许 http/https 协议
    
    Args:
        url: 要验证的 URL
        
    Raises:
        ValueError: 如果 URL 使用了不安全的协议（如 file://）
    """
    parsed = urlparse(url)
    if parsed.scheme not in ('http', 'https'):
        raise ValueError(f"不安全的 URL 协议: {parsed.scheme}。只允许 http 和 https。")


class CrawlError(Exception):
    """爬取相关错误的基类"""
    pass


class CrawlTimeoutError(CrawlError):
    """超时错误"""
    pass


class CrawlNetworkError(CrawlError):
    """网络错误"""
    pass


class InvalidURLError(CrawlError):
    """无效 URL 错误"""
    pass


class HTTPError(CrawlError):
    """HTTP 错误"""
    def __init__(self, message: str, status_code: int = None):
        super().__init__(message)
        self.status_code = status_code


def validate_url(url: str) -> None:
    """验证 URL 是否有效

    Args:
        url: 待验证的 URL

    Raises:
        InvalidURLError: URL 无效
    """
    if not url or not url.strip():
        raise InvalidURLError("URL 不能为空")

    url = url.strip()
    parsed = urlparse(url)

    if parsed.scheme not in ('http', 'https'):
        raise InvalidURLError(f"不支持的协议: {parsed.scheme}，只支持 http/https")

    if not parsed.netloc:
        raise InvalidURLError(f"无效的 URL: {url}")


@dataclass
class CrawlResult:
    """爬取结果数据结构

    Attributes:
        url: 爬取的 URL
        title: 页面标题
        markdown: Markdown 内容
        links: 页面中的链接
        status: 爬取状态 (success/failed)
        error: 错误信息（如果失败）
        depth: 爬取深度
        crawled_at: 爬取时间
    """

    url: str
    title: str
    markdown: str
    links: List[str] = field(default_factory=list)
    status: str = "success"
    error: Optional[str] = None
    depth: int = 0
    crawled_at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())

    def to_dict(self) -> Dict[str, Any]:
        """转换为字典"""
        return {
            "url": self.url,
            "title": self.title,
            "markdown": self.markdown,
            "links": self.links,
            "status": self.status,
            "error": self.error,
            "depth": self.depth,
            "crawled_at": self.crawled_at,
        }


class SmartCrawler:
    """智能爬虫

    基于 crawl4ai 提供单页和全站爬取功能。

    Example:
        >>> crawler = SmartCrawler()
        >>> result = await crawler.crawl_page("https://example.com")
        >>> print(result.markdown)
    """

    def __init__(self, verbose: bool = False):
        """初始化爬虫

        Args:
            verbose: 是否输出详细日志
        """
        self.verbose = verbose
        if verbose:
            logging.basicConfig(level=logging.DEBUG)

    async def crawl_page(
        self,
        url: str,
        format: str = "fit_markdown",
        wait_for: Optional[str] = None,
        timeout: int = 30,
        wait_until: str = "domcontentloaded",
        delay_before_return_html: float = 0.1,
    ) -> CrawlResult:
        """爬取单个页面

        Args:
            url: 目标 URL
            format: 输出格式 (fit_markdown/markdown_with_citations/raw_markdown)
            wait_for: 等待元素加载（CSS selector）
            timeout: 超时时间（秒）
            wait_until: 等待策略 (domcontentloaded/networkidle/load/commit)
            delay_before_return_html: 返回 HTML 前的额外延迟（秒），用于等待 JS 执行

        Returns:
            爬取结果

        Raises:
            CrawlTimeoutError: 超时
            CrawlNetworkError: 网络错误
            CrawlError: 其他爬取错误
        """
        try:
            from crawl4ai import AsyncWebCrawler
            from crawl4ai.async_configs import CrawlerRunConfig
        except ImportError:
            raise CrawlError("crawl4ai 未安装，请运行: pip install crawl4ai && crawl4ai-setup")

        # 验证 URL
        validate_url(url)

        try:
            logger.info(f"开始爬取: {url}")

            # 创建配置对象
            config = CrawlerRunConfig(
                wait_for=wait_for if wait_for else None,
                wait_until=wait_until,
                page_timeout=timeout * 1000,  # 转换为毫秒
                delay_before_return_html=delay_before_return_html,
            )

            async with AsyncWebCrawler(verbose=self.verbose) as crawler:
                result_container = await crawler.arun(
                    url=url,
                    config=config,
                )

                if not result_container.success:
                    error_msg = "Unknown error"
                    if hasattr(result_container, 'error_message'):
                        error_msg = result_container.error_message or error_msg
                    return CrawlResult(
                        url=url,
                        title="",
                        markdown="",
                        links=[],
                        status="failed",
                        error=error_msg,
                    )

                # crawl4ai 0.8.0: 结果是 CrawlResultContainer，需要获取第一个结果
                first_result = result_container[0] if hasattr(result_container, '__getitem__') else result_container

                # 获取标题 - 从 metadata 中获取
                title = ""
                if hasattr(first_result, 'metadata') and first_result.metadata:
                    title = first_result.metadata.get('title', '') or ""

                # 根据 format 选择输出格式
                # crawl4ai 0.8.0: 简化处理，直接使用 markdown 字符串
                markdown_str = str(result_container.markdown) if result_container.markdown else ""

                if format == "fit_markdown":
                    # 尝试获取 fit_markdown，如果不存在则使用原始 markdown
                    try:
                        fit_md = first_result.markdown.fit_markdown if hasattr(first_result.markdown, 'fit_markdown') else None
                        markdown = fit_md if fit_md else markdown_str
                    except (AttributeError, TypeError):
                        markdown = markdown_str
                elif format == "markdown_with_citations":
                    # 尝试获取带引用的 markdown
                    try:
                        if hasattr(first_result, 'markdown_v2') and first_result.markdown_v2:
                            citations_md = getattr(first_result.markdown_v2, 'markdown_with_citations', None)
                            markdown = citations_md if citations_md else markdown_str
                        else:
                            markdown = markdown_str
                    except (AttributeError, TypeError):
                        markdown = markdown_str
                else:
                    markdown = markdown_str

                # 提取链接
                links_dict = result_container.links if hasattr(result_container, 'links') else {}
                internal_links = links_dict.get("internal", []) if links_dict else []
                external_links = links_dict.get("external", []) if links_dict else []

                # 将链接对象转换为 URL 字符串
                all_links = []
                for link in internal_links:
                    if isinstance(link, dict):
                        all_links.append(link.get("href", ""))
                    else:
                        all_links.append(str(link))
                for link in external_links:
                    if isinstance(link, dict):
                        all_links.append(link.get("href", ""))
                    else:
                        all_links.append(str(link))

                logger.info(f"爬取完成: {url}, 标题: {title}")

                return CrawlResult(
                    url=url,
                    title=title,
                    markdown=markdown,
                    links=all_links,
                    status="success",
                )

        except asyncio.TimeoutError as e:
            logger.error(f"爬取超时: {url}")
            raise CrawlTimeoutError(f"爬取超时: {url}") from e
        except Exception as e:
            logger.error(f"爬取失败: {url}, 错误: {e}")
            raise CrawlError(f"爬取失败: {e}") from e

    async def crawl_site(
        self,
        start_url: str,
        max_depth: int = 2,
        max_pages: int = 50,
        include_external: bool = False,
        format: str = "fit_markdown",
        strategy: str = "auto",
    ) -> List[CrawlResult]:
        """爬取整个站点

        Args:
            start_url: 起始 URL
            max_depth: 最大爬取深度
            max_pages: 最大页面数量
            include_external: 是否包含外部链接
            format: 输出格式
            strategy: 爬取策略 (auto/sitemap/recursive)

        Returns:
            爬取结果列表
        """
        # 自动识别策略
        if strategy == "auto":
            strategy = self._detect_strategy(start_url)

        logger.info(f"开始全站爬取: {start_url}, 策略: {strategy}")

        # 根据策略执行爬取
        if strategy == "sitemap":
            return await self._crawl_from_sitemap(start_url, max_pages, format)
        elif strategy == "txt":
            return await self._crawl_from_txt(start_url, max_pages, format)
        else:
            return await self._crawl_recursive(
                start_url, max_depth, max_pages, include_external, format
            )

    def _detect_strategy(self, url: str) -> str:
        """检测爬取策略

        Args:
            url: URL

        Returns:
            策略名称
        """
        if url.endswith("sitemap.xml") or "sitemap" in url.lower():
            return "sitemap"
        elif url.endswith(".txt") and ("llms" in url.lower() or "links" in url.lower()):
            return "txt"
        else:
            return "recursive"

    async def _crawl_from_sitemap(
        self, sitemap_url: str, max_pages: int, format: str
    ) -> List[CrawlResult]:
        """从 sitemap.xml 爬取

        Args:
            sitemap_url: sitemap URL
            max_pages: 最大页面数
            format: 输出格式

        Returns:
            爬取结果列表
        """
        # 验证 URL
        validate_url(sitemap_url)

        try:
            import aiohttp
        except ImportError:
            import urllib.request
            import urllib.error
            # 使用同步请求作为回退
            try:
                # 验证 URL 协议安全性
                _validate_url_scheme(sitemap_url)
                with urllib.request.urlopen(sitemap_url, timeout=30) as response:  # nosec B310 - URL scheme validated
                    content = response.read().decode('utf-8')
            except urllib.error.HTTPError as e:
                raise HTTPError(f"获取 sitemap 失败: HTTP {e.code}", status_code=e.code)
            except urllib.error.URLError as e:
                raise CrawlNetworkError(f"获取 sitemap 失败: {e.reason}")
        else:
            async with aiohttp.ClientSession() as session:
                try:
                    async with session.get(sitemap_url) as response:
                        if response.status >= 400:
                            raise HTTPError(f"获取 sitemap 失败: HTTP {response.status}", status_code=response.status)
                        content = await response.text()
                except aiohttp.ClientError as e:
                    raise CrawlNetworkError(f"获取 sitemap 失败: {e}")

        # 解析 XML
        try:
            # Using defusedxml.ElementTree when available (fallback to xml.etree.ElementTree)
            root = ET.fromstring(content)  # nosec B314 - using defusedxml when available
            # 移除命名空间以简化解析
            for elem in root.iter():
                if '}' in elem.tag:
                    elem.tag = elem.tag.split('}', 1)[1]

            # 检查是否是 sitemap index（嵌套 sitemap）
            sitemap_locs = [loc.text for loc in root.findall('.//sitemap/loc') if loc.text]
            if sitemap_locs:
                logger.info(f"发现 sitemap index，包含 {len(sitemap_locs)} 个子 sitemap")
                # 递归获取所有子 sitemap 的 URL
                all_urls = []
                for sub_sitemap in sitemap_locs:
                    if len(all_urls) >= max_pages:
                        break
                    try:
                        sub_results = await self._crawl_from_sitemap(
                            sub_sitemap, max_pages - len(all_urls), format
                        )
                        # 只收集 URL，稍后统一爬取
                        for r in sub_results:
                            if len(all_urls) >= max_pages:
                                break
                            all_urls.append(r.url)
                    except Exception as e:
                        logger.warning(f"获取子 sitemap 失败: {sub_sitemap}, 错误: {e}")
                # 对于 sitemap index，直接返回已爬取的结果
                return sub_results if sub_results else []

            urls = [loc.text for loc in root.findall('.//loc') if loc.text]
        except ET.ParseError as e:
            logger.error(f"解析 sitemap 失败: {sitemap_url}")
            raise CrawlError(f"解析 sitemap XML 失败: {e}")

        if not urls:
            logger.warning(f"sitemap 中未发现任何 URL: {sitemap_url}")
            return []

        # 限制页面数量
        urls = urls[:max_pages]

        logger.info(f"从 sitemap 发现 {len(urls)} 个 URL")

        # 爬取所有 URL，添加延时
        results = []
        for i, url in enumerate(urls):
            try:
                result = await self.crawl_page(url, format=format)
                results.append(result)
            except Exception as e:
                logger.error(f"爬取失败: {url}, 错误: {e}")
                results.append(CrawlResult(
                    url=url,
                    title="",
                    markdown="",
                    status="failed",
                    error=str(e),
                ))

            if len(results) >= max_pages:
                break

            # 随机延迟 0.1~0.3 秒，模拟人类行为，避免触发速率限制
            await asyncio.sleep(random.uniform(0.1, 0.3))

        return results

    async def _crawl_from_txt(
        self, txt_url: str, max_pages: int, format: str
    ) -> List[CrawlResult]:
        """从 txt 文件爬取（llms-full.txt 等）

        Args:
            txt_url: txt 文件 URL
            max_pages: 最大页面数
            format: 输出格式

        Returns:
            爬取结果列表
        """
        # 验证 URL
        validate_url(txt_url)

        try:
            import aiohttp
        except ImportError:
            import urllib.request
            import urllib.error
            try:
                # 验证 URL 协议安全性
                _validate_url_scheme(txt_url)
                with urllib.request.urlopen(txt_url, timeout=30) as response:  # nosec B310 - URL scheme validated
                    content = response.read().decode('utf-8')
            except urllib.error.HTTPError as e:
                raise HTTPError(f"获取 txt 文件失败: HTTP {e.code}", status_code=e.code)
            except urllib.error.URLError as e:
                raise CrawlNetworkError(f"获取 txt 文件失败: {e.reason}")
        else:
            async with aiohttp.ClientSession() as session:
                try:
                    async with session.get(txt_url) as response:
                        if response.status >= 400:
                            raise HTTPError(f"获取 txt 文件失败: HTTP {response.status}", status_code=response.status)
                        content = await response.text()
                except aiohttp.ClientError as e:
                    raise CrawlNetworkError(f"获取 txt 文件失败: {e}")

        # 解析 URL（每行一个）
        urls = []
        for line in content.strip().split('\n'):
            line = line.strip()
            if line and (line.startswith('http://') or line.startswith('https://')):
                urls.append(line)

        if not urls:
            logger.warning(f"txt 文件中未发现任何 URL: {txt_url}")
            return []

        urls = urls[:max_pages]
        logger.info(f"从 txt 文件发现 {len(urls)} 个 URL")

        # 爬取所有 URL，添加延时
        results = []
        for i, url in enumerate(urls):
            try:
                result = await self.crawl_page(url, format=format)
                results.append(result)
            except Exception as e:
                logger.error(f"爬取失败: {url}, 错误: {e}")
                results.append(CrawlResult(
                    url=url,
                    title="",
                    markdown="",
                    status="failed",
                    error=str(e),
                ))

            if len(results) >= max_pages:
                break

            # 随机延迟 0.1~0.3 秒，模拟人类行为，避免触发速率限制
            await asyncio.sleep(random.uniform(0.1, 0.3))

        return results

    async def _crawl_recursive(
        self,
        start_url: str,
        max_depth: int,
        max_pages: int,
        include_external: bool,
        format: str,
    ) -> List[CrawlResult]:
        """递归爬取（BFS）

        Args:
            start_url: 起始 URL
            max_depth: 最大深度
            max_pages: 最大页面数
            include_external: 是否包含外部链接
            format: 输出格式

        Returns:
            爬取结果列表
        """
        # 验证起始 URL
        validate_url(start_url)

        # 解析基础域名
        parsed_start = urlparse(start_url)
        base_domain = parsed_start.netloc

        # BFS 队列: (url, depth)
        queue: List[tuple] = [(start_url, 0)]
        visited: Set[str] = set()
        results: List[CrawlResult] = []

        while queue and len(results) < max_pages:
            url, depth = queue.pop(0)

            # 跳过已访问的 URL
            if url in visited:
                continue

            # 跳过超过最大深度的 URL
            if depth > max_depth:
                continue

            visited.add(url)

            # 爬取页面
            try:
                result = await self.crawl_page(url, format=format)
                result.depth = depth
                results.append(result)

                logger.info(f"已爬取 {len(results)}/{max_pages} 页, 深度: {depth}")

                # 提取新链接加入队列
                if depth < max_depth:
                    for link in result.links:
                        if not link or link in visited:
                            continue

                        # 标准化链接
                        if link.startswith('/'):
                            link = urljoin(url, link)

                        parsed_link = urlparse(link)

                        # 检查是否为外部链接
                        is_external = parsed_link.netloc != base_domain

                        if is_external and not include_external:
                            continue

                        # 跳过非 HTTP 链接
                        if parsed_link.scheme not in ('http', 'https'):
                            continue

                        # 跳过锚点、javascript 等
                        if link.startswith('#') or link.startswith('javascript:'):
                            continue

                        queue.append((link, depth + 1))

            except Exception as e:
                logger.error(f"爬取失败: {url}, 错误: {e}")
                results.append(CrawlResult(
                    url=url,
                    title="",
                    markdown="",
                    status="failed",
                    error=str(e),
                    depth=depth,
                ))

            # 随机延迟 0.1~0.3 秒，模拟人类行为，避免触发速率限制
            await asyncio.sleep(random.uniform(0.1, 0.3))

        logger.info(f"递归爬取完成，共爬取 {len(results)} 页")
        return results

"""解析模块 - 内容格式化和后处理

提供 Markdown 格式化、引用添加等功能。
"""

from datetime import datetime, timezone
from typing import Dict, Any, List, Optional
import re


class ContentParser:
    """内容解析器

    用于格式化和后处理爬取的内容。

    Example:
        >>> parser = ContentParser()
        >>> formatted = parser.format_markdown(content, metadata)
        >>> print(formatted)
    """

    def format_markdown(
        self,
        content: str,
        metadata: Dict[str, Any],
        include_header: bool = True,
    ) -> str:
        """格式化 Markdown 输出

        Args:
            content: 原始 markdown 内容
            metadata: 元数据（URL、标题、时间等）
            include_header: 是否包含页头

        Returns:
            格式化后的 markdown
        """
        if not include_header:
            return content

        title = metadata.get("title", "Untitled")
        url = metadata.get("url", "")
        timestamp = metadata.get("timestamp", datetime.now(timezone.utc).isoformat())
        format_type = metadata.get("format", "markdown")

        header = f"""# {title}

## Metadata
- **URL**: {url}
- **Crawled at**: {timestamp}
- **Format**: {format_type}

---

"""
        return header + content

    def add_citations(self, content: str, links: List[str]) -> str:
        """添加引用列表

        Args:
            content: markdown 内容
            links: 链接列表

        Returns:
            带引用的 markdown
        """
        if not links:
            return content

        # 过滤空链接
        valid_links = [link for link in links if link and link.strip()]

        if not valid_links:
            return content

        citations = "\n\n---\n\n## References\n\n"
        for i, link in enumerate(valid_links, 1):
            citations += f"[{i}]: {link}\n"

        return content + citations

    def clean_markdown(self, content: str) -> str:
        """清理 Markdown 内容

        移除多余空行、修复格式问题等。

        Args:
            content: 原始 markdown

        Returns:
            清理后的 markdown
        """
        if not content:
            return ""

        # 移除多余空行（保留最多两个连续空行）
        content = re.sub(r'\n{4,}', '\n\n\n', content)

        # 移除行尾空白
        content = '\n'.join(line.rstrip() for line in content.split('\n'))

        # 确保文件以单个换行结尾
        content = content.strip() + '\n'

        return content

    def extract_title_from_markdown(self, content: str) -> Optional[str]:
        """从 Markdown 中提取标题

        Args:
            content: markdown 内容

        Returns:
            提取的标题或 None
        """
        if not content:
            return None

        # 查找第一个 h1 标题
        match = re.search(r'^#\s+(.+)$', content, re.MULTILINE)
        if match:
            return match.group(1).strip()

        return None

    def truncate_content(
        self,
        content: str,
        max_length: int = 10000,
        suffix: str = "\n\n... (内容已截断)"
    ) -> str:
        """截断内容

        Args:
            content: 原始内容
            max_length: 最大长度
            suffix: 截断后缀

        Returns:
            截断后的内容
        """
        if not content or len(content) <= max_length:
            return content

        # 在适当位置截断（尝试在段落边界）
        truncated = content[:max_length]

        # 尝试在最后一个段落结束处截断
        last_para = truncated.rfind('\n\n')
        if last_para > max_length * 0.8:
            truncated = truncated[:last_para]

        return truncated + suffix

    def merge_results(
        self,
        results: List[Dict[str, Any]],
        separator: str = "\n\n---\n\n"
    ) -> str:
        """合并多个爬取结果

        Args:
            results: 爬取结果列表
            separator: 分隔符

        Returns:
            合并后的 markdown
        """
        if not results:
            return ""

        parts = []
        for result in results:
            title = result.get("title", "Untitled")
            url = result.get("url", "")
            content = result.get("markdown", "")

            part = f"## {title}\n\n**Source**: {url}\n\n{content}"
            parts.append(part)

        return separator.join(parts)

"""搜索模块 - 提供 DuckDuckGo 网页搜索

这个模块封装了 DuckDuckGo 搜索功能，提供简洁的 API 用于网页搜索。
"""

from dataclasses import dataclass, asdict
from datetime import datetime, timezone
from typing import List, Optional
import logging

# 兼容新旧包名
try:
    from ddgs import DDGS
    from ddgs.exceptions import DDGSException as DuckDuckGoSearchException
except ImportError:
    from duckduckgo_search import DDGS
    from duckduckgo_search.exceptions import DuckDuckGoSearchException

logger = logging.getLogger(__name__)


class SearchError(Exception):
    """搜索相关错误的基类"""

    pass


class EmptyQueryError(SearchError):
    """空查询错误"""

    pass


class SearchNetworkError(SearchError):
    """网络错误"""

    pass


class RateLimitError(SearchError):
    """限流错误"""

    pass


@dataclass
class SearchResult:
    """搜索结果数据结构

    Attributes:
        title: 结果标题
        url: 结果 URL
        snippet: 结果摘要
        timestamp: 搜索时间戳 (ISO 格式)
    """

    title: str
    url: str
    snippet: str
    timestamp: str

    def to_dict(self) -> dict:
        """转换为字典"""
        return asdict(self)


class DuckDuckGoSearcher:
    """DuckDuckGo 搜索器

    使用 DuckDuckGo 进行网页搜索，无需 API key。

    Example:
        >>> searcher = DuckDuckGoSearcher()
        >>> results = searcher.search("python web scraping", num_results=5)
        >>> for r in results:
        ...     print(f"{r.title}: {r.url}")
    """

    def __init__(self, timeout: int = 10):
        """初始化搜索器

        Args:
            timeout: 请求超时时间（秒）
        """
        self.timeout = timeout
        self._ddgs: Optional[DDGS] = None

    @property
    def ddgs(self) -> DDGS:
        """懒加载 DDGS 实例"""
        if self._ddgs is None:
            try:
                self._ddgs = DDGS(timeout=self.timeout)
            except TypeError:
                # 新版 ddgs 包可能不支持 timeout 参数
                self._ddgs = DDGS()
        return self._ddgs

    def search(self, query: str, num_results: int = 10) -> List[SearchResult]:
        """执行搜索

        Args:
            query: 搜索关键词
            num_results: 返回结果数量 (1-100)

        Returns:
            搜索结果列表

        Raises:
            EmptyQueryError: 查询为空
            SearchNetworkError: 网络请求失败
            SearchError: 其他搜索错误
        """
        # 验证查询
        if not query or not query.strip():
            raise EmptyQueryError("搜索查询不能为空")

        query = query.strip()

        # 限制结果数量
        num_results = max(1, min(100, num_results))

        try:
            logger.info(f"搜索: '{query}', 期望结果数: {num_results}")

            # 执行搜索
            results = self.ddgs.text(
                query,
                max_results=num_results,
                safesearch="moderate",
            )

            # 获取当前时间戳
            timestamp = datetime.now(timezone.utc).isoformat()

            # 转换为 SearchResult 对象
            search_results = []
            for r in results:
                result = SearchResult(
                    title=r.get("title", ""),
                    url=r.get("href", ""),
                    snippet=r.get("body", ""),
                    timestamp=timestamp,
                )
                search_results.append(result)

            logger.info(f"搜索完成，返回 {len(search_results)} 条结果")
            return search_results

        except DuckDuckGoSearchException as e:
            logger.error(f"DuckDuckGo 搜索失败: {e}")
            raise SearchNetworkError(f"搜索请求失败: {e}") from e
        except Exception as e:
            logger.error(f"搜索时发生未知错误: {e}")
            raise SearchError(f"搜索失败: {e}") from e

    def search_news(self, query: str, num_results: int = 10) -> List[SearchResult]:
        """搜索新闻

        Args:
            query: 搜索关键词
            num_results: 返回结果数量

        Returns:
            新闻搜索结果列表
        """
        if not query or not query.strip():
            raise EmptyQueryError("搜索查询不能为空")

        query = query.strip()
        num_results = max(1, min(100, num_results))

        try:
            logger.info(f"搜索新闻: '{query}'")

            results = self.ddgs.news(query, max_results=num_results)
            timestamp = datetime.now(timezone.utc).isoformat()

            search_results = []
            for r in results:
                result = SearchResult(
                    title=r.get("title", ""),
                    url=r.get("url", ""),
                    snippet=r.get("body", ""),
                    timestamp=r.get("date", timestamp),
                )
                search_results.append(result)

            return search_results

        except DuckDuckGoSearchException as e:
            raise SearchNetworkError(f"新闻搜索请求失败: {e}") from e
        except Exception as e:
            raise SearchError(f"新闻搜索失败: {e}") from e