Python
[Python] 법령 개정 알림 서비스 구축하기 (3) - Streamlit 대시보드 구현
dev-grace
2025. 3. 10. 17:42
지난 글에 이어 법령 개정 알림 서비스 시리즈의 마지막 글에서는 Streamlit을 활용하여 법령 변경 이력을 조회하고 구독자를 관리할 수 있는 간단한 대시보드를 만들어 보겠습니다.
1. Streamlit 소개
Streamlit은 데이터 앱을 빠르게 만들 수 있는 Python 라이브러리입니다. 복잡한 웹 개발 지식 없이도 몇 줄의 코드로 대시보드를 구현할 수 있어 매우 편리합니다.
1-1. 설치하기
먼저 Streamlit을 설치합니다.
pip install streamlit pandas
2. 대시보드 구현하기
이제 `dashboard.py` 파일을 만들어 대시보드를 구현해 보겠습니다.
import streamlit as st
import pandas as pd
import altair as alt
from datetime import datetime, timedelta
import time
import os
import random # 데모 데이터용
# 이전에 만든 파일들 import
from database import LawDatabase
from monitor import check_law_updates
# 페이지 기본 설정
st.set_page_config(
page_title="법령 개정 알림 서비스",
page_icon="⚖️",
layout="wide",
initial_sidebar_state="expanded",
)
# 데이터베이스 연결
@st.cache_resource
def get_database():
return LawDatabase()
# 사이드바 메뉴
with st.sidebar:
st.title("⚖️ 법령 개정 알림 서비스")
st.divider()
selected = st.radio(
"메뉴",
options=["대시보드", "법령 목록", "변경 이력", "구독자 관리", "수동 업데이트"],
captions=[
"서비스 개요",
"등록된 법령 보기",
"법령 변경 내역",
"이메일 구독 관리",
"지금 업데이트 실행",
],
)
st.divider()
# 데이터베이스 객체 생성
db = get_database()
# 간단한 상태 표시
st.subheader("📊 현재 상태")
laws_count = len(db.get_all_laws())
changes_count = len(db.get_changes(limit=1000))
subscribers_count = len(db.get_subscribers())
col1, col2, col3 = st.columns(3)
col1.metric("법령", f"{laws_count}개")
col2.metric("변경", f"{changes_count}건")
col3.metric("구독자", f"{subscribers_count}명")
st.divider()
# 최근 업데이트 시간
st.subheader("📅 최근 확인")
if laws_count > 0:
laws = db.get_all_laws()
if laws and "last_crawled" in laws[0]:
last_check = laws[0]["last_crawled"]
st.info(f"최근 확인: {last_check}")
else:
st.warning("데이터 없음")
else:
st.warning("아직 데이터 없음")
# 대시보드 페이지
if selected == "대시보드":
st.title("📊 법령 개정 알림 서비스 대시보드")
st.info(
"""
이 서비스는 법령 개정 사항을 자동으로 감지하고 이메일로 알림을 보내는 시스템입니다.
대시보드에서 법령 현황과 변경 이력을 확인할 수 있습니다.
"""
)
# 통계 카드 (3열 배치)
col1, col2, col3 = st.columns(3)
with col1:
st.metric(label="등록된 법령", value=f"{laws_count}개", delta=None)
with col2:
st.metric(label="감지된 변경", value=f"{changes_count}건", delta=None)
with col3:
st.metric(label="현재 구독자", value=f"{subscribers_count}명", delta=None)
# 최근 변경 이력
st.subheader("최근 변경 이력")
changes = db.get_changes(limit=5)
if changes:
for i, change in enumerate(changes):
with st.expander(f"{change['law_title']} - {change['detected_at']}"):
st.write(f"**법령 ID:** {change['law_id']}")
st.write(f"**변경 유형:** {change['change_type']}")
st.write(f"**감지 시간:** {change['detected_at']}")
else:
st.info("아직 감지된 변경 이력이 없습니다.")
# 분석 섹션 (2열 배치)
st.subheader("법령 분석")
col1, col2 = st.columns(2)
with col1:
st.caption("변경이 많은 법령 Top 5")
if changes and len(changes) > 0:
# 법령별 변경 횟수 계산
law_changes = {}
for change in changes:
law_title = change["law_title"]
if law_title in law_changes:
law_changes[law_title] += 1
else:
law_changes[law_title] = 1
# 변경 횟수 순으로 정렬하여 상위 5개 추출
top_laws = sorted(law_changes.items(), key=lambda x: x[1], reverse=True)[:5]
# 데이터프레임 생성
df = pd.DataFrame(top_laws, columns=["법령명", "변경 횟수"])
# 차트 생성 (Altair 사용)
chart = (
alt.Chart(df)
.mark_bar()
.encode(
x=alt.X("변경 횟수:Q", title="변경 횟수"),
y=alt.Y("법령명:N", sort="-x", title="법령명"),
color=alt.Color(
"변경 횟수:Q", scale=alt.Scale(scheme="blues"), legend=None
),
)
.properties(height=250)
)
st.altair_chart(chart, use_container_width=True)
else:
st.info("변경 이력 데이터가 충분하지 않습니다.")
with col2:
st.caption("월별 변경 추이")
if changes and len(changes) > 3:
# 월별 변경 횟수 계산
monthly_changes = {}
for change in changes:
try:
date = datetime.strptime(change["detected_at"], "%Y-%m-%d %H:%M:%S")
month_key = f"{date.year}-{date.month:02d}"
if month_key in monthly_changes:
monthly_changes[month_key] += 1
else:
monthly_changes[month_key] = 1
except:
# 날짜 파싱 오류 무시
pass
# 데이터프레임 생성
months = sorted(monthly_changes.keys())
values = [monthly_changes[m] for m in months]
df = pd.DataFrame({"월": months, "변경 수": values})
# 라인 차트 생성
line_chart = (
alt.Chart(df)
.mark_line(point=True)
.encode(
x=alt.X("월:N", title="월"),
y=alt.Y("변경 수:Q", title="변경 건수"),
tooltip=["월", "변경 수"],
)
.properties(height=250)
)
st.altair_chart(line_chart, use_container_width=True)
else:
# 데모 데이터 생성 (실제 데이터가 부족한 경우)
months = []
values = []
for i in range(6):
# 6개월 전부터 현재까지
date = datetime.now() - timedelta(days=30 * i)
month_key = f"{date.year}-{date.month:02d}"
months.insert(0, month_key)
values.insert(0, random.randint(1, 10)) # 랜덤 데이터
df = pd.DataFrame({"월": months, "변경 수": values})
# 라인 차트 생성 (데모 데이터)
line_chart = (
alt.Chart(df)
.mark_line(point=True)
.encode(
x=alt.X("월:N", title="월"),
y=alt.Y("변경 수:Q", title="변경 건수"),
tooltip=["월", "변경 수"],
)
.properties(height=250)
)
st.altair_chart(line_chart, use_container_width=True)
st.caption("※ 위 차트는 데모 데이터입니다.")
# 법령 목록 페이지
elif selected == "법령 목록":
st.title("📋 등록된 법령 목록")
# 검색 기능
search_query = st.text_input("법령명 검색", placeholder="검색어를 입력하세요...")
# 법령 목록 가져오기
db = get_database()
laws = db.get_all_laws()
if laws:
# 검색 필터링
if search_query:
filtered_laws = [
law for law in laws if search_query.lower() in law["title"].lower()
]
else:
filtered_laws = laws
# 테이블로 표시
df = pd.DataFrame(filtered_laws)
df.columns = ["법령 ID", "법령명", "최종 업데이트", "최근 크롤링"]
# 가독성 개선
st.caption(f"등록된 법령 수: {len(laws)} (검색 결과: {len(filtered_laws)})")
st.dataframe(df, use_container_width=True, height=400)
# 특정 법령 상세 정보 조회
st.subheader("법령 상세 정보 조회")
col1, col2 = st.columns([3, 1])
with col1:
law_id = st.selectbox(
"법령 선택",
options=[law["id"] for law in filtered_laws],
format_func=lambda x: next(
(law["title"] for law in filtered_laws if law["id"] == x), x
),
)
with col2:
lookup_button = st.button("조회", use_container_width=True)
if lookup_button:
with st.spinner("정보를 불러오는 중..."):
law_detail = db.get_law(law_id)
if law_detail:
st.success(f"법령: {law_detail['title']}")
col1, col2 = st.columns(2)
with col1:
st.write(f"**법령 ID:** {law_detail['id']}")
with col2:
st.write(f"**최종 업데이트:** {law_detail['last_update']}")
st.write(f"**최근 크롤링:** {law_detail['last_crawled']}")
with st.expander("법령 내용"):
st.text_area(
"", law_detail["content"], height=300, disabled=True
)
else:
st.info("등록된 법령이 없습니다. '수동 업데이트'를 실행해 보세요.")
# 변경 이력 페이지
elif selected == "변경 이력":
st.title("🔄 법령 변경 이력")
# 변경 이력 가져오기
db = get_database()
changes = db.get_changes(limit=100) # 최근 100개까지만
if changes:
# 필터링 옵션
col1, col2 = st.columns(2)
with col1:
# 법령 목록 (중복 제거)
law_titles = ["전체"] + sorted(
list(set([change["law_title"] for change in changes]))
)
selected_law = st.selectbox("법령 선택", law_titles)
with col2:
# 변경 유형 목록 (중복 제거)
change_types = ["전체"] + sorted(
list(set([change["change_type"] for change in changes]))
)
selected_type = st.selectbox("변경 유형", change_types)
# 필터링 적용
if selected_law != "전체" and selected_type != "전체":
filtered_changes = [
change
for change in changes
if change["law_title"] == selected_law
and change["change_type"] == selected_type
]
elif selected_law != "전체":
filtered_changes = [
change for change in changes if change["law_title"] == selected_law
]
elif selected_type != "전체":
filtered_changes = [
change for change in changes if change["change_type"] == selected_type
]
else:
filtered_changes = changes
# 테이블로 표시
df = pd.DataFrame(filtered_changes)
df.columns = ["ID", "법령 ID", "법령명", "변경 유형", "감지 시간"]
st.caption(
f"총 변경 이력: {len(changes)}건 (필터링 결과: {len(filtered_changes)}건)"
)
st.dataframe(df, use_container_width=True, height=400)
# 변경 이력 시각화
st.subheader("변경 이력 분석")
col1, col2 = st.columns(2)
with col1:
st.caption("변경 유형별 분포")
type_counts = {}
for change in changes:
change_type = change["change_type"]
if change_type in type_counts:
type_counts[change_type] += 1
else:
type_counts[change_type] = 1
pie_data = pd.DataFrame(
{"변경 유형": type_counts.keys(), "건수": type_counts.values()}
)
st.bar_chart(pie_data, x="변경 유형", y="건수", use_container_width=True)
with col2:
st.caption("법령별 변경 분포")
law_counts = {}
for change in changes:
law_title = change["law_title"]
if law_title in law_counts:
law_counts[law_title] += 1
else:
law_counts[law_title] = 1
# 상위 5개만 표시
top_laws = sorted(law_counts.items(), key=lambda x: x[1], reverse=True)[:5]
law_data = pd.DataFrame(
{
"법령명": [l[0] for l in top_laws],
"변경 수": [l[1] for l in top_laws],
}
)
st.bar_chart(law_data, x="법령명", y="변경 수", use_container_width=True)
# 선택한 법령이 있으면 상세 분석
if selected_law != "전체":
st.subheader(f"{selected_law} 변경 이력")
law_changes = [
change for change in changes if change["law_title"] == selected_law
]
for change in law_changes:
with st.expander(f"{change['detected_at']} - {change['change_type']}"):
st.write(f"**법령 ID:** {change['law_id']}")
st.write(f"**변경 유형:** {change['change_type']}")
st.write(f"**감지 시간:** {change['detected_at']}")
else:
st.info("아직 감지된 변경 이력이 없습니다.")
# 구독자 관리 페이지
elif selected == "구독자 관리":
st.title("👥 구독자 관리")
# 현재 구독자 목록 표시
db = get_database()
subscribers = db.get_subscribers()
st.subheader("현재 구독자 목록")
if subscribers:
st.success(f"총 {len(subscribers)}명의 구독자가 등록되어 있습니다.")
# 테이블 형태로 표시
subscriber_df = pd.DataFrame({"이메일": subscribers})
st.dataframe(subscriber_df, use_container_width=True, hide_index=True)
# 구독자 삭제
st.subheader("구독자 삭제")
col1, col2 = st.columns([3, 1])
with col1:
delete_email = st.selectbox("삭제할 이메일 선택", subscribers)
with col2:
if st.button("삭제", use_container_width=True, type="primary"):
if db.remove_subscriber(delete_email):
st.success(f"{delete_email} 구독자가 삭제되었습니다.")
time.sleep(1)
st.rerun()
else:
st.error("구독자 삭제 중 오류가 발생했습니다.")
else:
st.info("현재 등록된 구독자가 없습니다.")
# 새 구독자 추가
st.subheader("새 구독자 추가")
col1, col2 = st.columns([3, 1])
with col1:
new_email = st.text_input("이메일 주소", placeholder="example@email.com")
with col2:
if st.button("구독자 추가", use_container_width=True):
if not new_email or "@" not in new_email:
st.error("유효한 이메일 주소를 입력해주세요.")
elif db.add_subscriber(new_email):
st.success(f"{new_email}이 구독자로 추가되었습니다.")
time.sleep(1)
st.rerun()
else:
st.warning(f"{new_email}은(는) 이미 등록된 이메일 주소입니다.")
# CSV 업로드로 구독자 일괄 추가
st.subheader("CSV로 구독자 일괄 추가")
st.caption("CSV 파일 형식: 한 줄에 하나의 이메일 주소")
st.text("예: example1@email.com\nexample2@email.com\nexample3@email.com")
uploaded_file = st.file_uploader("CSV 파일 업로드", type="csv")
if uploaded_file is not None:
try:
df = pd.read_csv(uploaded_file, header=None)
emails = df[0].tolist()
if st.button("일괄 추가", type="primary"):
added_count = 0
skipped_count = 0
for email in emails:
email = str(email).strip()
if "@" in email:
if db.add_subscriber(email):
added_count += 1
else:
skipped_count += 1
st.success(
f"처리 완료: {added_count}명 추가, {skipped_count}명 건너뜀 (이미 존재)"
)
time.sleep(1)
st.rerun()
except Exception as e:
st.error(f"CSV 파일 처리 중 오류가 발생했습니다: {str(e)}")
# 수동 업데이트 페이지
elif selected == "수동 업데이트":
st.title("🔄 법령 정보 수동 업데이트")
st.info(
"""
이 페이지에서는 법령 정보를 지금 수동으로 업데이트할 수 있습니다.
업데이트는 몇 분 정도 소요될 수 있으며, 변경 사항이 감지되면 구독자에게 이메일이 발송됩니다.
"""
)
# 최근 업데이트 정보
db = get_database()
laws = db.get_all_laws()
if laws:
last_update_times = [law.get("last_crawled", "알 수 없음") for law in laws]
last_update_times = [t for t in last_update_times if t != "알 수 없음"]
if last_update_times:
last_update = max(last_update_times)
st.success(f"최근 업데이트: {last_update}")
# 업데이트 옵션
st.subheader("업데이트 옵션")
col1, col2 = st.columns(2)
with col1:
update_limit = st.number_input(
"가져올 법령 수", min_value=1, max_value=50, value=5
)
with col2:
send_notifications = st.checkbox("변경 감지 시 알림 발송", value=True)
# 상태 표시용 컨테이너
status_container = st.empty()
# 업데이트 버튼
if st.button("지금 업데이트 실행", use_container_width=True, type="primary"):
# 상태 표시
status_container.info("업데이트 실행 중... 잠시만 기다려주세요.")
# 업데이트 실행
try:
# 업데이트 함수 호출 (실제 함수에 인자 전달 필요한 경우 수정)
with st.spinner("법령 정보 업데이트 중..."):
# 여기에 실제 업데이트 로직 추가 (예: check_law_updates(limit=update_limit, send_email=send_notifications))
check_law_updates()
# 성공 메시지
status_container.success("성공! 법령 정보 업데이트가 완료되었습니다.")
# 업데이트 후 통계 표시
new_laws = db.get_all_laws()
new_changes = db.get_changes(limit=1000)
st.write("**업데이트 후 통계:**")
st.write(f"- 등록된 법령: {len(new_laws)}개")
st.write(f"- 감지된 변경: {len(new_changes)}건")
except Exception as e:
status_container.error(
f"오류 발생! 업데이트 중 문제가 발생했습니다.\n{str(e)}"
)
# 스케줄링 안내
st.subheader("자동 업데이트 스케줄링")
st.info(
"""
법령 정보는 다음 일정에 따라 자동으로 업데이트됩니다:
- 매일 오전 9시
- 매일 오후 3시
이 일정은 `scheduler.py` 파일에서 수정할 수 있습니다.
"""
)
# 푸터
st.divider()
st.caption(
"© 2025 법령 개정 알림 서비스 | 이 서비스는 법령 정보를 수집하여 변경 사항을 알려주는 도구입니다."
)
# 데이터베이스 연결 닫기 (앱 종료 시)
# Streamlit은 자동으로 이 코드를 실행하지 않음 - 참고용
def cleanup():
db = get_database()
db.close()3. 대시보드 실행하기
Streamlit 앱을 실행하려면 다음 명령어를 사용합니다.
streamlit run dashboard.py
이 명령어를 실행하면 브라우저가 자동으로 열리고 대시보드가 표시됩니다. 기본적으로 `http://localhost:8501`에서 접근할 수 있습니다.
4. 대시보드 주요 기능 살펴보기
4-1. 메인 대시보드
메인 대시보드에서는 다음 정보를 한눈에 확인할 수 있습니다.
- 등록된 법령 수
- 감지된 변경 이력 수
- 현재 구독자 수
- 최근 변경 이력 목록
- 변경이 많은 법령 Top 5 그래프
4-2. 법령 목록
법령 목록 페이지는 다음 기능이 포함되어 있습니다.
- 모든 등록된 법령을 테이블 형태로 확인
- 특정 법령을 선택하여 상세 내용 조회
4-3. 변경 이력
변경 이력 페이지는 다음 기능이 포함되어 있습니다.
- 모든 변경 이력을 시간순으로 확인
- 특정 법령을 선택하여 해당 법령의 변경 이력만 필터링
4-4. 구독자 관리
구독자 관리 페이지는 다음 기능이 포함되어 있습니다.
- 현재 등록된 모든 구독자 목록 확인
- 구독자 추가 및 삭제 기능
4-5. 수동 업데이트
수동 업데이트 페이지는 다음 기능이 포함되어 있습니다.
- 버튼 클릭으로 법령 정보 즉시 업데이트
- 업데이트 과정과 결과 확인
5. 서비스 배포 및 활용 방법
실제 서비스로 배포하려면 다음과 같은 방법을 고려할 수 있습니다.
5-1. 로컬 서버에서 실행
작은 규모의 사용자를 대상으로 할 경우 로컬 서버에서 실행하는 것이 가장 간단합니다:
# 백그라운드에서 모니터링 실행
nohup python scheduler.py > logs/monitor.log 2>&1 &
# Streamlit 대시보드 실행
nohup streamlit run dashboard.py --server.port 8501 > logs/dashboard.log 2>&1 &
5-2. 클라우드 서비스 활용
AWS, GCP, Azure 등의 클라우드 서비스를 이용하여 24시간 운영할 수 있습니다.
- EC2 인스턴스 등에 코드 배포
- 시스템 서비스로 등록하여 자동 실행
- 도메인 연결 및 HTTPS 설정
5-3. Docker를 이용한 컨테이너화
Docker를 이용하여 컨테이너로 배포할 수도 있습니다. 다음은 간단한 Dockerfile 예시입니다.
FROM python:3.9-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
# 포트 설정
EXPOSE 8501
# 실행 명령
CMD ["streamlit", "run", "dashboard.py", "--server.port=8501", "--server.address=0.0.0.0"]
시리즈 전체 요약
이번 시리즈에서는
- 법령 정보를 크롤링하는 기본 기능을 구현했습니다.
- 데이터베이스를 연동하여 법령 정보와 변경 이력을 저장했습니다.
- 변경 감지 시 이메일로 알림을 보내는 기능을 만들었습니다.
- Streamlit을 이용해 간단한 대시보드를 구현했습니다.