token = os.environ["TELEGRAM_BOT_TOKEN"]
File "/usr/local/lib/python3.12/os.py", line 685, in __getitem__
raise KeyError(key) from None
KeyError: 'TELEGRAM_BOT_TOKEN'
```
### 원인
소스 코드에서는 `os.environ["..."]`으로 특정 환경변수를 요구하고 있으나, DashCloud 대시보드의 **환경변수** 설정에 해당 키가 등록되지 않았거나 오타가 있는 경우입니다.
### 1분 해결법
1. 프로젝트 상세 페이지 ➔ **환경 변수** 탭으로 이동합니다.
2. 누락된 환경변수 이름(`TELEGRAM_BOT_TOKEN`)을 오타 및 앞뒤 공백 없이 정확히 입력하고 토큰 값을 넣습니다.
3. 코드 작성 시 `os.environ["KEY"]` 대신 `os.environ.get("KEY")`를 사용하고 누락 시 명확한 에러 메시지를 출력하도록 방어 코드를 작성하는 것을 권장합니다:
```python
token = os.environ.get("TELEGRAM_BOT_TOKEN")
if not token:
logger.critical("필수 환경변수 TELEGRAM_BOT_TOKEN이 등록되지 않았습니다!")
sys.exit(1)
```
---
## 4. `telegram.error.Conflict: Terminated by other getUpdates request`
### 증상 및 로그
```text
telegram.error.Conflict: Conflict: terminated by other getUpdates request;
make sure that only one bot instance is running
```
### 원인
**동일한 텔레그램 봇 토큰으로 두 개 이상의 프로세스가 동시에 롱폴링을 요청**했을 때 텔레그램 서버가 기존 연결을 끊어버리는 현상입니다.
* 내 로컬 컴퓨터 터미널에서 봇을 켜둔 상태로 DashCloud에도 배포한 경우
* 다른 호스팅이나 이전 테스트 서버에 봇이 여전히 실행 중인 경우
### 1분 해결법
1. 내 로컬 컴퓨터 터미널이나 VS Code에서 실행 중인 파이썬 프로세스를 `Ctrl + C`로 완전히 종료합니다.
2. BotFather에서 봇 토큰을 재발급받아 대시보드 환경변수에 새로 입력하면 기존의 모든 유령 연결이 즉시 강제 종료됩니다.
---
## 5. `sqlite3.OperationalError: database is locked`
### 증상 및 로그
```text
sqlite3.OperationalError: database is locked
```
### 원인
여러 비동기 코루틴이나 스레드가 단일 SQLite 파일에 동시에 쓰기 트랜잭션을 시도하여 파일 락 충돌이 발생한 경우입니다.
### 1분 해결법
SQLite 연결을 생성할 때 **WAL(Write-Ahead Logging) 저널 모드**를 활성화하고 연결 타임아웃을 넉넉히 지정합니다.
```python title="database.py"
import sqlite3
def get_connection():
conn = sqlite3.connect("/app/data/bot.db", timeout=30.0)
# WAL 모드 활성화: 동시 읽기/쓰기 성능 및 충돌 방지 극대화
conn.execute("PRAGMA journal_mode=WAL;")
conn.execute("PRAGMA synchronous=NORMAL;")
return conn
```
---
## 6. `FileNotFoundError: [Errno 2] No such file or directory`
### 증상 및 로그
```text
FileNotFoundError: [Errno 2] No such file or directory: 'C:\\Users\\admin\\bot\\config.json'
```
### 원인
코드 내부에 로컬 PC 전용 절대 경로(`C:\Users\...` 또는 `/Users/name/...`)가 하드코딩되어 클라우드 리눅스 파일시스템에서 경로를 찾지 못하는 경우입니다.
### 1분 해결법
`pathlib.Path`를 사용하여 항상 현재 스크립트 실행 위치 기준의 상대 경로로 참조하세요.
```python
from pathlib import Path
BASE_DIR = Path(__file__).resolve().parent
CONFIG_PATH = BASE_DIR / "config.json"
DATA_DIR = BASE_DIR / "data"
```
---
## 7. `telegram.error.RetryAfter: Flood control exceeded` (429 에러)
### 증상 및 로그
```text
telegram.error.RetryAfter: Flood control exceeded. Retry in 18 seconds.
```
### 원인
텔레그램 봇 API의 초당 발송 한도(그룹 채팅 초당 20건, 개인 초당 30건)를 초과하여 대량의 메시지를 짧은 시간에 쏟아부었을 때 발생하는 레이트 리밋 차단입니다.
### 1분 해결법
1. 메시지를 즉시 전송하지 말고 비동기 큐(`asyncio.Queue`)에 담아 초당 20건 이하로 조율하여 순차 발송합니다.
2. `RetryAfter` 예외를 캐치하여 지정된 초만큼 대기한 뒤 재전송하는 재시도 로직을 적용합니다:
```python
from telegram.error import RetryAfter
import asyncio
async def safe_send_message(bot, chat_id, text):
try:
await bot.send_message(chat_id=chat_id, text=text)
except RetryAfter as e:
logger.warning(f"속도 제한 발생. {e.retry_after}초 후 재시도합니다.")
await asyncio.sleep(e.retry_after)
await bot.send_message(chat_id=chat_id, text=text)
```
---
## 8. `OOM_KILLED` (메모리 상한 초과 강제 종료)
### 증상 및 로그
```text
[DashCloud Monitor] Process exceeded memory limit (256MB).
[DashCloud Monitor] OS Kernel sent SIGKILL.
Service status: OOM_KILLED
```
### 원인
* Playwright/Selenium 크롤러에서 페이지를 이동할 때 브라우저 컨텍스트(`context.close()`)를 닫지 않아 Chromium DOM 힙이 누적된 경우
* 대용량 파일(CSV, 엑셀, 비디오)을 청크 단위 스트리밍이 아니라 한 번에 통째로 메모리에 로드(`read()`)한 경우
### 1분 해결법
1. 크롤러의 경우 각 수집 작업마다 `new_context()`를 생성하고 반드시 `finally: context.close()`를 호출하세요.
2. 대시보드의 **메트릭** 탭에서 메모리 사용량 그래프의 우상향 기울기를 확인하고 메모리 누수 지점을 진단하세요.
3. [헤드리스 Chromium 메모리 최적화 가이드](/docs/crawler/headless-chromium)를 참고하여 이미지/폰트 네트워크 차단을 적용하세요.
---
## 9. `OSError: [Errno 98] Address already in use`
### 증상 및 로그
```text
OSError: [Errno 98] Address already in use: ('0.0.0.0', 8080)
```
### 원인
텔레그램 롱폴링 봇이나 크롤러에 불필요하게 `app.run(port=8080)` 같은 웹서버 코드가 포함되어 포트 바인딩 충돌을 일으킨 경우입니다.
### 1분 해결법
롱폴링 봇과 크롤러는 웹 서버 포트를 열 필요가 없습니다. Flask/FastAPI 서버 기동 코드를 제거하고 순수 백그라운드 봇 스크립트로 실행하세요. 외부에서 웹훅을 받아야 하는 경우에만 포트를 바인딩합니다.
---
## 10. 스크린샷 캡처 시 한글 폰트가 깨짐 (`뛟뛟`, 사각형 기호)
### 증상
Playwright로 한국어 웹페이지를 캡처했을 때 한글 텍스트가 전부 네모 박스(`□□□`)로 깨져 나오는 현상입니다.
### 원인
리눅스 서버 환경에 기본 한글 폰트 패키지가 설치되어 있지 않은 경우입니다.
### 1분 해결법
DashCloud의 **Chromium + Python (`crawler-chromium`)** 런타임 프리셋을 선택하세요. Noto Sans CJK 한글 글꼴이 OS 레벨에 기본 탑재되어 있어 별도 설정 없이 완벽한 한글 렌더링을 지원합니다.
---
## 다음 추천 문서
## 텔레그램 봇
### 텔레그램 봇 아키텍처 및 생태계 개요
출처: https://dashcloud.kr/docs/telegram/overview
텔레그램 봇은 전 세계 메신저 플랫폼 중 개발자 친화성이 가장 뛰어난 봇 생태계를 갖추고 있습니다. 카카오톡이나 라인과 달리 복잡한 비즈니스 승인 절차나 유료 메시지 과금 없이, 누구나 **무료로 무제한 봇을 생성하고 메시지를 송수신**할 수 있습니다.
특히 텔레그램은 서버로 요청을 쏘는 **아웃바운드 롱폴링** 방식을 완벽히 지원하므로, 공인 도메인이나 SSL 인증서, 방화벽 포트 개방 없이도 DashCloud 클라우드 런타임에서 10초 만에 24시간 꺼지지 않는 봇을 구축할 수 있습니다.
---
## 텔레그램 봇의 통신 아키텍처
텔레그램 봇이 사용자와 메시지를 주고받는 구조는 텔레그램 클라우드 서버를 중간 브로커로 활용합니다.
```
┌──────────────┐ MTProto 암호화 프로토콜 ┌───────────────────────────┐
│ 사용자 폰/PC │ ◄─────────────────────────────────────► │ 텔레그램 공식 클라우드 │
│ 텔레그램 앱 │ │ (Telegram Bot API Server) │
└──────────────┘ └───────────────────────────┘
▲
│ HTTPS 443 (아웃바운드)
│ getUpdates 롱폴링
▼
┌───────────────────────────┐
│ DashCloud 격리 샌드박스 │
│ ├─ 내 파이썬 봇 프로세스 │
│ ├─ SQLite 로컬 영구 DB │
│ └─ 24/7 무중단 감시 데몬 │
└───────────────────────────┘
```
1. **사용자 발송**: 사용자가 텔레그램 앱에서 봇에게 `/start` 또는 질문을 전송하면, 텔레그램 공식 데이터센터 서버에 메시지가 즉시 적재됩니다.
2. **봇 업데이트 수신**: DashCloud에서 가동 중인 내 봇 프로세스가 텔레그램 서버로 `getUpdates` 요청을 보내 새로운 메시지를 실시간으로 가져옵니다(롱폴링).
3. **비즈니스 로직 처리**: 봇이 메시지를 파싱하고, 필요 시 SQLite DB를 조회하거나 ChatGPT/LLM API를 호출하여 답변을 생성합니다.
4. **결과 전송**: 봇이 `sendMessage` API를 호출하면 텔레그램 서버를 거쳐 0.05초 만에 사용자 화면에 응답이 출력됩니다.
봇이 텔레그램 서버로 먼저 접속을 시도하는 아웃바운드 롱폴링을 사용하기 때문에, DashCloud 컨테이너의 어떤 수신 포트도 외부에 노출할 필요가 없습니다. 외부의 DDoS나 해킹 공격으로부터 완벽히 안전합니다.
---
## 주요 라이브러리 생태계 비교 매트릭스
자신이 익숙한 언어와 프로젝트의 규모, 요구 성능에 맞춰 가장 적합한 라이브러리를 선택하세요.
| 라이브러리 | 언어 / 환경 | 비동기 지원 | 주요 특징 | 추천 프로젝트 |
| :--- | :--- | :--- | :--- | :--- |
| **`python-telegram-bot`** | Python 3.10+ | 네이티브 `asyncio` | • 가장 많은 공식 예제와 성숙한 커뮤니티
• `JobQueue`(스케줄링), `ConversationHandler`(대화 상태) 내장 | **입문자부터 실무 프로덕션 봇까지 90% 이상 추천** |
| **`aiogram`** | Python 3.11+ | 순수 `asyncio` 고속 | • 유한 상태 머신 FSM과 강력한 미들웨어 구조
• 초당 수백 건 이상의 고트래픽 처리 최적화 | 고성능 대형 봇, 복잡한 사용자 입력 폼 |
| **`telebot` (pyTelegramBotAPI)** | Python 3.8+ | 동기 / 비동기 겸용 | • 구조가 가장 단순하고 코드가 직관적
• 대화 상태 관리 기능은 다소 취약 | 단순 1줄 시스템 알림 발송 봇 |
| **`telegraf`** | Node.js (JS/TS) | Promise / Async | • Node.js 진영의 사실상 표준
• Express 스타일의 미들웨어 체이닝 구조 | 웹 개발자, Node.js 풀스택 워크플로우 |
| **`grammY`** | TypeScript / Deno | 순수 TS 최적화 | • 차세대 초고속 TS 네이티브 봇 프레임워크
• 모듈형 플러그인 시스템과 완벽한 타입 추론 | 타입스크립트 기반 대규모 엔터프라이즈 봇 |
---
## 완성형 프로덕션 보일러플레이트 코드
실제 운영 환경에 즉시 투입할 수 있도록 로깅, 에러 핸들러, 안전한 Graceful Shutdown 정상 종료 처리가 모두 포함된 완성형 코드입니다.
```python title="main.py"
import os
import sys
import logging
from telegram import Update
from telegram.ext import (
Application,
CommandHandler,
MessageHandler,
ContextTypes,
filters,
)
# 1. 로깅 구성 (DashCloud 실시간 웹 콘솔에 출력됨)
logging.basicConfig(
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
level=logging.INFO,
)
logger = logging.getLogger("TelegramBot")
# 2. 명령어 핸들러 정의
async def start_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
user = update.effective_user
logger.info(f"유저 [{user.id}] {user.first_name}님이 /start를 실행했습니다.")
await update.message.reply_text(
f"반갑습니다, {user.first_name}님!\n"
"DashCloud 24시간 클라우드 런타임에서 안전하게 가동 중입니다. 🚀\n\n"
"도움말을 보려면 /help 를 입력하세요."
)
async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
help_text = (
"📖 *명령어 안내*\n\n"
"/start - 봇 시작 및 인사말\n"
"/status - 봇 가동 상태 및 Uptime 확인\n"
"/help - 이 도움말 보기"
)
await update.message.reply_markdown(help_text)
async def status_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
await update.message.reply_text("🟢 시스템 상태: 정상 가동 중 (24/7 Cloud Running)")
# 3. 일반 텍스트 에코 핸들러
async def echo_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
user_text = update.message.text
logger.info(f"메시지 수신: {user_text}")
await update.message.reply_text(f"수신 확인: {user_text}")
# 4. 전역 에러 핸들러 (예외 발생 시 봇 다운 방지)
async def error_handler(update: object, context: ContextTypes.DEFAULT_TYPE):
logger.error("요청 처리 중 예외 발생:", exc_info=context.error)
def main():
token = os.environ.get("TELEGRAM_BOT_TOKEN")
if not token:
logger.critical("환경변수 TELEGRAM_BOT_TOKEN이 누락되었습니다!")
sys.exit(1)
logger.info("텔레그램 봇 인스턴스를 빌드합니다...")
app = Application.builder().token(token).build()
# 핸들러 등록
app.add_handler(CommandHandler("start", start_command))
app.add_handler(CommandHandler("help", help_command))
app.add_handler(CommandHandler("status", status_command))
app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, echo_message))
# 에러 핸들러 바인딩
app.add_error_handler(error_handler)
logger.info("24/7 무중단 롱폴링 루프를 가동합니다...")
# drop_pending_updates=True: 재시작 시 누적된 이전 구 메시지 중복 처리 방지
app.run_polling(drop_pending_updates=True)
if __name__ == "__main__":
main()
```
```python title="main.py (aiogram v3)"
import os
import sys
import logging
import asyncio
from aiogram import Bot, Dispatcher, types
from aiogram.filters import CommandStart, Command
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("AiogramBot")
dp = Dispatcher()
@dp.message(CommandStart())
async def handle_start(message: types.Message):
await message.answer(f"안녕하세요 {message.from_user.first_name}님! aiogram v3 봇이 가동 중입니다.")
@dp.message(Command("status"))
async def handle_status(message: types.Message):
await message.answer("시스템 헬스: 정상")
async def main():
token = os.environ.get("TELEGRAM_BOT_TOKEN")
if not token:
logger.critical("TELEGRAM_BOT_TOKEN이 필요합니다.")
sys.exit(1)
bot = Bot(token=token)
logger.info("aiogram 롱폴링 시작...")
await dp.start_polling(bot, drop_pending_updates=True)
if __name__ == "__main__":
asyncio.run(main())
```
```javascript title="index.js (Telegraf)"
import { Telegraf } from 'telegraf';
const token = process.env.TELEGRAM_BOT_TOKEN;
if (!token) {
console.error("TELEGRAM_BOT_TOKEN 환경변수가 필요합니다.");
process.exit(1);
}
const bot = new Telegraf(token);
bot.start((ctx) => {
ctx.reply(`반갑습니다, ${ctx.from.first_name}님! DashCloud Node.js 런타임에서 실행 중입니다.`);
});
bot.help((ctx) => {
ctx.reply("/start - 봇 시작\n/status - 상태 확인");
});
bot.command('status', (ctx) => {
ctx.reply(`가동 시간: ${Math.floor(process.uptime())}초 (정상)`);
});
bot.catch((err, ctx) => {
console.error(`업데이트 처리 중 오류 발생 (${ctx.updateType}):`, err);
});
// 안전한 프로세스 종료 처리
process.once('SIGINT', () => bot.stop('SIGINT'));
process.once('SIGTERM', () => bot.stop('SIGTERM'));
console.log("Telegraf 봇 가동 시작...");
bot.launch({ dropPendingUpdates: true });
```
---
## DashCloud에서 텔레그램 봇을 운영할 때의 핵심 이점
1. **노트북을 꺼도 24시간 1초도 쉬지 않음**: 집 컴퓨터의 Wi-Fi 단절이나 절전 모드와 완전히 독립된 클라우드 데이터센터에서 실행됩니다.
2. **영구 SQLite 스토리지 기본 제공**: `/app/data/users.db`에 저장한 유저 데이터, 포인트, 대화 세션이 서버가 재배포되어도 지워지지 않고 보존됩니다.
3. **비정상 종료 시 실시간 웹훅 통보**: 처리되지 않은 예외로 봇이 멈추면 관리자의 디스코드/텔레그램으로 크래시 로그와 함께 즉각 경보가 전송됩니다.
4. **대량 알림 429 레이트 리밋 방어선**: 초당 수백 건의 공지 메시지를 보내도 텔레그램 밴을 피할 수 있는 버퍼 큐 설계를 지원합니다.
---
## 관련 블로그
- [텔레그램 봇 24시간 무료 호스팅 완벽 가이드](/blog/telegram/free-hosting-guide) — Render·Heroku 대안부터 상시 가동까지
- [텔레그램 블로그 전체 보기](/blog/telegram)
---
## 다음 추천 문서
### BotFather 토큰 발급과 환경변수 설정
출처: https://dashcloud.kr/docs/telegram/botfather-setup
텔레그램 봇을 생성하고 프로그래밍 언어에서 제어하려면, 텔레그램 공식 관리 봇인 **@BotFather**로부터 고유한 **HTTP API 인증 토큰**을 발급받아야 합니다.
봇 생성부터 프로필 설정, 명령어 메뉴 등록, 그리고 DashCloud에 안전하게 암호화 환경변수로 주입하기까지의 전 과정을 상세히 다룹니다.
---
## 단계별 봇 생성 및 토큰 발급 5단계
텔레그램 모바일 앱 또는 데스크톱 앱의 검색창에 `@BotFather`를 검색합니다.
반드시 사용자 이름 옆에 **텔레그램 공식 인증 배지(파란색 체크마크)**가 붙어 있는 `@BotFather`를 선택해야 합니다. 공식 인증이 없는 유사 사칭 계정에 토큰을 노출하지 않도록 주의하세요.
대화방에 입장한 후 하단의 **시작** 버튼을 누르면 사용 가능한 명령어 목록이 안내됩니다.
채팅창에 `/newbot`을 입력하고 전송합니다.
1. **봇의 표시 이름 지정**:
사용자 화면에 표시될 친근한 이름을 입력합니다. 한글이나 이모지를 포함할 수 있습니다.
```text
> /newbot
Alright, a new bot. How are we going to call it? Please choose a name for your bot.
> 내 비서 알림봇 🤖
```
2. **봇의 고유 사용자명 Username 지정**:
전 세계에서 유일해야 하며, **반드시 `bot`으로 끝나야 합니다** (대소문자 무관, 언더스코어 `_` 허용).
```text
Good. Now let's choose a username for your bot. It must end in `bot`. Like this, for example: TetrisBot or tetris_bot.
> my_personal_secretary_bot
```
사용자 이름이 중복되지 않고 정상 생성되면 BotFather가 축하 메시지와 함께 다음과 같은 형식의 **HTTP API Token**을 발급해 줍니다.
```text title="발급된 토큰 예시"
Use this token to access the HTTP API:
7123456789:AAF_kX8y-1234567890abcdefghijklmnopqrst
Keep your token secure and store it safely, it can be used by anyone to control your bot.
```
* 앞부분의 숫자(`7123456789`)는 봇의 고유 식별 번호입니다.
* 콜론 뒤의 문자열(`AAF_kX8y-...`)은 봇을 원격 조작할 수 있는 비밀 암호키입니다.
* **이 토큰을 복사하여 안전한 곳에 보관합니다.**
사용자가 봇을 편리하게 사용할 수 있도록 프로필 설명과 채팅창 좌측 하단의 명령어 자동완성 메뉴를 등록합니다.
1. **명령어 메뉴 등록 (`/setcommands`)**:
BotFather에게 `/setcommands`를 입력하고 내 봇을 선택한 뒤 아래 형식으로 입력합니다:
```text
start - 봇 시작 및 기본 안내
status - 서버 및 봇 가동 상태 확인
help - 사용 가능한 도움말 보기
alert - 알림 수신 설정 변경
```
등록을 완료하면 사용자가 텔레그램 채팅창 좌측 하단의 `Menu` 버튼을 눌렀을 때 해당 명령어들이 깔끔하게 노출됩니다.
2. **소개 문구 등록 (`/setdescription`)**:
사용자가 봇 대화방에 처음 들어왔을 때 화면 중앙에 표시될 소개 글을 작성합니다.
3. **프로필 사진 등록 (`/setuserpic`)**:
봇의 아바타 이미지를 전송하여 프로필 사진으로 지정합니다.
봇이 특정 관리자(나 자신)에게만 시스템 장애 알림이나 개인 메시지를 보내게 하려면 내 텔레그램 계정의 고유 숫자 ID(`Chat ID`)를 알아야 합니다.
1. 텔레그램 검색창에 `@userinfobot`을 검색하고 대화를 시작합니다.
2. 봇이 즉시 내 프로필 정보와 함께 `Id: 123456789` 형태의 숫자를 알려줍니다.
3. 이 숫자를 복사해 둡니다 (환경변수 `TELEGRAM_ADMIN_CHAT_ID`로 활용).
---
## DashCloud에 토큰 안전하게 주입하기
발급받은 토큰을 절대 파이썬 코드(`main.py`)나 깃허브 공개 저장소에 직접 붙여넣지 마세요. 깃허브 봇 탐지기에 의해 토큰이 1분 만에 유출되어 계정이 해킹당할 수 있습니다.
DashCloud의 암호화 환경변수 기능을 사용해 안전하게 주입합니다.
1. DashCloud 대시보드에서 내 프로젝트 상세 페이지로 이동합니다.
2. 상단 탭 중 **환경변수** 메뉴를 선택합니다.
3. 아래 표와 같이 키-값 쌍을 입력하고 **저장**을 클릭합니다:
| 변수 이름 | 입력 값 (Value 예시) | 용도 설명 |
| :--- | :--- | :--- |
| `TELEGRAM_BOT_TOKEN` | `7123456789:AAF_kX8y-...` | 텔레그램 Bot API 통신용 마스터 인증 토큰 |
| `TELEGRAM_ADMIN_CHAT_ID` | `123456789` | 에러 알림 및 비상 리포트를 수신할 관리자 고유 ID |
| `TZ` | `Asia/Seoul` | 봇 내부 시간대를 한국 표준시 KST로 동기화 |
대시보드에 등록된 환경 변수는 데이터베이스에 평문으로 남지 않고 **AES-256-GCM** 방식으로 즉시 암호화됩니다. 프로젝트 샌드박스가 가동되는 찰나에만 프로세스 메모리에 직접 주입되므로 외부 유출이 원천 차단됩니다.
---
## 코드에서 환경변수 로드하기 (파이썬 예시)
```python title="main.py"
import os
import sys
import logging
from telegram.ext import Application
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# 환경변수 안전 로드
BOT_TOKEN = os.environ.get("TELEGRAM_BOT_TOKEN")
ADMIN_ID = os.environ.get("TELEGRAM_ADMIN_CHAT_ID")
if not BOT_TOKEN:
logger.critical("❌ 환경변수 TELEGRAM_BOT_TOKEN이 등록되지 않았습니다! 대시보드를 확인하세요.")
sys.exit(1)
logger.info("토큰 로드 성공. 봇 애플리케이션을 기동합니다.")
app = Application.builder().token(BOT_TOKEN).build()
```
---
## 토큰 유출 사고 시 1분 비상 대응 매뉴얼
만약 실수로 봇 토큰을 깃허브 공개 저장소나 디스코드 채팅창에 노출했다면, 악의적인 공격자가 봇의 모든 대화방에 스팸을 뿌리거나 채널 권한을 탈취할 수 있습니다. 즉시 다음 조치를 취하세요:
1. **BotFather에게 `/revoke` 전송**:
텔레그램 `@BotFather` 대화방에서 `/revoke`를 입력하고 유출된 봇을 선택합니다. 기존 토큰이 즉각 폐기되며 새로운 토큰이 재발급됩니다.
2. **DashCloud 환경변수 갱신**:
DashCloud 대시보드의 **환경 변수** 메뉴로 이동하여 `TELEGRAM_BOT_TOKEN` 값을 새로 발급받은 토큰으로 교체하고 저장합니다.
3. **원클릭 재시작**:
대시보드 우측 상단의 **서비스 재시작** 버튼을 누르면, 1초 만에 새 토큰으로 안전하게 24시간 가동이 재개됩니다.
---
## 다음 추천 문서
### 롱폴링 vs 웹훅 선택 및 전환 가이드
출처: https://dashcloud.kr/docs/telegram/polling-vs-webhook
텔레그램 봇이 사용자의 메시지를 수신하는 방법은 **롱폴링**과 **웹훅** 두 가지가 있습니다.
많은 초보 개발자가 "웹훅이 무조건 최신 기술이고 더 좋다"고 오해하지만, 실제 운영 환경에서는 인프라 구성 난이도, 보안 공격면, 트래픽 규모에 따라 명확한 트레이드오프가 존재합니다. 두 방식의 핵심 차이와 내 프로젝트에 가장 알맞은 선택법을 상세히 분석합니다.
DashCloud는 인바운드 포트와 공인 도메인을 제공하지 않으므로 **텔레그램 웹훅 방식은 사용할 수 없습니다.** 텔레그램 봇은 롱폴링으로 운영하세요. 아래 웹훅 코드는 자체 도메인·서버가 있는 환경을 위한 참고용입니다.
---
## 핵심 메커니즘 비교
```
[ 방식 1: 롱폴링 - 아웃바운드 연결 ]
내 봇 (DashCloud 컨테이너) ─────── 1. "새 메시지 있나요? (대기 30초)" ───────► 텔레그램 서버
내 봇 (DashCloud 컨테이너) ◄────── 2. "네, /start 메시지가 도착했습니다" ─────── 텔레그램 서버
(인바운드 포트 0개 / 방화벽 설정 불필요 / 사내망 완벽 동작)
[ 방식 2: 웹훅 - 인바운드 수신 ]
유저 메시지 발생 ─────────────────────────────────────────────────► 텔레그램 서버
내 봇 (공개 도메인) ◄────── 1. POST https://bot.example.com/webhook ──── 텔레그램 서버
(공인 도메인 필수 / SSL 인증서 필수 / 인바운드 443 포트 개방 필수)
```
---
## 8대 평가 항목별 상세 비교표
| 평가 항목 | 롱폴링 | 웹훅 |
| :--- | :--- | :--- |
| **통신 방향** | 봇 ➔ 텔레그램 서버 (**아웃바운드 단방향**) | 텔레그램 서버 ➔ 내 봇 (**인바운드 수신**) |
| **공인 도메인 / DNS** | **전혀 불필요** (즉시 시작 가능) | 고정된 FQDN 공인 도메인 필수 |
| **SSL/TLS 인증서** | **불필요** | 공인 CA(Let's Encrypt 등) 유효 인증서 필수 |
| **방화벽 및 포트 개방** | **모든 인바운드 포트 차단 가능 (최고 보안)** | 443, 8443, 80, 88 포트 중 하나 개방 필수 |
| **응답 지연** | 실시간 (수십 ms 이내) | 실시간 (즉시 트리거) |
| **로컬 디버깅 편의성** | 내 PC 터미널에서 코드 돌리면 즉시 동작 | ngrok이나 로컬 터널링 도구 연동 필수 |
| **동시 다중 인스턴스** | 단 1개의 인스턴스만 실행 가능 (중복 시 충돌) | 로드밸런서 뒤에 수십 대 서버 수평 확장 가능 |
| **추천 프로젝트** | 개인 봇, 퀀트 알림, 사내 봇, 커뮤니티 관리 (95%) | 초당 수백 건 이상 초대규모 사용자 트래픽 봇 |
---
## 어떤 방식을 선택해야 할까요?
### 롱폴링을 선택해야 하는 경우 (대부분의 경우 강력 권장)
* 도메인을 구매하거나 복잡한 Nginx 리버스 프록시, SSL 인증서 갱신을 신경 쓰고 싶지 않을 때
* DashCloud의 **제로 인바운드 포트** 보안 정책을 활용해 해킹 공격면을 원천 차단하고 싶을 때
* 개발 중 로컬 컴퓨터와 클라우드 서버 환경에서 코드 수정 없이 동일하게 테스트하고 싶을 때
* 일반적인 개인 봇, 알림 봇, 사내 업무 자동화 봇, 퀀트 매매 봇을 운영할 때
### 웹훅을 선택해야 하는 경우
* 초당 수백~수천 건 이상의 동시 메시지가 쏟아져 들어와 여러 대의 컨테이너로 부하를 분산해야 할 때
* 서버리스 아키텍처(AWS Lambda 등)에서 메시지가 올 때만 함수를 깨우는 구조로 운영할 때 (※ DashCloud는 24시간 상시 런타임이므로 롱폴링이 훨씬 경제적이고 안정적입니다)
---
## 프로덕션 구현 코드 비교
### 롱폴링 프로덕션 코드
별도의 웹 프레임워크(FastAPI, Flask) 없이 순수 파이썬 라이브러리만으로 가장 간결하고 견고하게 동작합니다.
```python title="main.py (롱폴링)"
import os
import sys
import logging
from telegram import Update
from telegram.ext import Application, CommandHandler, ContextTypes
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
await update.message.reply_text("롱폴링 모드로 24시간 정상 가동 중입니다!")
def main():
token = os.environ.get("TELEGRAM_BOT_TOKEN")
if not token:
logger.critical("TELEGRAM_BOT_TOKEN이 누락되었습니다.")
sys.exit(1)
app = Application.builder().token(token).build()
app.add_handler(CommandHandler("start", start))
# 롱폴링 가동 팁:
# 1. drop_pending_updates=True: 서버 재시작 시 오프라인 동안 쌓였던 구 메시지 중복 처리 방지
# 2. timeout=30: 텔레그램 서버가 새 메시지를 기다리는 롱폴링 타임아웃(초)
logger.info("텔레그램 롱폴링을 시작합니다...")
app.run_polling(drop_pending_updates=True, timeout=30)
if __name__ == "__main__":
main()
```
### 웹훅 프로덕션 코드 (FastAPI 기반)
공인 도메인과 포트가 준비되었을 때, 텔레그램 서버로부터 HTTP POST 요청을 받아 처리하는 구조입니다.
```python title="webhook_server.py (웹훅)"
import os
import sys
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request, Header, HTTPException
from telegram import Update
from telegram.ext import Application, CommandHandler, ContextTypes
BOT_TOKEN = os.environ.get("TELEGRAM_BOT_TOKEN")
WEBHOOK_URL = os.environ.get("WEBHOOK_URL") # 예: https://bot.example.com/telegram
# 보안 강화를 위한 비밀 토큰 헤더 검증
SECRET_TOKEN = os.environ.get("TELEGRAM_SECRET_TOKEN", "my-super-secret-passphrase")
ptb_app = Application.builder().token(BOT_TOKEN).build()
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
await update.message.reply_text("웹훅 모드로 응답했습니다!")
ptb_app.add_handler(CommandHandler("start", start))
@asynccontextmanager
async def lifespan(app: FastAPI):
# 서버 기동 시 텔레그램 서버에 웹훅 URL 등록
await ptb_app.initialize()
await ptb_app.bot.set_webhook(
url=WEBHOOK_URL,
secret_token=SECRET_TOKEN,
drop_pending_updates=True
)
yield
# 서버 종료 시 웹훅 해제
await ptb_app.bot.delete_webhook()
await ptb_app.shutdown()
fastapi_app = FastAPI(lifespan=lifespan)
@fastapi_app.post("/telegram")
async def telegram_webhook(
request: Request,
x_telegram_bot_api_secret_token: str = Header(None)
):
# 텔레그램 공식 요청인지 시크릿 토큰 대조
if x_telegram_bot_api_secret_token != SECRET_TOKEN:
raise HTTPException(status_code=403, detail="Unauthorized")
data = await request.json()
update = Update.de_json(data, ptb_app.bot)
await ptb_app.process_update(update)
return {"status": "ok"}
```
---
## 롱폴링 ↔ 웹훅 전환 시 주의점: "Conflict" 에러 방지
롱폴링과 웹훅을 번갈아 가며 테스트할 때 가장 많이 마주치는 실수는 **웹훅이 등록되어 있는 상태에서 롱폴링(`run_polling()`)을 실행**하는 것입니다.
이때 텔레그램 서버는 `telegram.error.Conflict: can't use getUpdates method while webhook is active` 에러를 뿜으며 연결을 거부합니다.
### 1분 해결법: 기존 웹훅 강제 삭제
파이썬 한 줄 스크립트나 브라우저 주소창에 아래 URL을 입력하여 기존 웹훅 설정을 깔끔하게 초기화하세요.
```bash
# curl 명령어로 웹훅 삭제
curl -F "drop_pending_updates=True" https://api.telegram.org/bot<내_토큰>/deleteWebhook
```
응답으로 `{"ok":true,"result":true,"description":"Webhook was deleted"}`가 반환되면 즉시 롱폴링 모드로 정상 복귀할 수 있습니다.
---
## 다음 추천 문서
### SQLite로 상태와 대화 기록 영구화하기
출처: https://dashcloud.kr/docs/telegram/sqlite-persistence
텔레그램 봇을 운영할 때 사용자의 대화 기록, 포인트 잔액, 구독 설정, 일일 이용량 카운터를 메모리 변수(`dict`, `list`)에만 보관하면, 서버가 재배포되거나 재시작되는 순간 모든 데이터가 완전히 증발합니다.
외부 클라우드 데이터베이스(AWS RDS, Supabase 등)를 매달 수만 원씩 들여 구축할 필요 없이, DashCloud의 **영구 디스크 볼륨과 내장 SQLite**를 조합하여 비용 0원에 초저지연 로컬 데이터 저장소를 운영하는 방법을 상세히 다룹니다.
---
## 왜 DashCloud + SQLite 조합이 최고일까요?
1. **외부 DB 고정비 0원**: 매달 1~3만 원씩 청구되는 관리형 PostgreSQL/MySQL 인스턴스 비용이 전혀 발생하지 않습니다.
2. **마이크로초(µs) 단위 초고속 쿼리**: 네트워크를 거치지 않고 DashCloud 고성능 NVMe SSD에서 직접 읽고 쓰므로 0.1ms 미만의 초고속 응답을 보장합니다.
3. **영구 볼륨 디렉터리(`/app/data`)**: 새 코드를 커밋하고 재배포해도 물리 디스크에 보존되어 데이터 유실이 원천 차단됩니다.
4. **단일 파일로 손쉬운 백업**: `bot.db` 파일 하나만 복사하면 어디서든 로컬 PC나 다른 환경으로 데이터를 100% 복원할 수 있습니다.
---
## 동시성 잠금 방지: WAL 모드 필수 설정
비동기 텔레그램 봇에서 여러 사용자가 동시에 메시지를 보낼 때 가장 흔하게 발생하는 에러가 `sqlite3.OperationalError: database is locked`입니다.
기본 저널 DELETE 모드는 쓰기 트랜잭션이 진행되는 동안 다른 모든 읽기 쿼리를 차단하기 때문입니다. 이를 해결하기 위해 반드시 **WAL(Write-Ahead Logging) 모드**를 활성화해야 합니다.
```sql title="SQLite 필수 PRAGMA 설정"
-- 1. WAL 모드 활성화 (읽기와 쓰기가 서로를 블로킹하지 않음)
PRAGMA journal_mode = WAL;
-- 2. 동기화 수준 최적화 (성능과 내구성의 최적 균형)
PRAGMA synchronous = NORMAL;
-- 3. 메모리 캐시 크기 증설 (약 64MB 할당)
PRAGMA cache_size = -64000;
-- 4. 외래 키 제약 조건 활성화
PRAGMA foreign_keys = ON;
```
---
## 완성형 비동기 SQLite 매니저 보일러플레이트
파이썬 비동기 프레임워크와 궁합이 가장 뛰어난 `aiosqlite` 라이브러리를 활용한 프로덕션급 데이터베이스 관리자 모듈입니다.
```text title="requirements.txt에 추가"
aiosqlite>=0.20.0
```
```python title="database.py"
import os
import logging
from pathlib import Path
import aiosqlite
logger = logging.getLogger("Database")
# DashCloud 영구 볼륨 디렉터리 경로 설정
BASE_DIR = Path(__file__).resolve().parent
DATA_DIR = BASE_DIR / "data"
DATA_DIR.mkdir(parents=True, exist_ok=True)
DB_PATH = DATA_DIR / "bot_database.db"
class DatabaseManager:
def __init__(self, db_path: Path = DB_PATH):
self.db_path = db_path
async def get_connection(self) -> aiosqlite.Connection:
"""연결 풀 생성 및 WAL 모드 자동 주입"""
conn = await aiosqlite.connect(self.db_path, timeout=30.0)
conn.row_factory = aiosqlite.Row # 딕셔너리처럼 컬럼명으로 접근 가능
await conn.execute("PRAGMA journal_mode = WAL;")
await conn.execute("PRAGMA synchronous = NORMAL;")
return conn
async def init_tables(self):
"""테이블 스키마 생성 및 초기화"""
async with await self.get_connection() as conn:
# 1. 사용자 정보 테이블
await conn.execute("""
CREATE TABLE IF NOT EXISTS users (
user_id INTEGER PRIMARY KEY,
username TEXT,
first_name TEXT,
points INTEGER DEFAULT 100,
is_admin INTEGER DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
last_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
""")
# 2. 대화 기록 테이블 (대화형 AI 챗봇 메모리용)
await conn.execute("""
CREATE TABLE IF NOT EXISTS chat_history (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER,
role TEXT CHECK(role IN ('user', 'assistant')),
content TEXT NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY(user_id) REFERENCES users(user_id) ON DELETE CASCADE
);
""")
await conn.commit()
logger.info("SQLite 테이블 초기화 완료 (WAL 모드 활성화)")
async def upsert_user(self, user_id: int, username: str, first_name: str):
"""유저 접속 시 자동 등록 또는 방문 일시 갱신"""
async with await self.get_connection() as conn:
await conn.execute("""
INSERT INTO users (user_id, username, first_name, last_seen)
VALUES (?, ?, ?, CURRENT_TIMESTAMP)
ON CONFLICT(user_id) DO UPDATE SET
username = excluded.username,
first_name = excluded.first_name,
last_seen = CURRENT_TIMESTAMP;
""", (user_id, username, first_name))
await conn.commit()
async def get_user(self, user_id: int):
"""특정 유저 정보 조회"""
async with await self.get_connection() as conn:
async with conn.execute("SELECT * FROM users WHERE user_id = ?", (user_id,)) as cursor:
return await cursor.fetchone()
# 싱글톤 인스턴스 생성
db = DatabaseManager()
```
---
## 텔레그램 봇 핸들러 연동 예시
생성한 데이터베이스 매니저를 `main.py`의 명령어 핸들러와 결합합니다.
```python title="main.py"
import os
import sys
import logging
from telegram import Update
from telegram.ext import Application, CommandHandler, ContextTypes
from database import db
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
async def start_handler(update: Update, context: ContextTypes.DEFAULT_TYPE):
user = update.effective_user
# 1. DB에 유저 정보 자동 기록 (영구 저장)
await db.upsert_user(
user_id=user.id,
username=user.username or "",
first_name=user.first_name
)
# 2. 유저 정보 조회
user_data = await db.get_user(user.id)
points = user_data["points"]
await update.message.reply_text(
f"반갑습니다 {user.first_name}님!\n"
f"현재 보유 포인트: {points}P\n"
f"회원 정보가 영구 디스크(/app/data)에 안전하게 보존됩니다."
)
async def post_init(application: Application):
"""봇 기동 시 비동기 DB 테이블 초기화"""
await db.init_tables()
def main():
token = os.environ.get("TELEGRAM_BOT_TOKEN")
if not token:
logger.critical("TELEGRAM_BOT_TOKEN이 누락되었습니다.")
sys.exit(1)
app = (
Application.builder()
.token(token)
.post_init(post_init) # 테이블 자동 생성 훅 바인딩
.build()
)
app.add_handler(CommandHandler("start", start_handler))
app.run_polling(drop_pending_updates=True)
if __name__ == "__main__":
main()
```
---
## 무중단 온라인 백업과 웹 파일 다운로드
### 1. 파이썬 스케줄러로 무중단 온라인 백업 (`sqlite3.backup`)
SQLite는 가동 중인 데이터베이스를 서비스 중단 없이 안전하게 다른 파일로 복사하는 `backup()` API를 제공합니다.
```python title="backup_cron.py"
import sqlite3
import shutil
from datetime import datetime
from pathlib import Path
def create_online_backup():
source_db = "/app/data/bot_database.db"
backup_dir = Path("/app/data/backups")
backup_dir.mkdir(parents=True, exist_ok=True)
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
target_file = backup_dir / f"backup_{timestamp}.db"
# 잠금 없이 온라인 상태에서 안전하게 복제
with sqlite3.connect(source_db) as src, sqlite3.connect(target_file) as dst:
src.backup(dst)
print(f"백업 완료: {target_file}")
```
### 2. 대시보드 웹 파일 관리자에서 DB 다운로드
* DashCloud 대시보드의 프로젝트 상세 ➔ **파일** 메뉴로 이동합니다.
* `/app/data/bot_database.db` 파일을 클릭하면, 브라우저에서 즉시 내 PC로 다운로드하여 `DB Browser for SQLite` 같은 무료 도구로 테이블과 데이터를 시각적으로 열람할 수 있습니다.
---
## 다음 추천 문서
### 타임아웃, 예외 처리와 무중단 복구
출처: https://dashcloud.kr/docs/telegram/error-handling
봇을 24시간 상시 운영할 때 프로세스가 중단되는 가장 흔한 원인은 문법 버그가 아니라, **외부 네트워크의 일시적 순단(순시 단절), 텔레그램 API 타임아웃, 사용자가 봇을 차단했을 때 발생하는 예외**를 코드에서 적절히 처리하지 못했기 때문입니다.
처리되지 않은 예외 하나로 봇 전체가 크래시되지 않도록 방어하는 프로덕션급 에러 핸들링 기법을 상세히 알아봅니다.
---
## 텔레그램 봇 5대 핵심 예외 유형과 대응 전략
| 예외 클래스 | 발생 원인 | 기본 동작 결과 | 권장 대응 전략 |
| :--- | :--- | :--- | :--- |
| **`telegram.error.TimedOut`** | 텔레그램 서버 응답 지연 (네트워크 지연) | 즉시 프로세스 크래시 위험 | 일시적 현상이므로 즉시 무시하고 다음 폴링 루프 진행 |
| **`telegram.error.NetworkError`** | 호스트 인터넷 순시 단절, DNS 룩업 지연 | 소켓 강제 단절 | 지수 백오프(1초 ➔ 2초 ➔ 4초) 기반 자동 재접속 |
| **`telegram.error.RetryAfter`** | 429 Too Many Requests 초당 발송 한도 초과 | 거래소/텔레그램 일시 차단 | `e.retry_after` 초만큼 정확히 대기 후 재전송 |
| **`telegram.error.Forbidden`** | 사용자가 봇을 차단했거나 방에서 내보냄 | 전송 실패 크래시 | DB에서 해당 유저의 활성 플래그를 `is_active=0`으로 변경 후 제외 |
| **`telegram.error.BadRequest`** | 수정할 내용이 이전과 동일하거나 텍스트 누락 | 파라미터 에러 | 단순 로깅 후 정상 패스 (치명적 오류 아님) |
---
## 전역 에러 핸들러 구축
`python-telegram-bot` v21에서는 모든 핸들러에서 발생하는 예외를 한곳으로 모아 처리하는 `add_error_handler` 기능을 제공합니다.
어떤 예외가 터져도 봇 프로세스가 죽지 않으며, 에러 트레이스백 전문을 관리자의 개인 텔레그램 대화방으로 즉각 보고합니다.
```python title="error_handlers.py"
import html
import json
import logging
import traceback
import os
from telegram import Update
from telegram.constants import ParseMode
from telegram.ext import ContextTypes
logger = logging.getLogger("ErrorHandler")
ADMIN_CHAT_ID = os.environ.get("TELEGRAM_ADMIN_CHAT_ID")
async def global_error_handler(update: object, context: ContextTypes.DEFAULT_TYPE):
"""모든 처리되지 않은 예외를 가로채어 로깅하고 관리자에게 보고"""
logger.error("요청 처리 중 치명적 예외 발생:", exc_info=context.error)
# 1. 상세 트레이스백 텍스트 추출
tb_list = traceback.format_exception(None, context.error, context.error.__traceback__)
tb_string = "".join(tb_list)
# 2. 업데이트 정보 요약
update_str = update.to_dict() if isinstance(update, Update) else str(update)
# 3. 관리자에게 보낼 긴급 경보 메시지 조립
message = (
"🚨 봇 시스템 예외 경보 발생!\n\n"
f"에러 내용: {html.escape(str(context.error))}\n\n"
f"트레이스백:\n{html.escape(tb_string[-2000:])}"
)
# 관리자 채팅방으로 에러 알림 발송 (관리자 방 발송 자체가 실패해도 봇은 유지)
if ADMIN_CHAT_ID:
try:
await context.bot.send_message(
chat_id=int(ADMIN_CHAT_ID),
text=message,
parse_mode=ParseMode.HTML
)
except Exception as e:
logger.error(f"관리자 알림 전송 실패: {e}")
# 4. 사용자에게 친절한 안내 메시지 전송 (옵션)
if isinstance(update, Update) and update.effective_message:
try:
await update.effective_message.reply_text(
"⚠️ 요청을 처리하는 도중 일시적인 시스템 오류가 발생했습니다. 관리자에게 자동 보고되었습니다."
)
except Exception:
pass
```
---
## 지수 백오프 재연결 데코레이터
외부 API 호출이나 네트워크 전송 시 일시적인 타임아웃에 대처하는 자동 재시도 데코레이터입니다.
```python title="retry.py"
import asyncio
import functools
import logging
from telegram.error import TimedOut, NetworkError, RetryAfter
logger = logging.getLogger("Retry")
def with_retry(max_retries: int = 3, base_delay: float = 1.0):
"""네트워크 타임아웃 발생 시 지수 백오프로 재시도하는 데코레이터"""
def decorator(func):
@functools.wraps(func)
async def wrapper(*args, **kwargs):
retries = 0
while True:
try:
return await func(*args, **kwargs)
except RetryAfter as e:
# 429 속도 제한: 텔레그램이 지시한 초만큼 대기
logger.warning(f"429 속도 제한. {e.retry_after}초 후 재시도합니다.")
await asyncio.sleep(e.retry_after + 0.5)
except (TimedOut, NetworkError) as e:
retries += 1
if retries > max_retries:
logger.error(f"최대 재시도 횟수({max_retries}회) 초과: {e}")
raise
delay = base_delay * (2 ** (retries - 1))
logger.warning(f"네트워크 오류 ({e}). {delay}초 후 재시도 ({retries}/{max_retries})...")
await asyncio.sleep(delay)
return wrapper
return decorator
# 사용 예시:
# @with_retry(max_retries=3)
# async def send_report(bot, chat_id, text):
# await bot.send_message(chat_id=chat_id, text=text)
```
---
## 사용자의 봇 차단(`Forbidden`) 안전 처리
대량 알림을 보낼 때 특정 사용자가 봇을 차단했으면 `telegram.error.Forbidden: Bot was blocked by the user` 에러가 발생합니다.
이를 제대로 예외 처리하지 않으면 전체 발송 루프가 중간에 멈추어 나머지 사용자에게 알림이 가지 않는 사고가 발생합니다.
```python title="safe_sender.py"
import logging
from telegram.error import Forbidden, BadRequest
logger = logging.getLogger("SafeSender")
async def send_notification_to_user(bot, user_id: int, text: str, db):
try:
await bot.send_message(chat_id=user_id, text=text)
return True
except Forbidden:
# 사용자가 봇을 차단함: DB에서 알림 수신 거부 처리하여 다음 루프부터 제외
logger.info(f"유저 [{user_id}]가 봇을 차단했습니다. 수신 비활성화 처리합니다.")
await db.execute("UPDATE users SET is_active = 0 WHERE user_id = ?", (user_id,))
return False
except BadRequest as e:
if "Chat not found" in str(e):
logger.warning(f"유저 [{user_id}] 대화방이 존재하지 않습니다.")
await db.execute("UPDATE users SET is_active = 0 WHERE user_id = ?", (user_id,))
else:
logger.error(f"유저 [{user_id}] 전송 실패: BadRequest - {e}")
return False
```
---
## 텔레그램 봇 완성형 메인 진입점 연동
```python title="main.py"
import os
import logging
from telegram.ext import Application, CommandHandler
from error_handlers import global_error_handler
logging.basicConfig(level=logging.INFO)
async def test_error(update, context):
# 의도적인 0 나누기 에러 테스트
result = 1 / 0
def main():
token = os.environ.get("TELEGRAM_BOT_TOKEN")
app = Application.builder().token(token).build()
# 핸들러 및 전역 에러 핸들러 바인딩
app.add_handler(CommandHandler("crash_test", test_error))
app.add_error_handler(global_error_handler)
app.run_polling(drop_pending_updates=True)
if __name__ == "__main__":
main()
```
---
## 다음 추천 문서
### 대량 알림 발송 봇과 비동기 속도 조율 큐
출처: https://dashcloud.kr/docs/telegram/broadcast-and-queue
쇼핑몰 특가 알림, 급등주 시세 알림, 커뮤니티 공지 등 수천~수만 명의 사용자에게 동시에 메시지를 뿌려야 하는 대량 발송 브로드캐스트 봇을 운영할 때 가장 큰 기술적 장벽은 **텔레그램의 엄격한 API 호출 호출 속도 제한 정책**입니다.
단순 `for` 루프로 수천 명에게 `send_message()`를 무작정 발송하면 불과 1초 만에 **429 Too Many Requests (Flood Control Exceeded)** 에러가 터지고 봇 토큰이 수십 분간 정지당합니다.
텔레그램 정책 한도를 완벽히 준수하면서 누락 없이 대량 메시지를 순차 처리하는 비동기 메시지 버퍼 큐 아키텍처를 구현해 봅니다.
---
## 텔레그램 공식 메시지 발송 한도 규정
텔레그램 Bot API 공식 문서에 명시된 3대 제약 사항입니다:
1. **전역 발송 속도 상한**: 봇 하나당 전체 사용자를 대상으로 **초당 최대 30건(30 msg/sec)**까지 발송 가능합니다.
2. **동일 그룹/채널 발송 한도**: 같은 그룹 채팅방이나 채널에는 **분당 최대 20건(1건당 3초 간격)**을 초과할 수 없습니다.
3. **개인 유저 발송 한도**: 1명의 개인 유저에게는 **초당 1건** 이상 연속 발송을 지양해야 합니다.
공식 한도는 초당 30건이지만, 네트워크 핑 지연과 패킷 버스팅으로 인해 일시적으로 30건을 초과하는 경우가 발생합니다. 프로덕션 환경에서는 발송 속도를 **초당 25건(건당 0.04초 딜레이)**으로 제한하여 100% 안전 마진을 확보하는 것이 정석입니다.
---
## 비동기 생산자-소비자 큐 아키텍처
알림을 생성하는 비즈니스 로직과 실제 텔레그램 서버로 메시지를 쏘는 전송 엔진을 완전히 분리합니다.
```
[ 알림 생산자 ]
├─ 크롤러: "특가 상품 발견!" ──┐
├─ 퀀트: "비트코인 5% 급락!" ──┼──► [ asyncio.PriorityQueue (인메모리 버퍼) ]
└─ 관리자: "전체 서버 공지" ───┘ │
▼
[ 전송 소비자 워커 ]
├─ 초당 25건 엄격한 속도 제어
├─ 우선순위 높은 메시지 우선 인출
├─ 429 감지 시 즉시 슬로우다운
└─ 차단 유저 DB 자동 필터링
│
▼ (HTTPS Outbound)
[ 텔레그램 공식 Bot API ]
```
1. **생산자**: 가격 변동이나 새 이벤트가 발생하면 즉시 큐(`queue.put_nowait()`)에 메시지를 밀어 넣고 다음 작업으로 즉시 복귀합니다 (블로킹 0초).
2. **소비자 워커**: 백그라운드 태스크로 상시 실행되며, 큐에서 메시지를 하나씩 꺼내 초당 25건 속도로 조율하며 텔레그램으로 전송합니다.
---
## 완성형 프로덕션 메시지 큐 모듈 (`broadcast_queue.py`)
우선순위 큐와 자동 재시도, 429 동적 대기 로직이 모두 포함된 완성형 파이썬 모듈입니다.
```python title="broadcast_queue.py"
import asyncio
import logging
from dataclasses import dataclass, field
from typing import Any
from telegram import Bot
from telegram.error import RetryAfter, Forbidden, BadRequest, NetworkError
logger = logging.getLogger("BroadcastQueue")
@dataclass(order=True)
class QueuedMessage:
priority: int # 1: 긴급(비상 알림), 2: 중요(체결), 3: 일반(마케팅)
chat_id: int = field(compare=False)
text: str = field(compare=False)
retries: int = field(default=0, compare=False)
class TelegramBroadcastQueue:
def __init__(self, bot: Bot, rate_limit: int = 25):
"""
:param bot: 텔레그램 Bot 인스턴스
:param rate_limit: 초당 최대 발송 건수 (기본 25건 권장)
"""
self.bot = bot
self.delay = 1.0 / rate_limit # 25건 기준 건당 0.04초
self.queue = asyncio.PriorityQueue()
self.is_running = False
self.worker_task = None
def push(self, chat_id: int, text: str, priority: int = 3):
"""메시지를 큐에 비동기 적재 (생산자)"""
msg = QueuedMessage(priority=priority, chat_id=chat_id, text=text)
self.queue.put_nowait(msg)
async def start(self):
"""백그라운드 전송 워커 기동"""
if not self.is_running:
self.is_running = True
self.worker_task = asyncio.create_task(self._worker())
logger.info(f"메시지 브로드캐스트 큐 워커 기동 완료 (초당 {int(1/self.delay)}건 제어)")
async def stop(self):
"""남은 메시지 전송 후 안전 종료"""
self.is_running = False
if self.worker_task:
self.worker_task.cancel()
try:
await self.worker_task
except asyncio.CancelledError:
pass
logger.info("메시지 브로드캐스트 큐 워커가 안전하게 중지되었습니다.")
async def _worker(self):
"""큐에서 메시지를 인출해 속도를 조율하며 전송하는 소비자"""
while self.is_running:
msg: QueuedMessage = await self.queue.get()
try:
await self.bot.send_message(chat_id=msg.chat_id, text=msg.text)
await asyncio.sleep(self.delay) # 0.04초 지연으로 초당 25건 조율
except RetryAfter as e:
# 텔레그램 서버가 지시한 초만큼 대기 후 큐에 재인입
logger.warning(f"429 속도 제한 격발! {e.retry_after}초 동안 큐를 일시 정지합니다.")
await asyncio.sleep(e.retry_after + 1.0)
await self.queue.put(msg)
except Forbidden:
# 유저가 봇을 차단함 (재시도 없이 폐기)
logger.info(f"유저 [{msg.chat_id}]가 봇을 차단하여 메시지를 폐기합니다.")
except (NetworkError, BadRequest) as e:
if msg.retries < 3:
msg.retries += 1
logger.warning(f"전송 실패({e}). 재시도 카운트({msg.retries}/3) 증가 후 재인입")
await asyncio.sleep(1.0)
await self.queue.put(msg)
else:
logger.error(f"최종 전송 실패 유저 [{msg.chat_id}]: {e}")
finally:
self.queue.task_done()
```
---
## 텔레그램 봇 메인 코드 연동 예시
```python title="main.py"
import os
import asyncio
import logging
from telegram import Update
from telegram.ext import Application, CommandHandler, ContextTypes
from broadcast_queue import TelegramBroadcastQueue
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# 전역 큐 인스턴스 홀더
broadcast_queue = None
async def broadcast_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""관리자가 전체 유저에게 공지를 보낼 때 호출하는 핸들러"""
admin_id = int(os.environ.get("TELEGRAM_ADMIN_CHAT_ID", 0))
if update.effective_user.id != admin_id:
await update.message.reply_text("⛔ 관리자 전용 명령어입니다.")
return
# 가상 사용자 1,000명 리스트 (실제로는 SQLite에서 조회)
user_list = [10000000 + i for i in range(1000)]
notice_text = "📢 [긴급 공지] DashCloud 인프라 정기 점검이 완료되었습니다."
await update.message.reply_text(f"{len(user_list)}명의 사용자에게 순차 발송을 시작합니다...")
# 1초 만에 1,000개 메시지를 큐에 등록 (블로킹 없음)
for uid in user_list:
broadcast_queue.push(chat_id=uid, text=notice_text, priority=2)
await update.message.reply_text("큐 적재 완료! 백그라운드 워커가 초당 25건으로 안전하게 전송합니다.")
async def post_init(application: Application):
"""앱 기동 시 큐 워커 실행"""
global broadcast_queue
broadcast_queue = TelegramBroadcastQueue(bot=application.bot, rate_limit=25)
await broadcast_queue.start()
async def post_shutdown(application: Application):
"""앱 종료 시 큐 워커 정상 중지"""
if broadcast_queue:
await broadcast_queue.stop()
def main():
token = os.environ.get("TELEGRAM_BOT_TOKEN")
app = (
Application.builder()
.token(token)
.post_init(post_init)
.post_shutdown(post_shutdown)
.build()
)
app.add_handler(CommandHandler("broadcast", broadcast_command))
app.run_polling(drop_pending_updates=True)
if __name__ == "__main__":
main()
```
---
## 1만 명 이상 대규모 발송 시 성능 팁
1. **DB 조회 시 청크 분할**: `SELECT user_id FROM users`로 1만 개 레코드를 한 번에 가져오지 말고, `yield` 제너레이터나 `LIMIT 500 OFFSET ...`으로 500개씩 청크 단위로 인출하여 큐에 적재하세요 (RAM 절약).
2. **발송 성공 여부 비동기 로깅**: 전송이 완료된 메시지는 SQLite DB에 즉시 매 건 UPDATE 치지 말고, 100건 단위로 묶어서 배치 커밋하여 디스크 I/O를 최적화하세요.
3. **DashCloud CPU 메트릭 관찰**: 대시보드의 **메트릭** 탭에서 대량 발송 중 CPU 사용량이 30% 미만으로 평온하게 유지되는지 모니터링할 수 있습니다.
---
## 다음 추천 문서
## 크롤러 / 스크래퍼
### 크롤러 / 스크래퍼 아키텍처 및 도구 선택 가이드
출처: https://dashcloud.kr/docs/crawler/overview
최신 웹 환경(React, Vue, Next.js 기반의 싱글 페이지 애플리케이션)에서 데이터를 주기적으로 수집하는 작업은 과거의 단순 HTML 파싱보다 훨씬 복잡한 인프라 요구조건을 수반합니다.
자바스크립트 동적 실행, 무한 스크롤, WAF 웹 방화벽의 봇 식별 장치, 그리고 브라우저를 장시간 켜두었을 때 발생하는 **Chromium 메모리 누수**를 극복하지 못하면, 크롤러는 며칠을 버티지 못하고 크래시됩니다.
DashCloud가 제공하는 크롤러 전용 런타임의 특징과, 대상 사이트의 특성에 맞춰 최적의 라이브러리를 선택하는 가이드를 상세히 다룹니다.
---
## 데이터 수집 도구별 기술 스택 매트릭스
모든 수집 작업에 무거운 헤드리스 브라우저를 띄울 필요는 없습니다. 대상 웹사이트의 데이터 렌더링 방식에 따라 적절한 도구를 조합하는 것이 서버 자원과 수집 속도를 극대화하는 비결입니다.
| 프레임워크 도구 | 렌더링 엔진 | 1회 수집 속도 | 메모리 점유율 | 자바스크립트 실행 | 추천 수집 대상 |
| :--- | :--- | :--- | :--- | :--- | :--- |
| **`Playwright`** | Chromium (헤드리스) | 보통 (0.8~2.5초) | 250~450MB | **완벽 실행 (동적 SPA)** | 쇼핑몰 옵션가, 로그인 필요 사이트, 무한 스크롤 |
| **`curl_cffi`** | 없음 (L7 C-클라이언트) | **초고속 (0.05~0.2초)** | 20~40MB | 불가 (원시 HTML/JSON) | 일반 REST API, 차단 강도가 높은 정적 쇼핑몰 |
| **`httpx` / `requests`** | 없음 (순수 파이썬) | **초고속 (0.05~0.2초)** | 15~30MB | 불가 (원시 HTML/JSON) | 공공 데이터 포털, 단순 뉴스/게시판, 피드 |
| **`Selenium`** | 웹드라이버 기반 | 느림 (1.5~4.0초) | 400~700MB | 실행 가능 | 레거시 파이프라인 유지보수 (신규 권장 안 함) |
신규 프로젝트라면 무조건 **Playwright**를 권장합니다. Selenium에 비해 실행 속도가 2~3배 빠르고, 네트워크 패킷 가로채기와 이미지/폰트 차단, 로그인 세션 저장이 파이썬 네이티브 비동기(`async/await`)로 훨씬 우아하게 지원됩니다.
---
## DashCloud `Chromium + Python` 전용 런타임의 강점
DashCloud는 크롤러 개발자가 리눅스 가상머신을 세팅할 때 겪는 번거로운 종속성 문제를 해결한 `crawler-chromium` 런타임 프리셋을 기본 지원합니다.
1. **Chromium 브라우저 바이너리 기본 내장**:
일반 도커나 VPS에서 `playwright install`을 실행할 때 수백 메가바이트의 브라우저를 다운로드하느라 빌드가 5분씩 걸리던 문제를 없앴습니다. 런타임 이미지에 최신 Chromium이 사전 빌드되어 10초 만에 즉시 배포됩니다.
2. **리눅스 필수 C 라이브러리 완비**:
우분투 환경에서 Chromium 구동에 필요한 20여 종의 핵심 공유 라이브러리(`libx11-xcb`, `libasound2`, `libnss3` 등)가 시스템 레벨에 완벽히 설치되어 있어 의존성 오류가 발생하지 않습니다.
3. **Noto Sans CJK 한글 폰트 기본 내장**:
웹페이지 스크린샷이나 PDF 캡처 시 한글 텍스트가 네모 박스(`□□□`, 뛟뛟)로 깨지는 고질적인 리눅스 폰트 누락 문제를 원천 차단했습니다.
4. **좀비 프로세스 자동 청소**:
크롤러 스크립트가 비정상 종료되어도 백그라운드에 고립된 자식 Chromium 프로세스를 호스트 데몬이 감지하여 메모리를 즉시 회수합니다.
---
## 완성형 스타터 보일러플레이트 코드
```python title="crawler_playwright.py"
import os
import sys
import logging
from pathlib import Path
from playwright.sync_api import sync_playwright
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("PlaywrightCrawler")
# 영구 저장소 디렉터리
OUTPUT_DIR = Path("/app/data/screenshots")
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
def crawl_dynamic_page(url: str):
logger.info(f"Playwright 브라우저 가동 시작: {url}")
with sync_playwright() as p:
# 1. 헤드리스 Chromium 런칭
browser = p.chromium.launch(
headless=True,
args=[
"--no-sandbox",
"--disable-dev-shm-usage", # 메모리 부족 방지
"--disable-gpu"
]
)
# 2. 독립된 브라우저 컨텍스트 생성 (캐시/쿠키 격리)
context = browser.new_context(
viewport={"width": 1920, "height": 1080},
user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36"
)
page = context.new_page()
try:
# 3. 네트워크 레벨 이미지/폰트 차단 (메모리와 대역폭 80% 절감)
# 스크린샷이 필요한 경우 route 차단을 해제하세요
# page.route("**/*", lambda route: route.abort() if route.request.resource_type in ["image", "media"] else route.continue_())
# 4. 페이지 이동 (DOM 렌더링 대기)
page.goto(url, wait_until="domcontentloaded", timeout=30000)
# 특정 데이터 셀렉터가 뜰 때까지 대기
page.wait_for_selector("body", timeout=10000)
# 페이지 타이틀 및 본문 텍스트 추출
title = page.title()
logger.info(f"수집 성공: 페이지 제목 = {title}")
# 한글 스크린샷 캡처 (Noto Sans CJK 폰트 적용)
screenshot_path = OUTPUT_DIR / "sample_capture.png"
page.screenshot(path=str(screenshot_path), full_page=False)
logger.info(f"스크린샷 저장 완료: {screenshot_path}")
except Exception as e:
logger.error(f"수집 실패: {e}")
raise
finally:
# 5. [필수] 컨텍스트 및 브라우저 명시적 해제 (메모리 누수 방지)
context.close()
browser.close()
logger.info("Chromium 프로세스가 안전하게 종료되었습니다.")
if __name__ == "__main__":
crawl_dynamic_page("https://news.naver.com")
```
```python title="crawler_httpx.py"
import asyncio
import logging
import httpx
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("HttpxCrawler")
HEADERS = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
"Accept": "application/json, text/plain, */*",
"Accept-Language": "ko-KR,ko;q=0.9,en-US;q=0.8",
}
async def fetch_product_api(client: httpx.AsyncClient, product_id: int):
url = f"https://api.example.com/products/{product_id}"
try:
response = await client.get(url, timeout=10.0)
if response.status_code == 200:
data = response.json()
logger.info(f"상품 [{product_id}] 수집 성공: {data.get('name')} - {data.get('price')}원")
return data
else:
logger.warning(f"상품 [{product_id}] 비정상 응답: {response.status_code}")
except httpx.RequestError as e:
logger.error(f"통신 에러 발생 [{product_id}]: {e}")
return None
async def main():
async with httpx.AsyncClient(headers=HEADERS, http2=True) as client:
tasks = [fetch_product_api(client, pid) for pid in range(1001, 1010)]
results = await asyncio.gather(*tasks)
logger.info(f"총 {len([r for r in results if r])}개 상품 수집 완료.")
if __name__ == "__main__":
asyncio.run(main())
```
---
## 장기 무중단 크롤러 설계의 4대 원칙
1. **컨텍스트 수명주기 격리**: 단일 브라우저 인스턴스를 유지하더라도, 루프를 돌 때마다 `context = browser.new_context()`를 새로 생성하고 `finally: context.close()`로 닫아야 DOM 가비지가 디스크 힙에 쌓이지 않습니다.
2. **불필요한 리소스 차단**: CSS, 이미지, 폰트, 구글 애널리틱스 스크립트 전송을 가로채서 중단시키면 대역폭과 CPU 연산량이 70% 이상 절감됩니다.
3. **가변 주기 스케줄링**: 정확히 5.0초마다 요청을 보내는 기계적 패턴은 웹 방화벽의 봇 탐지기에 즉시 포착됩니다. 불규칙한 지연(랜덤 지터)을 두어 탐지를 회피하세요.
4. **차분 감지 기반 저장**: 모든 페이지 내용을 매번 디스크에 쓰지 말고, 직전 수집 결과와 가격이나 재고 상태가 달라졌을 때만 SQLite에 기록하여 스토리지 수명을 보존하세요.
---
## 관련 블로그
- [파이썬 웹 크롤러를 24시간 무료로 자동 실행하는 방법](/blog/crawler/python-scraper-cloud-automation) — PC를 꺼도 매일 정기 수집
- [크롤러 블로그 전체 보기](/blog/crawler)
---
## 다음 추천 문서
### 헤드리스 Playwright 메모리 최적화 및 누수 방지
출처: https://dashcloud.kr/docs/crawler/headless-chromium
Playwright를 사용해 1회성 스크립트를 실행할 때는 메모리 문제를 체감하기 어렵습니다. 하지만 수백 개 쇼핑몰 상품 페이지를 매 5분마다 24시간 내내 순회하는 크롤러를 운영하다 보면, 처음에는 150MB 수준이던 메모리가 서서히 우상향하여 반나절 만에 1GB를 돌파하고 결국 **메모리 초과로 프로세스가 강제 종료되는 현상**을 마주하게 됩니다.
Chromium 브라우저 엔진의 내부 캐시 축적 원리를 이해하고, 24시간 가동해도 메모리 사용량이 일정하게 유지되도록 통제하는 4대 엔지니어링 기법을 상세히 다룹니다.
---
## Chromium이 장시간 실행 시 메모리를 누수시키는 원인
1. **V8 자바스크립트 엔진의 분리된 DOM 노드 보관**:
페이지 이동 시 이전 페이지의 복잡한 자바스크립트 이벤트 리스너와 DOM 트리 잔여물이 가비지 컬렉션되지 않고 힙에 남아 누적됩니다.
2. **내부 이미지 디코딩 및 렌더링 캐시**:
브라우저는 빠른 재렌더링을 위해 내려받은 그래픽 리소스와 폰트 글리프를 내부 메모리에 캐싱하며, 이 캐시는 일반적인 방법으로 비워지지 않습니다.
3. **GPU 프로세스 및 IPC 통신 파편화**:
헤드리스 모드라 하더라도 내부적으로 소프트웨어 GL 렌더러가 작동하며 프로세스 간 통신 파이프에 버퍼가 쌓입니다.
---
## 최적화 기법 1: 브라우저 재사용 + 컨텍스트 수집 주기별 격리
브라우저 자체(`browser = launch()`)를 띄우는 작업은 1~2초의 CPU 오버헤드가 발생하므로 매번 띄우기엔 부담스럽습니다.
**가장 이상적인 프로덕션 패턴은 브라우저는 단 1회 띄워두고, 매 수집 루프마다 새로운 컨텍스트(`browser.new_context()`)를 생성한 뒤 작업 직후 `finally` 블록에서 즉시 닫는 것**입니다.
```python title="context_isolation.py"
from playwright.sync_api import sync_playwright
with sync_playwright() as p:
# 1. 브라우저 인스턴스는 1개만 실행
browser = p.chromium.launch(
headless=True,
args=["--no-sandbox", "--disable-dev-shm-usage"]
)
while True:
# 2. [핵심] 수집 작업마다 독립된 가상 브라우저 컨텍스트 생성
context = browser.new_context()
page = context.new_page()
try:
page.goto("https://target-site.com", timeout=30000)
data = extract_data(page)
save_to_database(data)
except Exception as e:
logger.error(f"수집 실패: {e}")
finally:
# 3. [핵심] 컨텍스트를 닫는 순간 해당 세션의 모든 DOM 힙과 캐시가 100% 즉시 증발
context.close()
```
`page = context.new_page()`를 루프 바깥에 하나만 만들어두고 `page.goto()`를 수백 번 반복하는 코드는 최악의 안티패턴입니다. 수집 페이지가 바뀔 때마다 메모리가 수십 MB씩 영구 누적되어 몇 시간 안에 서버가 다운됩니다.
---
## 최적화 기법 2: 네트워크 레벨 리소스 차단 (대역폭 80% 절감)
크롤러가 필요로 하는 것은 웹페이지의 텍스트 데이터(가격, 제목, 재고 수량)이지 화려한 배너 이미지나 동영상, 광고 트래커가 아닙니다.
Playwright의 `page.route()` 기능을 활용해 불필요한 리소스 요청을 네트워크 진입 단계에서 강제로 차단(`route.abort()`)합니다.
```python title="resource_abort.py"
def block_unnecessary_resources(page):
"""이미지, 미디어, 웹폰트 다운로드를 네트워크 단에서 즉각 거절"""
def route_filter(route):
# 차단할 리소스 타입 정의
if route.request.resource_type in ["image", "media", "font"]:
route.abort()
# 구글 애널리틱스 등 불필요한 외부 추적기 차단
elif any(domain in route.request.url for domain in ["google-analytics.com", "doubleclick.net", "facebook.net"]):
route.abort()
else:
route.continue_()
page.route("**/*", route_filter)
```
* **메모리 절감 효과**: 이미지 렌더링을 위한 버퍼 메모리가 생성되지 않아 1개 페이지당 메모리 점유율이 60~70% 감소합니다.
* **수집 속도 향상**: 대용량 이미지 다운로드 대기 시간이 사라져 `page.goto()` 완료 시간이 2.5초에서 0.4초로 단축됩니다.
---
## 최적화 기법 3: 프로세스 주기적 자동 재활용
Chromium 자체의 C++ 엔진 레벨 메모리 파편화는 파이썬 코드 레벨에서 100% 완벽히 제거하기 어렵습니다.
실전 프로덕션 크롤러에서 가장 안정적인 설계는 **수집 작업을 500회 완료했거나 가동 후 12시간이 경과하면 프로세스가 스스로 정상 종료(`os._exit(0)`)하고, DashCloud의 무중단 데몬 감시자가 즉시 깨끗한 새 프로세스로 재기동하도록 맡기는 것**입니다.
```python title="auto_recycle_worker.py"
import os
import time
import psutil
import logging
logger = logging.getLogger("RecycleWorker")
MAX_TASKS_PER_PROCESS = 500 # 500회 작업 후 자율 재시작
MAX_UPTIME_SECONDS = 12 * 3600 # 12시간 가동 후 자율 재시작
MAX_MEMORY_MB = 400 # 메모리가 400MB를 넘으면 자율 재시작
def get_current_memory_mb():
"""현재 프로세스의 실제 물리 메모리 RSS 점유량 측정"""
process = psutil.Process(os.getpid())
return process.memory_info().rss / (1024 * 1024)
def check_lifecycle_and_recycle(task_count, start_time):
current_mem = get_current_memory_mb()
uptime = time.time() - start_time
if task_count >= MAX_TASKS_PER_PROCESS or uptime >= MAX_UPTIME_SECONDS or current_mem >= MAX_MEMORY_MB:
logger.info(
f"자율 재활용 임계치 도달 (작업: {task_count}회, 가동: {int(uptime)}초, 메모리: {current_mem:.1f}MB). "
"프로세스를 정상 종료하여 메모리를 100% 초기화합니다."
)
# DashCloud 런타임이 프로세스 종료를 감지하면 깨끗한 컨테이너로 자동 재기동함
os._exit(0)
```
---
## 최적화 기법 4: 완성형 무중단 크롤러 보일러플레이트
위의 3가지 최적화 기법이 모두 결합된 프로덕션급 파이썬 크롤러 템플릿입니다.
```python title="production_crawler.py"
import os
import time
import logging
from playwright.sync_api import sync_playwright
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("CrawlerEngine")
def run_crawler_pipeline():
start_time = time.time()
task_count = 0
with sync_playwright() as p:
logger.info("Chromium 엔진 기동...")
browser = p.chromium.launch(
headless=True,
args=[
"--no-sandbox",
"--disable-dev-shm-usage",
"--disable-gpu",
"--disable-extensions",
]
)
while True:
# 1. 독립 컨텍스트 생성
context = browser.new_context(viewport={"width": 1280, "height": 720})
page = context.new_page()
# 2. 이미지/미디어 네트워크 차단
page.route(
"**/*",
lambda route: route.abort()
if route.request.resource_type in ["image", "media", "font"]
else route.continue_()
)
try:
# 3. 데이터 수집 실행
page.goto("https://news.ycombinator.com", timeout=20000, wait_until="domcontentloaded")
title = page.title()
task_count += 1
logger.info(f"[{task_count}회차] 수집 성공: {title}")
except Exception as e:
logger.error(f"수집 예외 발생: {e}")
finally:
# 4. 컨텍스트 및 페이지 메모리 즉시 회수
context.close()
# 5. 자율 재시작 체크 (12시간 경과 또는 500회 수집)
if task_count >= 500 or (time.time() - start_time) > 43200:
logger.info("메모리 완전 초기화를 위해 프로세스를 정상 종료하고 재기동합니다.")
browser.close()
os._exit(0)
# 다음 수집 주기까지 대기 (예: 30초)
time.sleep(30)
if __name__ == "__main__":
run_crawler_pipeline()
```
---
## 다음 추천 문서
### 로그인 세션 및 쿠키 영구화
출처: https://dashcloud.kr/docs/crawler/session-persistence
회원 전용 도매몰 가격, 사내 인트라넷 게시판, 또는 SMS OTP 등 2단계 인증이 걸려 있는 웹사이트의 데이터를 수집할 때 매번 스크립트에서 아이디와 비밀번호를 입력하고 로그인하는 것은 매우 비효율적이며 보안상으로도 위험합니다.
Playwright의 **storage_state 세션 상태 보존** 기능을 활용하면, 1회 인증으로 획득한 세션 쿠키와 localStorage 토큰 토큰을 DashCloud의 영구 디스크에 저장해 두고, 서버가 재시작되어도 **로그인 상태를 유지하며 즉각 수집을 재개**할 수 있습니다.
---
## 세션 영구화의 핵심 동작 메커니즘
```
[ 1회성 초기 인증 (내 로컬 PC) ]
개발자가 브라우저를 열고 로그인 (SMS/OTP 직접 입력)
│
▼
context.storage_state(path="session.json") 호출
│ (쿠키 + localStorage 토큰이 JSON 파일로 추출됨)
▼
대시보드 또는 CLI로 DashCloud 영구 볼륨(/app/data/session.json)에 업로드
─────────────────────────────────────────────────────────────
[ 24/7 무중단 수집 (DashCloud 런타임) ]
browser.new_context(storage_state="/app/data/session.json")
│ (로그인 페이지를 거치지 않고 인증 완료 상태로 즉시 브라우저 로드)
▼
회원 전용 마이페이지 또는 특가 페이지로 다이렉트 이동
```
---
## 1단계: 로컬에서 1회성 세션 파일 생성하기
내 로컬 컴퓨터에서 브라우저 창(`headless=False`)을 띄우고 사람이 직접 로그인을 완료한 뒤 세션 파일을 추출하는 1회용 파이썬 스크립트입니다.
```python title="generate_session.py (로컬 실행용)"
from pathlib import Path
from playwright.sync_api import sync_playwright
SESSION_FILE = Path("session.json")
def create_initial_session():
with sync_playwright() as p:
# 사람이 직접 로그인할 수 있도록 화면에 브라우저 표시
browser = p.chromium.launch(headless=False)
context = browser.new_context()
page = context.new_page()
print("로그인 페이지로 이동합니다. 브라우저에서 직접 로그인을 완료하세요...")
page.goto("https://target-service.com/login")
# 로그인 완료 후 이동하는 메인 페이지 URL 또는 특정 요소가 뜰 때까지 대기
# 예: 마이페이지 링크나 로그아웃 버튼이 보일 때까지 대기 (최대 120초)
page.wait_for_selector("a[href='/logout']", timeout=120000)
print("로그인이 감지되었습니다! 세션 상태를 저장합니다.")
# [핵심] 쿠키와 로컬스토리지 전체를 session.json으로 덤프
context.storage_state(path=str(SESSION_FILE))
print(f"세션 파일 저장 완료: {SESSION_FILE.resolve()}")
context.close()
browser.close()
if __name__ == "__main__":
create_initial_session()
```
생성된 `session.json` 파일을 DashCloud 프로젝트의 `/app/data/session.json` 위치에 업로드합니다 (웹 대시보드의 **파일** 메뉴에서 드래그앤드롭 가능).
---
## 2단계: DashCloud에서 세션 주입하여 24시간 수집하기
저장된 `session.json` 파일이 존재할 경우 자동으로 브라우저 컨텍스트에 주입하여 로그인 과정을 건너뛰고 바로 타깃 페이지를 수집하는 프로덕션 코드입니다.
```python title="authenticated_crawler.py"
import os
import logging
from pathlib import Path
from playwright.sync_api import sync_playwright
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("AuthCrawler")
SESSION_PATH = Path("/app/data/session.json")
def run_authenticated_crawler():
if not SESSION_PATH.exists():
logger.error(f"세션 파일({SESSION_PATH})이 존재하지 않습니다! 먼저 세션을 업로드하세요.")
return
with sync_playwright() as p:
browser = p.chromium.launch(headless=True)
# [핵심] 저장된 세션 상태를 주입하여 컨텍스트 생성
context = browser.new_context(
storage_state=str(SESSION_PATH),
viewport={"width": 1920, "height": 1080}
)
page = context.new_page()
try:
logger.info("회원 전용 페이지로 직접 이동합니다...")
page.goto("https://target-service.com/wholesale/products", timeout=30000)
# 1. 세션 만료 여부 검증 (로그아웃 버튼 존재 확인)
if not page.is_visible("a[href='/logout']"):
logger.warning("⚠️ 세션이 만료되어 비로그인 상태로 리다이렉트되었습니다!")
handle_session_expired()
return
# 2. 회원 전용 데이터 수집 수행
items = page.query_selector_all(".wholesale-item")
logger.info(f"성공: {len(items)}개 도매 상품 정보를 수집했습니다.")
except Exception as e:
logger.error(f"크롤링 도중 예외 발생: {e}")
finally:
context.close()
browser.close()
def handle_session_expired():
# 텔레그램이나 디스코드로 관리자에게 세션 만료 알림 전송
logger.critical("관리자 텔레그램으로 재로그인 요청 알림을 발송합니다.")
if __name__ == "__main__":
run_authenticated_crawler()
```
---
## 3단계: 세션 자동 갱신 파이프라인
일반적인 웹사이트는 마지막 접속 후 7일~30일이 경과하면 보안 정책상 세션을 강제 만료시킵니다.
세션 만료를 방지하고 최신 쿠키로 유지하려면, **주기적으로 메인 페이지를 방문하고 갱신된 쿠키를 다시 `storage_state`로 덮어쓰는 Keep-Alive 루프**를 구성해야 합니다.
```python title="keep_alive.py"
def refresh_and_save_session(page, context, session_path: Path):
"""주기적으로 페이지를 방문하여 갱신된 세션 토큰을 영구 디스크에 저장"""
try:
page.goto("https://target-service.com/my-account", timeout=20000)
# 최신 세션 상태를 다시 디스크에 덮어쓰기
context.storage_state(path=str(session_path))
logger.info("세션 상태가 성공적으로 갱신되었습니다.")
except Exception as e:
logger.warning(f"세션 갱신 실패: {e}")
```
---
## 보안 주의사항
* `session.json` 파일에는 내 계정의 실제 로그인 쿠키가 포함되어 있으므로 **절대 깃허브 공개 저장소에 커밋하지 마세요.**
* 반드시 프로젝트의 `.gitignore` 파일에 `data/` 및 `*.json`을 추가하세요.
* DashCloud는 `/app/data` 디렉터리를 외부와 격리된 전용 물리 볼륨에 안전하게 저장하므로 안전성이 보장됩니다.
---
## 다음 추천 문서
### 403 차단 우회와 웹 방화벽 봇 탐지 회피 전략
출처: https://dashcloud.kr/docs/crawler/anti-blocking
파이썬의 `requests`나 기본 `Playwright`로 데이터 수집을 시도했을 때 응답 코드 **403 Forbidden**이나 Cloudflare의 보안 확인(Turnstile/대기 화면)이 나타나는 이유는 단순히 `User-Agent` 문자열 때문이 아닙니다.
최신 웹 보안 엔진은 브라우저의 내부 자바스크립트 변수(`navigator.webdriver`), 그래픽 렌더러 지문, 그리고 요청 간격의 기계적 규칙성까지 다층적으로 검증합니다. 실전 수집 환경에서 차단율을 90% 이상 낮추는 핵심 방어 기법들을 상세히 다룹니다.
---
## 최신 웹 방화벽이 크롤러를 식별하는 4가지 기준
1. **자동화 플래그 노출 (`navigator.webdriver`)**:
Playwright나 Selenium이 실행한 브라우저는 자바스크립트 엔진 내부에 `navigator.webdriver = true`라는 플래그를 기본적으로 들고 있습니다. 보안 스크립트는 이 변수 하나만으로 즉시 봇 판정을 내리고 차단합니다.
2. **그래픽 렌더러 지문 (WebGL / Canvas)**:
가상 서버 환경에서는 실제 외장 그래픽 카드가 없기 때문에 WebGL 렌더러 이름으로 `SwiftShader`나 `llvmpipe` 같은 소프트웨어 가상 드라이버 이름이 노출되어 자동화 샌드박스임이 들통납니다.
3. **기계적인 고정 주기 요청**:
정확히 3.0초나 5.0초마다 오차 없이 요청을 반복하는 행위는 행동 분석 엔진이 크롤러를 식별하는 가장 손쉬운 지표입니다.
4. **취약한 절대 경로 XPath 파싱**:
사이트가 약간만 레이아웃을 개편해도 `/html/body/div[2]/div[1]/span` 같은 절대 경로는 즉시 파싱 에러를 발생시켜 크롤러를 중단시킵니다.
---
## 회피 기법 1: `playwright-stealth` 자동화 플래그 은폐
`playwright-stealth` 모듈은 브라우저 컨텍스트가 생성되는 순간, 자동화 탐지에 사용되는 내부 자바스크립트 프로퍼티를 실제 일반 사용자의 크롬 브라우저와 동일하게 위장합니다.
```text title="requirements.txt에 추가"
playwright>=1.43.0
playwright-stealth>=1.0.6
```
```python title="stealth_example.py"
from playwright.sync_api import sync_playwright
from playwright_stealth import stealth_sync
with sync_playwright() as p:
browser = p.chromium.launch(headless=True)
# 일반 크롬 브라우저와 동일한 설정으로 컨텍스트 생성
context = browser.new_context(
viewport={"width": 1920, "height": 1080},
user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
locale="ko-KR",
timezone_id="Asia/Seoul"
)
page = context.new_page()
# [핵심] stealth 모듈 적용: navigator.webdriver 숨김, WebGL/플러그인 위장
stealth_sync(page)
# 봇 탐지 테스트 사이트 접속 검증
page.goto("https://bot.sannysoft.com", timeout=30000)
print("스텔스 모드 적용 완료. WebDriver 플래그가 완벽히 숨겨졌습니다.")
context.close()
browser.close()
```
---
## 회피 기법 2: 불규칙한 가변 지연 시간 (랜덤 지터)
인간의 웹 브라우징은 글을 읽는 속도와 마우스 움직임에 따라 페이지 간 체류 시간이 불규칙합니다.
고정된 `time.sleep(5)` 대신, 정규분포(가우시안 분포) 기반의 불규칙한 지연 시간을 적용하여 기계적 패턴을 제거합니다.
```python title="human_delay.py"
import time
import random
import logging
logger = logging.getLogger("Delay")
def human_like_delay(mean: float = 4.0, std_dev: float = 1.2, min_delay: float = 2.0):
"""
평균 4.0초, 표준편차 1.2초의 자연스러운 가변 지연 시간 생성
(실제 지연: 2.1초 ~ 6.5초 사이로 자연스럽게 분산됨)
"""
delay = random.gauss(mean, std_dev)
delay = max(min_delay, delay) # 최소 대기 시간 보장
logger.debug(f"자연스러운 인간형 딜레이 적용: {delay:.2f}초 대기")
time.sleep(delay)
```
---
## 회피 기법 3: 변경에 강한 복합 셀렉터 및 구조화 데이터 활용
쇼핑몰이나 포털 사이트가 디자인을 리뉴얼하더라도 크롤러가 부러지지 않도록 데이터 위치를 추적하는 우선순위 원칙입니다.
```
[ 셀렉터 설계 3대 우선순위 ]
1순위 (가장 안전):