🧵 Python으로 Threads API 자동화하기

 

 

🧵 Python으로 Threads API 자동화하기

 

 

threads 게시 예제

 

목표: 이미지와 텍스트를 Threads API를 통해 자동으로 게시하고, Access Token을 주기적으로 갱신하여 안정적인 운영 시스템을 구축합니다.

---

1. 프로젝트 개요

• Threads API를 활용한 이미지 업로드 및 게시
• SQLite로 Access Token 저장 및 발급일 관리
• 자동 토큰 갱신 (60일 유효 기간 감지)

2. 초기 Access Token 저장 (최초 1회)

import sqlite3
from datetime import datetime

DB_PATH = "token_store.db"
INIT_TOKEN = "여기에 초기 access_token 입력"
ISSUE_DATE = datetime(2025, 6, 17)

def initialize_token():
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS tokens (
            id INTEGER PRIMARY KEY,
            access_token TEXT NOT NULL,
            issued_at TEXT NOT NULL
        )
    ''')
    cursor.execute("DELETE FROM tokens")
    cursor.execute("INSERT INTO tokens (access_token, issued_at) VALUES (?, ?)", 
                   (INIT_TOKEN, ISSUE_DATE.isoformat()))
    conn.commit()
    conn.close()

initialize_token()

3. Access Token 자동 갱신 함수

import requests
from datetime import datetime, timedelta

PAGE_ID = "페이지 ID 입력"

def get_token():
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    cursor.execute("SELECT access_token, issued_at FROM tokens ORDER BY id DESC LIMIT 1")
    row = cursor.fetchone()
    conn.close()
    return row if row else (None, None)

def save_token(new_token):
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    cursor.execute("INSERT INTO tokens (access_token, issued_at) VALUES (?, ?)", 
                   (new_token, datetime.utcnow().isoformat()))
    conn.commit()
    conn.close()

def refresh_token_if_needed():
    token, issued_at = get_token()
    if not token:
        raise Exception("토큰이 없습니다.")
    issued_time = datetime.fromisoformat(issued_at)
    if datetime.utcnow() - issued_time > timedelta(days=55):
        print("🔄 갱신 시도 중...")
        response = requests.get(
            f"https://graph.threads.net/v1.0/{PAGE_ID}/refresh_access_token",
            params={"access_token": token}
        )
        if response.status_code == 200:
            new_token = response.json().get("access_token")
            save_token(new_token)
            print("✅ 토큰 갱신 완료")
            return new_token
        else:
            raise Exception("❌ 갱신 실패")
    return token

4. 게시글 업로드 & 발행 함수

def upload_and_publish(image_url, text, page_id, access_token):
    create_url = f"https://graph.threads.net/v1.0/{page_id}/threads"
    payload = {
        "media_type": "IMAGE",
        "image_url": image_url,
        "text": text,
        "access_token": access_token
    }
    create_response = requests.post(create_url, data=payload)
    if create_response.status_code == 200:
        creation_id = create_response.json().get("id")
        print(f"✅ 업로드 성공. creation_id: {creation_id}")
        publish_url = f"https://graph.threads.net/v1.0/{page_id}/threads_publish"
        publish_response = requests.post(publish_url, params={
            "creation_id": creation_id,
            "access_token": access_token
        })
        if publish_response.status_code == 200:
            print("🎉 게시 완료")
        else:
            print("❌ 게시 실패:", publish_response.text)
    else:
        print("❌ 업로드 실패:", create_response.text)

5. 전체 실행 흐름

if __name__ == "__main__":
    from datetime import datetime
    DB_PATH = "token_store.db"
    PAGE_ID = "page id 입력"
    IMAGE_URL = "http://이미지서버.myddns.me:8059/image_sender1.png"
    TEXT = "#python #앱 https://play.google.com/store/apps/details?id=com.billcoreatech.opdgang1127&pcampaignid=web_share"

    try:
        token = refresh_token_if_needed()
        upload_and_publish(IMAGE_URL, TEXT, PAGE_ID, token)
    except Exception as e:
        print("🚨 오류 발생:", e)

📝 마무리

이렇게 Python과 SQLite를 이용해 Threads API 자동화를 구현했습니다. 토큰 갱신과 게시 흐름이 깔끔하게 처리되기 때문에, 향후 자동 배포나 백그라운드 업로드도 손쉽게 확장할 수 있어요.

 

** 테스트을 위해서는 테스터 등록을 해야 합니다. 또한 등록된 테스터의 계정에서 승인 해야 하는 데, 내가 관리 하는 계정의 경우

아래 그림과 같이 등록된 테스터 목록에 나오는 "웹사이트 권한" 링크를 타고 들어가면 바로 테스트 초대 을 찾을 수 있으니 해당 메뉴에서 승인을 해 주면 바로 적용 됩니다.