サンプルコード集

用途別に「コピペで動く」スニペットを集めました。API の網羅的な仕様は API リファレンス、概念面の解説は ガイド をご覧ください。

概要
前提 hub が http://your-hub.example:8000 (LAN) もしくは http://localhost:8000 (ローカル Docker) で起動済みであること。paprika-client は pip install paprika-client でインストールできます。

準備・接続

SDK は async と sync の両方を提供します。基本は async_paprika.connect():

import asyncio
from paprika_client import async_paprika

async def main():
    # 引数を省略すると環境変数 PAPRIKA_HUB を見る
    # → paprika-runner 内なら何も書かなくて OK
    async with async_paprika.connect("http://your-hub.example:8000") as cli:
        info = await cli.health()
        print(info)         # {'ok': True, 'store': 'redis', 'workers': 25}

asyncio.run(main())

以降の例では cli は connect 済みの client、page はセッションを開いた状態を想定します。

セッションを開く / 閉じる

パターン 1: async with (推奨)

ブロック抜けで close() が自動で走る。スクリプトの 9 割はこの形で OK。

async with cli.session("https://example.com") as page:
    await page.click("a.login")
    # ブロック終端で lane が解放される

パターン 2: 明示 open / close

例外を握って自分でリトライしたい場合や、複数のスクリプトをまたいでセッションを使い回したい時。

page = await cli.open_session(
    initial_url="https://example.com",
    use_profile="default",      # 事前 upload した Chrome profile
    idle_ttl_s=300,             # 無操作 TTL (秒)
    absolute_ttl_s=3600,        # 絶対 TTL (秒)
    parent_job_id=None,         # 親 job を持たせると assets がそこに紐づく
    worker_id=None,             # 特定 worker を狙うなら "host-xxxx"
    lane_hint=None,             # 特定 lane を狙うなら 0/1
)
try:
    await page.click(...)
finally:
    await page.close()

パターン 3: 初期 URL 無しで開いて goto

async with cli.session() as page:
    await page.goto("https://example.com/")
    await page.goto("https://example.com/page2")
async with cli.session() as page:
    # 移動
    reply = await page.goto("https://example.com/article")
    # reply["status"] == "OK" → CDP は届いた
    # reply["result"]["response"]["status"] == 200 → HTTP ステータス

    # 現在の URL / title / lane
    st = await page.state()
    # {'url': 'https://example.com/article',
    #  'title': '記事タイトル',
    #  'lane_idx': 0,
    #  'visited_count': 1}

    title = await page.title()
    print(title)

戻る / 進む は CDP の history API を直接呼べる JS evaluate で実現できますが、サイトの普通の操作系 (リンククリック / goto) で十分なケースがほとんどです。

クリック・入力・キー操作

async with cli.session("https://example.com/login") as page:
    # ----- クリック -----
    await page.click("button.submit")              # CSS セレクタ
    await page.click("#login-link")
    await page.click('[data-paprika-id="42"]')     # outline の @42 番号

    # ----- 入力 -----
    await page.fill("input#email", "me@example.com")

    # ${PW} はワーカー側で置換されるのでログにも LLM 履歴にも値が出ない
    import os
    await page.fill("input#password", "${PW}",
                    variables={"PW": os.environ["PASSWORD"]})

    # 既にフォーカス済みの要素に直接タイプ
    await page.type("hello world")

    # ----- キー -----
    await page.press("Enter")
    await page.press("Backspace", count=3)
    await page.press("Ctrl+A")
    await page.press("ArrowDown", count=5)         # メニューの選択肢移動
    await page.press("Ctrl+Shift+T")               # 複合修飾子
セレクタが脆い時は outline の @N page.outline() が返すテキストには各要素に [@N] が振られています。[data-paprika-id="N"] で参照すると、レイアウト変更に強く、LLM 生成スクリプトと同じ識別子で操作できます。

スクロール・待機

await page.scroll("down", pixels=800)
await page.scroll("up", pixels=400)
await page.scroll("bottom")                        # ページ末尾まで

await page.wait_for(seconds=2)                     # 単純な sleep (client 側)
await page.wait_for(ms=500)                        # ms 指定も可

# 要素が出るまで待つ (タイムアウト 30 秒、ms 単位指定も可)
await page.wait_for_selector(".result")
await page.wait_for_selector("#chart", timeout=10)

DOM を見る (outline / links)

# LLM agent が読むのと同じテキスト outline
out = await page.outline()
# 例:
#   [@1] <button> 'ログイン'
#   [@2] <a href="https://..."> '記事を読む'
#   [@3] <input placeholder="search">
#   ...

# ページ内のリンクを全部
links = await page.links()
# [{"href": "https://...", "text": "..."}, ...]

urls = await page.links(urls_only=True)
# ["https://...", "https://...", ...]

# クロール済み URL (このセッション内)
visited = await page.visited_urls()

スクリーンショット

# 現在の viewport を PNG bytes で返す (Playwright と同じ形)
png = await page.screenshot()

# ローカルファイルに保存
png = await page.screenshot(path="step1.png")

# parent_job_id を持ったセッションなら gallery に流す
# (admin UI の Live > Screenshot タブに即時反映)
png = await page.screenshot(label="after-login")

「HTML + screenshot + outline」をまとめて記録するなら capture():

snap = await page.capture("step1")
# snap = {"label": "step1", "step": 0, "html_path": "...", "png_path": "...", ...}

アセット (画像・動画) を取る

paprika は worker 側で「ページが読み込んだ画像・動画・音声」を 受動的に キャプチャしています。page.assets() で取り出せます。

async with cli.session("https://example.com/gallery", parent_job_id="abc123") as page:
    await page.scroll("bottom")                # 遅延読み込みを発火させる
    await page.wait_for(seconds=2)

    # 画像 URL を全件 (既定: kind="image")
    images = await page.assets()

    # 動画だけ
    videos = await page.assets(kind="video")

    # 全種類 + メタデータ付き
    everything = await page.assets(kind=None, details=True)
    # [
    #   {"name": "foo.jpg", "url": "http://hub/.../foo.jpg",
    #    "size": 12345, "mime": "image/jpeg", "kind": "image",
    #    "source_url": "...", "page_url": "..."},
    #   ...
    # ]
job バインドが必要 assets() は parent job のディレクトリにファイルを保存します。cli.session(..., parent_job_id="...") または paprika-runner 環境で実行してください。普通のフェッチが目的なら次の cli.fetch() 経由が簡単です。

動画をダウンロード

パターン A: ワンショット Fetch (管理画面の「動画をダウンロード」と等価)

async with async_paprika.connect() as cli:
    job = await cli.fetch(
        "https://example.com/video-page",
        download_video=True,            # ← 管理画面のチェック相当
        scroll=True,
    )
    print("job_id:", job["job_id"])

    videos = await cli.job_assets(job["job_id"], kind="video")
    for v in videos:
        print(v["url"], v.get("name"))

download_video=True で worker は:

パターン B: セッション内で明示的に page.download_video()

async with cli.session("https://example.com/video-page",
                       parent_job_id="abc123") as page:
    # 現在ページの URL に対して yt-dlp を実行
    result = await page.download_video()

    # または特定 URL を referer 付きで
    result = await page.download_video(
        url="https://cdn.example.com/clip.m3u8",
        referer="https://example.com/",
        timeout_s=1800,
    )
    # result = {"ok": True, "files": ["clip.mp4"], "file_count": 1,
    #           "message": "...", "url": "..."}

    from urllib.parse import quote
    for name in result["files"]:
        print(f"http://your-hub.example:8000/jobs/abc123/assets/{quote(name, safe='')}")

パターン C: ハイブリッド (deep-trace ON + 明示呼び出し)

job = await cli.create_job(
    "https://example.com/video-page",
    mode="fetch",
    scroll=True,
    download_video=True,
    keep_session=True,        # 終了後もセッションを残す
)

ログイン必須サイト

方法 1: Chrome プロファイルを使う (推奨)

事前にあなたのブラウザの User Data ディレクトリを paprika-client upload-profile でアップロードしておくと、そのセッションは既にログイン済みの状態で開きます。

async with cli.session(
    "https://example.com/dashboard",
    use_profile="default",          # アップロード時に付けた名前
) as page:
    # クッキー・localStorage・拡張機能・ログインセッション全部
    # 引き継いだ状態で開く
    await page.click("a.my-profile")

方法 2: 都度ログイン

import os

async with cli.session("https://example.com/login") as page:
    await page.fill("input[name=email]", os.environ["EMAIL"])
    await page.fill("input[name=password]", "${PW}",
                    variables={"PW": os.environ["PASSWORD"]})
    await page.click("button[type=submit]")
    await page.wait_for_selector(".dashboard", timeout=30)
    # ここから先はログイン済み

ログイン → 操作 → 動画 (フルフロー)

import asyncio, os
from paprika_client import async_paprika

async def main():
    async with async_paprika.connect("http://your-hub.example:8000") as cli:
        # 親 job を作っておく (assets を集約させる)
        job = await cli.create_job(
            "about:blank",
            mode="fetch",
            download_video=True,
            keep_session=True,
            wait=False,           # 投げっぱなしで session を直接操作
        )

        async with cli.session(
            initial_url="https://example.com/login",
            use_profile="default",
            parent_job_id=job["job_id"],
        ) as page:
            # 1. ログイン
            await page.fill("input[name=email]", os.environ["EMAIL"])
            await page.fill("input[name=password]", "${PW}",
                            variables={"PW": os.environ["PASSWORD"]})
            await page.click("button[type=submit]")
            await page.wait_for_selector(".dashboard", timeout=30)

            # 2. 目的のページへ
            await page.goto("https://example.com/movies/1234")
            await page.scroll("down", pixels=600)
            await page.wait_for(seconds=2)

            # 3. 再生開始 (agent に任せる)
            await page.agent("プレイヤーの再生ボタンを押して再生開始", max_steps=3)
            await page.wait_for(seconds=5)        # m3u8 が動き出すまで

            # 4. 動画 DL
            res = await page.download_video()
            print("saved:", res["files"])

            # 5. ついでに画像も
            imgs = await page.assets(kind="image")
            print(f"{len(imgs)} images")

asyncio.run(main())

LLM (ask / observe / agent)

ローカルの状態は outline() で読めますが、自然言語で問いたいときは LLM 系プリミティブが便利です。

page.ask() — 真偽値を問う

has_form = await page.ask("ログインフォームが見えていますか？")
if not has_form:
    await page.click("a.login")

page.observe() — 要素を探す

cands = await page.observe("'いいね' ボタン")
# cands = [Candidate(selector="...", text="...", confidence=0.9), ...]
if cands:
    await page.click(cands[0])         # Candidate をそのまま click に渡せる

page.agent() — 観察 + 実行を委任

# 最大 5 ステップまで observe → act のループを LLM に任せる
result = await page.agent(
    "クッキー同意ダイアログを閉じて、トップの '記事を読む' を押す",
    max_steps=5,
    engine="auto",      # = "qwen"（Qwen-VL ビジョンエンジン）
)
# result["status"] == "OK" / "FAIL"
# result["steps"] = [{"action": "click", "selector": "...", ...}, ...]
料金注意 agent() / ask() / observe() は LLM API を呼びます。決定論的に書ける箇所は click(selector) 等の primitive のみで構成し、LLM 系は「サイト固有の解析が要る所」だけに局所適用してください。

エラー処理

from paprika_client import PaprikaError, PaprikaActionError

async with cli.session("https://example.com") as page:
    try:
        await page.click("button.does-not-exist")
    except PaprikaActionError as e:
        # 要素が見つからない / クリック不可能 / タイムアウト
        print("click failed:", e)

    try:
        await page.wait_for_selector(".never-appears", timeout=5)
    except PaprikaActionError:
        # 5 秒以内に出ない → 代替フローへ
        await page.goto("https://example.com/fallback")

# 接続エラー / 502 / HTTP レベルの失敗は PaprikaError
try:
    async with async_paprika.connect("http://nope:8000") as cli:
        await cli.health()
except PaprikaError as e:
    print("hub unreachable:", e)

同期版で書く

ノートブックや既存の同期コードと混ぜる場合は sync_paprika。await を全部外した同じ API です。

from paprika_client import sync_paprika

with sync_paprika.connect("http://your-hub.example:8000") as cli:
    job = cli.fetch("https://example.com/article", download_video=True)

    for url in cli.job_images(job.job_id):
        print(url)

    with cli.session("https://example.com") as page:
        page.click("text=ログイン")
        page.fill("input#email", "me@example.com")
        page.click("button[type=submit]")
        print(page.title())

        res = page.download_video()
        print(res["files"])

paprika-runner 環境

管理画面の Submit で code モード を選んでスクリプトを投げると、worker 内の paprika-runner が以下を自動で設定してから python script.py を実行します:

import asyncio
from paprika_client import async_paprika

async def main():
    # 引数 0 で OK -- runner 環境変数から HUB と JOB_ID を読む
    async with async_paprika.connect() as cli:
        async with cli.session("https://example.com") as page:
            # 自動で parent_job_id がバインドされる
            await page.goto("https://example.com/article")
            await page.scroll("bottom")
            # 画像は管理画面の Gallery に出る
            imgs = await page.assets(kind="image")
            print(f"{len(imgs)} images captured")

asyncio.run(main())

ローカル開発で同じスクリプトを動かしたい時は、環境変数だけセットすれば OK:

import os
os.environ["PAPRIKA_HUB"] = "http://your-hub.example:8000"
# PAPRIKA_JOB_ID をセットしない → assets() は使えないが click / goto は動く