오늘의 이야기

토스쇼핑 쉐어링크 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 기준으로 조정하세요.