오늘의 이야기

토스쇼핑 쉐어링크 API Client (작업일지)

 

토스쇼핑 쉐어링크 Open API (공식 문서) 파이썬 연동 프로젝트입니다.

쉐어링크 Open API는 제휴사가 토스쇼핑 상품을 소개하고, 발급받은 추적 링크(쉐어링크)를 통해 발생한 구매를 수익으로 정산받을 수 있게 해주는 서버-투-서버 API입니다.

이 API는 아직 시범 운영 중입니다. 일 사용 상한·엔드포인트·응답 필드가 바뀔 수 있어, 이 클라이언트는 모르는 응답 필드를 무시하고 일 사용 상한 초과 시 전체가 멈추지 않도록 방어적으로 만들어져 있습니다.

준비물

코드를 실행하기 전에 쉐어링크 크리에이터 어드민(sharelink.toss.im)에서 아래를 미리 발급/등록해야 합니다 (문서: 연동 시작하기).

1. Access Key / Secret Key 발급 — API 연동 메뉴에서 발급받습니다. Secret Key는 발급 직후 1회만 표시되니 안전하게 보관하세요.


2. 호출 서버의 출발지 IP 등록 — 등록되지 않은 IP에서 호출하면 SHARELINK_OPENAPI_ACCESS_DENIED 가 발생합니다. 등록할 IP는 내 PC가 아니라 실제로 API를 호출하는 서버의 IP입니다.


3. Publisher UUID (publisherId) — 링크 발급 시 필요하며, 인증 정보와 함께 안내받습니다.

설치

python -m venv .venv
source .venv/bin/activate   # Windows: .venv\Scripts\activate
pip install -r requirements.txt

.env.example 을 .env 로 복사한 뒤 발급받은 값을 채워주세요.

cp .env.example .env

빠른 시작

문서의 빠른 시작 5단계(토큰 발급 → 연결 확인 → 상품 목록 조회 → 쉐어링크 발급 → 게시)를 그대로 옮긴 예제입니다.

python examples/quickstart.py

카테고리 기반으로 상위 상품에 일괄로 링크를 발급하는 예제도 있습니다.

python examples/list_category_best_and_issue_links.py

기본 사용법

from sharelink import ShareLinkClient, ShareLinkConfig

config = ShareLinkConfig.from_env()          # .env 에서 읽기
client = ShareLinkClient.with_file_cache(config)  # 발급받은 링크를 로컬 JSON에 캐싱

# 연결 확인
client.health()

# 지금 많이 팔리는 상품 5개
page = client.get_best_selling(size=5)
for item in page.items:
    print(item.display_name, item.display_price, item.taca_item_id)

# 쉐어링크 발급 (수익 집계는 이 링크로만 이뤄집니다. productUrl 은 추적되지 않습니다)
link = client.issue_link(taca_item_id=page.items[0].taca_item_id)
print(link.short_url)

페이지 신경 쓰지 않고 전체 순회하기

목록 API 3종(카테고리 베스트 / 베스트 / 하루특가)은 커서 기반 페이징입니다. iter_* 메서드를 쓰면 페이지 경계를 신경 쓰지 않고 순회할 수 있습니다.

for item in client.iter_best_selling(size=30):
    ...

카테고리 조회

categories = client.get_categories()          # 트리 전체
for node in categories[0].flatten():           # 재귀적으로 평탄화
    print(node.category_id, node.display_name)

page = client.get_best_categories(category_id=101, size=30)

상품 상세 조회 (최대 30건, 자동 분할 지원)

items, not_found_ids = client.get_product_detail(taca_item_ids=[12345, 12346])

# 30건 넘는 목록은 자동으로 나눠서 조회
items, not_found_ids = client.get_product_details_bulk(all_ids)

대량 링크 발급

links = client.issue_links_bulk(taca_item_ids=[12345, 12346], on_error="skip")
# links 는 taca_item_ids 와 같은 길이/순서. 발급 실패한 자리는 None.

이 클라이언트가 문서 규칙을 반영한 부분

문서(공통 규약)가 요구하는 방어적 구현을 기본으로 갖추고 있습니다.

• 토큰 재사용: expires_in 동안 캐싱하고, 만료 60초 전에만 자동 재발급합니다. 호출마다 새로 발급받지 않습니다.


• resultType 기준 판별: HTTP 상태 코드가 아니라 resultType / error.errorCode 로 성공·실패를 판별합니다.


• 에러별 재시도 정책:
  
  
  
  • INVALID_ARGUMENT, SHARELINK_OPENAPI_ACCESS_DENIED, SHARELINK_OPENAPI_QUOTA_EXCEEDED, HTTP 401 → 재시도하지 않음 (원인을 고쳐야 함 / 자정 KST 리셋 대기)
  
  
  • HTTP 429 → Retry-After 헤더를 우선 따라 재시도
  
  
  • HTTP 500, 본문 errorCode: "500" → 지수 백오프 재시도
  
  
  • 알 수 없는 errorCode → 재시도하지 않고 로그로 남김 (향후 필드/코드 추가에 대비)
  
  
  
  


• 모르는 응답 필드 무시: 모델 파싱이 필드 추가로 깨지지 않습니다.


• 일 사용 상한 대응: 발급받은 쉐어링크를 로컬 JSON(.sharelink_cache/)에 캐싱해 같은 상품을 다시 발급받지 않도록 합니다. 카테고리·랭킹도 직접 캐싱해 저장해 두고 재사용하는 것을 권장합니다(문서 권장 사항).


• productUrl을 게시글에 쓰지 않도록 구분: 조회 API 응답의 product_url은 추적되지 않는 일반 링크입니다. 게시글에는 반드시 issue_link()로 받은 short_url / origin_url을 사용하세요.

프로젝트 구조

sharelink-api-client/
├── sharelink/
│   ├── __init__.py       # 공개 API export
│   ├── config.py         # 환경 설정 (Access/Secret Key, publisherId, production/alpha)
│   ├── auth.py            # OAuth2 client_credentials 토큰 관리(캐싱/재발급)
│   ├── client.py          # ShareLinkClient — 전체 엔드포인트 + 재시도/에러 매핑
│   ├── models.py          # 응답 모델 (Category, ProductCard, ProductDetail, IssuedLink)
│   ├── cache.py           # 발급 링크 로컬 캐시 (JSON 파일)
│   └── exceptions.py      # 에러 코드별 예외 클래스
├── examples/
│   ├── quickstart.py                          # 문서 '빠른 시작' 5단계 그대로
│   └── list_category_best_and_issue_links.py  # 카테고리 베스트 + 대량 링크 발급
├── tests/
│   └── test_client.py    # 토큰 캐싱, 에러 매핑, 페이징에 대한 단위 테스트 (requests mock)
├── requirements.txt
├── requirements-dev.txt
├── .env.example
└── README.md

참고: 환경(운영/알파)

문서 간 안내가 약간 다릅니다.

• 연동 시작하기: "Open API는 운영 환경에서만 제공되며, 별도의 테스트(알파) 환경은 없다"


• 빠른 시작: "먼저 알파(테스트) 환경에서 확인 후 운영으로 옮기길 권장. 주소만 https://alpha-sharelink.toss.im/openapi 로 바꾸면 됨"

이 클라이언트는 두 안내를 모두 반영해 SHARELINK_ENVIRONMENT=alpha 를 지원하도록 만들어 뒀습니다. 실제 알파 환경 동작 여부는 담당자에게 확인하시고, 불확실하면 기본값인 production 을 사용하세요.

호출 제한 / 일 사용 상한 요약

항목	기준
호출 속도	파트너 단위 10rps, 순간 버스트 최대 30 (전 엔드포인트 합산)
일 사용 상한 (조회 상품)	하루 10,000개
일 사용 상한 (신규 링크 발급)	하루 10,000개
리셋	매일 자정 (KST)

두 상한은 서로 독립적이며, 응답을 저장해 재사용하면 상한에 거의 닿지 않는다고 문서에 안내되어 있습니다.

instagram test

threads_ids : access_key, secret_key, publisher_id를 환경변수로 관리 (.env의 INSTAGRAM_* 참고)

인증값은 더 이상 이 파일에 두지 않습니다. .env의 INSTAGRAM_ACCESS_TOKEN / INSTAGRAM_APP_SECRET / INSTAGRAM_PUBLISHER_ID 를 채운 뒤 아래처럼 사용하세요.

curl -X POST \
  'https://graph.instagram.com/v25.0/me/messages' \
  -H "Authorization: Bearer $INSTAGRAM_ACCESS_TOKEN" \
  -H 'Content-Type: application/json' \
  -d '{
    "message": {
        "text": "Hello World test"
    },
    "recipient": {
        "id": "<RECIPIENT_IGSID>"
    }
  }'

파이썬으로 동일하게 테스트하려면:

python examples/instagram_send_test.py --recipient <RECIPIENT_IGSID>

테스트용 IGSID(recipient id) 구하기

recipient.id 로 아무 값이나 넣으면 IGApiException (code 100, subcode 2534014, "요청한 사용자를 찾을 수 없습니다") 이 발생합니다. 이 계정에 실제로 DM을 보낸 적 있는 사용자의 IGSID만 사용할 수 있으며, 이는 웹훅 이벤트로만 확인할 수 있습니다.

# 1) .env 의 INSTAGRAM_WEBHOOK_VERIFY_TOKEN 을 아무 문자열로 채운다
# 2) 로컬 수신 서버 실행
python examples/instagram_webhook_receiver.py

# 3) 별도 터미널에서 ngrok 등으로 공개 HTTPS 주소 생성
ngrok http 8000

# 4) Meta 개발자 콘솔 > 앱 > Instagram > Webhooks 에서
#    콜백 URL = https://<ngrok-주소>/webhook, Verify Token = .env 값과 동일하게 등록,
#    구독 필드에서 messages 체크

# 5) 연결된 인스타그램 계정으로 실제 DM을 보내면, 수신 서버 콘솔에 IGSID(sender.id)가 출력된다

ngrok 없이 상시로 띄워두고 싶다면 gcf/instagram_webhook/ 에 동일한 역할을 하는 Google Cloud Functions(gen2, HTTP 트리거) 버전이 있습니다.

cd gcf/instagram_webhook
gcloud functions deploy instagram-webhook \
  --gen2 \
  --runtime=python312 \
  --region=asia-northeast3 \
  --source=. \
  --entry-point=instagram_webhook \
  --trigger-http \
  --allow-unauthenticated \
  --set-env-vars=INSTAGRAM_WEBHOOK_VERIFY_TOKEN=<.env 의 INSTAGRAM_WEBHOOK_VERIFY_TOKEN 과 동일한 값>

배포 후 출력되는 URL을 Meta 콘솔의 웹훅 콜백 URL로 등록하면 됩니다. 수신된 이벤트와 IGSID는 Cloud Logging에서 확인할 수 있습니다.

받은 IGSID/메시지는 gcf/instagram_webhook/ 안의 firebase-adminsdk 서비스 계정 키로 인증해 Firebase Realtime Database(instagram_dm_events/<sender_id>)에도 저장됩니다.

Business Login (정식 토큰 발급)

INSTAGRAM_ACCESS_TOKEN을 Graph API Explorer 등에서 임시로 받은 경우, 메시징 권한이 없거나 금방 만료될 수 있습니다. 정식으로는 Business Login for Instagram OAuth 플로우를 거쳐야 합니다.

# 1) gcf/instagram_oauth_callback 배포 (redirect_uri는 일단 비워서 배포 후 URL부터 확보)
cd gcf/instagram_oauth_callback
gcloud functions deploy instagram-oauth-callback \
  --gen2 --runtime=python312 --region=asia-northeast3 \
  --source=. --entry-point=instagram_oauth_callback \
  --trigger-http --allow-unauthenticated \
  --set-env-vars=INSTAGRAM_APP_ID=<.env 의 INSTAGRAM_APP_ID>,INSTAGRAM_APP_SECRET=<.env 의 INSTAGRAM_APP_SECRET>,INSTAGRAM_OAUTH_REDIRECT_URI=<배포 후 출력되는 URL>

# 2) 배포로 나온 URL을 .env 의 INSTAGRAM_OAUTH_REDIRECT_URI 에 채우고, 같은 값으로 재배포
#    (redirect_uri 는 실제 호출 시 정확히 일치해야 하므로 한 번 더 --set-env-vars 로 갱신)

# 3) Meta 앱 대시보드 > Instagram > API setup with Instagram Login > Business Login settings 에서
#    OAuth redirect URI        = <배포 URL>/
#    Deauthorize callback URL  = <배포 URL>/deauthorize
#    Data Deletion Request URL = <배포 URL>/data-deletion

# 4) 브라우저에서 로그인 동의 진행
#    https://www.instagram.com/oauth/authorize
#      ?client_id=<INSTAGRAM_APP_ID>
#      &redirect_uri=<INSTAGRAM_OAUTH_REDIRECT_URI>
#      &scope=instagram_business_basic,instagram_business_manage_messages
#      &response_type=code

# 5) 동의하면 이 함수가 code -> 단기 -> 장기(60일) 토큰까지 자동 교환해서
#    Firebase Realtime Database `instagram_oauth/<ig_user_id>` 에 저장합니다.
#    그 값을 .env 의 INSTAGRAM_ACCESS_TOKEN 으로 옮겨주세요.

Threads OAuth (쉐어링크 자동 게시용, THREADS_AUTOMATION_CHECKLIST.md 2번)

Instagram Business Login과 동일한 패턴입니다. gcf/threads_oauth_callback/이 code -> 단기 -> 장기(60일) 토큰 교환을 처리합니다.

# 1) gcf/threads_oauth_callback 배포 (redirect_uri는 일단 비워서 배포 후 URL부터 확보)
cd gcf/threads_oauth_callback
gcloud functions deploy threads-oauth-callback \
  --gen2 --runtime=python312 --region=asia-northeast3 \
  --source=. --entry-point=threads_oauth_callback \
  --trigger-http --allow-unauthenticated \
  --set-env-vars=THREADS_APP_ID=<.env 의 THREADS_APP_ID>,THREADS_APP_SECRET=<.env 의 THREADS_APP_SECRET>,THREADS_OAUTH_REDIRECT_URI=<배포 후 출력되는 URL>

# 2) 배포로 나온 URL을 .env 의 THREADS_OAUTH_REDIRECT_URI 에 채우고, 같은 값으로 재배포
#    (redirect_uri 는 실제 호출 시 정확히 일치해야 하므로 한 번 더 --set-env-vars 로 갱신)

# 3) Meta 앱 대시보드 > Threads > API setup with Threads > Threads Login settings 에서
#    OAuth redirect URI = <배포 URL>/

# 4) 브라우저에서 로그인 동의 진행
#    https://threads.net/oauth/authorize
#      ?client_id=<THREADS_APP_ID>
#      &redirect_uri=<THREADS_OAUTH_REDIRECT_URI>
#      &scope=threads_basic,threads_content_publish
#      &response_type=code

# 5) 동의하면 이 함수가 code -> 단기 -> 장기(60일) 토큰까지 자동 교환해서
#    Firebase Realtime Database `threads_oauth/<threads_user_id>` 에 저장합니다.
#    그 값을 .env 의 THREADS_ACCESS_TOKEN 으로 옮겨주세요.
#
#    장기 토큰은 60일 후 만료됩니다. 자동 갱신은 아래 threads-token-refresh 함수가 처리합니다.

Threads 장기 토큰 자동 갱신 (Cloud Scheduler)

gcf/threads_token_refresh/가 Firebase RTDB threads_oauth/ 아래 저장된 모든 계정의 토큰을 주기적으로 갱신합니다. OAuth 콜백과 달리 공개 엔드포인트가 아니며, Cloud Scheduler의 OIDC 토큰으로만 호출할 수 있습니다.

# 1) 최초 1회: 프로젝트에 Cloud Scheduler API 활성화
gcloud services enable cloudscheduler.googleapis.com

# 2) 비공개 함수 배포 (--no-allow-unauthenticated)
cd gcf/threads_token_refresh
gcloud functions deploy threads-token-refresh \
  --gen2 --runtime=python312 --region=asia-northeast3 \
  --source=. --entry-point=threads_token_refresh \
  --trigger-http --no-allow-unauthenticated

# 3) 함수를 호출할 서비스 계정에 invoker 권한 부여
#    (기본 컴퓨트 서비스 계정을 그대로 재사용. 다른 서비스 계정을 쓴다면 그 계정으로 교체)
gcloud run services add-iam-policy-binding threads-token-refresh \
  --region=asia-northeast3 \
  --member="serviceAccount:<프로젝트번호>-compute@developer.gserviceaccount.com" \
  --role="roles/run.invoker"

# 4) Cloud Scheduler 작업 등록 (예: 매주 일요일 03:00 KST)
gcloud scheduler jobs create http threads-token-refresh-weekly \
  --location=asia-northeast3 \
  --schedule="0 3 * * 0" \
  --time-zone="Asia/Seoul" \
  --uri=<2번에서 배포된 함수 URL> \
  --http-method=POST \
  --oidc-service-account-email="<프로젝트번호>-compute@developer.gserviceaccount.com" \
  --oidc-token-audience=<2번에서 배포된 함수 URL>

수동으로 즉시 실행해보고 싶다면:

gcloud scheduler jobs run threads-token-refresh-weekly --location=asia-northeast3

실행 결과(계정별 성공/실패)는 Cloud Logging에서 threads-token-refresh 함수 로그로 확인할 수 있습니다.

Threads 게시 (THREADS_AUTOMATION_CHECKLIST.md 3번)

threads/ 패키지가 미디어 컨테이너 생성 → (이미지인 경우) 처리 상태 폴링 → 발행까지의 2단계 발행 흐름을 처리합니다. sharelink 클라이언트와 같은 방식으로 에러를 분류해 재시도합니다 (threads/exceptions.py 참고).

from threads import ThreadsClient

client = ThreadsClient.from_env()  # .env 의 THREADS_ACCESS_TOKEN / THREADS_USER_ID 사용

# 텍스트 게시물: 컨테이너 생성 + 발행을 한 번에 처리
media_id = client.publish_text("쉐어링크로 발급받은 short_url 을 여기에 넣습니다")

# 이미지 게시물: 생성 -> 처리 완료 대기 -> 발행까지 한 번에 처리
media_id = client.publish_image(
    "https://example.com/public-image.jpg",  # Threads 가 접근 가능한 공개 URL
    text="상품 소개 문구 + short_url",
)

단계를 직접 제어하고 싶다면(예: 여러 컨테이너를 먼저 만들어두고 나중에 한꺼번에 발행):

creation_id = client.create_image_container("https://example.com/public-image.jpg", text="...")
client.wait_until_container_ready(creation_id)  # FINISHED 될 때까지 폴링
media_id = client.publish_container(creation_id)

CAROUSEL(media_type=CAROUSEL, 여러 이미지 묶음)은 아직 구현하지 않았습니다.

게시 문구 생성 (THREADS_AUTOMATION_CHECKLIST.md 4번)

threads/content.py가 상품 정보 + 쉐어링크를 게시 문구로 조합합니다. product.product_url은 이 모듈이 아예 참조하지 않으므로 실수로 쓸 일이 없습니다. 광고/제휴 마케팅 고지 문구가 기본으로 포함되고, 500자 제한 초과 시 링크·고지는 그대로 두고 상품 설명만 줄여서 맞춥니다.

from sharelink import ShareLinkClient, ShareLinkConfig
from threads import ThreadsClient, build_post_content

share_client = ShareLinkClient.with_file_cache(ShareLinkConfig.from_env())
threads_client = ThreadsClient.from_env()

product = share_client.get_best_selling(size=1).items[0]
link = share_client.issue_link(taca_item_id=product.taca_item_id)

content = build_post_content(product, link, hashtags=["오늘의특가"])
if content.image_url:
    threads_client.publish_image(content.image_url, text=content.text)
else:
    threads_client.publish_text(content.text)

중복 게시 방지 / 상태 관리 (THREADS_AUTOMATION_CHECKLIST.md 5번)

threads/store.py의 PostStore가 상품별 게시 상태를 SQLite(.threads_store/posts.db)에 저장합니다. threads/publisher.py의 publish_once()가 예약(reserve) → 발행 → 상태 확정까지 순서를 고정해서, 이미 게시된 상품은 건너뛰고 "발행은 됐는데 응답만 실패"한 애매한 상태에서는 자동 재시도 대신 예외를 올립니다.

from threads import PostStore, publish_once

store = PostStore()  # 기본 경로: .threads_store/posts.db

media_id = publish_once(
    threads_client,
    store,
    product.taca_item_id,
    content,
    daily_limit=10,  # 최근 24시간 게시 건수가 이 값 이상이면 DailyPostLimitReached
)
if media_id is None:
    print("이미 게시된 상품이라 건너뛰었습니다.")

PendingPostConflictError가 나면(이전 시도의 결과를 알 수 없는 상태) Threads 앱에서 실제로 게시됐는지 직접 확인한 뒤 store.resolve_pending(taca_item_id, published=True, media_id=...) 또는 published=False로 결과를 확정해야 다음 시도가 가능합니다.

자동화 트리거: 라즈베리파이 + cron (THREADS_AUTOMATION_CHECKLIST.md 6번)

scripts/post_today_deals.py가 get_today_deals(그날 하루만 파는 특가) 상품을 순회하며 품절/이미 게시된 상품을 건너뛰고 --count건을 새로 게시합니다. 실행 로그는 파일에 남기지 않고 표준출력/에러로만 출력합니다.

# 라즈베리파이(Linux)에서 최초 1회
git clone <이 저장소> ~/sharelink-api-client
cd ~/sharelink-api-client/shareLink/sharelink-api-client  # 실제 경로는 클론 방식에 따라 다를 수 있음
python3 -m venv .venv
.venv/bin/pip install -r requirements.txt
cp .env.example .env   # SHARELINK_*, THREADS_ACCESS_TOKEN, THREADS_USER_ID 채우기

# 수동으로 한 번 실행해서 확인
.venv/bin/python scripts/post_today_deals.py --count 5 --daily-limit 10

crontab -e에 등록 (매일 오전 9시 KST, 신규 게시 최대 5건, 24시간 총 게시 상한 10건):

MAILTO=you@example.com
0 9 * * * cd /home/pi/sharelink-api-client/shareLink/sharelink-api-client && .venv/bin/python scripts/post_today_deals.py --count 5 --daily-limit 10 >> /dev/null 2>&1

MAILTO를 설정해두면 스크립트가 예외로 중단됐을 때(0이 아닌 종료 코드) cron이 표준출력/에러를 그 주소로 메일 발송합니다 — 별도 로그 파일이나 알림 시스템 없이 실패를 알아챌 수 있는 최소한의 장치입니다. 라즈베리파이의 타임존이 Asia/Seoul이 아니라면 sudo timedatectl set-timezone Asia/Seoul로 맞추거나 cron 시간을 UTC 기준으로 조정하세요.

오늘의 이야기

#스하리1000명프로젝트,
Bị lạc ở Hàn Quốc? Ngay cả khi bạn không nói được tiếng Hàn, ứng dụng này vẫn giúp bạn đi lại dễ dàng.
Chỉ cần nói ngôn ngữ của bạn—nó sẽ dịch, tìm kiếm và hiển thị kết quả bằng ngôn ngữ của bạn.
Tuyệt vời cho du khách! Hỗ trợ hơn 10 ngôn ngữ bao gồm tiếng Anh, tiếng Nhật, tiếng Trung, tiếng Việt, v.v.
Hãy thử nó ngay bây giờ!
https://play.google.com/store/apps/details?id=com.billcoreatech.opdgang1127

오늘의 이야기

#스하리1000명프로젝트,
Nawala sa Korea? Kahit na hindi ka nagsasalita ng Korean, tinutulungan ka ng app na ito na madaling makalibot.
Sabihin lang ang iyong wika—ito ay nagsasalin, naghahanap, at nagpapakita ng mga resulta pabalik sa iyong wika.
Mahusay para sa mga manlalakbay! Sinusuportahan ang 10+ wika kabilang ang English, Japanese, Chinese, Vietnamese, at higit pa.
Subukan ito ngayon!
https://play.google.com/store/apps/details?id=com.billcoreatech.opdgang1127

오늘의 이야기

Build Log

토스쇼핑 쉐어링크를 Threads에 자동 게시하기까지

상품 조회부터 추적 링크 발급, 게시 문구 생성, 중복 방지, cron 자동화까지 — 하나의 파이프라인을 만들며 실제로 부딪힌 설계 결정과 구현 기법을 정리했다.

Python SQLite Cloud Functions gen2 Cloud Scheduler Threads Graph API Raspberry Pi · cron

---

무엇을 만들었나

체크리스트 8개 항목 — 전부 완료

토스쇼핑 쉐어링크 Open API로 상품을 조회하고 추적 링크(short_url)를 발급한 뒤, 그 링크로 게시 문구를 만들어 Threads에 자동으로 올리는 파이프라인이다. 인증, 발행, 문구 생성, 중복 방지, 트리거까지 전 구간을 직접 설계했다.

01

Meta 앱 · 권한 준비

DONE

02

OAuth 인증 흐름

DONE

03

2단계 발행 로직

DONE

04

콘텐츠 생성 규칙

DONE

05

중복 게시 방지

DONE

06

자동화 트리거

DONE

07

시크릿 관리

DONE

08

실게시 검증 · 테스트

DONE

순서대로 만든 게 아니라 2 → 3 → 6번을 먼저 뚫고, 4·5·7·8은 그 과정에서 자연히 같이 끝났다.

전체 흐름

get_today_deals() → issue_link() → build_post_content() → publish_once() → PostStore

핵심 구현 기술

코드로 남길 만했던 결정들

1. HTTP 상태 코드가 아니라 응답 바디로 성공/실패를 판정한다

쉐어링크 API는 HTTP 200이어도 바디의 resultType이 FAIL일 수 있다. 반대로 Threads(Meta Graph API)는 HTTP 상태와 error 객체 유무로 판정한다. 두 API가 실패를 표현하는 방식이 완전히 다르다는 걸 먼저 인정하고, 각자에 맞는 판정 함수를 따로 짰다 — 하나의 공통 추상화로 억지로 묶지 않았다.

sharelink/client.py result-based dispatch

if result_type == "SUCCESS":
    return payload.get("success", {})

if result_type == "FAIL":
    error_code = error.get("errorCode")
    if error_code == "INVALID_ARGUMENT":
        raise ShareLinkInvalidArgumentError(...)
    if error_code == _RETRYABLE_ERROR_CODE and attempt < max_retries:
        raise _RetryableBodyError(...)  # 상위 루프에서 백오프 재시도

2. 에러마다 재시도 정책이 다르다 — 하나로 뭉뚱그리지 않는다

"실패하면 재시도"가 아니라, 에러의 성격에 따라 재시도 여부가 갈린다. 두 클라이언트 모두 같은 원칙을 따르되 각 API의 실제 에러 표현에 맞게 구현했다.

상황	ShareLink	Threads	정책
인증 실패	HTTP 401	OAuthException (190)	재시도 금지 · 토큰 재발급 필요
호출 속도 제한	HTTP 429	code 4/17/32/613	Retry-After 존중, 백오프 재시도
일시적 서버 오류	HTTP 500	is_transient: true	지수 백오프 재시도
파라미터 오류	INVALID_ARGUMENT	기타 4xx	재시도 금지 · 결과가 항상 같음
미지의 에러 코드	문서에 없는 코드	문서에 없는 코드	재시도 없이 예외 (베타 API 대비)

3. 발행은 2단계다 — 컨테이너를 만들고, 기다리고, 발행한다

Threads는 글을 바로 올리지 않는다. 미디어 컨테이너를 먼저 만들고(특히 이미지는 비동기 처리), 상태가 FINISHED가 될 때까지 폴링한 뒤에야 발행 API를 호출할 수 있다. TEXT는 폴링이 필요 없어서 분기했다.

threads/client.py create → poll → publish

def publish_image(self, image_url, text=None, *, poll_interval_seconds=2.0, timeout_seconds=60.0):
    creation_id = self.create_image_container(image_url, text)
    self.wait_until_container_ready(creation_id, poll_interval_seconds, timeout_seconds)
    return self.publish_container(creation_id)

def wait_until_container_ready(self, creation_id, *, poll_interval_seconds, timeout_seconds):
    deadline = time.monotonic() + timeout_seconds
    while True:
        status, error_message = self.get_container_status(creation_id)
        if status == "FINISHED":
            return
        if status in ("ERROR", "EXPIRED"):
            raise ThreadsContainerError(creation_id, status, error_message)
        if time.monotonic() >= deadline:
            raise ThreadsContainerTimeoutError(creation_id, status)
        time.sleep(poll_interval_seconds)

4. "쓰면 안 되는 필드"는 코드에서 아예 안 보이게 만든다

쉐어링크 API가 돌려주는 product_url은 추적이 안 되는 일반 링크다. 실수로 이걸 게시글에 쓰면 수익이 전혀 집계되지 않는다. 주석으로 "쓰지 마세요"라고 적어두는 대신, 문구 생성 모듈이 product_url을 아예 import도 참조도 하지 않게 만들었다. 리뷰나 주의력에 기대지 않고, 구조로 막는 편이 항상 더 오래 간다.

threads/content.py structural enforcement

# product.product_url 은 이 모듈 어디에서도 참조하지 않는다.
# 게시 링크는 IssuedLink.short_url 하나만 쓴다.
def _build_tail(link, *, hashtags, include_disclosure):
    parts = [link.short_url]
    if hashtags:
        parts.append(" ".join(_normalize_hashtag(t) for t in hashtags))
    if include_disclosure:
        parts.append(AD_DISCLOSURE)
    return "\n\n".join(parts)

글자 수(500자) 초과 처리도 같은 원칙이다. 링크와 광고 고지는 절대 잘리면 안 되는 부분이므로, 초과분은 항상 상품 설명(head) 쪽에서만 줄인다.

threads/content.py truncate head, never tail

def _compose(head, tail):
    text = f"{head}\n\n{tail}"
    if len(text) <= MAX_POST_LENGTH:
        return text

    budget = MAX_POST_LENGTH - len(tail) - 2
    if budget < len(_TRUNCATION_MARK):
        raise ContentGenerationError("링크·고지만으로 이미 글자 수를 초과합니다.")
    truncated_head = head[:budget - len(_TRUNCATION_MARK)].rstrip() + _TRUNCATION_MARK
    return f"{truncated_head}\n\n{tail}"

5. "발행됐는지 모르겠다"는 상태를 SQLite로 정직하게 다룬다

가장 신경 쓴 부분이다. 네트워크가 발행 요청 도중 끊기면, 서버에는 이미 게시됐는데 클라이언트만 실패로 아는 상황이 생길 수 있다. 이걸 자동으로 재시도하면 중복 게시로 이어진다. 그래서 상태를 3단계로 나누고, 애매한 상태는 자동으로 넘어가지 않게 막았다.

reserve() → pending → published / failed

threads/store.py idempotent reservation

def reserve(self, taca_item_id, *, daily_limit=None):
    existing = self.get(taca_item_id)
    if existing is not None:
        if existing.status == "published":
            raise AlreadyPostedError(taca_item_id, existing.media_id)
        if existing.status == "pending":
            # 이전 시도의 결과를 모른다 — 자동 재시도로 넘어가지 않는다
            raise PendingPostConflictError(taca_item_id, existing.creation_id, existing.reserved_at)
        # status == "failed" 는 재시도 허용

    if daily_limit is not None and self.count_published_since(24.0) >= daily_limit:
        raise DailyPostLimitReached(...)

pending에서 멈춘 건은 자동으로 풀리지 않는다. Threads 앱을 직접 열어 실제로 게시됐는지 확인한 뒤 resolve_pending()으로 사람이 결과를 확정해야 다음 시도가 가능하다. 번거롭지만, 중복 게시보다는 훨씬 낫다.

6. private Cloud Function + Cloud Scheduler로 토큰을 조용히 갱신한다

Threads 장기 토큰은 60일마다 만료된다. OAuth 콜백은 공개 엔드포인트여야 하지만, 토큰 갱신 함수는 그럴 필요가 없다 — 오히려 공개해두면 안 된다. 그래서 갱신 함수는 --no-allow-unauthenticated로 배포하고, Cloud Scheduler의 OIDC 토큰으로만 호출되도록 IAM invoker 권한을 딱 그 서비스 계정에만 부여했다.

Cloud Scheduler (매주 일 03:00 KST) → OIDC 토큰 → threads-token-refresh (private) → Firebase RTDB 갱신

겪었던 이슈와 주의사항

코드보다 오래 걸렸던 것들

Infra

gcloud services enable cloudscheduler.googleapis.com 직후 바로 스케줄러 작업을 만들면 SERVICE_DISABLED 에러가 난다. API 목록에는 즉시 나타나도, 실제로 쓸 수 있게 되기까지 전파 지연이 있다 — 90초 정도 기다렸다가 재시도하면 해결된다.

IAM

--no-allow-unauthenticated로 배포한 Cloud Function은 배포한 서비스 계정이라도 자동으로 호출 권한을 갖지 않는다. gcloud run services add-iam-policy-binding ... --role=roles/run.invoker 를 명시적으로 실행해야 Cloud Scheduler가 실제로 호출할 수 있다.

Meta OAuth

redirect URI를 Threads Login 설정에만 등록해도 error_code
1349168(차단된 URL)이 날 수 있다. 앱 대시보드의 App Domains에 도메인만 따로 등록해야 화이트리스트가 완성된다 — 둘 다 챙겨야 한다.

Legal

광고/제휴 고지 문구(AD_DISCLOSURE)는 상수 하나로 관리해서 바꾸기는 쉽게 만들었지만, 문구 자체가 표시광고법 요건을 충분히 충족하는지는 엔지니어링으로 해결되는 문제가 아니다. 코드는 "고지를 빠뜨리지 않는 것"까지만 보장하고, 문구의 적절성은 계속 법무 검토 대상으로 남겨뒀다.

Habit

실제 계정에 처음 게시하기 전, 발행 API를 호출하지 않고 문구·이미지·글자수만 출력하는 드라이런 스크립트를 먼저 돌렸다. 공개적으로 노출되는 작업(실계정 게시, 프로덕션 배포)은 결과를 미리 보여주고 확인받은 뒤에 실행하는 습관이 결국 시간을 아껴줬다.

테스트 전략

네트워크 없이, 그러나 촘촘하게

두 API 클라이언트 모두 실제 HTTP를 타지 않는다. requests.Session을 흉내 낸 FakeSession에 미리 정해둔 응답 큐를 넣어두고, 재시도 횟수까지 len(session.calls)로 정확히 검증한다. 상태 저장소 테스트는 PostStore(":memory:")로 디스크 I/O 없는 SQLite를 썼다 — 파일 경로 정리나 정리(teardown) 걱정 없이 상태 전이만 순수하게 검증할 수 있다.

파일	기법	검증 포인트
test_client.py	FakeSession	토큰 재사용, 재시도별 정확한 호출 횟수
test_threads_client.py	FakeSession	폴링 분기, 인증/한도/서버 에러 분류
test_content.py	고정 fixture	product_url 미사용, 500자 트렁케이션
test_store.py / test_publisher.py	in-memory SQLite	중복 게시 방지, pending 충돌, 일일 한도

돌아보며

가장 오래 고민한 건 화려한 기능이 아니라 실패를 어떻게 정직하게 다룰지였다 — 어떤 에러는 재시도해도 되고, 어떤 에러는 절대 안 되고, 어떤 상태는 사람이 봐야만 한다. 이 세 가지를 구분하는 것만으로 코드의 절반이 결정됐다.

다음으로 손댈 곳은 CAROUSEL(여러 이미지) 발행과, 라즈베리파이 운영 중에 실제로 pending 충돌이 얼마나 자주 나는지 지켜보는 것 — 지금은 이론적으로만 대비해둔 경로라 실전 데이터가 필요하다.

토스쇼핑 쉐어링크 × Threads 자동 게시 파이프라인 빌드로그 · Python 3.12

오늘의 이야기

#스하리1000명프로젝트,
迷失在韓國？即使您不會說韓語，這個應用程式也可以幫助您輕鬆出行。
只需說出您的語言即可 - 它會翻譯、搜尋並以您的語言顯示結果。
非常適合旅行者！支援英語、日語、中文、越南語等10多種語言。
現在就試試吧！
https://play.google.com/store/apps/details?id=com.billcoreatech.opdgang1127

오늘의 이야기

#스하리1000명프로젝트,
迷失在韩国？即使您不会说韩语，这个应用程序也可以帮助您轻松出行。
只需说出您的语言即可 - 它会翻译、搜索并以您的语言显示结果。
非常适合旅行者！支持英语、日语、中文、越南语等10多种语言。
现在就试试吧！
https://play.google.com/store/apps/details?id=com.billcoreatech.opdgang1127

오늘의 이야기

#스하리1000명프로젝트,
韓国で迷子になりましたか？韓国語が話せなくても、このアプリを使えば簡単に移動できます。
あなたの言語で話すだけで、翻訳、検索が行われ、結果があなたの言語で表示されます。
旅行者に最適!英語、日本語、中国語、ベトナム語などを含む 10 以上の言語をサポートします。
今すぐ試してみましょう！
https://play.google.com/store/apps/details?id=com.billcoreatech.opdgang1127

오늘의 이야기

#스하리1000명프로젝트,
한국에서 길을 잃었나요? 한국어를 못하더라도 이 앱을 사용하면 쉽게 돌아다닐 수 있습니다.
귀하의 언어로 말하면 귀하의 언어로 번역, 검색 및 결과가 표시됩니다.
여행자에게 좋습니다! 영어, 일본어, 중국어, 베트남어 등 10개 이상의 언어를 지원합니다.
지금 사용해 보세요!
https://play.google.com/store/apps/details?id=com.billcoreatech.opdgang1127

오늘의 이야기

#스하리1000명프로젝트,
Perso in Corea? Anche se non parli coreano, questa app ti aiuta a muoverti facilmente.
Basta parlare la tua lingua: traduce, cerca e mostra i risultati nella tua lingua.
Ottimo per i viaggiatori! Supporta oltre 10 lingue tra cui inglese, giapponese, cinese, vietnamita e altre.
Provalo adesso!
https://play.google.com/store/apps/details?id=com.billcoreatech.opdgang1127

오늘의 이야기

#스하리1000명프로젝트,
Perdu en Corée ? Même si vous ne parlez pas coréen, cette application vous aide à vous déplacer facilement.
Parlez simplement votre langue : il traduit, recherche et affiche les résultats dans votre langue.
Idéal pour les voyageurs ! Prend en charge plus de 10 langues, dont l'anglais, le japonais, le chinois, le vietnamien et plus encore.
Essayez-le maintenant !
https://play.google.com/store/apps/details?id=com.billcoreatech.opdgang1127

오늘의 이야기

#스하리1000명프로젝트,
¿Perdido en Corea? Incluso si no hablas coreano, esta aplicación te ayuda a moverte fácilmente.
Simplemente hable su idioma: traduce, busca y muestra resultados en su idioma.
¡Genial para viajeros! Admite más de 10 idiomas, incluidos inglés, japonés, chino, vietnamita y más.
¡Pruébalo ahora!
https://play.google.com/store/apps/details?id=com.billcoreatech.opdgang1127