--- title: "Prefect" description: "Prefect と TwexAPI 読み取り専用タスクで、スケジュールされた Twitter 検索、プロフィール、タイムライン、トレンドワークフローを Python で構築する。" --- Prefect は Python コード向けのオープンソースワークフローオーケストレーターです。[`prefect-x-api-scraper`](https://github.com/twexapi-dev/prefect-x-api-scraper) で繰り返し可能なツイート検索、プロフィール参照、タイムライン更新、トレンド確認を行えます。コレクションは読み取り専用です。6つの非同期 Prefect タスクを提供し、各タスクは正規 TwexAPI JSON レスポンスを Python 辞書として返します。キーワード、ハッシュタグ、アカウント、日付、X クエリ演算子を検索。数値 ID から1件の公開ツイートを取得。名前、ユーザー名、トピックで公開 X アカウントを検索。スクリーン名で1件の公開プロフィールを取得。ユーザータイムラインページから最近のツイートを取得。国、トピック、コンテンツタグ別にグローバルトレンドツイートを取得。リサーチ、エンリッチメント、ダッシュボード、アラート、インデックス作成にこのコレクションを使用してください。書き込み、フォロワーページネーション、エクスポートには直接 REST、[Python SDK](/sdks/python)、または [MCP](/mcp/overview) を使用してください。 ## インストール Python 3.10 以降を使用してください。 ```bash python -m pip install "prefect>=3.0.0" "prefect-x-api-scraper" ``` Prefect インストール後、認証情報ブロックを登録します。 ```bash prefect block register -m prefect_x_api_scraper ``` ## TwexAPI API キーを保存 [TwexAPI ダッシュボード](https://twexapi.io/dashboard) から API キーを作成し、`TwexApiCredentials` ブロック内に保存します。 ```python from prefect_x_api_scraper import TwexApiCredentials credentials = TwexApiCredentials( api_key="YOUR_API_KEY", base_url="https://api.twexapi.io", timeout_seconds=30, ) credentials.save("twexapi", overwrite=True) ``` Prefect ブロックはフローとデプロイメント間で型付き設定を保存します。API キーをデプロイメント YAML、フローパラメータ、ログ、リポジトリに置かないでください。 ## 適切な Prefect タスクを選択 `POST /twitter/advanced_search/page` を呼び出し。`query`, `cursor`, `sort_by`, ページネーションフィールドを受け付け。 `POST /v2/tweet/detail` を呼び出し。1つの数値ツイート ID を渡す。 `GET /twitter/search-user/{keyword}/{target_count}` を呼び出し。プロフィールクエリを受け付け。 `GET /twitter/{screen_name}/about` を呼び出し。`@` 付き/なしスクリーン名を受け付け。 `POST /twitter/{screen_name}/timeline/page` を呼び出し。カーソルとページサイズをサポート。 `GET /twitter/global-trending/tweets` を呼び出し。`country`, `topic`, `content`, `count` を受け付け。 ## Python で Twitter 自動化フローを構築このフローは最近の投稿を検索し、ツイート行を正規化します。ID、著者、タイムスタンプ、メトリクス、URL、ページネーション状態を保持します。 ```python from __future__ import annotations from typing import Any from prefect import flow, task from prefect_x_api_scraper import TwexApiCredentials, search_tweets @task async def normalize_tweet_page(page: dict[str, Any]) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] for tweet in page.get("data", {}).get("tweets", page.get("tweets", [])): if not isinstance(tweet, dict): continue author = tweet.get("author") or tweet.get("user") rows.append( { "tweet_id": tweet.get("id") or tweet.get("tweet_id"), "text": tweet.get("text") or tweet.get("full_text"), "created_at": tweet.get("created_at") or tweet.get("createdAt"), "author": author if isinstance(author, dict) else {}, "like_count": tweet.get("like_count") or tweet.get("likeCount"), "repost_count": tweet.get("retweet_count") or tweet.get("retweetCount"), "reply_count": tweet.get("reply_count") or tweet.get("replyCount"), } ) return rows @flow(name="TwexAPI Twitter Search") async def social_signal_flow() -> dict[str, Any]: credentials = TwexApiCredentials.load("twexapi") page = await search_tweets( credentials, '"workflow orchestration" lang:en -filter:retweets', sort_by="Latest", ) rows = await normalize_tweet_page(page) return { "tweet_rows": rows, "has_more": page.get("has_more") or page.get("has_next_page", False), "next_cursor": page.get("next_cursor") or page.get("nextCursor"), } ``` asyncio で実行: ```python import asyncio result = asyncio.run(social_signal_flow()) ``` ## 焦点を絞ったツイート検索クエリを記述完全一致フレーズには `"workflow orchestration"` を使用。 1アカウントのツイートには `from:PrefectIO` を使用。ハッシュタグ一致には `#prefect #python` を使用。クエリ内で `since:` と `until:` 日付を使用。オリジナル投稿が重要な場合は `-filter:retweets` を使用。時系列モニタリングには `sort_by="Latest"` を使用します。エンゲージメント順発見には `sort_by="Top"` を使用します。ランキングは変わる可能性があるため、ツイート ID を永続化してください。 ## Twitter 検索パイプラインをスケジュール `.serve()` で長時間実行ローカルプロセスを使用します。 ```python from prefect.schedules import Cron if __name__ == "__main__": social_signal_flow.serve( name="twexapi-social-signals", schedule=Cron("0 * * * *", timezone="UTC"), ) ``` Docker、Kubernetes、サーバーレスワーカーにはワークプールデプロイメントを使用します。 ```bash prefect deploy social_signal_flow.py:social_signal_flow \ --name twexapi-social-signals \ --pool production ``` ## ツイートとプロフィール結果をページネーションカーソルページネーションはページ番号推測なしで大規模検索を継続します。元のリクエストは変更せず、返されたカーソルのみを渡します。 ```python from typing import Any, Optional from prefect import flow from prefect_x_api_scraper import TwexApiCredentials, search_tweets @flow async def collect_tweet_pages(query: str) -> list[dict[str, Any]]: credentials = TwexApiCredentials.load("twexapi") rows_by_id: dict[str, dict[str, Any]] = {} cursor: Optional[str] = None while True: page = await search_tweets( credentials, query=query, sort_by="Latest", cursor=cursor, ) tweets = page.get("data", {}).get("tweets", page.get("tweets", [])) for tweet in tweets: if isinstance(tweet, dict): tweet_id = tweet.get("id") or tweet.get("tweet_id") if tweet_id: rows_by_id[str(tweet_id)] = tweet cursor_value = page.get("next_cursor") or page.get("nextCursor") has_more = bool(page.get("has_more") or page.get("has_next_page", False)) if not has_more or not cursor_value: break cursor = str(cursor_value) return list(rows_by_id.values()) ``` カーソルをデコードしないでください。不透明な文字列として扱います。各ページ永続化後にカーソルを保存してください。 ## スケジュール実行をべき等に `tweet_id` でツイート行を upsert。本文をキーにしない。数値ユーザー ID でプロフィール行を upsert。各コミットページ後に `has_more` と `next_cursor` を保存。 ## 一時的障害のみリトライ Prefect はリトライ遅延、ジッター、条件付きリトライをサポートします。無効な入力を変更なしでリトライしないでください。 ```python from prefect_x_api_scraper import search_tweets search_recent_tweets = search_tweets.with_options( name="Search Recent Tweets", retries=4, retry_delay_seconds=[5, 15, 45, 120], ) ``` ## ドキュメント化されたエラーをルーティング | ステータス | 対処法 | | ------ | ------ | | `400` | 欠落クエリまたは不正入力を修正。変更なしでリトライしない。 | | `401` | 認証情報ブロックから有効な API キーを読み込み。 | | `403` | 次回実行前にアカウントアクセスまたはクレジットを解決。 | | `404` | ツイート ID、ユーザー名、スクリーン名を確認。 | | `429` | スケジュールを遅くし、ジッター遅延でリトライ。 | 完全な復旧ガイダンスは [Error Handling](/guides/error-handling) と [Rate Limits](/guides/rate-limits) を参照してください。 | `5xx` | 上限付きで一時的障害をリトライ。 | ## MCP 計画代替計画段階で [Twexapi MCP](/mcp/overview) を使用してエンドポイントを発見し、選択ルートを Prefect タスクまたは REST 呼び出しにコード化します。推奨計画プロンプト: ```txt Use Twexapi MCP explore to choose the best endpoint for daily AI trend collection. Return the exact method, relative path, query parameters, pagination fields, expected response fields, and retry guidance. Do not execute write endpoints. ``` ## Prefect コレクションか直接 TwexAPI API か 6つのサポート読み取りには `prefect-x-api-scraper` を選択してください。ブロック、非同期呼び出し、検証、タスクメタデータを提供します。以下には直接 REST、[Python SDK](/sdks/python)、または MCP を使用してください。 * ツイート、返信、引用、いいね、フォロー、DM アクション * フォロワーとフォローエクスポート * リストとコミュニティ定期チェックには Prefect フローまたは REST ページネーションをスケジュールしてください。TwexAPI はネイティブモニターエンドポイントを公開していません。TwexAPI は非同期 CSV/JSON 抽出ジョブを別製品として公開していません。書き込みは明示的な承認の背後に置いてください。書き込みアクションをリトライする前に確認済み ID を保存してください。 ## ソースと次のステップ * [prefect-x-api-scraper source](https://github.com/twexapi-dev/prefect-x-api-scraper) * [Prefect schedules](https://docs.prefect.io/v3/concepts/schedules) * [Prefect task retries](https://docs.prefect.io/v3/how-to-guides/workflows/retries) * [Advanced Twitter Search](/api-reference/search-endpoints/get-data-page-twitter-advanced-search-page-post) * [Global Trending Tweets](/api-reference/trending-endpoints/get-global-trending-tweets-api-twitter-global-trending-tweets-get) * [Agent MCP Handoff](/mcp/agent-handoff)