AI 자동화 연구소

실패했다고 다시 부르면, 더 나빠집니다 외부 API를 호출하는 코드를 짜다 보면 자연스럽게 이런 생각이 듭니다. "실패하면 몇 번 더 해보면 되겠지." 그래서 for 문으로 세 번 감싸고, 사이에 sleep 을 하나 넣습니다. 이게 위험합니다. 실패에는 다시 시도하면 성공하는 것과, 백 번을 해도 똑같은 것이 있습니다. 후자를 반복하면 응답 시간만 늘어나고, 최악의 경우 API 제공자로부터 호출량 제한에 걸립니다. 문제를 해결하려던 재시도가 새로운 문제를 만드는 셈입니다. 이 글은 블로그 자동 발행 서비스를 운영하면서 실제로 적용한 재시도 정책을 정리한 것입니다. Python과 httpx 를 예로 들지만, 원리는 어떤 언어에서든 같습니다. HTTP 상태코드의 의미를 어느 정도 알고 있다고 가정합니다. 상태코드별 처리표 결론부터 놓겠습니다. 제가 실제로 쓰고 있는 정책입니다. 상태코드 의미 처리 400 요청이 잘못됨 재시도하지 않음 401 인증 실패 토큰 갱신 후 1회만 재시도 403 권한 없음 재시도하지 않음 404 대상이 없음 재시도하지 않음 429 호출량 초과 Retry-After 확인 후 짧으면 대기, 길면 포기 5xx 서버 쪽 문제 지수 백오프 + 지터로 재시도 표를 관통하는 기준은 하나입니다. "내가 똑같이 다시 보내면, 결과가 달라질 여지가 있는가." 있으면 재시도하고, 없으면 즉시 포기합니다. 왜 이렇게 나눴나 400, 403, 404는 내 요청이 문제입니다. 필드 이름...

여러 외부 API를 호출하는 서버라면, 요청을 기다리는 동안 다른 일을 처리하는 비동기 방식이 유리합니다. 파이썬에서 비동기 HTTP 호출에 널리 쓰이는 httpx의 AsyncClient 사용법을, 실제 운영 중인 수집·발행 코드에서 쓰는 형태를 기준으로 정리합니다. 이 글의 가정 : async/await(파이썬 비동기 문법) 기본 개념을 알고 있고, Python 3.8 이상 환경을 가정합니다. 설치는 pip install httpx 입니다. AsyncClient 기본 사용 httpx는 동기 방식과 비동기 방식을 모두 지원합니다. 비동기에서는 AsyncClient 를 async with 블록으로 열고 await 로 요청을 보냅니다. import httpx async with httpx.AsyncClient(timeout=30) as client: resp = await client.post(url, headers=headers, json=payload) data = resp.json() async with 로 열면 블록이 끝날 때 클라이언트가 자동으로 정리됩니다. 타임아웃 설정 외부 API가 응답하지 않을 때 무한정 기다리지 않도록 timeout 을 지정하는 것이 중요합니다. 호출의 성격에 따라 값을 다르게 둘 수 있습니다. 예를 들어 인증 토큰 발급처럼 빨라야 하는 호출은 약 10초, 데이터 수집은 약 30초, 생성 시간이 긴 AI 언어 모델 호출은 약 40초처럼 맥락에 맞춰 정합니다. 응답 처리와 에러 스킵 여러 요청을 다룰 때는, 하나가 실패해도 전체가 멈추지 않도록 상태 코드를 확인하고 건너뛰는 방식이 실용적입니다. async with httpx.AsyncClient(timeout=30) as client: resp = await client.get(api_url, params={"q": keyword}) if resp.status_code != 200: prin...

블로그 글을 코드로 자동 발행하려면 Blogger API와 OAuth 2.0 인증이 필요합니다. 실제로 이 블로그를 운영하는 자동화 서버가 사용하는 흐름을 기준으로, 액세스 토큰 발급부터 글 생성과 공개 전환까지 정리합니다. 이 글의 가정 : Google 계정과 Blogger 블로그가 있고, Google Cloud에서 OAuth 클라이언트(client_id, client_secret)와 refresh token을 이미 발급받았다고 가정합니다. 발급 절차 자체는 Google 공식 인증 흐름을 따릅니다. 비밀값은 코드에 직접 쓰지 말고 환경변수로 관리합니다. OAuth 2.0 refresh token 흐름 refresh token(장기 보관용 재발급 토큰)은 한 번 발급받아 두고, API를 호출할 때마다 짧은 수명의 access token(실제 요청에 쓰는 접근 토큰)으로 교환해서 사용합니다. 즉 "refresh token → access token → API 호출" 순서입니다. 액세스 토큰 발급 토큰 교환은 Google 토큰 엔드포인트에 grant_type=refresh_token 으로 요청합니다. 아래는 비동기 HTTP 클라이언트 httpx를 사용한 예입니다. import os, httpx TOKEN_URL = "https://oauth2.googleapis.com/token" API_BASE = "https://www.googleapis.com/blogger/v3" async def get_access_token(): async with httpx.AsyncClient(timeout=10.0) as client: resp = await client.post(TOKEN_URL, data={ "client_id": os.environ["BLOGGER_CLIENT_ID"], "clie...

백그라운드에서 일정 간격으로 도는 작업(주기 수집, 정리 배치 등)이 필요할 때, 별도의 크론 서버 없이 FastAPI 앱 안에서 처리하는 방법을 정리합니다. 실제 운영 중인 유튜브 쇼츠 자동화 서버에서 트렌드 수집을 주기적으로 돌리는 코드를 기준으로 설명합니다. 이 글의 가정 : FastAPI 앱이 이미 동작하고 있고, Python 3.10 이상 환경을 가정합니다. 패키지는 pip install apscheduler 로 설치합니다. APScheduler란 APScheduler(Advanced Python Scheduler, 파이썬에서 예약 작업을 관리하는 라이브러리)는 "언제 무엇을 실행할지"를 코드로 등록해두면 그 시점에 함수를 대신 호출해 줍니다. FastAPI가 비동기(async) 기반이므로, 같은 이벤트 루프에서 도는 AsyncIOScheduler 를 사용합니다. 앱 생명주기에 스케줄러 붙이기 스케줄러는 앱이 켜질 때 시작하고, 꺼질 때 정리해야 합니다. FastAPI의 startup / shutdown 이벤트에 연결합니다. from apscheduler.schedulers.asyncio import AsyncIOScheduler scheduler = AsyncIOScheduler(timezone="Asia/Seoul") @app.on_event("startup") async def startup(): scheduler.start() @app.on_event("shutdown") async def shutdown(): scheduler.shutdown() timezone 을 명시하면 서버 로캘과 무관하게 의도한 시간대로 동작합니다. interval 작업 등록하기 가장 흔한 형태는 "일정 간격마다 반복(interval)"입니다. 수집 함수를 일정 주기로 등록합니다. @app.on_event("startup...

유튜브 쇼츠를 직접 촬영하지 않고 Python으로 자동 생성하는 시스템을 실제로 운영하고 있다. UGREEN DXP4800PLUS NAS에서 YouTube API로 트렌드 데이터를 수집하고, 성장률 분석 → 주제 선정 → 대본 생성 → 음성 합성 → 영상 합성까지 자동으로 처리한다. 현재 약 1,100개 영상 데이터를 수집 중이며, 30분마다 트렌드 분석을 수행한다. 이 글은 그 파이프라인 전체 구조와 실제로 막혔던 부분들을 정리한다. 전체 파이프라인 구조 YouTube API ↓ PostgreSQL (영상 데이터 저장) ↓ 성장률 분석 (6시간 / 24시간) ↓ AI 언어 모델 (대본 생성) ↓ Edge-TTS (한국어 음성 합성) ↓ FFmpeg (자막 + 배경 합성) ↓ drafts/ 폴더 저장 ↓ 수동 검토 ↓ YouTube 업로드 각 단계가 독립적인 Python 스크립트로 구성돼 있어서 문제가 생긴 단계만 수정하고 나머지는 그대로 쓸 수 있다. 1. PostgreSQL에서 트렌드 주제 선정 단순히 조회수 높은 영상이 아니라 6시간 성장률 기준으로 지금 뜨는 주제를 고른다. 성장률 계산 구조는 이전 글 PostgreSQL로 YouTube 조회수 성장률 계산하기 에서 다뤘다. import asyncpg async def get_trending_topics(): conn = await asyncpg.connect(DATABASE_URL) rows = await conn.fetch(""" SELECT v.title, v.content_cluster, s.growth_6h FROM videos v JOIN video_snapshots s ON v.video_id = s.video_id WHERE s.recorded_at = (SELECT MAX(recorded_at) FROM video_snapshots) ...

UGREEN DXP4800PLUS NAS에서 FastAPI 기반 쇼츠 트렌드 수집 시스템을 운영하고 있다. 현재 DB에는 영상 700개 이상, 스냅샷 40,000개 이상이 쌓여 있고, 30분마다 자동으로 수집·저장이 돌아간다. 이 구조를 만들면서 FastAPI와 PostgreSQL 연동에서 막혔던 부분들을 정리한다. 단순한 예제 코드가 아니라 Docker Compose 환경에서 실제로 동작하는 구성을 기준으로 설명한다. 현재 운영 환경 NAS: UGREEN DXP4800PLUS (Debian 12 기반) FastAPI: Docker 컨테이너로 실행 PostgreSQL: 동일 Docker Compose 내 별도 컨테이너 ORM: SQLAlchemy 2.0 (비동기, asyncpg 드라이버) DB명: shorts / 주요 테이블: videos, snapshots 1. Docker Compose 구성 FastAPI와 PostgreSQL을 같은 Compose 파일 안에 두면 컨테이너 이름으로 서로 통신할 수 있다. services: db: image: postgres:15 environment: POSTGRES_DB: shorts POSTGRES_USER: user POSTGRES_PASSWORD: password volumes: - postgres_data:/var/lib/postgresql/data healthcheck: test: ["CMD-SHELL", "pg_isready -U user -d shorts"] interval: 5s retries: 5 api: build: . ports: - "8105:8000" environment: DATABASE_URL: postgresql+asyncp...

  처음에는 단순히 API 몇 개 호출하면 끝날 줄 알았다. UGREEN DXP4800PLUS NAS 위에서 유튜브 쇼츠 자동화 시스템을 구축하면서 YouTube Data API를 쓰기 시작했는데, 실제로 운영해보니 쿼터 관리가 더 큰 문제였다. YouTube Data API를 사용해 쇼츠 수집기나 트렌드 분석 시스템을 만들다 보면 쿼터(Quota) 부족 오류를 한 번쯤 겪게 된다. 하루 10,000유닛이면 사실상 무제한이라고 생각했는데, 글로벌 AI 키워드를 몇 개 추가하고 나서 하루도 지나지 않아 쿼터 부족 오류가 발생했다. 로그를 확인해보니 범인은 search.list 였다. 같은 API라도 videos.list 는 1유닛인데 search.list 는 100유닛이다. 이 차이를 이해하기 전까지는 쿼터가 왜 사라지는지 전혀 감이 오지 않았다. 약 700개 영상, 4만 개 이상의 스냅샷을 수집하는 과정에서 부딪힌 쿼터 제한과 해결 방법을 정리한다. YouTube Data API 기본 제한 사항 YouTube Data API v3는 무료로 제공되지만 일일 쿼터(Quota)가 존재한다. 기본 할당량: 하루 10,000 유닛 주요 API별 소비 유닛: videos.list : 1 유닛 search.list : 100 유닛 channels.list : 1 유닛 search.list 가 videos.list 보다 100배 비싸다. 처음에 이걸 몰라서 쿼터를 빠르게 소진했다. 실제로 겪은 문제들 문제 1: search.list 남용으로 쿼터 소진 초기 수집기는 이런 구조였다. # 초기 구조 - 쿼터 낭비 search_results = search . list ( q = "쇼츠" , type = "video" ) # 100 유닛 for video in search_results : detail = videos . list ( id = video . id ) # 1 유닛씩 s...

필자는 UGREEN DXP4800PLUS NAS 환경에서 Docker 기반 유튜브 쇼츠 자동화 시스템을 운영하고 있다. FastAPI와 PostgreSQL로 트렌드 수집 대시보드를 구축하던 중 모든 API 엔드포인트가 404를 반환하는 문제를 겪었다. 처음에는 PostgreSQL 연결 문제라고 생각했다. API가 전부 404를 반환하니 DB 연결이 끊어진 줄 알았다. docker compose logs로 컨테이너 로그를 뒤졌고, curl로 직접 호출해보니 FastAPI까지 요청이 들어오지도 않았다. 결국 원인은 app.mount("/") 한 줄이었다. 수정하고 나서 허탈했다. 30분을 날렸다.  app.mount("/") 선언 위치 하나 때문이었다. 수정 후 즉시 정상 동작했다. 이 글에서는 해당 오류의 원인과 해결 방법, 그리고 FastAPI 내부 동작 원리까지 정리한다. 문제 상황 아래와 같이 코드를 작성했을 때 /api/trends 엔드포인트가 404를 반환한다. from fastapi import FastAPI from fastapi . staticfiles import StaticFiles app = FastAPI ( ) # 정적 파일 마운트 app . mount ( "/" , StaticFiles ( directory = "static" , html = True ) , name = "static" ) # API 엔드포인트 선언 @app . get ( "/api/trends" ) async def get_trends ( ) : return { "status" : "success" } 코드 자체는 문법적으로 문제가 없다. 하지만 실행하면 /api/trends 호출 시 404가 반환된다. 원인 분석 FastAPI는 라우팅 처리 시 선언 순서대로 매칭 을 시도한다....

 집에 NAS가 있다면 유튜브 쇼츠 자동화 시스템을 직접 구축할 수 있다. 이 글은 UGREEN DXP4800PLUS NAS 위에서 실제로 운영 중인 쇼츠 자동화 파이프라인 구축 과정을 기록한 실전 후기다. 현재 운영 현황 (2026년 6월 기준) 수집 영상: 676개 저장 스냅샷: 38,134개 성장률 계산 성공률: 95.1% 트렌드 클러스터: 10종 (IT_DEVICE, AI/자동화, 연예/엔터 등) 자동 영상 생성: Video Factory v2 개발 중 단순 성공 사례가 아니다. FastAPI 404 오류, 분류기 정확도 문제, 영상 생성 실패까지 실제 겪은 삽질과 해결 과정을 함께 담았다. 전체 파이프라인 구조 YouTube API 수집 (30분마다) ↓ PostgreSQL 저장 ↓ 성장률 계산 ↓ content_cluster 분류 (10종) ↓ format_type 분류 (6종) ↓ 트렌드 대시보드 ↓ AI 대본 자동 생성 ↓ Edge-TTS 음성 생성 ↓ FFmpeg 영상 합성 ↓ drafts 폴더 저장 처음에는 간단한 스크립트 몇 개면 끝날 줄 알았다. 실전은 달랐다. FastAPI app.mount("/") 사용 시 API 404 오류 해결 방법 FastAPI에서 정적 파일을 서빙할 때 이런 코드를 썼다. # 오류 코드 - API가 전부 404 app . mount ( "/" , StaticFiles ( directory = "static" , html = True ) , name = "static" ) @app . get ( "/api/trends" ) async def get_trends ( ) : . . . API 호출마다 404가 났다. 원인은 라우팅 우선순위였다. app.mount("/") 가 상단에 있으면 뒤에 선언된 모든 엔드포인트가 무시된다. # 해결 코드 - 항상 moun...