--- title: "Prefect" description: "Prefect와 TwexAPI 읽기 전용 작업으로 Python에서 스케줄 Twitter 검색, 프로필, 타임라인, 트렌드 워크플로를 구축합니다." --- Prefect는 Python 코드용 오픈 소스 워크플로 오케스트레이터입니다. 반복 가능한 트윗 검색, 프로필 조회, 타임라인 새로고침, 트렌드 확인에는 [`prefect-x-api-scraper`](https://github.com/twexapi-dev/prefect-x-api-scraper)를 사용하세요. 이 컬렉션은 읽기 전용입니다. 6개의 비동기 Prefect 작업을 제공합니다. 각 작업은 표준 TwexAPI JSON 응답을 Python 딕셔너리로 반환합니다. 키워드, 해시태그, 계정, 날짜, X 쿼리 연산자를 검색합니다. 숫자 ID로 공개 트윗 하나를 가져옵니다. 이름, 사용자명, 주제로 공개 X 계정을 찾습니다. 스크린 이름으로 공개 프로필 하나를 조회합니다. 사용자 타임라인 페이지에서 최근 트윗을 가져옵니다. 국가, 주제, 콘텐츠 태그별 글로벌 인기 트윗을 조회합니다. 연구, 데이터 보강, 대시보드, 알림, 인덱싱에는 이 컬렉션을 사용하세요. 쓰기, 팔로워 페이지네이션, 보내기에는 직접 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 블록은 플로우와 배포 전반에 걸쳐 타입이 지정된 구성을 저장합니다. 배포 YAML, 플로우 매개변수, 로그, 저장소에 API 키를 넣지 마세요. ## 적절한 Prefect 작업 선택 `POST /twitter/advanced_search/page`를 호출합니다. `query`, `cursor`, `sort_by`, 페이지네이션 필드를 받습니다. `POST /v2/tweet/detail`을 호출합니다. 숫자 트윗 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"`을 사용합니다. 한 계정의 트윗에는 `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` | 스케줄을 늦추고 지터가 적용된 지연으로 재시도합니다. | | `5xx` | 상한이 있는 재시도로 일시적 실패를 처리합니다. | 전체 복구 가이드는 [Error Handling](/guides/error-handling)과 [Rate Limits](/guides/rate-limits)를 참조하세요. ## 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는 비동기 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)