토스쇼핑 쉐어링크를 Threads에 자동 게시하기까지
상품 조회부터 추적 링크 발급, 게시 문구 생성, 중복 방지, cron 자동화까지 — 하나의 파이프라인을 만들며 실제로 부딪힌 설계 결정과 구현 기법을 정리했다.
무엇을 만들었나
체크리스트 8개 항목 — 전부 완료
토스쇼핑 쉐어링크 Open API로 상품을 조회하고 추적 링크(short_url)를 발급한 뒤, 그 링크로 게시 문구를 만들어 Threads에 자동으로 올리는 파이프라인이다. 인증, 발행, 문구 생성, 중복 방지, 트리거까지 전 구간을 직접 설계했다.
DONE
DONE
DONE
DONE
DONE
DONE
DONE
DONE
순서대로 만든 게 아니라 2 → 3 → 6번을 먼저 뚫고, 4·5·7·8은 그 과정에서 자연히 같이 끝났다.
전체 흐름
핵심 구현 기술
코드로 남길 만했던 결정들
1. HTTP 상태 코드가 아니라 응답 바디로 성공/실패를 판정한다
쉐어링크 API는 HTTP 200이어도 바디의 resultType이 FAIL일 수 있다. 반대로 Threads(Meta Graph API)는 HTTP 상태와 error 객체 유무로 판정한다. 두 API가 실패를 표현하는 방식이 완전히 다르다는 걸 먼저 인정하고, 각자에 맞는 판정 함수를 따로 짰다 — 하나의 공통 추상화로 억지로 묶지 않았다.
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는 폴링이 필요 없어서 분기했다.
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도 참조도 하지 않게 만들었다. 리뷰나 주의력에 기대지 않고, 구조로 막는 편이 항상 더 오래 간다.
# 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) 쪽에서만 줄인다.
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단계로 나누고, 애매한 상태는 자동으로 넘어가지 않게 막았다.
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 권한을 딱 그 서비스 계정에만 부여했다.
겪었던 이슈와 주의사항
코드보다 오래 걸렸던 것들
gcloud services enable cloudscheduler.googleapis.com 직후 바로 스케줄러 작업을 만들면 SERVICE_DISABLED 에러가 난다. API 목록에는 즉시 나타나도, 실제로 쓸 수 있게 되기까지 전파 지연이 있다 — 90초 정도 기다렸다가 재시도하면 해결된다.
--no-allow-unauthenticated로 배포한 Cloud Function은 배포한 서비스 계정이라도 자동으로 호출 권한을 갖지 않는다. gcloud run services add-iam-policy-binding ... --role=roles/run.invoker 를 명시적으로 실행해야 Cloud Scheduler가 실제로 호출할 수 있다.
redirect URI를 Threads Login 설정에만 등록해도 error_code(차단된 URL)이 날 수 있다. 앱 대시보드의 App Domains에 도메인만 따로 등록해야 화이트리스트가 완성된다 — 둘 다 챙겨야 한다.
1349168
광고/제휴 고지 문구(AD_DISCLOSURE)는 상수 하나로 관리해서 바꾸기는 쉽게 만들었지만, 문구 자체가 표시광고법 요건을 충분히 충족하는지는 엔지니어링으로 해결되는 문제가 아니다. 코드는 "고지를 빠뜨리지 않는 것"까지만 보장하고, 문구의 적절성은 계속 법무 검토 대상으로 남겨뒀다.
실제 계정에 처음 게시하기 전, 발행 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 충돌이 얼마나 자주 나는지 지켜보는 것 — 지금은 이론적으로만 대비해둔 경로라 실전 데이터가 필요하다.
