[카테고리:] General

typebot이라는 오픈소스를 이용해 챗봇을 만들기가 매우 쉽습니다.

사내에서 챗봇에 대한 니즈가 있어서 오픈소스를 찾아서 이거 저거 써봤는데 typebot이 가장 좋더군요.

우선 chatgpt가 알려주는 typebot의 설명입니다.

Typebot은 대화형 봇을 구축하고 운영하기 위한 플랫폼으로, 코드 작성 없이도 다양한 대화형 봇을 설계하고 배포할 수 있는 환경을 제공합니다. 주로 사용자와의 상호작용을 간소화하거나 고객 지원, 정보 제공, 자동화된 프로세스를 처리하기 위해 사용됩니다.

아래는 Typebot의 주요 특징과 기능, 사용 사례를 정리한 내용입니다.

---

1. Typebot의 주요 특징

1.1. 코드 없이 쉽게 설계 가능

• 드래그 앤 드롭 인터페이스를 통해 복잡한 대화 흐름을 설계할 수 있습니다.
• 초보자도 이해하기 쉬운 직관적인 UI 제공.

1.2. 멀티 플랫폼 지원

• Slack, WhatsApp, 웹사이트, Facebook Messenger 등 다양한 플랫폼과의 통합이 가능.
• 웹 임베드(embed) 기능을 통해 웹사이트에 쉽게 추가 가능.

1.3. 강력한 커스터마이징

• 봇의 대화 스타일, 답변 논리, 조건 등을 세부적으로 설정 가능.
• 사용자 입력에 따라 동적으로 반응하는 대화 흐름 설계 가능.

1.4. API와의 통합

• 외부 API와의 연동을 통해 데이터를 가져오거나 동적으로 데이터를 처리 가능.
• 예: 데이터베이스 조회, AI API 연동 등.

1.5. 분석 및 통계

• 사용자와의 대화 데이터를 분석하여 성능 개선 가능.
• 사용량, 클릭률, 대화 성공률 등을 시각화하여 제공.

---

2. 주요 기능

2.1. 대화 흐름 설계

• 질문-응답 시퀀스를 블록으로 정의.
• 조건 분기와 반복 로직 지원.

2.2. 사용자 입력 처리

• 버튼 선택, 텍스트 입력, 날짜 선택 등의 다양한 입력 방식을 지원.
• 사용자의 입력을 변수로 저장하여 다른 블록에서 재사용 가능.

2.3. 멀티미디어 지원

• 이미지, 동영상, 링크, 파일 등을 전송 가능.

2.4. 외부 서비스 연동

• Zapier, Integromat 같은 자동화 플랫폼과 통합.
• Google Sheets, CRM, 이메일 서비스 등과 연동.

2.5. 다국어 지원

• 여러 언어로 대화를 설정 가능.
• 다국어를 사용하는 글로벌 사용자층에 적합.

---

3. 사용 사례

3.1. 고객 지원

• 자주 묻는 질문(FAQ) 자동화.
• 간단한 문제 해결 봇 구축.

3.2. 예약 시스템

• 사용자가 날짜와 시간을 선택하면 예약을 처리.
• 예약 확인 이메일 또는 알림 발송.

3.3. 리드 생성

• 웹사이트 방문자와의 대화를 통해 잠재 고객 데이터를 수집.
• 설문조사나 의견 수집.

3.4. 교육

• 대화형 학습 모듈 제공.
• 퀴즈 및 테스트 자동화.

---

4. 장점

1. 사용 편의성: 프로그래밍 지식 없이도 빠르게 설정 가능.
2. 유연성: 다양한 플랫폼과 통합 가능.
3. 확장성: 외부 API와 연동하여 기능을 무한히 확장할 수 있음.
4. 비용 효율성: 비교적 저렴한 비용으로 기업용 봇 구축 가능.

이제 설치에 대해 알아 봅시다.

1. 도커 설치(도커가 제일 만만하고 편하죠~)
   sudo apt update && sudo apt install -y apt-transport-https ca-certificates curl software-properties-common && curl -fsSL https://download.docker.com/linux/ubuntu/gpg | sudo gpg –dearmor -o /usr/share/keyrings/docker-archive-keyring.gpg && echo “deb [arch=$(dpkg –print-architecture) signed-by=/usr/share/keyrings/docker-archive-keyring.gpg] https://download.docker.com/linux/ubuntu $(lsb_release -cs) stable” | sudo tee /etc/apt/sources.list.d/docker.list > /dev/null && sudo apt update && sudo apt install -y docker-ce docker-ce-cli containerd.io docker-compose-plugin && sudo usermod -aG docker $USER && newgrp docker
   
   2. typebot 디렉토리 생성
   sudo mkdir typebot

3. docker-compose.yml 생성
version: ‘3.3’

volumes:
db-data:

services:
typebot-db:
image: postgres:16
restart: always
volumes:
– db-data:/var/lib/postgresql/data
environment:
– POSTGRES_DB=typebot
– POSTGRES_PASSWORD=typebot
healthcheck:
test: [“CMD-SHELL”, “pg_isready -U postgres”]
interval: 5s
timeout: 5s
retries: 5
typebot-builder:
image: baptistearno/typebot-builder:latest
restart: always
depends_on:
typebot-db:
condition: service_healthy
ports:
– ‘8080:3000’
extra_hosts:
– ‘host.docker.internal:host-gateway’
env_file: .env

typebot-viewer:
image: baptistearno/typebot-viewer:latest
depends_on:
typebot-db:
condition: service_healthy
restart: always
ports:
– ‘8081:3000’
env_file: .env

4. .env 파일 생성
Make sure to change this to your own random string of 32 characters (https://docs.typebot.io/self-hosting/deploy/docker#2-add-the-required-configuration)

ENCRYPTION_SECRET=do+UspMmB/rewbX2K/rskFmtgGSSZ8Ta

DATABASE_URL=postgresql://postgres:typebot@typebot-db:5432/typebot

NODE_OPTIONS=–no-node-snapshot

NEXTAUTH_URL=(서버 ip 혹은 도메인 기입)(관리자 페이지)
NEXT_PUBLIC_VIEWER_URL=(서버 ip 혹은 도메인 기입)(챗봇 뷰어)

ADMIN_EMAIL=(관리자 이메일)

5. docker compose up -d 명령어로 실행


* 주의사항: smtp를 꼭 설정해줘야 합니다. 그래야 관리자 페이지로 들어가거나 할때 매직링크(이메일로 코드 날라오면 해당 코드로 접속) 사용이 가능합니다.

로그인을 하면 나오는 첫 화면.. 저기서 + 버튼을 눌러서 챗봇을 만들 수 있습니다.

챗봇 크레에이터 버튼을 누르면 나오는 화면, 첫번째는 스크래치를 이용해서 블럭 쌓기 처럼 챗봇을 만들 수 있는 메뉴.

두번째는 예제 파일을 이용해서 만드는 경우. 우리는 두번째 예제 파일을 살펴볼껍니다.

템플릿 메뉴를 누르면 나오는 화면입니다. 좌측은 여러가지 예제가 있고, 예제를 선택하면 우측에는 예제가 실행되는 화면이 나옵니다. 이것을 선택해 보도록 하죠.

자 그러면 이렇게 스크래치 화면이 나오고 우측 하단에 머머리 아저씨의 동영상 설명이 나옵니다. 좌측에는 여러가지 메뉴 박스가 있고 해당 기능을 이용해서 계속해서 챗봇을 구축해 나가면 됩니다.

몇번만 해 보면 감이 오실꺼에요, 매우 쉽습니다. 물론 고급기능을 사용하기에는 부족할 수 있지만, 일반적인 챗봇은 대부분 구현이 가능할 정도로 쉽고 정교하게 구축이 가능합니다.

이번 포스트에서는 Python을 사용하여 시스템 로그를 모니터링하고, 웹 대시보드로 표시하며, Telegram으로 알림을 보내는 통합 시스템을 구현하는 방법을 알아보겠습니다.

1. 프로젝트 개요

이 시스템은 다음과 같은 주요 기능들을 제공합니다:

• 시스템 로그 실시간 모니터링
• SQLite 데이터베이스에 로그 저장
• Flask 기반 웹 대시보드
• Telegram을 통한 알림 전송
• 주기적인 로그 통계 보고

2. 필요한 라이브러리 및 초기 설정

주요 라이브러리들을 임포트하고 필요한 상수들을 정의합니다. Telegram 봇 토큰과 채팅 ID는 실제 값으로 교체해야 합니다.

3. 데이터베이스 설정

SQLite 데이터베이스를 설정하고 로그를 저장할 테이블을 생성합니다. check_same_thread=False는 다중 스레드 환경에서 데이터베이스 접근을 가능하게 합니다.

4. Telegram 알림 기능

Telegram API를 사용하여 메시지를 전송하는 함수입니다. 예외 처리를 통해 전송 실패 시 적절한 에러 메시지를 출력합니다.

5. 로그 파싱 및 처리

로그 라인을 파싱하여 구조화된 데이터로 변환하는 함수입니다. 시간, 호스트명, 프로그램명, 로그 레벨, 메시지를 추출합니다.

6. 실시간 로그 모니터링

follow 함수는 파일의 새로운 라인을 지속적으로 모니터링하고, log_monitor 함수는 이를 파싱하여 데이터베이스에 저장합니다.

7. 웹 대시보드 구현

Flask를 사용하여 웹 대시보드를 구현했습니다. 메인 페이지와 로그를 제공하는 API 엔드포인트를 제공합니다.

 

8. 실행 코드

두 개의 백그라운드 스레드(로그 모니터링, 통계 전송)를 시작하고 Flask 애플리케이션을 실행합니다.

import time
import os
import requests
import sqlite3
import threading
from flask import Flask, render_template_string, request, jsonify

# 텔레그램 토큰 및 채팅 ID 하드코딩
TELEGRAM_TOKEN = “YOUR_TELEGRAM_TOKEN”
CHAT_ID = “YOUR_CHAT_ID”

# 텔레그램으로 전송할 로그 레벨
TELEGRAM_LEVELS = [‘경고’, ‘마이너’, ‘메이저’, ‘치명적’]

# 로그 파일 경로 설정
LOG_FILE_PATH = “/var/log/syslog”

# Flask 애플리케이션 설정
app = Flask(__name__)

# 데이터베이스 설정
def create_db_connection():
conn = sqlite3.connect(‘logs.db’, check_same_thread=False)
conn.row_factory = sqlite3.Row # 딕셔너리 형식으로 결과 반환
return conn

# 데이터베이스 초기화
conn = create_db_connection()
cursor = conn.cursor()
cursor.execute(”’
CREATE TABLE IF NOT EXISTS logs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT,
hostname TEXT,
program TEXT,
level TEXT,
message TEXT
)
”’)
conn.commit()

# 텔레그램 메시지 전송 함수
def send_to_telegram(message):
url = f”https://api.telegram.org/bot{TELEGRAM_TOKEN}/sendMessage”
data = {
‘chat_id’: CHAT_ID,
‘text’: message
}
try:
response = requests.post(url, data=data)
if not response.ok:
print(f”텔레그램 메시지 전송 실패: {response.text}”)
except Exception as e:
print(f”텔레그램 메시지 전송 중 오류 발생: {e}”)

# 데이터베이스에 로그 저장
def save_to_db(conn, timestamp, hostname, program, level, message):
try:
cursor = conn.cursor()
cursor.execute(”’
INSERT INTO logs (timestamp, hostname, program, level, message)
VALUES (?, ?, ?, ?, ?)
”’, (timestamp, hostname, program, level, message))
conn.commit()
except Exception as e:
print(f”데이터베이스 저장 중 오류 발생: {e}”)

# 로그 라인 파싱
def parse_log_line(line):
try:
parts = line.split(‘ ‘)
if len(parts) < 5:
return None

timestamp = ‘ ‘.join(parts[:3])
hostname = parts[3]
program_part = parts[4]
program = program_part.split(‘[‘)[0].strip(‘:’)
message = ‘ ‘.join(parts[5:])
level = “정보”

for l in TELEGRAM_LEVELS:
if f”[{l}]” in message:
level = l
message = message.replace(f”[{l}]”, “”).strip()
break

return {
‘timestamp’: timestamp,
‘hostname’: hostname,
‘program’: program,
‘level’: level,
‘message’: message
}
except Exception as e:
print(f”로그 파싱 중 예외 발생: {e}”)
return None

# 로그 파일 실시간 읽기
def follow(file_path):
with open(file_path, ‘r’, encoding=’utf-8′) as f:
f.seek(0, os.SEEK_END)
while True:
line = f.readline()
if line:
yield line.strip()
else:
time.sleep(0.1)

# 로그 모니터링 스레드
def log_monitor():
thread_conn = create_db_connection()
for line in follow(LOG_FILE_PATH):
parsed = parse_log_line(line)
if parsed:
save_to_db(
thread_conn,
parsed[‘timestamp’],
parsed[‘hostname’],
parsed[‘program’],
parsed[‘level’],
parsed[‘message’]
)

# 3분마다 로그 개수 텔레그램으로 전송
def send_log_count():
thread_conn = create_db_connection()
while True:
cursor = thread_conn.cursor()
cursor.execute(“SELECT COUNT(*) as count FROM logs”)
log_count = cursor.fetchone()[‘count’]
send_to_telegram(f”지난 3분 동안 데이터베이스에 쌓인 로그 개수: {log_count}”)
time.sleep(180)

# Flask 웹 라우트
@app.route(‘/’)
def index():
return render_template_string(TEMPLATE)

@app.route(‘/api/logs’)
def get_logs():
cursor = conn.cursor()
cursor.execute(“SELECT timestamp, hostname, program, level, message FROM logs ORDER BY timestamp DESC LIMIT 20”)
logs = cursor.fetchall()
return jsonify([dict(log) for log in logs])

# HTML 템플릿
TEMPLATE = ”’
<!DOCTYPE html>
<html lang=”ko”>
<head>
<meta charset=”UTF-8″>
<title>로그 대시보드</title>
<style>
table { width: 100%; border-collapse: collapse; }
th, td { border: 1px solid #ddd; padding: 8px; }
th { background-color: #f2f2f2; text-align: left; }
</style>
<script>
async function fetchLogs() {
const response = await fetch(‘/api/logs’);
const logs = await response.json();
const tableBody = document.getElementById(‘logs-body’);
tableBody.innerHTML = ”;
logs.forEach(log => {
const row = `<tr>
<td>${log.timestamp}</td>
<td>${log.hostname}</td>
<td>${log.program}</td>
<td>${log.level}</td>
<td>${log.message}</td>
</tr>`;
tableBody.innerHTML += row;
});
}
setInterval(fetchLogs, 5000); // 5초마다 갱신
window.onload = fetchLogs;
</script>
</head>
<body>
<h1>로그 대시보드</h1>
<table>
<thead>
<tr>
<th>시간</th>
<th>호스트명</th>
<th>프로그램</th>
<th>레벨</th>
<th>메시지</th>
</tr>
</thead>
<tbody id=”logs-body”></tbody>
</table>
</body>
</html>
”’

# Flask 앱 실행
if __name__ == ‘__main__’:
threading.Thread(target=log_monitor, daemon=True).start()
threading.Thread(target=send_log_count, daemon=True).start()
app.run(host=’0.0.0.0′, port=5000)