from scrapling.fetchers import Fetcher, FetcherSession

# 单次请求
page = Fetcher.get('https://quotes.toscrape.com/')
quotes = page.css('.quote .text::text').getall()

# 会话请求（保持 Cookie）
with FetcherSession(impersonate='chrome') as session:
    page = session.get('https://quotes.toscrape.com/', stealthy_headers=True)
    data = page.css('.quote .text::text').getall()

from scrapling.fetchers import StealthyFetcher, StealthySession

# 单次请求（自动打开浏览器，完成后关闭）
page = StealthyFetcher.fetch(
    'https://nopecha.com/demo/cloudflare',
    solve_cloudflare=True
)
data = page.css('#padded_content a').getall()

# 会话请求（保持浏览器会话）
with StealthySession(headless=True, solve_cloudflare=True) as session:
    page = session.fetch('https://nopecha.com/demo/cloudflare', google_search=False)
    data = page.css('#padded_content a').getall()

from scrapling.fetchers import DynamicFetcher, DynamicSession

# 单次请求
page = DynamicFetcher.fetch(
    'https://quotes.toscrape.com/',
    headless=True,
    network_idle=True
)
data = page.css('.quote .text::text').getall()

# 会话请求
with DynamicSession(headless=True, disable_resources=False, network_idle=True) as session:
    page = session.fetch('https://quotes.toscrape.com/', load_dom=False)
    # XPath 或 CSS 选择器都可以
    data = page.xpath('//span[@class="text"]/text()').getall()

from scrapling.spiders import Spider, Response

class QuotesSpider(Spider):
    name = "quotes"
    start_urls = ["https://quotes.toscrape.com/"]
    concurrent_requests = 10  # 并发数（默认 4）

    async def parse(self, response: Response):
        for quote in response.css('.quote'):
            yield {
                "text": quote.css('.text::text').get(),
                "author": quote.css('.author::text').get(),
            }

        # 跟进分页：follow 接收 URL 字符串
        next_page = response.css('.next a')
        if next_page:
            yield response.follow(next_page[0].attrib['href'])

# 启动爬虫
result = QuotesSpider().start()
print(f"Scraped {len(result.items)} quotes")
result.items.to_json("quotes.json")

from scrapling.spiders import Spider, Request, Response
from scrapling.fetchers import FetcherSession, AsyncStealthySession

class MultiSessionSpider(Spider):
    name = "multi"
    start_urls = ["https://example.com/"]

    def configure_sessions(self, manager):
        # 添加不同类型的会话
        manager.add("fast", FetcherSession(impersonate="chrome"))
        manager.add("stealth", AsyncStealthySession(headless=True), lazy=True)

    async def parse(self, response: Response):
        for link in response.css('a::attr(href)').getall():
            if "protected" in link:
                # 受保护页面走 stealth 会话
                yield Request(link, sid="stealth")
            else:
                # 快速页面走 fast 会话
                yield Request(link, sid="fast", callback=self.parse)

# 带 crawldir 启动，启用检查点
result = QuotesSpider(crawldir="./crawl_data").start()

# 按 Ctrl+C 优雅停止 - 进度自动保存
# 之后传同一个 crawldir 重新运行，自动从上次位置恢复
result = QuotesSpider(crawldir="./crawl_data").start()

class StreamingSpider(Spider):
    name = "streaming"
    start_urls = ["https://example.com/"]

    async def parse(self, response: Response):
        for item in response.css('.item'):
            yield {"title": item.css('h2::text').get()}

# 流式处理（边爬边拿结果，附带实时统计）
async for item in StreamingSpider().stream():
    print(item)
    # 适合：UI 展示、数据管道、长时间爬取

class GentleSpider(Spider):
    name = "gentle"
    start_urls = ["https://example.com/"]
    autothrottle_enabled = True       # 默认关闭
    autothrottle_start_delay = 5.0    # 起始延迟（秒）
    autothrottle_max_delay = 60.0     # 延迟上限（秒）

from scrapling.fetchers import Fetcher

page = Fetcher.get('https://quotes.toscrape.com/')

# CSS 选择器
quotes = page.css('.quote')

# XPath 选择器
quotes = page.xpath('//div[@class="quote"]')

# BeautifulSoup 风格
quotes = page.find_all('div', class_='quote')

# 文本搜索
quotes = page.find_by_text('quote', tag='div')

from scrapling.parser import Selector

page = Selector("<html>...</html>")

# 链式选择
first_quote = page.css('.quote')[0]
author = first_quote.css('.author::text').get()

# 元素关系导航（siblings / children 是属性，不用加括号）
parent = first_quote.parent
siblings = first_quote.siblings
children = first_quote.children

# 保存元素特征（网站改版后用于重新定位）
page = StealthyFetcher.fetch('https://example.com/', headless=True)
products = page.css('.product', auto_save=True)

# 后续使用（网站结构可能已变化）
# Scrapling 会按保存的特征，用相似度算法找到元素的新位置
products = page.css('.product', adaptive=True)

first_item = page.css('.product')[0]

# 查找相似元素
similar_items = first_item.find_similar()

# 查找下方元素
below_items = first_item.below_elements()

from scrapling.fetchers import Fetcher, ProxyRotator

# 1. 单个代理：proxy 接收一个 URL 字符串
page = Fetcher.get(
    'https://example.com/',
    proxy="http://username:password@localhost:8030"
)

# 2. 按域名配置：proxies 接收字典
page = Fetcher.get(
    'https://example.com/',
    proxies={"example.com": "http://proxy1:8080"}
)

# 3. 代理轮换：传给 proxy_rotator 参数
rotator = ProxyRotator(["http://p1:8080", "http://p2:8080", "http://p3:8080"])
page = Fetcher.get('https://example.com/', proxy_rotator=rotator)

from random import choice
from scrapling.fetchers import ProxyRotator

# 自定义策略：函数签名 (proxies, current_index) -> (proxy, next_index)
def random_rotation(proxies, current_index):
    return choice(proxies), current_index

rotator = ProxyRotator(
    ["http://p1:8080", "http://p2:8080", "http://p3:8080"],
    strategy=random_rotation
)

pip install "scrapling[ai]"
scrapling install   # 首次使用还需安装浏览器依赖

{
  "mcpServers": {
    "ScraplingServer": {
      "command": "scrapling-mcp"
    }
  }
}

# 先定位可执行文件路径
which scrapling-mcp
# 注册到 Claude Code（路径替换为上一步的输出，如 /Users/<用户名>/.venv/bin/scrapling-mcp）
claude mcp add ScraplingServer /Users/<用户名>/.venv/bin/scrapling-mcp

from scrapling.fetchers import Fetcher

markdown = Fetcher.get("https://example.com").markdown(main_content_only=True)

# 仅解析器（无请求功能）
pip install scrapling

# 含请求功能（fetchers + spiders + 浏览器控制）
pip install "scrapling[fetchers]"

# 安装浏览器依赖（Playwright Chromium 及系统依赖、指纹组件）
scrapling install        # 正常安装
scrapling install --force  # 强制重装

from scrapling.cli import install

install([], standalone_mode=False)          # 正常安装
install(["--force"], standalone_mode=False) # 强制重装

# 从 DockerHub 拉取
docker pull pyd4vinci/scrapling

# 或从 GitHub Registry
docker pull ghcr.io/d4vinci/scrapling:latest

# 首次运行把响应缓存到磁盘，之后直接回放，
# 迭代 parse() 逻辑时不再重复请求目标服务器
class DevSpider(Spider):
    name = "dev"
    start_urls = ["https://example.com/"]
    development_mode = True             # 默认 False
    development_cache_dir = "./cache/"  # 可选，自定义缓存目录

class PoliteSpider(Spider):
    name = "polite"
    robots_txt_obey = True   # 遵守 Disallow / Crawl-delay / Request-rate
    download_delay = 1       # 每次请求间隔（秒）

# 重要爬取任务：降低并发 + 启用检查点
class ImportantSpider(Spider):
    name = "important"
    concurrent_requests = 5  # 降低并发（默认 4，按目标站点承受力调）

# crawldir 是构造参数，不是类属性
spider = ImportantSpider(crawldir="./checkpoints/")
result = spider.start()
# 按 Ctrl+C 优雅停止，进度自动保存
# 之后传同一个 crawldir 重新运行，从断点继续

# 提取为 Markdown（默认提取 body 内容；.txt 输出纯文本，.html 输出 HTML）
scrapling extract get 'https://example.com' content.md

# 指定 CSS 选择器
scrapling extract get 'https://example.com' content.txt \
    --css-selector '#main-content'

# 使用隐身模式
scrapling extract stealthy-fetch 'https://nopecha.com/demo/cloudflare' \
    captchas.html --css-selector '#padded_content a' \
    --solve-cloudflare

# 启动交互式爬虫 Shell（IPython 环境，内置 Scrapling 快捷方式）
scrapling shell

pip install "scrapling[fetchers]"
scrapling install