안녕하세요.
해당 프로젝트의 마지막 포스팅입니다..!
이번 포스팅에서는 Flask를 이용해서
이전 포스팅에서 만들어 두었던
Crawling, Summarization 기능을 Web Service로 제공해보겠습니다.
1. Template
1. 프로젝트 상에서 video id와 채팅 수집 시간을 입력하는 View
2. 요약 내용을 보여주는 View
3. 요약 History를 보여주는 View
3가지의 View를 생성하겠습니다.
1-1. index.html
index.html 입니다.
페이지를 들어가면 가장 처음 노출되는 페이지입니다.
비디오 ID, 수집기간을 입력받습니다.
<!DOCTYPE html>
<html lang="ko">
<head>
<meta charset="UTF-8">
<title>실시간 유튜브 채팅 요약하기(한국어, 영어)</title>
</head>
<body>
<h1>실시간 유튜브 채팅 요약하기(한국어, 영어)</h1>
<form action="/start" method="post">
<div><label for="video_id">비디오 ID:</label>
<input type="text" id="video_id" name="video_id" required></div>
<div><span>* https://www.youtube.com/watch?v=비디오 ID << 를 입력해주세요!</span></div><br></br>
<div><label for="collect_time">수집 시간(sec):</label>
<input type="number" id="collect_time" name="collect_time" value="10" required></div>
<div><span>* 수집 시간 간격을 입력해주세요.</span></div>
<input type="submit" value="요약 시작">
</form>
</body>
</html>
해당 id로 입력된 video_id, collect_time 데이터는
Flask 구현 코드에서
@app.route('/start', methods=['POST'])의
데코레이터 이후의 함수에서 전달받아 사용할 수 있습니다.
@app.route('/start', methods=['POST'])
def start_summary():
# 폼 데이터 가져오기
video_id = request.form['video_id']
collect_time = int(request.form['collect_time'])
index.html 화면 구성
1-2. summary.html
비디오 정보,
수집한 채팅,
요약 정보,
긍정, 부정에 대한 정보
를 나타내는 페이지입니다.
<style></style> 태그 사이는 CSS 입니다. 해당 코드에서 html 구성의 스타일을 정해주었고,
<script></script> 태그 사이는 JavaScript로 html을 유동적으로 변경할 수 있게 해주었습니다.
script와 CSS로
긍정이 많은 요약이라면 파란색 화면으로 3번 깜빡이게 함.
등의 UI적인 기능을 추가할 수 있습니다.
<!DOCTYPE html>
<html lang="ko">
<head>
<meta charset="UTF-8">
<title>{{ video_title }} - 요약 결과</title>
<style>
/* 공통 스타일 */
body {
font-family: 'Arial', sans-serif;
margin: 0;
padding: 20px;
background-color: #f9f9f9; /* 기본 배경색 */
color: #333333; /* 기본 텍스트 색 */
}
.container {
max-width: 800px;
margin: 0 auto;
padding: 20px;
border-radius: 8px;
box-shadow: 0 2px 10px rgba(0, 0, 0, 0.1);
background-color: white;
}
h2 {
margin: 0;
padding-bottom: 10px;
}
/* 테마 별 스타일 */
.safari {
background-color: #f0f8e7;
color: #2c3e50;
}
/* 스크롤 가능한 채팅 스타일 */
.chat-scroll-container {
max-height: 200px; /* 최대 높이 설정 */
overflow-y: auto; /* 세로 스크롤바 활성화 */
border: 1px solid #ccc;
border-radius: 8px;
padding: 10px;
background-color: #f9f9f9;
margin-top: 20px;
}
/* 카드 스타일 */
.chat-card {
background-color: #f1f1f1;
border-radius: 8px;
padding: 5px;
margin-bottom: 5px;
box-shadow: 0 2px 5px rgba(0, 0, 0, 0.2);
}
.chat-author {
font-weight: bold;
color: #3498db;
}
.chat-time {
font-size: 0.8em;
color: #666;
margin-top: 5px;
}
/* 작은 원형 로더 스타일 */
.small-loader {
border: 8px solid #f3f3f3;
border-top: 8px solid #3498db;
border-radius: 50%;
width: 15px;
height: 15px;
animation: spin 2s linear infinite;
display: inline-block;
margin-left: 10px;
}
@keyframes spin {
0% { transform: rotate(0deg); }
100% { transform: rotate(360deg); }
}
/* 바 그래프 스타일 */
.bar-container {
display: none;
align-items: center;
margin-top: 20px;
height: 50px;
}
.bar {
position: relative;
height: 100%;
border-radius: 5px;
transition: width 1s ease;
}
.positive-bar {
background-color: #3498db;
display: flex;
justify-content: center;
align-items: center;
color: white;
font-weight: bold;
}
.negative-bar {
background-color: #e74c3c;
display: flex;
justify-content: center;
align-items: center;
color: white;
font-weight: bold;
}
/* 배경색 애니메이션 */
.positive-background {
animation: backgroundPulseBlue 2s infinite;
animation-iteration-count: 3; /* 3번 반복 */
}
.negative-background {
animation: backgroundPulseRed 2s infinite;
animation-iteration-count: 3; /* 3번 반복 */
}
@keyframes backgroundPulseBlue {
0% { background-color: white; }
50% { background-color: #3498db; }
100% { background-color: white; }
}
@keyframes backgroundPulseRed {
0% { background-color: white; }
50% { background-color: #e74c3c; }
100% { background-color: white; }
}
/* 로딩 메시지 스타일 */
#loading-message {
display: none;
justify-content: center;
align-items: center;
margin-top: 20px;
}
#loading-text {
font-size: 1em;
color: #333;
}
/* 버튼 스타일 */
button {
background-color: #3498db;
color: white;
padding: 10px 20px;
border: none;
border-radius: 5px;
cursor: pointer;
margin-top: 10px;
}
button:hover {
background-color: #2980b9;
}
</style>
</head>
<body class="safari">
<div class="container">
<div id="notice-container">
<div><span>유튜브 실시간 채팅 요약이 반복 진행됩니다.</span></div>
<div><span>채팅 수집 시간: {{ collect_time }}sec</span></div>
<div><span>* 요약 중지 - 요약을 멈춤 </span></div>
<div><span>* 히스토리 보기 - 요약 기록 확인 </span></div>
<div id="loader" class="small-loader"></div>
<button id="toggle-button" onclick="stopSummarization()">요약 중지</button>
</div>
<!-- 비디오 정보 표시 -->
<div id="video-info">
<h2>유튜브 라이브</h2>
<div id="video-title">유튜브 라이브 방송명: {{ video_title }}</div>
<div id="video-author">채널 명: {{ video_author }}</div>
<div id="video-published">라이브 시작일: {{ video_published }}</div>
</div>
<div class="clearfix">
<div id="chat-container">
<h2>채팅 내용</h2>
<div id="chat" class="chat-scroll-container"></div>
</div>
<div id="result-container">
<h2>요약 결과</h2>
<div id="summary-container">
<p id="summary">{{ summary_result }}</p>
</div>
<h2>민심</h2>
<!-- 긍정 및 부정 비율을 시각화하는 바 그래프 -->
<div id="bar-graph" class="bar-container">
<div class="bar positive-bar" id="positive-bar" style="width: 0%;">긍정: 0%</div>
<div class="bar negative-bar" id="negative-bar" style="width: 0%;">부정: 0%</div>
</div>
</div>
</div>
<div><a href="/history" target="_blank">히스토리 보기</a></div>
<div><a href="/" onclick="stopAndRetry();">다시 시도하기</a></div>
</div>
<script>
let isSummarizing = true;
const sessionId = "{{ session_id }}";
function hideLoadingMessage() {
document.getElementById('loading-message').style.display = 'none';
}
function updateSummary() {
fetch('/update_summary', {
method: 'POST',
headers: {'Content-Type': 'application/json'},
body: JSON.stringify({session_id: sessionId})
})
.then(response => response.json())
.then(data => {
document.getElementById('summary').innerText = data.summary;
// 긍정/부정 비율에 따른 배경색 변경
const positiveRatio = data.positive_ratio;
const negativeRatio = data.negative_ratio;
const summaryContainer = document.getElementById('summary-container');
if (positiveRatio > negativeRatio) {
summaryContainer.classList.add('positive-background');
summaryContainer.classList.remove('negative-background');
} else if (negativeRatio > positiveRatio) {
summaryContainer.classList.add('negative-background');
summaryContainer.classList.remove('positive-background');
} else {
summaryContainer.classList.remove('positive-background');
summaryContainer.classList.remove('negative-background');
}
// 바 그래프 업데이트
updateBarGraph(positiveRatio, negativeRatio);
});
}
function updateBarGraph(positiveRatio, negativeRatio) {
const total = positiveRatio + negativeRatio;
const positiveBar = document.getElementById('positive-bar');
const negativeBar = document.getElementById('negative-bar');
const barGraph = document.getElementById('bar-graph');
if (total === 0) {
barGraph.style.display = 'none';
} else {
barGraph.style.display = 'flex';
if (positiveRatio === 100) {
positiveBar.style.width = '100%';
positiveBar.innerText = `긍정: ${positiveRatio}%`;
negativeBar.style.display = 'none';
} else if (negativeRatio === 100) {
negativeBar.style.width = '100%';
negativeBar.innerText = `부정: ${negativeRatio}%`;
positiveBar.style.display = 'none';
} else {
const positiveWidth = (positiveRatio / total) * 100;
const negativeWidth = (negativeRatio / total) * 100;
positiveBar.style.width = positiveWidth + '%';
positiveBar.innerText = `긍정: ${positiveRatio}%`;
positiveBar.style.display = 'flex';
negativeBar.style.width = negativeWidth + '%';
negativeBar.innerText = `부정: ${negativeRatio}%`;
negativeBar.style.display = 'flex';
}
}
}
function updateChat() {
fetch('/get_chat', {
method: 'POST',
headers: {'Content-Type': 'application/json'},
body: JSON.stringify({session_id: sessionId})
})
.then(response => response.json())
.then(data => {
const chatContainer = document.getElementById('chat');
chatContainer.innerHTML = '';
const chatMessages = data.chat.split('\n');
chatMessages.slice(1).forEach((message) => {
if (message.trim()) {
const [author, content, timestamp] = message.split(',');
if (author && content && timestamp) {
const card = `<div class="chat-card"><span class="chat-author">${author}</span><p>${content}</p><span class="chat-time">${timestamp}</span></div>`;
chatContainer.innerHTML += card;
}
}
});
});
}
function updateVideoInfo() {
fetch('/update_video_info', {
method: 'POST',
headers: {'Content-Type': 'application/json'},
body: JSON.stringify({session_id: sessionId})
})
.then(response => response.json())
.then(data => {
document.getElementById('video-title').innerText = '유튜브 라이브 방송: ' + data.video_title;
document.getElementById('video-author').innerText = '채널 명: ' + data.video_author;
document.getElementById('video-published').innerText = '라이브 시작일: ' + data.video_published;
});
}
function stopSummarization() {
if (isSummarizing) {
fetch('/stop', {
method: 'POST',
headers: {'Content-Type': 'application/json'},
body: JSON.stringify({session_id: sessionId})
})
.then(response => response.json())
.then(data => {
isSummarizing = false;
document.getElementById('loader').style.display = 'none';
document.getElementById('toggle-button').innerText = '요약 시작';
document.getElementById('toggle-button').onclick = startSummarization;
});
}
}
function startSummarization() {
if (!isSummarizing) {
fetch('/resume', {
method: 'POST',
headers: {'Content-Type': 'application/json'},
body: JSON.stringify({session_id: sessionId})
})
.then(response => response.json())
.then(data => {
isSummarizing = true;
document.getElementById('loader').style.display = 'inline-block';
document.getElementById('toggle-button').innerText = '요약 중지';
document.getElementById('toggle-button').onclick = stopSummarization;
startInterval();
});
}
}
function stopAndRetry() {
stopSummarization()
}
function checkSummarizingStatus() {
fetch('/is_summarizing', {
method: 'POST',
headers: {'Content-Type': 'application/json'},
body: JSON.stringify({session_id: sessionId})
})
.then(response => response.json())
.then(data => {
isSummarizing = data.is_summarizing;
const loadingMessage = document.getElementById('loading-message');
const barGraph = document.getElementById('bar-graph');
if (!isSummarizing) {
document.getElementById('loader').style.display = 'none';
document.getElementById('toggle-button').innerText = '요약 시작';
document.getElementById('toggle-button').onclick = startSummarization;
loadingMessage.style.display = 'flex'; // 로딩 메시지 표시
barGraph.style.display = 'none'; // 바 그래프 숨기기
} else {
document.getElementById('loader').style.display = 'inline-block';
document.getElementById('toggle-button').innerText = '요약 중지';
document.getElementById('toggle-button').onclick = stopSummarization;
loadingMessage.style.display = 'none'; // 로딩 메시지 숨기기
barGraph.style.display = 'flex'; // 바 그래프 표시
}
});
}
let interval;
function startInterval() {
interval = setInterval(() => {
if (isSummarizing) {
updateChat();
updateSummary();
updateVideoInfo();
checkSummarizingStatus();
} else {
clearInterval(interval);
}
}, 1000);
}
window.onload = () => {
updateChat();
updateSummary();
updateVideoInfo();
startInterval();
}
</script>
</body>
</html>
summary.html 화면 구성
1-3. history.html
summary.html 페이지에서 "요약 히스토리" <a>태그 하이퍼링크를 클릭하면
보여지는 페이지입니다.요약 할때마다 저장하는 요약 내용을 카드 형식으로 보여주는 스타일로 구성하였습니다.
<!DOCTYPE html>
<html lang="ko">
<head>
<meta charset="UTF-8">
<title>요약 히스토리</title>
<style>
/* 공통 스타일 */
body {
font-family: 'Arial', sans-serif;
margin: 0;
padding: 20px;
background-color: #f9f9f9; /* 기본 배경색 */
color: #333333; /* 기본 텍스트 색 */
}
.container {
max-width: 800px;
margin: 0 auto;
}
h1 {
text-align: center;
margin-bottom: 30px;
}
/* 카드 스타일 */
.summary-card {
background-color: #fff;
border-radius: 8px;
padding: 20px;
margin-bottom: 20px;
box-shadow: 0 2px 10px rgba(0, 0, 0, 0.1);
}
.summary-content {
line-height: 1.6;
}
.summary-number {
font-weight: bold;
margin-bottom: 10px;
font-size: 1.2em;
color: #3498db;
}
/* 긍정/부정 비율 스타일 */
.sentiment {
margin-top: 10px;
font-weight: bold;
}
.positive {
color: #27ae60;
}
.negative {
color: #e74c3c;
}
</style>
</head>
<body>
<div class="container">
<h1>요약 히스토리</h1>
{% for summary in history %}
<div class="summary-card">
<div class="summary-content">
{% for line in summary.split('\n') %}
<p>{{ line }}</p>
{% endfor %}
</div>
</div>
{% endfor %}
</div>
</body>
</html>
history.html 화면 구성
2. Flask
Flask로 최종적인 서비스 동작을 구현했습니다.
제가 구현한 메서드들에 대한 설명을 위주로 진행하겠습니다.
pip 에서 Flask를 설치해야 Flask를 사용할 수 있습니다.
pip install flask
index() 메서드를 생성해주었습니다.
'/'이라는 URL이 온다면 index.html을 렌더링을 진행합니다.
(다시 시도하기 <a>태그에서 href를 "/"로 두면 index.html 페이지가 노출됨.)
@app.route('/')
def index():
return render_template('index.html')
start_summary() 메서드입니다.
'/start' 라는 URL이 온다면 실행이 되며,
video_id, collect_time을 POST request로 받을 수 있습니다.
threading 라이브러리를 사용해서 Thread를 생성해서 run_summarizer()라는
메서드를 실행해줍니다.
멀티스레딩을 해준이유: 일단 다중 접속자가 있다고 가정하고, 크롤링->요약 작업을 하는 동안에
다른 사용자의 동작도 진행하고 싶었음.
(그러나, 사용자가 1명 이상이면 GPU RAM이 터짐...
최대한 서비스에 초점을 두고 코딩했습니다..)
session_id는 각 사용자의 id라고 생각하시면 될 것 같습니다.!
@app.route('/start', methods=['POST'])
def start_summary():
# 폼 데이터 가져오기
video_id = request.form['video_id']
collect_time = int(request.form['collect_time'])
# 세션 ID 생성
session_id = str(uuid.uuid4())
# 입력 프롬프트 생성 (크롤링 완료 후 데이터를 기반으로 생성)
prompt_template = """
{comments}
위의 유튜브 라이브 채팅 댓글을 읽고, 주제와 대화의 흐름을 파악하여 아래의 형식으로 부드럽고 친근한 어조로 존댓말로 요약해줘:
1. 첫 번째 요약 내용
2. 두 번째 요약 내용
3. 세 번째 요약 내용
요약 예시:
1. 시청자들은 스트리밍 중 게스트 등장에 신기해 합니다.
2. 시청자들은 스트리머의 시간을 끌고 고민하는 모습이 지루합니다.
3. 시청자들은 게스트들의 대화 내용이 너무 재미있습니다.
긍정:XX/부정:XX
- 위의 형식에 맞춰 요약 내용을 작성해 주고, 흐름을 이해할 수 있을 정도의 구체적인 키워드를 포함해줘.
- 요약 시 부드럽고 자연스러운 표현을 사용해줘.
- 긍정과 부정의 비율은 전체 100%를 기준으로 계산해줘.
- 다른 불필요한 말은 생략하고, 위의 형식으로만 출력해줘.
"""
# 세션별 파라미터 저장
session_params[session_id] = {
'video_id': video_id,
'collect_time': collect_time,
'prompt_template': prompt_template
}
# 요약 작업을 별도의 스레드에서 실행 (크롤링 완료 후 요약)
summarizer_thread = threading.Thread(
target=run_summarizer,
args=(session_id,)
)
summarizer_thread.start()
# 결과 페이지로 리디렉션 (session_id와 collect_time을 URL 파라미터로 전달)
return redirect(url_for('summary', session_id=session_id, collect_time=collect_time))
여기서 redirect된 summary는 아래의 Flask 코드에서 전달 받게 됩니다.
요약이 완료된 데이터는 summary.html로 전달되고 페이지 렌더링이 다시 진행됩니다.
@app.route('/summary')
def summary():
session_id = request.args.get('session_id')
collect_time = request.args.get('collect_time')
with data_lock:
video_info = video_info_dict.get(session_id, {
'video_title': ' ',
'video_author': ' ',
'video_published': ' '
})
summary_result = result_dict.get(session_id, {}).get('summary', '')
return render_template('summary.html',
session_id=session_id,
collect_time=collect_time,
video_title=video_info['video_title'],
video_author=video_info['video_author'],
video_published=video_info['video_published'],
summary_result=summary_result
)
위의 start_summary()에서 Thread에 동작하도록 입력한 메서드입니다.
크롤링 후에 요약을 진행하는 코드로 구성됩니다.
def run_summarizer(session_id):
try:
with status_lock:
summarizing_status[session_id] = True
# 세션별 파라미터 가져오기
params = session_params.get(session_id)
if not params:
print("세션 파라미터를 찾을 수 없습니다.")
return
video_id = params['video_id']
collect_time = params['collect_time']
prompt_template = params['prompt_template']
# 채팅 파일 경로 설정
chat_file_path = f'./data/{video_id}_chat.csv'
while summarizing_status.get(session_id, False):
# 채팅 크롤링 및 비디오 정보 가져오기
print("=== 채팅 크롤링 시작 ===")
crawler = Chat_Crawler(
collect_time=collect_time,
youtube_api_key=api_key,
video_id=video_id
)
# 비디오 정보는 한 번만 가져옴
if session_id not in video_info_dict:
video_info = crawler.get_video()
video_info_dict[session_id] = {
"video_title": video_info.title,
"video_author": video_info.author,
"video_published": video_info.published
}
else:
video_info = video_info_dict[session_id]
# 채팅 크롤링 수행
crawler.do_crawling()
print("=== 채팅 크롤링 완료 ===")
# 크롤링한 채팅 데이터를 새로 읽어서 세션별로 저장
with open(chat_file_path, 'r', encoding='utf-8') as f:
chat_content = f.read()
# 각 세션별로 크롤링된 채팅 데이터를 저장
with data_lock:
chat_contents[session_id] = chat_content
result_dict[session_id] = {"summary": ""}
# 최신 크롤링된 데이터를 기반으로 입력 프롬프트 생성
prompt = prompt_template.format(comments=chat_content)
# 요약 결과 생성
print("=== 요약 생성 시작 ===")
def should_stop():
with status_lock:
return not summarizing_status.get(session_id, False)
summary_result, positive_ratio, negative_ratio = summarizer.summarize(prompt, should_stop=should_stop)
print("=== 요약 생성 완료 ===")
# 요약 결과 저장
with data_lock:
result_dict[session_id]["summary"] = summary_result
result_dict[session_id]["positive_ratio"] = positive_ratio
result_dict[session_id]["negative_ratio"] = negative_ratio
history.append(summary_result)
# 상태 확인 후 대기 (다음 크롤링 및 요약 작업까지 대기)
print(f"Waiting {collect_time} seconds before next iteration.")
wait_time = 0
while wait_time < collect_time:
with status_lock:
if not summarizing_status.get(session_id, False):
print("Summarization stopped.")
return # 중지 명령을 받으면 함수를 종료
time.sleep(1)
wait_time += 1
except Exception as e:
print(f"에러 발생: {str(e)}")
with data_lock:
result_dict[session_id] = {"summary": f"에러 발생: {str(e)}"}
finally:
with status_lock:
summarizing_status[session_id] = False
update_video_info() 메서드는
스트리밍 정보를 받아와서 json형태로 html에 전달해줍니다.
@app.route('/update_video_info', methods=['POST'])
def update_video_info():
session_id = request.json.get('session_id')
with data_lock:
video_info = video_info_dict.get(session_id, {})
return jsonify(video_info)
/stop 이라는 URL이 오면
해당 사용자의 동작을 중지하고
UI갱신을 위해 html로 jsonify 상태 전달해줍니다.
@app.route('/stop', methods=['POST'])
def stop_summary():
session_id = request.json.get('session_id')
with status_lock:
if session_id in summarizing_status:
summarizing_status[session_id] = False
return jsonify({'status': 'stopped'})
중지와 마찬가지로 작업 재개에 대한 코드입니다.
이미 실행 중인지 체크하고
실행 중이지 않다면 Thread를 다시 실행하도록 작업(메서드)을 넣어줍니다.
status_lock은 요약 스레드간의 충돌을 막기 위한 lock Flag입니다.
@app.route('/resume', methods=['POST'])
def resume_summary():
session_id = request.json.get('session_id')
with status_lock:
if summarizing_status.get(session_id, False):
return jsonify({'status': 'already running'})
else:
summarizing_status[session_id] = True
# 요약 작업을 별도의 스레드에서 실행
summarizer_thread = threading.Thread(
target=run_summarizer,
args=(session_id,)
)
summarizer_thread.start()
return jsonify({'status': 'resumed'})
요약 정보를 가져오고 Update를 진행하는 메서드입니다.
긍정/부정에 대한 비율 정보도 따로 처리해서 반환해주었습니다.
data_lock은 스레드간의 공유 데이터를 보호하기 위한 lock Flag입니다.
@app.route('/update_summary', methods=['POST'])
def update_summary():
session_id = request.json.get('session_id')
with data_lock:
summary_data = result_dict.get(session_id, {})
summary_result = summary_data.get("summary", "요약 중....")
# 긍정/부정 비율도 반환
positive_ratio = summary_data.get("positive_ratio", 50) # 기본값 50
negative_ratio = summary_data.get("negative_ratio", 50) # 기본값 50
return jsonify({
'summary': summary_result,
'positive_ratio': positive_ratio,
'negative_ratio': negative_ratio
채팅 데이터에 대한 갱신을 담당하는 메서드입니다.
chat 데이터를 html에 전달합니다.
@app.route('/get_chat', methods=['POST'])
def get_chat():
session_id = request.json.get('session_id')
with data_lock:
chat_content = chat_contents.get(session_id, '')
return jsonify({'chat': chat_content})
히스토리에 대한 처리를 담당하는 메서드입니다.
history.html의 렌더링을 진행합니다.
@app.route('/history')
def view_history():
with data_lock:
return render_template('history.html', history=history)
Flask 전체 코드입니다.
from flask import Flask, render_template, request, redirect, url_for, jsonify
from dotenv import load_dotenv
import os
import threading
from chat_crawler import Chat_Crawler # 채팅 크롤링 클래스
from summarizer import CommentSummarizer # 요약 클래스
import time
import uuid
app = Flask(__name__)
app.secret_key = 'your_secret_key'
# .env 파일에서 API 키 로드
load_dotenv()
api_key = os.getenv('YOUTUBE_API_KEY')
# 모델을 애플리케이션 시작 시 전역 변수로 로드
summarizer = CommentSummarizer()
# 공유 데이터와 락 초기화
summarizing_status = {}
result_dict = {}
chat_contents = {}
video_info_dict = {} # 비디오 정보 저장
session_params = {} # 세션별 파라미터 저장
history = []
status_lock = threading.Lock()
data_lock = threading.Lock()
chat_file_path = ""
def load_chat(chat_file_path):
if chat_file_path:
with open(chat_file_path, 'r', encoding='utf-8') as f:
comments = f.read()
return comments
else:
return ""
def run_summarizer(session_id):
try:
with status_lock:
summarizing_status[session_id] = True
# 세션별 파라미터 가져오기
params = session_params.get(session_id)
if not params:
print("세션 파라미터를 찾을 수 없습니다.")
return
video_id = params['video_id']
collect_time = params['collect_time']
prompt_template = params['prompt_template']
# 채팅 파일 경로 설정
chat_file_path = f'./data/{video_id}_chat.csv'
while summarizing_status.get(session_id, False):
# 채팅 크롤링 및 비디오 정보 가져오기
print("=== 채팅 크롤링 시작 ===")
crawler = Chat_Crawler(
collect_time=collect_time,
youtube_api_key=api_key,
video_id=video_id
)
# 비디오 정보는 한 번만 가져옴
if session_id not in video_info_dict:
video_info = crawler.get_video()
video_info_dict[session_id] = {
"video_title": video_info.title,
"video_author": video_info.author,
"video_published": video_info.published
}
else:
video_info = video_info_dict[session_id]
# 채팅 크롤링 수행
crawler.do_crawling()
print("=== 채팅 크롤링 완료 ===")
# 크롤링한 채팅 데이터를 새로 읽어서 세션별로 저장
with open(chat_file_path, 'r', encoding='utf-8') as f:
chat_content = f.read()
# 각 세션별로 크롤링된 채팅 데이터를 저장
with data_lock:
chat_contents[session_id] = chat_content
result_dict[session_id] = {"summary": "요약 중..."}
# 최신 크롤링된 데이터를 기반으로 입력 프롬프트 생성
prompt = prompt_template.format(comments=chat_content)
# 요약 결과 생성
print("=== 요약 생성 시작 ===")
def should_stop():
with status_lock:
return not summarizing_status.get(session_id, False)
summary_result, positive_ratio, negative_ratio = summarizer.summarize(prompt, should_stop=should_stop)
print("=== 요약 생성 완료 ===")
# 요약 결과 저장
with data_lock:
result_dict[session_id]["summary"] = summary_result
result_dict[session_id]["positive_ratio"] = positive_ratio
result_dict[session_id]["negative_ratio"] = negative_ratio
history.append(summary_result)
# 상태 확인 후 대기 (다음 크롤링 및 요약 작업까지 대기)
print(f"Waiting {collect_time} seconds before next iteration.")
wait_time = 0
while wait_time < collect_time:
with status_lock:
if not summarizing_status.get(session_id, False):
print("Summarization stopped.")
return # 중지 명령을 받으면 함수를 종료
time.sleep(1)
wait_time += 1
except Exception as e:
print(f"에러 발생: {str(e)}")
with data_lock:
result_dict[session_id] = {"summary": f"에러 발생: {str(e)}"}
finally:
with status_lock:
summarizing_status[session_id] = False
@app.route('/')
def index():
return render_template('index.html')
@app.route('/start', methods=['POST'])
def start_summary():
# 폼 데이터 가져오기
video_id = request.form['video_id']
collect_time = int(request.form['collect_time'])
# 세션 ID 생성
session_id = str(uuid.uuid4())
# 입력 프롬프트 생성 (크롤링 완료 후 데이터를 기반으로 생성)
prompt_template = """
{comments}
위의 유튜브 라이브 채팅 댓글을 읽고, 주제와 대화의 흐름을 파악하여 아래의 형식으로 부드럽고 친근한 어조로 존댓말로 요약해줘:
1. 첫 번째 요약 내용
2. 두 번째 요약 내용
3. 세 번째 요약 내용
요약 예시:
1. 시청자들은 스트리밍 중 게스트 등장에 신기해 합니다.
2. 시청자들은 스트리머의 시간을 끌고 고민하는 모습이 지루합니다.
3. 시청자들은 게스트들의 대화 내용이 너무 재미있습니다.
긍정:XX/부정:XX
- 위의 형식에 맞춰 요약 내용을 작성해 주고, 흐름을 이해할 수 있을 정도의 구체적인 키워드를 포함해줘.
- 요약 시 부드럽고 자연스러운 표현을 사용해줘.
- 긍정과 부정의 비율은 전체 100%를 기준으로 계산해줘.
- 다른 불필요한 말은 생략하고, 위의 형식으로만 출력해줘.
"""
# 세션별 파라미터 저장
session_params[session_id] = {
'video_id': video_id,
'collect_time': collect_time,
'prompt_template': prompt_template
}
# 요약 작업을 별도의 스레드에서 실행 (크롤링 완료 후 요약)
summarizer_thread = threading.Thread(
target=run_summarizer,
args=(session_id,)
)
summarizer_thread.start()
# 결과 페이지로 리디렉션 (session_id와 collect_time을 URL 파라미터로 전달)
return redirect(url_for('summary', session_id=session_id, collect_time=collect_time))
@app.route('/summary')
def summary():
session_id = request.args.get('session_id')
collect_time = request.args.get('collect_time')
with data_lock:
video_info = video_info_dict.get(session_id, {
'video_title': ' ',
'video_author': ' ',
'video_published': ' '
})
summary_result = result_dict.get(session_id, {}).get('summary', '')
return render_template('summary.html',
session_id=session_id,
collect_time=collect_time,
video_title=video_info['video_title'],
video_author=video_info['video_author'],
video_published=video_info['video_published'],
summary_result=summary_result
)
@app.route('/update_video_info', methods=['POST'])
def update_video_info():
session_id = request.json.get('session_id')
with data_lock:
video_info = video_info_dict.get(session_id, {})
return jsonify(video_info)
@app.route('/stop', methods=['POST'])
def stop_summary():
session_id = request.json.get('session_id')
with status_lock:
if session_id in summarizing_status:
summarizing_status[session_id] = False
return jsonify({'status': 'stopped'})
@app.route('/resume', methods=['POST'])
def resume_summary():
session_id = request.json.get('session_id')
with status_lock:
if summarizing_status.get(session_id, False):
return jsonify({'status': 'already running'})
else:
summarizing_status[session_id] = True
# 요약 작업을 별도의 스레드에서 실행
summarizer_thread = threading.Thread(
target=run_summarizer,
args=(session_id,)
)
summarizer_thread.start()
return jsonify({'status': 'resumed'})
@app.route('/update_summary', methods=['POST'])
def update_summary():
session_id = request.json.get('session_id')
with data_lock:
summary_data = result_dict.get(session_id, {})
summary_result = summary_data.get("summary", "요약 중....")
# 긍정/부정 비율도 반환
positive_ratio = summary_data.get("positive_ratio", 50) # 기본값 50
negative_ratio = summary_data.get("negative_ratio", 50) # 기본값 50
return jsonify({
'summary': summary_result,
'positive_ratio': positive_ratio,
'negative_ratio': negative_ratio
})
@app.route('/get_chat', methods=['POST'])
def get_chat():
session_id = request.json.get('session_id')
with data_lock:
chat_content = chat_contents.get(session_id, '')
return jsonify({'chat': chat_content})
@app.route('/is_summarizing', methods=['POST'])
def is_summarizing():
session_id = request.json.get('session_id')
with status_lock:
is_summarizing = summarizing_status.get(session_id, False)
return jsonify({'is_summarizing': is_summarizing})
@app.route('/history')
def view_history():
with data_lock:
return render_template('history.html', history=history)
if __name__ == "__main__":
# 외부 접속 가능하게 설정
app.run(host='0.0.0.0', port=5000, debug=False, use_reloader=False)
결과
최종적인 기능을 정리해보면 아래와 같습니다.
1. 유튜브 라이브 채팅의 비디오 ID를 입력하고, 채팅 수집 시간을 설정함.
2. 요약은 계속해서 반복됨
3.채팅 결과와 요약 결과가 표시되며, 그래프를 통해 채팅 분위기에 따른 긍정/부정 감정이 시각화됨.
4.감정이 긍정적일 경우, 요약 결과가 파란색으로 깜빡이고, 부정적일 경우 빨간색으로 깜빡임.
5. "히스토리 보기"를 클릭하여 요약 기록을 확인할 수 있음.
아직은 프로토타입 정도의 프로젝트지만,
계속 보완을 하면 유튜브 채팅 분석 프로그램으로 발전도 가능하겠다고 생각이 들었습니다!
시간이 있을때 계속 보완해 나가겠습니다.
해당 프로젝트는 Git에 올려두었습니다.
참고하실 분은 참고해주세요.
github :
https://github.com/AIKONG2024/Real-time-YouTube-Live-Chat-Summarization
GitHub - AIKONG2024/Real-time-YouTube-Live-Chat-Summarization: YouTube Live Chat Summarization with Gemma
YouTube Live Chat Summarization with Gemma. Contribute to AIKONG2024/Real-time-YouTube-Live-Chat-Summarization development by creating an account on GitHub.
github.com
#GemmaSprint
'인공지능 개발하기 > AI Projects' 카테고리의 다른 글
| [Gemma2 모델을 이용한 유튜브 실시간 채팅 요약 프로젝트] 2) Chat Summarization with LLM (0) | 2024.10.02 |
|---|---|
| [Gemma2 모델을 이용한 유튜브 실시간 채팅 요약 프로젝트] 1) YouTube Live Chat Crawling (2) | 2024.10.02 |
| [YOLOv5 프로젝트] 특정 사람 얼굴 인식하기(2) (1) | 2024.03.30 |
| [YOLOv5 프로젝트] 특정 사람 얼굴 인식하기(1) (3) | 2024.03.29 |
