ガイド

よくあるタスクを動くコードで紹介します。全関数の仕様は API リファレンス、用途別の動くスニペット集はサンプルコード集をご覧ください。

概要

このページは 「何ができて、どう書くか」のガイド。コードのコピペサンプルはサンプル集、関数の網羅仕様は API リファレンス。
用語に迷ったら用語集、エラーはエラーリファレンス、よくある質問は FAQ。
ジョブの実行モード（fetch / codegen-loop / rerun）の選び方は FAQ: モード選択。

画像を一括で取得

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://your-hub.example: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://your-hub.example: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://your-hub.example: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://your-hub.example: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 + download_video=True を使います。worker が iframe / ネスト iframe の通信を観測し、HLS/DASH の配信動画を yt-dlp で 1 本の mp4 にまとめます:

job = await cli.fetch("https://example.com/clips", download_video=True)
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")

→ 詳細はこちら（サンプル集: 動画をダウンロード）

なぜ動画は「再生発火」が要るのか

画像は 1 枚 = 1 ファイル（<img src loading="lazy">）なので、ページを開けば URL が出そろい、そのまま取得できます。動画は 配信方式によって取得の難しさが変わります。大きく 2 種類です。

1. プログレッシブ配信（直リンクの mp4）

<video src="movie.mp4"> のように 1 本のファイル をそのまま配信する方式。画像と同じで、URL さえ分かればそのままダウンロードできます。いちばん簡単なケースです。

2. ストリーミング配信（HLS / DASH）

多くの動画サイトはこちらです。動画を 数秒ごとの小さなセグメント（.ts / .m4s）に分割し、その並び順を書いた マニフェスト（HLS は .m3u8、DASH は .mpd）で配信します。

ブラウザはマニフェストを読み、セグメントを順に取得して繋ぎながら再生します（回線に応じて画質を切り替える＝ABR）。
つまり「動画ファイルへの直リンク」は存在せず、再生して初めてマニフェストとセグメントの URL が通信に現れます。
だから「ページを開いただけ」では URL が取れないことがあり、再生を発火させる必要があります。

マニフェスト (.m3u8 / .mpd)
 ├─ segment0.ts
 ├─ segment1.ts   ← これらを順に取得して連結 = 1 本の動画
 └─ ...

Paprika はどう取得するか

通信トレース — download_video: true のとき、ページ（および埋め込み iframe / ネスト iframe）の通信を監視し、.m3u8 / .mpd / .mp4 を検出します。
必要なら再生を発火 — ストリーミングはマニフェストが出ないことがあるので、AI モード（mode: codegen-loop）に「動画を再生してから取得して」と指示すると、page.agent() がプレイヤーを操作して再生し、URL を出させます。
yt-dlp で取得 + 連結 — 見つかったマニフェストを yt-dlp に渡し、セグメントをまとめて 1 本の動画ファイルにして保存します。保存後は 画像と同じ「アセット」 として gallery や assets.json に並びます。

await page.agent("メイン動画を再生して")   # 必要なら再生を発火
await page.download_video()               # 検出した m3u8 / mp4 を取得・連結して保存

取れないケース (DRM) — かつ取得を試みてはいけない

DRM（Widevine / FairPlay / PlayReady など）で暗号化された配信は、復号鍵がブラウザの保護領域（CDM）にあるため、Paprika では取得できません。Netflix・Amazon Prime Video・Disney+・U-NEXT・ABEMA プレミアム等の多くの有料配信サービスが該当します。

Paprika は DRM の回避・解除を一切行いません。これは単なる技術的制約ではなく、法的・倫理的に許されない行為だからです（著作権法第30条第1項第2号 / 第120条の2、不正アクセス禁止法、DMCA §1201、EU 著作権指令第6条）。DRM で守られたコンテンツが必要な場合は、配信元サービスが公式に提供する手段（公式アプリのダウンロード機能、API、再販ライセンスなど）を利用してください。

厳禁 DRM を回避する目的で Paprika を使うこと、回避を試みるスクリプトを生成・実行することは固く禁止します。違反はサポート対象外です。

一度ログインして Cookie を Host レジストリに保存すれば、以後は自動で再利用されます。継続運用の選択肢は Bridge 拡張（手動 push）／use_profile（フルプロファイル）／Host レシピ（自動再ログイン）の 3 通りです。

# 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 スタイルの取得・待機・入力デバイスが使えます。 Locator（page.locator(...) / get_by_*）の完全な API は Locator リファレンスを参照。

# 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）で実際にファイルを渡すので確実です。

→ 詳細はこちら（サンプル集: クリック・入力・キー操作 / スクロール・待機 / DOM を見る）

サイトを巡回（walk）

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

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}")   # 各ページで保存

asyncio.run(main())

例: walk + `page.extract()` で構造化データを集める

各ページを巡回しつつ、その場で Pydantic スキーマに沿って LLM に抽出させる組み合わせ。画像だけでなく「タイトル・著者・公開日・タグ」のような構造化情報をクロール中に集約できます。取得不能なページは skip して続行。

import asyncio
import json
from pydantic import BaseModel
from paprika_client import async_paprika, walk

class Article(BaseModel):
    title: str
    author: str | None = None
    published_at: str | None = None
    tags: list[str] = []

async def main():
    async with async_paprika.connect() as cli:
        async with cli.session("https://news.example.com",
                               parent_job_id="news-crawl") as page:
            collected: list[dict] = []
            async for visit in walk(
                page,
                target_pages=30,
                same_domain=True,
                allow_paths=["/articles/"],
                deny_paths=["/login", "/tag/"],
            ):
                try:
                    # extract() は outline をコンテキストに LLM へ投げ、
                    # JSON を返し → Pydantic で検証してから型付きで戻る
                    art = await page.extract(
                        "記事のタイトル・著者・公開日・タグを抽出。"
                        "見つからない項目は null。",
                        Article,
                    )
                    collected.append({"url": visit.url, **art.model_dump()})
                    print(f"[{visit.n:3}] {art.title}")
                except Exception as e:
                    # スキーマ違反 / LLM 不達は 1 件単位で skip。
                    print(f"[{visit.n:3}] skipped: {e}")

            with open("articles.json", "w", encoding="utf-8") as f:
                json.dump(collected, f, ensure_ascii=False, indent=2)
            print(f"total: {len(collected)} articles")

asyncio.run(main())

extract と walk の相性

list[Article] を渡せばリスト形式のページ（記事一覧）も一括取得できます: arts = await page.extract("記事一覧を取得", list[Article])
失敗ページを止めずに続行したい場合は try/except で PaprikaActionError を捕捉（スキーマ違反 / JSON パース不能はこれ）。
収集中に画像も併せて保存するなら 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)`	スクリプト内の局所的な不確実部分（年齢ゲート突破、再生ボタン探し、ログイン）。CSS が効かない画面も Qwen-VL が「見て」操作
`mode="codegen-loop"`	「このサイトを巡回」のような大きめタスクを抽象的な言葉で操作

# スクリプト内で 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")

→ 詳細はこちら（サンプル集: LLM (ask / observe / agent)）

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 → 同期版をご覧ください。

→ 詳細はこちら（サンプル集: 同期版で書く）

ガイド

画像を一括で取得

既存ジョブから取得

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

動画を取得

なぜ動画は「再生発火」が要るのか

1. プログレッシブ配信（直リンクの mp4）

2. ストリーミング配信（HLS / DASH）

Paprika はどう取得するか

取れないケース (DRM) — かつ取得を試みてはいけない

ログイン必須サイト

セッションで操作

DOM 取得・待機・入力

サイトを巡回（walk）

例: walk + page.extract() で構造化データを集める

LLM の使い分け

Simple Macro

同期版で書く

例: walk + `page.extract()` で構造化データを集める