728x90
반응형
MICEMore 크롤링 서버 개요
MICEMore 앱에 표시되는 MICE 행사 데이터는 어디서 올까요? 바로 크롤링 서버가 전국 8개 행사 사이트에서 자동으로 데이터를 수집하고, 정규화하여 Firebase에 동기화합니다. 이 포스트에서는 Python으로 구축된 MICEMore 크롤링 서버의 전체 아키텍처를 상세히 분석합니다.
기술 스택 한눈에 보기
| 영역 | 기술 | 선택 이유 |
|---|---|---|
| 언어 | Python 3.7+ | 풍부한 스크래핑 라이브러리 생태계 |
| DB | SQLite3 (WAL 모드) | 설치 불필요, 파일 기반, 충분한 성능 |
| HTTP 서버 | http.server + ThreadingMixIn | 외부 프레임워크 없이 경량 서버 |
| 스크래핑 | requests + BeautifulSoup4 + lxml | HTML/JSON 파싱 표준 조합 |
| 스케줄링 | schedule (6시간 간격) | 간단한 주기 작업 실행 |
| 클라우드 동기화 | Firebase Admin SDK → Firestore | Flutter 앱과 데이터 공유 |
| 프론트엔드 | Vanilla HTML5/CSS3/JS (SPA) | 프레임워크 없는 순수 웹 대시보드 |
| 실시간 통신 | Server-Sent Events (SSE) | 서버 → 클라이언트 단방향 스트림 |
| 외부 접속 | Cloudflare Tunnel / ngrok | 로컬 서버를 외부에 안전하게 노출 |
핵심 설계 철학: 외부 의존성 최소화! requests, beautifulsoup4, lxml, schedule, firebase-admin 단 5개만 사용하고, 나머지는 모두 Python 표준 라이브러리입니다.
크롤러 시스템: 8개 크롤러 상세 분석
크롤러(Crawler)란?
크롤러는 웹사이트를 자동으로 방문해서 데이터를 수집하는 프로그램입니다. 사람이 브라우저로 사이트에 접속해서 정보를 복사하는 것을 프로그램이 대신 해주는 것이죠. 마치 로봇 비서가 여러 신문사를 돌아다니며 기사를 스크랩해오는 것과 같습니다.
8개 크롤러 한눈에 보기
| 크롤러 | 소스 | 방식 | 파싱 기법 |
|---|---|---|---|
| ExcoCrawler | EXCO 대구전시컨벤션센터 | HTML 스크래핑 | CSS 셀렉터 + 상세페이지 보강 |
| WeMiceCrawler | 위마이스 | JSON REST API | API 응답 파싱 |
| HicoCrawler | HICO 경주화백컨벤션센터 | HTML 스크래핑 | data-* 속성 + 그리드 카드 |
| PohangCrawler | 포항시 관광 | HTML 캘린더 | CSS 셀렉터 + 날짜 추출 |
| GyeongjuCrawler | 경주시 관광 | HTML 리스트 | dl/dd 구조 파싱 |
| DaeguFestivalCrawler | 대구관광 축제 | HTML 스크래핑 | .festival_box 카드 파싱 |
| VisitKoreaCrawler | 한국관광공사 | JSON API (GET/POST) | 지역코드 필터링 |
| EventbriteCrawler | Eventbrite | JSON-LD | Schema.org 구조화 데이터 |
크롤링 방식 3가지 비교
크롤러마다 데이터를 가져오는 방식이 다릅니다. 크게 3가지로 나뉩니다:
1. HTML 스크래핑 (5개 크롤러)
웹 브라우저에 표시되는 HTML 코드를 직접 읽어서 필요한 정보를 추출하는 방식입니다.
import requests
from bs4 import BeautifulSoup
# 1. 웹페이지 HTML 가져오기
response = requests.get("https://exco.co.kr/events")
html = response.text
# 2. BeautifulSoup으로 HTML 파싱 (구조화)
soup = BeautifulSoup(html, "lxml") # lxml은 빠른 파서
# 3. CSS 셀렉터로 원하는 요소 찾기
event_cards = soup.select(".event-card") # class="event-card"인 요소들
for card in event_cards:
title = card.select_one(".title").text.strip()
date = card.select_one(".date").text.strip()
venue = card.select_one(".venue").text.strip()
print(f"행사: {title}, 날짜: {date}, 장소: {venue}")2. JSON REST API (2개 크롤러)
사이트가 제공하는 API를 호출해서 정형화된 JSON 데이터를 받는 방식입니다. HTML 파싱보다 안정적입니다.
import requests
# API 호출 - 깔끔한 JSON 데이터 수신
response = requests.get(
"https://api.wemice.com/events",
params={"region": "daegu", "page": 1}
)
data = response.json() # JSON을 Python dict로 변환
for event in data["events"]:
title = event["title"]
start_date = event["startDate"]
venue = event["venue"]["name"]3. JSON-LD Schema.org (1개 크롤러)
JSON-LD는 HTML 페이지 안에 숨겨진 구조화된 데이터입니다. 검색엔진(Google)이 읽을 수 있도록 표준 형식으로 작성됩니다.
# HTML 안에 이런 script 태그가 숨겨져 있음
# <script type="application/ld+json">
# {"@type": "Event", "name": "Tech Conference", ...}
# </script>
soup = BeautifulSoup(html, "lxml")
# JSON-LD 스크립트 태그 찾기
script_tag = soup.find("script", type="application/ld+json")
event_data = json.loads(script_tag.string) # JSON 파싱
title = event_data["name"]
start_date = event_data["startDate"]
venue = event_data["location"]["name"]| 방식 | 장점 | 단점 | 비유 |
|---|---|---|---|
| HTML 스크래핑 | 어떤 사이트든 가능 | HTML 구조 변경 시 깨짐 | 신문 오려서 스크랩 |
| JSON API | 안정적, 정형화 | API가 있어야만 가능 | 기자에게 직접 기사 요청 |
| JSON-LD | 표준화된 구조 | 모든 사이트에 있진 않음 | 기사 뒤 요약본 읽기 |
데이터베이스: SQLite3 최적화 전략
SQLite란?
SQLite는 파일 하나가 곧 데이터베이스인 경량 DB입니다. MySQL이나 PostgreSQL처럼 별도 서버를 설치할 필요 없이, events.db라는 파일 하나만 있으면 됩니다.
| 비교 | MySQL/PostgreSQL | SQLite |
|---|---|---|
| 설치 | 서버 설치 필요 | 설치 불필요 (Python 내장) |
| 실행 | 별도 프로세스로 실행 | 앱 내부에서 직접 실행 |
| 저장 | 서버 디렉토리에 저장 | 단일 파일 (events.db) |
| 적합한 용도 | 대규모 웹서비스 | 로컬 앱, 크롤링 서버 |
| 비유 | 대형 도서관 | 개인 서재 |
events 테이블 구조 (20개 컬럼)
CREATE TABLE events (
id INTEGER PRIMARY KEY AUTOINCREMENT,
title TEXT NOT NULL, -- 행사 제목
source TEXT NOT NULL, -- 출처 (exco, wemice, hico...)
source_id TEXT NOT NULL, -- 원본 사이트의 고유 ID
category TEXT, -- 카테고리 (축제, 전시회, 컨퍼런스...)
region TEXT, -- 지역 (대구, 경주, 포항...)
venue TEXT, -- 장소명
start_date TEXT, -- 시작일 (YYYY-MM-DD)
end_date TEXT, -- 종료일
description TEXT, -- 상세 설명
image_url TEXT, -- 대표 이미지 URL
source_url TEXT, -- 원본 페이지 URL
organizer TEXT, -- 주최자
price TEXT, -- 가격 정보
tags TEXT, -- 태그 (JSON 배열)
latitude REAL, -- GPS 위도
longitude REAL, -- GPS 경도
status TEXT DEFAULT "active", -- 상태
created_at TEXT, -- 생성 시간
updated_at TEXT, -- 수정 시간
UNIQUE(source, source_id) -- 중복 방지 제약조건
);UNIQUE(source, source_id) 중복 방지 원리
같은 행사가 여러 번 크롤링되어도 중복 저장되지 않게 하는 복합 유니크 제약조건입니다.
-- source + source_id 조합이 유일해야 함
-- 예시:
-- ("exco", "EVT001") → 최초 삽입 OK
-- ("exco", "EVT001") → 중복! INSERT 실패 (무시)
-- ("hico", "EVT001") → source가 다르므로 OK
-- INSERT OR IGNORE: 중복이면 에러 없이 건너뜀
INSERT OR IGNORE INTO events (source, source_id, title, ...)
VALUES (?, ?, ?, ...); -- ? = 파라미터 바인딩 (SQL Injection 방지)SQLite 성능 최적화 4가지
| 최적화 | 설정 | 효과 | 비유 |
|---|---|---|---|
| WAL 모드 | PRAGMA journal_mode=WAL | 읽기/쓰기 동시 가능 | 도서관에서 책 빌리는 동안 다른 사람도 열람 가능 |
| 캐시 크기 | PRAGMA cache_size=8000 (8MB) | 자주 쓰는 데이터 메모리에 보관 | 자주 보는 책을 책상 위에 놓기 |
| 동기화 모드 | PRAGMA synchronous=NORMAL | 쓰기 속도 향상 (약간의 안전성 트레이드오프) | 매번 금고 잠그기 vs 가끔 잠그기 |
| 스레드별 커넥션 | threading.local() | 멀티스레드 안전 | 각 직원에게 개인 열쇠 지급 |
WAL(Write-Ahead Logging) 모드란?
WAL은 "변경사항을 별도 로그 파일에 먼저 기록"하는 방식입니다. 기본 모드에서는 쓰기 중 읽기가 차단되지만, WAL 모드에서는 읽기와 쓰기가 동시에 가능합니다.
import sqlite3
import threading
# 스레드별 독립 커넥션 관리
_thread_local = threading.local()
def get_connection():
"""현재 스레드 전용 DB 커넥션 반환"""
if not hasattr(_thread_local, "connection"):
conn = sqlite3.connect("events.db")
conn.execute("PRAGMA journal_mode=WAL") # WAL 모드
conn.execute("PRAGMA cache_size=8000") # 8MB 캐시
conn.execute("PRAGMA synchronous=NORMAL") # 빠른 동기화
conn.row_factory = sqlite3.Row # dict처럼 접근
_thread_local.connection = conn
return _thread_local.connectioncrawl_logs 테이블: 크롤링 이력 추적
CREATE TABLE crawl_logs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
source TEXT NOT NULL, -- 어떤 크롤러가 실행했는지
status TEXT NOT NULL, -- success / error
total_found INTEGER, -- 발견한 행사 수
new_added INTEGER, -- 새로 추가된 수
updated INTEGER, -- 업데이트된 수
duration_seconds REAL, -- 소요 시간
error_message TEXT, -- 에러 메시지 (있다면)
created_at TEXT DEFAULT (datetime("now"))
);API 서버: 프레임워크 없이 구축한 REST API
왜 Flask/Django를 안 쓰나요?
MICEMore 크롤링 서버는 Python 표준 라이브러리의 http.server만으로 API를 구축했습니다. 외부 프레임워크 없이도 충분한 기능을 구현할 수 있고, 의존성을 최소화하여 배포와 유지보수가 간단합니다.
from http.server import HTTPServer, BaseHTTPRequestHandler
from socketserver import ThreadingMixIn
# ThreadingMixIn: 요청마다 별도 스레드에서 처리
# → 여러 클라이언트가 동시에 접속해도 서로 기다리지 않음
class ThreadedHTTPServer(ThreadingMixIn, HTTPServer):
daemon_threads = True # 메인 종료 시 스레드도 함께 종료
class APIHandler(BaseHTTPRequestHandler):
def do_GET(self):
# URL 경로에 따라 분기 처리
if self.path == "/api/events":
self.handle_events()
elif self.path == "/api/stats":
self.handle_stats()
elif self.path == "/api/stream":
self.handle_sse_stream()
elif self.path == "/dashboard":
self.serve_dashboard()
else:
self.send_error(404)
def do_POST(self):
if self.path == "/api/crawl":
self.trigger_crawl() # 수동 크롤링 실행
# 서버 시작
server = ThreadedHTTPServer(("0.0.0.0", 8080), APIHandler)
server.serve_forever()API 엔드포인트 상세
| 엔드포인트 | 메서드 | 설명 | 캐시 TTL |
|---|---|---|---|
| /dashboard | GET | 실시간 대시보드 UI (HTML) | ETag |
| /api/events | GET | 이벤트 목록 (region/category 필터) | 10초 |
| /api/stats | GET | 통계 (소스별/지역별) | 30초 |
| /api/crawl-logs | GET | 크롤링 이력 | 30초 |
| /api/stream | GET | SSE 실시간 스트림 | 없음 |
| /api/live | GET | 최근 50개 이벤트 | 없음 |
| /api/health | GET | 서버 상태 확인 | 60초 |
| /api/crawl | POST | 수동 크롤링 트리거 | 없음 |
성능 최적화 전략 4가지
1. Gzip 압축 (70-80% 대역폭 절감)
import gzip
def send_response(self, data, content_type="application/json"):
body = json.dumps(data).encode("utf-8")
# 256바이트 이상이면 Gzip 압축 적용
if len(body) > 256 and "gzip" in self.headers.get("Accept-Encoding", ""):
body = gzip.compress(body)
self.send_header("Content-Encoding", "gzip")
self.send_header("Content-Type", content_type)
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
# 효과: 100KB JSON → 20-30KB로 압축 (70-80% 절감)2. TTL 캐시 + 이벤트 기반 무효화
TTL(Time To Live) 캐시는 "일정 시간 동안 같은 응답을 재사용"하는 것입니다. 크롤링이 완료되면 캐시를 자동으로 갱신합니다.
import time
class TTLCache:
def __init__(self):
self._cache = {} # {key: (data, expire_time)}
def get(self, key, ttl_seconds):
"""캐시에서 데이터 조회. 만료되었으면 None 반환"""
if key in self._cache:
data, expire_time = self._cache[key]
if time.time() < expire_time: # 아직 유효
return data
return None # 만료되었거나 없음
def set(self, key, data, ttl_seconds):
"""캐시에 데이터 저장 (TTL 설정)"""
self._cache[key] = (data, time.time() + ttl_seconds)
def invalidate(self, key=None):
"""크롤링 완료 시 캐시 무효화"""
if key:
self._cache.pop(key, None)
else:
self._cache.clear() # 전체 캐시 초기화
# 사용 예시
cache = TTLCache()
# /api/events 요청 처리
def handle_events(self):
cached = cache.get("events", ttl_seconds=10) # 10초 캐시
if cached:
return self.send_json(cached) # 캐시 히트!
events = db.query_events() # DB 조회
cache.set("events", events, ttl_seconds=10)
return self.send_json(events)3. ETag + 304 Not Modified (대시보드 캐싱)
ETag는 "콘텐츠의 지문(fingerprint)"입니다. 브라우저가 이전에 받은 ETag를 보내면, 서버는 콘텐츠가 변경되지 않았을 때 304 (변경 없음)만 응답하여 데이터 전송을 생략합니다.
import hashlib
def serve_dashboard(self):
html = load_dashboard_html()
# ETag 생성 (HTML 내용의 해시값)
etag = hashlib.md5(html.encode()).hexdigest()
# 브라우저가 보낸 ETag와 비교
if self.headers.get("If-None-Match") == etag:
self.send_response(304) # 변경 없음! 본문 전송 생략
self.end_headers()
return
# 변경되었으면 새 데이터 전송
self.send_response(200)
self.send_header("ETag", etag)
self.send_header("Content-Type", "text/html")
self.end_headers()
self.wfile.write(html.encode())4. Rate Limiting (IP당 120req/60s)
Rate Limiting은 "특정 IP에서 너무 많은 요청이 오면 차단"하는 보안 기능입니다. 슬라이딩 윈도우 방식으로 구현됩니다.
from collections import defaultdict, deque
import time
class RateLimiter:
def __init__(self, max_requests=120, window_seconds=60):
self.max_requests = max_requests
self.window = window_seconds
self.requests = defaultdict(deque) # IP별 요청 시간 기록
def is_allowed(self, ip):
now = time.time()
timestamps = self.requests[ip]
# 윈도우 밖의 오래된 요청 제거
while timestamps and timestamps[0] < now - self.window:
timestamps.popleft()
if len(timestamps) >= self.max_requests:
return False # 429 Too Many Requests
timestamps.append(now)
return True이벤트 버스: 실시간 통신의 핵심
이벤트 버스(Event Bus)란?
이벤트 버스는 "앱 내부의 방송 시스템"입니다. 크롤러가 "크롤링 완료!"라고 외치면, 대시보드, 캐시 관리자, SSE 스트림 등 관심 있는 모든 컴포넌트가 동시에 이를 듣고 반응합니다. 이것을 Observer/Pub-Sub 패턴이라고 합니다.
| 비유 | 이벤트 버스 |
|---|---|
| 학교 방송 | 방송실(Publisher)에서 안내하면, 교실(Subscriber)마다 스피커로 동시에 들음 |
| 유튜브 구독 | 채널(Publisher)에 영상 올리면, 구독자(Subscriber)에게 알림 |
from collections import deque
import threading
import queue
class EventBus:
def __init__(self):
self._subscribers = [] # SSE 클라이언트 목록
self._buffer = deque(maxlen=200) # 최근 200개 이벤트 보관
self._lock = threading.Lock() # 스레드 안전 보장
def publish(self, event_type, data):
"""이벤트 발행 (크롤러가 호출)"""
event = {"type": event_type, "data": data, "time": time.time()}
with self._lock:
self._buffer.append(event) # 버퍼에 보관
# 모든 구독자에게 전달
for subscriber_queue in self._subscribers:
try:
subscriber_queue.put_nowait(event)
except queue.Full:
pass # 큐가 가득 차면 건너뜀
def subscribe(self):
"""새 SSE 클라이언트 구독"""
q = queue.Queue(maxsize=50)
with self._lock:
self._subscribers.append(q)
return q
def unsubscribe(self, q):
"""SSE 클라이언트 구독 해제"""
with self._lock:
self._subscribers.remove(q)
# 사용 예시
event_bus = EventBus()
# 크롤러에서 이벤트 발행
event_bus.publish("crawl_done", {
"source": "exco",
"new_events": 5,
"duration": 3.2
})이벤트 타입 5가지
| 이벤트 | 발생 시점 | 데이터 내용 |
|---|---|---|
| crawl_start | 크롤링 시작 | 크롤러 이름, 시작 시간 |
| crawl_done | 크롤링 완료 | 새 행사 수, 업데이트 수, 소요시간 |
| crawl_error | 크롤링 실패 | 에러 메시지, 크롤러 이름 |
| event_new | 새 행사 발견 | 행사 제목, 장소, 날짜 |
| event_updated | 행사 정보 변경 | 변경된 필드, 행사 ID |
Server-Sent Events (SSE): 실시간 스트림
SSE란?
SSE는 서버에서 클라이언트로 단방향으로 데이터를 계속 보내는 기술입니다. WebSocket과 달리 HTTP만으로 동작하고, 브라우저가 자동 재연결을 지원합니다.
| 비교 | 일반 HTTP | SSE | WebSocket |
|---|---|---|---|
| 방향 | 요청-응답 (1회) | 서버 → 클라이언트 (단방향) | 양방향 |
| 연결 | 매번 새로 연결 | 한번 연결 후 유지 | 한번 연결 후 유지 |
| 자동 재연결 | 없음 | 브라우저 내장 | 직접 구현 필요 |
| 프로토콜 | HTTP | HTTP | WS (별도 프로토콜) |
| 적합한 용도 | 일반 API | 실시간 알림, 대시보드 | 채팅, 게임 |
# 서버 측: SSE 스트림 전송
def handle_sse_stream(self):
self.send_response(200)
self.send_header("Content-Type", "text/event-stream") # SSE 타입
self.send_header("Cache-Control", "no-cache") # 캐시 금지
self.send_header("Connection", "keep-alive") # 연결 유지
self.end_headers()
# 이벤트 버스 구독
client_queue = event_bus.subscribe()
try:
while True:
# 이벤트가 올 때까지 대기 (최대 30초)
try:
event = client_queue.get(timeout=30)
# SSE 형식으로 전송
msg = f"event: {event['type']}\n"
msg += f"data: {json.dumps(event['data'])}\n\n"
self.wfile.write(msg.encode())
self.wfile.flush()
except queue.Empty:
# 30초 동안 이벤트 없으면 하트비트 전송
self.wfile.write(b": heartbeat\n\n")
self.wfile.flush()
except (BrokenPipeError, ConnectionResetError):
pass # 클라이언트 연결 종료
finally:
event_bus.unsubscribe(client_queue) # 구독 해제
# 클라이언트 측: JavaScript에서 SSE 수신
# const eventSource = new EventSource("/api/stream");
# eventSource.addEventListener("crawl_done", (e) => {
# const data = JSON.parse(e.data);
# updateDashboard(data);
# });데이터 정규화: 8개 소스를 하나의 형식으로
정규화(Normalization)란?
8개의 서로 다른 사이트에서 가져온 데이터는 형식이 제각각입니다. 정규화란 이 다양한 형식을 하나의 통일된 형식으로 변환하는 과정입니다.
마치 한국, 미국, 일본에서 온 편지를 모두 한국어로 번역하는 것과 같습니다.
카테고리 정규화 (7종 표준화)
# 각 사이트마다 카테고리 표현이 다름
# EXCO: "전시/박람회", 위마이스: "Exhibition", 경주: "전시"
# → 모두 "전시회"로 통일!
CATEGORY_MAP = {
# 원본 → 표준화
"전시": "전시회",
"전시/박람회": "전시회",
"Exhibition": "전시회",
"expo": "전시회",
"축제/페스티벌": "축제",
"Festival": "축제",
"학술대회": "컨퍼런스",
"Conference": "컨퍼런스",
"교육": "세미나",
"Seminar": "세미나",
"Workshop": "워크숍",
"모임": "네트워킹",
}
# 표준 카테고리: 축제, 전시회, 컨퍼런스, 세미나, 워크숍, 네트워킹, 기타
def normalize_category(raw_category):
return CATEGORY_MAP.get(raw_category, "기타")지역 정규화 (8종)
| 표준 지역 | 매핑되는 원본 값들 |
|---|---|
| 대구 | "대구", "대구광역시", "Daegu", "대구시" |
| 포항 | "포항", "포항시", "Pohang" |
| 경주 | "경주", "경주시", "Gyeongju" |
| 경산 | "경산", "경산시" |
| 구미 | "구미", "구미시" |
| 안동 | "안동", "안동시" |
| 김천 | "김천", "김천시" |
| 경북 | "경북", "경상북도" (위 지역 외 경북 전체) |
날짜 정규화 (6+ 포맷 지원)
사이트마다 날짜 형식이 전부 다릅니다. 이걸 모두 YYYY-MM-DD 형식으로 통일합니다.
from datetime import datetime
def normalize_date(raw_date):
"""다양한 날짜 형식을 YYYY-MM-DD로 변환"""
formats = [
"%Y-%m-%d", # 2025-03-16
"%Y%m%d", # 20250316
"%Y.%m.%d", # 2025.03.16
"%Y/%m/%d", # 2025/03/16
"%m.%d.(%a)", # 3.16.(월) - 요일 포함
"%Y년 %m월 %d일", # 2025년 03월 16일
]
for fmt in formats:
try:
dt = datetime.strptime(raw_date.strip(), fmt)
return dt.strftime("%Y-%m-%d") # 표준 형식으로 반환
except ValueError:
continue
return None # 파싱 실패
# 예시:
# normalize_date("20250316") → "2025-03-16"
# normalize_date("2025.03.16") → "2025-03-16"
# normalize_date("2025년 03월 16일") → "2025-03-16"유효성 검증: 필수 필드 6개
REQUIRED_FIELDS = ["title", "source_id", "venue", "start_date", "category", "region"]
def validate_event(event):
"""이벤트 데이터 유효성 검증"""
for field in REQUIRED_FIELDS:
if not event.get(field):
raise ValueError(f"필수 필드 누락: {field}")
# 날짜 형식 검증
try:
datetime.strptime(event["start_date"], "%Y-%m-%d")
except ValueError:
raise ValueError(f"잘못된 날짜 형식: {event['start_date']}")
# 카테고리 검증
valid_categories = ["축제", "전시회", "컨퍼런스", "세미나", "워크숍", "네트워킹", "기타"]
if event["category"] not in valid_categories:
event["category"] = "기타" # 잘못된 값은 "기타"로
return TrueFirebase 동기화: 크롤링 데이터를 Flutter 앱으로
동기화 흐름
크롤링 서버 (Python)
|
[SQLite DB]
|
Firebase Admin SDK
|
[Firestore]
|
Flutter 앱 (MICEMore)
|
사용자 화면에 행사 표시import firebase_admin
from firebase_admin import credentials, firestore
# Firebase 초기화 (서비스 계정 키 사용)
cred = credentials.Certificate("firebase-key.json")
firebase_admin.initialize_app(cred)
db = firestore.client()
def sync_to_firebase(events):
"""크롤링된 이벤트를 Firestore에 동기화"""
batch = db.batch() # Batch Write로 효율적 전송
count = 0
for event in events:
# source + source_id로 고유 문서 ID 생성
doc_id = f"{event['source']}_{event['source_id']}"
doc_ref = db.collection("crawled_events").document(doc_id)
batch.set(doc_ref, {
"title": event["title"],
"category": event["category"],
"region": event["region"],
"venue": event["venue"],
"startDate": event["start_date"],
"endDate": event["end_date"],
"description": event["description"],
"imageUrl": event["image_url"],
"sourceUrl": event["source_url"],
"source": event["source"],
"updatedAt": firestore.SERVER_TIMESTAMP,
}, merge=True) # merge=True: 기존 데이터와 병합
count += 1
if count % 450 == 0: # Batch 최대 500개 제한
batch.commit()
batch = db.batch()
batch.commit() # 나머지 전송
print(f"Firebase 동기화 완료: {count}건")보안 전략
SQL Injection 방지: 파라미터 바인딩
SQL Injection은 악의적인 사용자가 URL 파라미터에 SQL 코드를 삽입해서 데이터베이스를 조작하는 공격입니다.
# 위험한 코드 (SQL Injection 취약)
region = request.params["region"] # 사용자 입력
query = f"SELECT * FROM events WHERE region = '{region}'" # 직접 삽입!
# 만약 region = "'; DROP TABLE events; --" 이면?
# → SELECT * FROM events WHERE region = ''; DROP TABLE events; --'
# → events 테이블 삭제!
# 안전한 코드 (파라미터 바인딩)
query = "SELECT * FROM events WHERE region = ?"
cursor.execute(query, (region,)) # ? 자리에 안전하게 삽입
# SQL 코드가 아닌 순수 문자열로 처리됨 → 공격 불가보안 체크리스트
| 보안 항목 | 구현 방법 | 효과 |
|---|---|---|
| SQL Injection | 파라미터 바인딩 (?) | DB 조작 공격 차단 |
| Rate Limiting | IP당 120req/60s | DDoS/브루트포스 방지 |
| CORS 헤더 | Access-Control-Allow-Origin 설정 | 허용된 도메인만 API 호출 |
| Firebase 키 보안 | .gitignore에 추가 | 키 유출 방지 |
CLI 실행 모드: 7가지 명령어
MICEMore 크롤링 서버는 CLI(Command Line Interface)로 다양한 모드를 지원합니다.
import sys
def main():
if len(sys.argv) < 2:
command = "serve" # 기본 모드
else:
command = sys.argv[1]
if command == "serve":
start_api_server() # API 서버 시작
start_auto_crawling() # 6시간마다 자동 크롤링
elif command == "crawl":
run_all_crawlers() # 1회 크롤링 후 종료
elif command == "api":
start_api_server() # API 서버만 (크롤링 없음)
elif command == "tunnel":
start_api_server()
start_cloudflare_tunnel() # 외부 접속 터널
elif command == "stats":
print_statistics() # 통계 출력
elif command == "export":
export_to_json() # JSON 파일 내보내기
elif command == "init":
initialize_database() # DB 초기화
if __name__ == "__main__":
main()| 명령어 | 용도 | 사용 시나리오 |
|---|---|---|
| python main.py serve | API + 자동 크롤링 | 운영 서버 (기본 모드) |
| python main.py crawl | 1회 크롤링 | 수동 데이터 수집 |
| python main.py api | API 서버만 | 크롤링 없이 데이터 조회 |
| python main.py tunnel | API + Cloudflare 터널 | 외부에서 접속 시 |
| python main.py stats | 통계 출력 | 현재 데이터 현황 확인 |
| python main.py export | JSON 내보내기 | 데이터 백업/분석 |
| python main.py init | DB 초기화 | 최초 설치 시 |
외부 의존성: 단 5개
# requirements.txt
requests>=2.28.0 # HTTP 요청 (웹페이지 다운로드)
beautifulsoup4>=4.11.0 # HTML 파싱 (데이터 추출)
lxml>=4.9.0 # 빠른 XML/HTML 파서
schedule>=1.1.0 # 주기적 작업 스케줄링
firebase-admin>=6.0.0 # Firebase Firestore 동기화
# 나머지는 모두 Python 표준 라이브러리:
# http.server, sqlite3, json, threading,
# collections, hashlib, gzip, time, sys 등전체 아키텍처 요약
[8개 웹사이트] ──스크래핑──> [크롤러들]
|
[데이터 정규화]
|
+-----------+-----------+
| |
[SQLite DB] [이벤트 버스]
| |
[API 서버] [SSE 스트림]
| |
[REST API] [실시간 대시보드]
|
[Firebase 동기화]
|
[Firestore DB]
|
[Flutter 앱 (MICEMore)]핵심 정리
| 개념 | 핵심 포인트 |
|---|---|
| 크롤러 | 8개 사이트에서 HTML/JSON/JSON-LD 3가지 방식으로 데이터 수집 |
| SQLite | WAL 모드 + 8MB 캐시 + 스레드별 커넥션으로 최적화 |
| API 서버 | 프레임워크 없이 http.server + ThreadingMixIn으로 구축 |
| 캐시 | TTL 캐시 + ETag + Gzip 압축으로 성능 극대화 |
| 이벤트 버스 | Thread-safe Pub/Sub 패턴으로 실시간 이벤트 전파 |
| SSE | 서버 → 클라이언트 단방향 실시간 스트림 |
| 데이터 정규화 | 카테고리 7종, 지역 8종, 날짜 6+ 포맷 표준화 |
| Firebase 동기화 | Batch Write + merge로 Firestore에 효율적 업로드 |
| 보안 | 파라미터 바인딩, Rate Limiting, CORS, 키 관리 |
| 의존성 | 외부 패키지 단 5개, 나머지 표준 라이브러리만 사용 |
728x90
반응형
