## 1. Telethon 소개
Telethon은 Telegram의 공식 API를 사용하여 Python으로 Telegram 봇과 사용자 클라이언트를 개발할 수 있게 해주는 비동기 라이브러리입니다. MTProto 프로토콜을 기반으로 하며, Bot API보다 더 많은 기능을 제공합니다.
### 주요 특징
- **비동기 처리**: async/await 패턴 지원
- **완전한 API 접근**: Telegram의 모든 기능 활용 가능
- **사용자 계정 지원**: 봇뿐만 아니라 일반 사용자 계정으로도 동작
- **세션 관리**: 로그인 상태 유지
- **미디어 처리**: 파일 업로드/다운로드 지원
## 2. 설치 및 설정
### 2.1 라이브러리 설치
```bash
pip install telethon
```
### 2.2 API 키 발급
1. [my.telegram.org](https://my.telegram.org/)에 접속
2. 휴대폰 번호로 로그인
3. API Development Tools 선택
4. 애플리케이션 정보 입력 후 `api_id`와 `api_hash` 발급
### 2.3 기본 설정
```python
from telethon import TelegramClient
import asyncio
# API 정보 설정
api_id = 'YOUR_API_ID'
api_hash = 'YOUR_API_HASH'
phone_number = 'YOUR_PHONE_NUMBER'
# 클라이언트 생성
client = TelegramClient('session_name', api_id, api_hash)
```
## 3. 기본 사용법
### 3.1 클라이언트 시작
```python
async def main():
# 클라이언트 시작
await client.start(phone=phone_number)
# 본인 정보 확인
me = await client.get_me()
print(f"안녕하세요, {me.first_name}!")
# 실행
with client:
client.loop.run_until_complete(main())
```
### 3.2 메시지 보내기
```python
async def send_message():
# 특정 사용자에게 메시지 보내기
await client.send_message('username', '안녕하세요!')
# 채팅 ID로 메시지 보내기
await client.send_message(chat_id, '메시지 내용')
# 파일과 함께 메시지 보내기
await client.send_file('username', '/path/to/file.jpg', caption='사진입니다')
```
### 3.3 메시지 수신
```python
@client.on(events.NewMessage(pattern='/start'))
async def start_handler(event):
"""명령어 처리"""
await event.respond('봇이 시작되었습니다!')
@client.on(events.NewMessage)
async def message_handler(event):
"""모든 메시지 처리"""
print(f"받은 메시지: {event.text}")
sender = await event.get_sender()
print(f"보낸 사람: {sender.first_name}")
```
## 4. 고급 기능
### 4.1 채팅 및 사용자 정보 가져오기
```python
async def get_chat_info():
# 채팅 목록 가져오기
async for dialog in client.iter_dialogs():
print(f"채팅: {dialog.name}, ID: {dialog.id}")
# 특정 채팅의 참가자 목록
async for user in client.iter_participants('chat_username'):
print(f"사용자: {user.first_name} {user.last_name}")
```
### 4.2 메시지 히스토리 조회
```python
async def get_message_history():
# 최근 10개 메시지 가져오기
messages = await client.get_messages('username', limit=10)
for message in messages:
print(f"{message.date}: {message.text}")
# 특정 날짜 범위의 메시지
from datetime import datetime, timedelta
yesterday = datetime.now() - timedelta(days=1)
messages = await client.get_messages(
'username',
offset_date=yesterday,
limit=100
)
```
### 4.3 인라인 키보드 사용
```python
from telethon.tl.types import KeyboardButtonCallback
from telethon import Button
async def send_inline_keyboard():
buttons = [
[Button.inline('옵션 1', b'option1')],
[Button.inline('옵션 2', b'option2')],
[Button.url('웹사이트', 'https://example.com')]
]
await client.send_message(
'username',
'옵션을 선택하세요:',
buttons=buttons
)
@client.on(events.CallbackQuery)
async def callback_handler(event):
"""인라인 버튼 클릭 처리"""
if event.data == b'option1':
await event.answer('옵션 1을 선택했습니다!')
elif event.data == b'option2':
await event.answer('옵션 2를 선택했습니다!')
```
### 4.4 파일 처리
```python
async def handle_files():
# 파일 다운로드
messages = await client.get_messages('username', limit=10)
for message in messages:
if message.media:
path = await client.download_media(message)
print(f"파일 다운로드: {path}")
# 여러 파일 업로드
files = [
'image1.jpg',
'document.pdf',
'video.mp4'
]
await client.send_file('username', files, caption='여러 파일입니다')
```
## 5. 에러 처리
### 5.1 일반적인 예외 처리
```python
from telethon.errors import SessionPasswordNeededError, FloodWaitError
import asyncio
async def safe_operation():
try:
await client.start(phone=phone_number)
except SessionPasswordNeededError:
# 2단계 인증이 설정된 경우
password = input('2단계 인증 비밀번호를 입력하세요: ')
await client.start(password=password)
except FloodWaitError as e:
# API 호출 제한에 걸린 경우
print(f"{e.seconds}초 후에 다시 시도하세요")
await asyncio.sleep(e.seconds)
```
### 5.2 연결 안정성
```python
async def robust_client():
while True:
try:
await client.run_until_disconnected()
except Exception as e:
print(f"연결 오류: {e}")
print("5초 후 재연결 시도...")
await asyncio.sleep(5)
```
## 6. 실제 프로젝트 예제
### 6.1 간단한 에코 봇
```python
from telethon import TelegramClient, events
import asyncio
class EchoBot:
def __init__(self, api_id, api_hash, phone):
self.client = TelegramClient('echo_bot', api_id, api_hash)
self.phone = phone
self.setup_handlers()
def setup_handlers(self):
@self.client.on(events.NewMessage(pattern='/start'))
async def start(event):
await event.respond(
'안녕하세요! 메시지를 보내면 그대로 따라합니다.'
)
@self.client.on(events.NewMessage)
async def echo(event):
if not event.text.startswith('/'):
await event.respond(f"에코: {event.text}")
async def run(self):
await self.client.start(phone=self.phone)
print("에코 봇이 시작되었습니다...")
await self.client.run_until_disconnected()
# 실행
if __name__ == "__main__":
bot = EchoBot(api_id, api_hash, phone_number)
asyncio.run(bot.run())
```
### 6.2 채팅 백업 도구
```python
import json
from datetime import datetime
class ChatBackup:
def __init__(self, client):
self.client = client
async def backup_chat(self, chat_username, output_file):
messages_data = []
async for message in self.client.iter_messages(chat_username):
message_data = {
'id': message.id,
'date': message.date.isoformat(),
'text': message.text,
'sender_id': message.sender_id,
'media_type': type(message.media).__name__ if message.media else None
}
messages_data.append(message_data)
with open(output_file, 'w', encoding='utf-8') as f:
json.dump(messages_data, f, ensure_ascii=False, indent=2)
print(f"백업 완료: {len(messages_data)}개 메시지가 {output_file}에 저장됨")
# 사용 예제
async def backup_example():
backup_tool = ChatBackup(client)
await backup_tool.backup_chat('chat_username', 'backup.json')
```
## 7. 최적화 및 모범 사례
### 7.1 성능 최적화
```python
# 메시지 일괄 처리
async def batch_send_messages(chat, messages):
for i in range(0, len(messages), 20): # 20개씩 배치
batch = messages[i:i+20]
tasks = [client.send_message(chat, msg) for msg in batch]
await asyncio.gather(*tasks)
await asyncio.sleep(1) # API 제한 방지
# 세션 재사용
client = TelegramClient('session_name', api_id, api_hash)
# 'session_name.session' 파일이 생성되어 로그인 상태 유지
```
### 7.2 보안 모범 사례
```python
import os
from dotenv import load_dotenv
# 환경 변수 사용
load_dotenv()
API_ID = os.getenv('API_ID')
API_HASH = os.getenv('API_HASH')
PHONE = os.getenv('PHONE')
# .env 파일 예시:
# API_ID=your_api_id
# API_HASH=your_api_hash
# PHONE=your_phone_number
```
### 7.3 로깅 설정
```python
import logging
# 로깅 설정
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('telethon.log'),
logging.StreamHandler()
]
)
# Telethon 로깅 레벨 조정
logging.getLogger('telethon').setLevel(logging.WARNING)
```
## 8. 배포 및 운영
### 8.1 Docker 컨테이너화
```dockerfile
FROM python:3.9-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
CMD ["python", "bot.py"]
```
### 8.2 시스템 서비스 등록 (Linux)
```ini
# /etc/systemd/system/telethon-bot.service
[Unit]
Description=Telethon Bot
After=network.target
[Service]
Type=simple
User=botuser
WorkingDirectory=/path/to/bot
ExecStart=/usr/bin/python3 bot.py
Restart=always
[Install]
WantedBy=multi-user.target
```
### 8.3 모니터링
```python
import psutil
import asyncio
async def monitor_bot():
while True:
cpu_percent = psutil.cpu_percent()
memory_info = psutil.virtual_memory()
if cpu_percent > 80 or memory_info.percent > 80:
await client.send_message(
'admin_username',
f'⚠️ 시스템 경고\nCPU: {cpu_percent}%\n메모리: {memory_info.percent}%'
)
await asyncio.sleep(300) # 5분마다 체크
```
## 9. 문제 해결
### 9.1 일반적인 문제들
**세션 파일 문제**
```python
# 세션 파일 삭제 후 재인증
import os
if os.path.exists('session_name.session'):
os.remove('session_name.session')
```
**API 제한 해결**
```python
from telethon.errors import FloodWaitError
import asyncio
async def safe_api_call(func, *args, **kwargs):
try:
return await func(*args, **kwargs)
except FloodWaitError as e:
print(f"API 제한, {e.seconds}초 대기")
await asyncio.sleep(e.seconds)
return await func(*args, **kwargs)
```
**인코딩 문제**
```python
# UTF-8 인코딩 명시적 처리
import codecs
with codecs.open('file.txt', 'r', encoding='utf-8') as f:
content = f.read()
```
## 10. 참고 자료
- **공식 문서**: [docs.telethon.dev](https://docs.telethon.dev/)
- **예제 코드**: [GitHub 저장소](https://github.com/LonamiWebs/Telethon)
- **Telegram API 문서**: [core.telegram.org](https://core.telegram.org/api)
- **커뮤니티**: [@TelethonChat](https://t.me/TelethonChat)
이 가이드를 통해 Telethon을 활용한 다양한 Telegram 애플리케이션을 개발할 수 있습니다. 프로젝트 규모와 요구사항에 따라 적절한 패턴과 구조를 선택하여 사용하시기 바랍니다.