ガイド

よくあるタスクを動くコードで紹介します。全関数の仕様は API リファレンス をご覧ください。

画像を一括で取得

URL を 1 つ渡すだけです。cli.fetch() が実行から完了まで行います。

import asyncio
from paprika_client import async_paprika

async def main():
    async with async_paprika.connect() as cli:
        job = await cli.fetch("https://example.com/article", scroll=True)
        images = await cli.job_images(job["job_id"])        # 画像URL一覧
        print(len(images), "枚")
        await cli.download_job_assets(job["job_id"], "out/images")  # 保存

asyncio.run(main())
遅延ロード対策 多くのサイトは画面に入った画像だけ読み込みます。scroll=True（既定）でスクロールさせ、長いページは scroll_max=12000 など大きめに設定してください。

既存ジョブから取得

管理 UI や別スクリプトで実行済みのジョブから、あとで回収します:

jobs   = await cli.list_jobs()                 # 一覧（新しい順）
job_id = jobs[0]["job_id"]
images = await cli.job_images(job_id)          # 画像URL一覧
rows   = await cli.job_assets(job_id, details=True)  # メタ付き(size/source_url/mime)
await cli.download_job_assets(job_id, f"out/{job_id}")

画像一覧を 1 件ずつ表示（for で出力）

収集した画像をまとめて確認したいときの基本パターンです。cli.job_assets(..., details=True) で メタ情報込みの dict のリストを取得し、for で 1 件ずつ出力します。各エントリは name / url / size / size_h / mime / source_url（元のページ上の画像 URL）/ page_url / kind を持ちます。

import asyncio
from paprika_client import async_paprika

async def main():
    async with async_paprika.connect() as cli:
        # 1) ページを取得して画像を集める
        job = await cli.fetch("https://en.wikipedia.org/wiki/Cat", scroll=True)

        # 2) 画像一覧（メタ付き dict のリスト）を取得
        rows = await cli.job_assets(job["job_id"], kind="image", details=True)

        # 3) for で 1 件ずつ詳細を出力
        print(f"{len(rows)} 枚の画像:")
        for i, a in enumerate(rows, 1):
            print(f"[{i:3}] {a['size_h']:>10}  {a['mime'] or '-':<12}  {a['name']}")
            print(f"      URL:  {a['url']}")
            if a.get("source_url"):
                print(f"      元:   {a['source_url']}")

asyncio.run(main())

実行結果（例 — 最初の 4 件 / 全 49 件）:

49 枚の画像:
[  1]     3.0 KB  image/jpeg    120px-Felis_chaus_-_1700-…jpg
      URL:  http://paprika.lan:8000/jobs/356ad248c57c/assets/120px-Felis_chaus_-_1700-…jpg
      元:   https://upload.wikimedia.org/.../120px-Felis_chaus_-_1700-…jpg
[  2]     6.5 KB  image/jpeg    120px-Gustav_chocolate.jpg
      URL:  http://paprika.lan:8000/jobs/356ad248c57c/assets/120px-Gustav_chocolate.jpg
      元:   https://upload.wikimedia.org/.../120px-Gustav_chocolate.jpg
[  3]     8.9 KB  image/jpeg    120px-Orange_tabby_cat_…jpg
      URL:  http://paprika.lan:8000/jobs/356ad248c57c/assets/120px-Orange_tabby_cat_…jpg
      元:   https://upload.wikimedia.org/.../120px-Orange_tabby_cat_…jpg
[  4]     5.9 KB  image/jpeg    120px-Sheba1.JPG
      URL:  http://paprika.lan:8000/jobs/356ad248c57c/assets/120px-Sheba1.JPG
      元:   https://upload.wikimedia.org/.../120px-Sheba1.JPG
…
(残り 45 枚)
応用
  • details=False（既定）なら URL 文字列のリスト → さらに簡単な for u in urls: print(u)
  • kind=None で画像以外も含める（動画は kind="video"、音声は "audio"）
  • 条件でフィルタしてダウンロード: if a["size"] > 100_000: ...
  • 同期版なら from paprika_client import sync_paprika で await を外すだけ（API → 同期版）

動画を取得

ページ上の動画ファイルをまとめて取得するなら fetch + play_videos を使います:

job = await cli.fetch("https://example.com/clips", play_videos=True, wait_seconds=30)
videos = await cli.job_assets(job["job_id"], kind="video")
await cli.download_job_assets(job["job_id"], "out/videos", kind="video")

HLS/DASH の配信動画を 1 本の mp4 として取得したいときは、セッションで download_video()（yt-dlp）を使います:

async with cli.session("https://video.example/watch/123",
                       parent_job_id="video-grab") as page:
    await page.download_video()                # 現ページを yt-dlp
    await page.save_assets("out/videos", kind="video")

ログイン必須サイト

一度ログインして Cookie を Host レジストリに保存すれば、以後は自動で再利用されます。

# 1) セッションでログイン（手動 noVNC でも page 操作でも）
async with cli.session("https://market.example.com/login",
                       parent_job_id="login") as page:
    await page.fill("input[name=email]", "user@example.com")
    await page.fill("input[name=password]", "******")
    await page.click("button[type=submit]")
    await page.save_cookies_to_host(all_cookies=True)   # Cookie を保存

# 2) 以後は cookies_from で会員ページを収集
job = await cli.fetch("https://market.example.com/item/xxx",
                      cookies_from="market.example.com")
await cli.download_job_assets(job["job_id"], "out/item")

セッションで操作

クリックや入力を挟んでから取得したいとき。Playwright と同じ書き方です。

async with cli.session("https://news.ycombinator.com") as page:
    await page.locator(".athing .titleline > a").click()
    await page.scroll()                       # 遅延ロードを発火
    srcs = await page.assets()                # このページの画像URL
    await page.save_assets("out/images")
parent_job_id が必須 page.assets() / save_assets() は画像の保存先となる親ジョブが要ります。 手元実行なら cli.session(url, parent_job_id="任意のID") を渡してください （runner 上では PAPRIKA_JOB_ID で自動）。

DOM 取得・待機・入力

page.evaluate() を土台に、Playwright スタイルの取得・待機・入力デバイスが使えます。

# JS 実行
title = await page.evaluate("document.title")

# 取得
txt  = await page.text_content("h1")
href = await page.get_attribute("a", "href")
n    = await page.count(".item")

# 待機
await page.wait_for_selector("#result")                 # 出現を待つ
await page.wait_for_selector(".spinner", state="detached")

# 入力デバイス
await page.hover(".menu")
await page.select_option("select#country", "JP")
await page.check("#agree")
await page.set_input_files("input[type=file]", "photo.jpg")

# Locator（遅延解決・チェーン）
rows = page.locator(".item")
for r in await rows.all():
    print(await r.get_attribute("data-id"))
await page.get_by_text("ログイン").click()
「実マウス」ではない点に注意 hover / select_option / check などの入力系は、 実際にマウスを動かしているのではなく、JavaScript からその要素にイベントを発火させる方式で動きます。 ほとんどのサイトはこれで反応しますが、まれに「人間が本当にクリックしたか」を厳しくチェックする画面（広告ゲート・一部の動画再生ボタン等）には効きません。 そのときは page.agent()（LLM が画面を見て操作）か、noVNC で人手操作してください。 set_input_files（ファイルアップロード）だけは別ルート（CDP）で実際にファイルを渡すので確実です。

サイトを巡回（walk）

「このサイトを N 件たどって各ページの画像を保存」のようなクロールは、walk() に任せると キュー・重複除去・ドメイン/パス制限・オフスコープ redirect 対応まで対応しています。

from paprika_client import async_paprika, walk

async def main():
    async with async_paprika.connect() as cli:
        async with cli.session("https://example.com",
                               parent_job_id="crawl") as page:
            async for visit in walk(page, target_pages=50, same_domain=True,
                                    deny_paths=["/login", "/cart"]):
                print(visit.n, visit.depth, visit.url)
                await page.save_assets(f"out/{visit.n:04d}")   # 各ページで保存

主なオプション: target_pages（上限）/ same_domain・allowed_domains（範囲）/ allow_paths・deny_paths（フィルタ）/ order（bfs・dfs）/ max_depth / persist_state（attempt 跨ぎ再開）。全項目は API → サイト巡回 をご覧ください。

手書きループより walk 「リンクを集めて for で回す」を自前で書くと、重複・無限ループ・別ドメイン流出でハマりがちです。walk() はそれらを内蔵しています。

LLM の使い分け

使い方向いてるタスク
page.agent(goal)スクリプト内の局所的な不確実部分（年齢ゲート突破、再生ボタン探し、ログイン）
mode="codegen-loop"「このサイトを巡回」のような大きめタスクを抽象的な言葉で操作
mode="vision-agent"CSS セレクタが効かない / 動的レンダリング / 視覚情報必須のサイト
 
# スクリプト内で 1 ステップだけ LLM に任せる
async with cli.session("https://example.com") as page:
    if await page.ask("年齢確認ダイアログが出ている?"):
        await page.agent("年齢確認の「はい」を押す", max_steps=3)
    await page.capture("after-gate")

Simple Macro

コードを書かずに、管理 UI 上で「開く → クリック → 入力 → 保存」を行のように積んで実行できます （内部では paprika-client の Python に compile されて rerun モードで走ります）。 詳しくは管理 UI の Macro タブを参照してください。

同期版で書く

async/await を使わずに書きたい場合（ノートブック、簡単なスクリプト、既存の同期コード）は、 sync_paprika を使うと await を全部外した同じコードで記述できます。

from paprika_client import sync_paprika

with sync_paprika.connect() as cli:
    job = cli.fetch("https://example.com/article")
    for url in cli.job_images(job["job_id"]):
        print(url)

    with cli.session("https://example.com") as page:
        page.click("text=ログイン")
        print(page.title())

メソッドは async 版と同一です。詳細は API → 同期版 をご覧ください。

← 戻る
はじめに