datascience / predict_module.py
yearye's picture
Update predict_module.py
9fcfedf verified
import pandas as pd
import numpy as np
import requests
import cv2
import mediapipe as mp
import torch
from PIL import Image
from io import BytesIO
from joblib import load
from transformers import AutoTokenizer, AutoModelForSequenceClassification
from youtube_transcript_api import YouTubeTranscriptApi
# ๋ชจ๋ธ ๋ฐ ๊ธฐํƒ€ ํŒŒ์ผ ๋กœ๋“œ
model = load('view_predictor.joblib')
_, _, le_cat = load('label_encoders.joblib')
feature_cols = load('features.joblib')
# ๊ฐ์„ฑ ๋ถ„์„ ๋ชจ๋ธ
senti_model_name = "nlp04/korean_sentiment_analysis_kcelectra"
senti_tokenizer = AutoTokenizer.from_pretrained(senti_model_name)
senti_model = AutoModelForSequenceClassification.from_pretrained(senti_model_name)
senti_model.eval()
def sentiment_score(text):
if not text or pd.isna(text):
return 0.0
with torch.no_grad():
inputs = senti_tokenizer(text, return_tensors="pt", truncation=True)
outputs = senti_model(**inputs)
probs = torch.softmax(outputs.logits, dim=1).squeeze()
try:
return round(float(probs[2]) * 100, 1) # Positive
except IndexError:
return round(float(probs[1]) * 100, 1)
category_dict = {
'์Œ์‹': ['์ฏ”์–‘', '์ฐจ๋ฐฅ์—ด๋ผ', '๋จน๋ฐฉ', '๋‹จ๊ณจ', '์•„์นจ', '์žฅ์‚ฌ', '๋งŒ๋“ค๊ธฐ', '์นผ๋กœ๋ฆฌ', '๋ฒ ์ด๊ธ€', '๊ณฑ์ฐฝ', '์Šคํ…Œ์ดํฌ', '๊ณ ๊ธฐ',
'์‚ผ๊ฒน์‚ด', '์„ฑ์‹ฌ๋‹น', 'ํŽธ์˜์ ', '์ด์˜์ž', '๋ผ๋ฉด', '๊น€๋ฐฅ', '์น˜ํ‚จ', '๋ง›์ง‘', '์ง‘๋ฐฅ', '๋–ก๋ณถ์ด', '์Œ์‹', '๊น€์น˜',
'๊ด‘์–ด', '๋งŒ๋‘', '๋ƒ‰๋ฉด', '์ฒ ํŒ', '๋ผ์ง€', '์š”๋ฆฌ', '๊ฐ„์‹', 'ํšŒ์‹', '์ˆ ์ž๋ฆฌ', '๋ ˆ์‹œํ”ผ', '๊น€์น˜์ฐŒ๊ฐœ'],
'์—ฐ์˜ˆ/์œ ๋ช…์ธ': ['์ตœํ™”์ •', '์ดํ•ด๋ฆฌ', '๊ฐœ๊ทธ๋งจ', '๊ฐ•๋ฏผ๊ฒฝ', '๋‹ค๋น„์น˜', '์ด์ง€ํ˜œ', '์—ฌ์ž', '์•„์ด๋Œ', '๋‹ค๋‚˜์นด', '์ œ๋‹ˆ',
'์œ ์žฌ์„', 'ํ•‘๊ณ„๊ณ ', '์กฐ์„ธํ˜ธ', '์žฅ์˜๋ž€', '๊น€๊ตฌ๋ผ', '๊น€์˜์ฒ ', '์—ฐ์˜ˆ์ธ', '๋ฐฐ์šฐ', '์Šคํƒ€', '์ถœ์—ฐ', '์„ญ์™ธ',
'๊ฐ€์ˆ˜', '๋…ธ๋ž˜', '์ฝ˜์„œํŠธ', '์ด์Šน์ฒ '],
'๊ต์œก/๊ณต๋ถ€': ['์ผ์ฐจ๋ฐฉ์ •์‹', '์ด์ฐจ๋ฐฉ์ •์‹', '๋‹ฎ์Œ', '์ธ์ˆ˜๋ถ„ํ•ด', '์ง€์ˆ˜', '๋งž์ถค๋ฒ•', 'ํ•œ๊ตญ์‚ฌ', '๊ณผํ•™', '๊ณผ์™ธ', '์ˆ˜ํ•™',
'์ˆ˜์—…', '๊ณต๋ถ€', '์—ญ์‚ฌ', '๊ณต๋ถ€์™•', '์ˆ˜๋Šฅ', 'ํ€ด์ฆˆ', '์Šคํ„ฐ๋””', '์„ ์ƒ๋‹˜', '์‹œํ—˜', '์ง€์‹', '๋ฌธ์ œ',
'์ผ์ฐจํ•จ์ˆ˜', '์ด์ฐจํ•จ์ˆ˜', '๋ฐฉ์ •์‹', '๊ฒ€์ •๊ณ ์‹œ', '์˜์–ด', '๊ตญ์–ด', 'ํ•œ๊ตญ์–ด', '์„œ์šธ๋Œ€'],
'์—ฌํ–‰/์žฅ์†Œ': ['๋‘๋ฐ”์ด', 'ํœด๊ฐ€', '์ „๊ตญ', '์—ฌํ–‰', 'ํˆฌ์–ด', '์„ธ๊ณ„', '์ง€ํ•˜์ฒ ', 'ํ•œ๊ฐ•', '์นดํŽ˜', '์ฝ”์Šค', 'ํ•˜์™€์ด',
'๋„์ฟ„', '๋ชฝ๊ณจ', '์ผ๋ณธ', '์˜ค์‚ฌ์นด', '์ œ์ฃผ', '์ „์ฃผ', '์ œ์ฃผ๋„', '์„œ์šธ', '๋ฏผ๋ฐ•', '๋ฏธ๊ตญ', '๋Œ€๋งŒ',
'ํŒŒ๋ฆฌ', '์ŠคํŽ˜์ธ', '์šธ๋ฆ‰๋„', 'ํ™์ฝฉ'],
'์ผ์ƒ/๊ฐ€์กฑ': ['๊ฐ€์กฑ', '์—„๋งˆ', '์•„๋น ', '๋‚จํŽธ', '์ž์‹', '๋ชจ๋…€', 'ํ˜ผ์ž', 'ํ•˜๋ฃจ', '์ผ์ƒ', '์‚ฌ๋žŒ', '์•„์ด', '๊ณต์œ ',
'ํ˜„์žฅ', '๋ถ€๋ถ€', '๊ฐ€์žฅ', '์–ด๋จธ๋‹ˆ', '์กฐ์นด', '๊ฐ€์„', '์•„๋“ค', '๊ฒฐํ˜ผ์‹'],
'์ฝ˜ํ…์ธ /์œ ํŠœ๋ธŒ': ['์˜ˆ๋Šฅ', '์‹œ์ฆŒ', '๋ฆฌ๋ทฐ', '๋ผ์ด๋ธŒ', '๋ฐฉ์†ก', '์˜์ƒ', '์ฑ„๋„', '๊ฒŒ์ž„', '์œ ํŠœ๋ธŒ', '์ƒ๋ฐฉ์†ก',
'์ดฌ์˜', '์ฝ˜ํ…์ธ ', '๋Œ“๊ธ€', '์‡ผํ•‘'],
'์ •์น˜': ['๋Œ€์„ ', '๊ณต์•ฝ', '์•ˆ์ฒ ์ˆ˜', '๊ตญํšŒ', '์ •์น˜', '๋Œ€ํ†ต๋ น', '์„ ๊ฑฐ', '์ •๋‹น', '์˜์›'],
'๊ฒฝ์ œ': ['์ฃผ์‹', '๋น„ํŠธ์ฝ”์ธ', '์ฝ”์ธ', '์„ ๋ฌผ', '๋ถ€์ž', 'ํˆฌ์ž', '๊ฒฝ์ œ', '๊ธˆ์œต', '๊ด‘๊ณ ', '๋Œ€์ถœ', '์€ํ–‰', '์‹œ์žฅ'],
'๊ฑด๊ฐ•/์šด๋™': ['์šด๋™', '๊ฑด๊ฐ•', '๋‹ค์ด์–ดํŠธ', 'ํ—ฌ์Šค', '์ŠคํŠธ๋ ˆ์นญ', '์š”๊ฐ€', '์ฒด๋ ฅ', 'ํ”ผํŠธ๋‹ˆ์Šค', '๋‹ฌ๋ฆฌ๊ธฐ', '๊ทผ๋ ฅ', '์‹๋‹จ'],
'์ธ๊ฐ„๊ด€๊ณ„/๊ณ ๋ฏผ': ['์—ฐ์• ', '๊ณ ๋ฐฑ', '์†Œ๊ฐœํŒ…', '๋ฐ์ดํŠธ', '์†”๋กœ', '๊ณ ๋ฏผ']
}
# ์ œ๋ชฉ ํ‚ค์›Œ๋“œ ๊ธฐ๋ฐ˜ ๋ถ„๋ฅ˜ ํ•จ์ˆ˜
def classify_by_keywords(title, keyword_dict):
for category, keywords in keyword_dict.items():
for keyword in keywords:
if keyword in title:
return category
return None
# ์œ ํŠœ๋ธŒ ์นดํ…Œ๊ณ ๋ฆฌ + ํ‚ค์›Œ๋“œ ๊ธฐ๋ฐ˜์œผ๋กœ ์‚ฌ์šฉ์ž ์นดํ…Œ๊ณ ๋ฆฌ ๋ถ„๋ฅ˜
def map_category(category_id, title, api_key):
# ์œ ํŠœ๋ธŒ ์นดํ…Œ๊ณ ๋ฆฌ ์ด๋ฆ„ ๊ฐ€์ ธ์˜ค๊ธฐ
url = f'https://www.googleapis.com/youtube/v3/videoCategories?part=snippet&id={category_id}&regionCode=KR&key={api_key}'
try:
res = requests.get(url).json()
yt_category = res['items'][0]['snippet']['title']
except:
yt_category = "๊ธฐํƒ€"
# ์œ ํŠœ๋ธŒ ์นดํ…Œ๊ณ ๋ฆฌ๋ช… โ†’ ์‚ฌ์šฉ์ž ์นดํ…Œ๊ณ ๋ฆฌ ๋งคํ•‘
category_map = {
"์˜ํ™”/์• ๋‹ˆ๋ฉ”์ด์…˜": "์ฝ˜ํ…์ธ /์œ ํŠœ๋ธŒ",
"์Œ์•…": "์—ฐ์˜ˆ/์œ ๋ช…์ธ",
"์—”ํ„ฐํ…Œ์ธ๋จผํŠธ": "์ฝ˜ํ…์ธ /์œ ํŠœ๋ธŒ",
"์ฝ”๋ฏธ๋””": "์ฝ˜ํ…์ธ /์œ ํŠœ๋ธŒ",
"์ธ๋ฌผ/๋ธ”๋กœ๊ทธ": "์—ฐ์˜ˆ/์œ ๋ช…์ธ",
"๊ฒŒ์ž„": "์ฝ˜ํ…์ธ /์œ ํŠœ๋ธŒ",
"๋…ธํ•˜์šฐ/์Šคํƒ€์ผ": "์ผ์ƒ/๊ฐ€์กฑ",
"๋‰ด์Šค/์ •์น˜": "์ •์น˜",
"๊ต์œก": "๊ต์œก/๊ณต๋ถ€",
"๊ณผํ•™/๊ธฐ์ˆ ": "๊ต์œก/๊ณต๋ถ€",
"์Šคํฌ์ธ ": "๊ฑด๊ฐ•/์šด๋™",
"์ž๋™์ฐจ": "๊ธฐํƒ€",
"๋™๋ฌผ": "๊ธฐํƒ€",
"์—ฌํ–‰": "์—ฌํ–‰/์žฅ์†Œ"
}
mapped_category = category_map.get(yt_category, None)
# ํ‚ค์›Œ๋“œ ๊ธฐ๋ฐ˜ ๋ณด์™„ ๋ถ„๋ฅ˜
keyword_category = classify_by_keywords(title, category_dict)
# ์ตœ์ข… ์šฐ์„ ์ˆœ์œ„ ์ ์šฉ
return keyword_category or mapped_category or "๊ธฐํƒ€"
def hue_to_color_group(hue_value):
if 0 <= hue_value < 15 or hue_value >= 345:
return "๋นจ๊ฐ• ๊ณ„์—ด"
elif 15 <= hue_value < 45:
return "์ฃผํ™ฉ/๋…ธ๋ž‘ ๊ณ„์—ด"
elif 45 <= hue_value < 75:
return "์—ฐ๋‘/์ดˆ๋ก ๊ณ„์—ด"
elif 75 <= hue_value < 165:
return "์ดˆ๋ก/ํ•˜๋Š˜ ๊ณ„์—ด"
elif 165 <= hue_value < 255:
return "ํŒŒ๋ž‘/๋‚จ์ƒ‰ ๊ณ„์—ด"
elif 255 <= hue_value < 285:
return "๋ณด๋ผ ๊ณ„์—ด"
elif 285 <= hue_value < 345:
return "๋ถ„ํ™ ๊ณ„์—ด"
else:
return "๊ธฐํƒ€"
def analyze_thumbnail(thumbnail_url):
response = requests.get(thumbnail_url)
img = Image.open(BytesIO(response.content)).convert('RGB')
img_np = np.array(img)
hsv = cv2.cvtColor(img_np, cv2.COLOR_RGB2HSV)
hue_avg = int(np.mean(hsv[:, :, 0]) * 2)
# ์–ผ๊ตด ์ˆ˜ ๊ฒ€์ถœ
mp_face = mp.solutions.face_detection
with mp_face.FaceDetection(model_selection=1, min_detection_confidence=0.3) as fd:
results = fd.process(cv2.cvtColor(img_np, cv2.COLOR_RGB2BGR))
face_count = len(results.detections) if results.detections else 0
return hue_to_color_group(hue_avg), face_count, hue_avg
def predict_views(video_id, api_key):
url = f'https://www.googleapis.com/youtube/v3/videos?part=snippet,statistics&id={video_id}&key={api_key}'
res = requests.get(url).json()
item = res['items'][0]
title = item['snippet']['title']
published_at = item['snippet']['publishedAt']
category_id = item['snippet'].get('categoryId', '')
thumbnail_url = item['snippet']['thumbnails']['high']['url']
views = int(item['statistics'].get('viewCount', 0))
# ๊ฒŒ์‹œ์ผ ์ •๋ณด
dt = pd.to_datetime(published_at)
hour = dt.hour
weekday = dt.dayofweek
# ์ž๋ง‰ ์ˆ˜
def count_manual_subtitles(video_id):
ppl = YouTubeTranscriptApi.list_transcripts(video_id)
manual = [t for t in ppl if not t.is_generated]
return len(manual)
caption_count = count_manual_subtitles(video_id)
# ์ธ๋„ค์ผ ๋ถ„์„
hue_group, face_count, hue_value = analyze_thumbnail(thumbnail_url)
# ๊ฐ์„ฑ ์ ์ˆ˜
senti = sentiment_score(title)
# ์นดํ…Œ๊ณ ๋ฆฌ ์ด๋ฆ„ ๋งคํ•‘
user_category = map_category(category_id, title, api_key)
# Label Encoding
if user_category not in le_cat.classes_:
user_category = '๊ธฐํƒ€'
cat_encoded = le_cat.transform([user_category])[0]
# ์˜ˆ์ธก
X_input = pd.DataFrame([{
'์‹œ๊ฐ„๋Œ€': hour,
'์š”์ผ': weekday,
'์ž๋ง‰์ˆ˜': caption_count,
'์นดํ…Œ๊ณ ๋ฆฌ': cat_encoded,
'Hue': hue_value,
'์ธ๋„ค์ผ ์–ผ๊ตด ์ˆ˜': face_count,
'๊ฐ์„ฑ์ ์ˆ˜': senti
}])
pred_log = model.predict(X_input[feature_cols])[0]
predicted_views = int(np.expm1(pred_log))
return {
'์ œ๋ชฉ': title,
'์˜ˆ์ธก ์กฐํšŒ์ˆ˜': predicted_views,
'์‹ค์ œ ์กฐํšŒ์ˆ˜': views,
'์นดํ…Œ๊ณ ๋ฆฌ': user_category,
'์‹œ๊ฐ„๋Œ€': hour,
'์š”์ผ': weekday,
'์ž๋ง‰์ˆ˜': caption_count,
'์ธ๋„ค์ผ ์–ผ๊ตด ์ˆ˜': face_count,
'๊ฐ์„ฑ์ ์ˆ˜': senti,
'Hue ๊ทธ๋ฃน': hue_group,
'Hue ๊ฐ’': hue_value,
'์ธ๋„ค์ผ URL': thumbnail_url
}
#1. ์ถ”์ธก ํ•จ์ˆ˜
def extract_features_from_video_id(video_id, api_key):
info = predict_views(video_id, api_key)
return pd.DataFrame([{
'์‹œ๊ฐ„๋Œ€': info['์‹œ๊ฐ„๋Œ€'],
'์š”์ผ': info['์š”์ผ'],
'์ž๋ง‰์ˆ˜': info['์ž๋ง‰์ˆ˜'],
'์นดํ…Œ๊ณ ๋ฆฌ': le_cat.transform([info['์นดํ…Œ๊ณ ๋ฆฌ']])[0],
'Hue': info['Hue ๊ฐ’'],
'์ธ๋„ค์ผ ์–ผ๊ตด ์ˆ˜': info['์ธ๋„ค์ผ ์–ผ๊ตด ์ˆ˜'],
'๊ฐ์„ฑ์ ์ˆ˜': info['๊ฐ์„ฑ์ ์ˆ˜']
}])
# 2. ์˜ˆ์ธก ํ•จ์ˆ˜
def predict_view_count(model, features):
pred_log = model.predict(features[feature_cols])[0]
return int(np.expm1(pred_log))
# 3. ์‹œ๊ฐํ™” ํ•จ์ˆ˜
def visualize_result(video_id, features, predicted_view_count, info):
์š”์ผ_ํ…์ŠคํŠธ = ['์›”', 'ํ™”', '์ˆ˜', '๋ชฉ', '๊ธˆ', 'ํ† ', '์ผ'][features['์š”์ผ'].values[0]]
html = f"""
<div style="background-color: #111; color: white; padding: 20px; border-radius: 10px; max-width: 600px; font-family: Arial, sans-serif;">
<h2>๐ŸŽฏ ์˜ˆ์ธก ์กฐํšŒ์ˆ˜: {predicted_view_count:,}ํšŒ</h2>
<h3>๐Ÿ“Œ ์˜์ƒ ์ œ๋ชฉ: {info['์ œ๋ชฉ']}</h3>
<img src="{info['์ธ๋„ค์ผ URL']}" alt="์ธ๋„ค์ผ ์ด๋ฏธ์ง€" style="width: 100%; border-radius: 10px; margin-bottom: 20px;"/>
<ul style="list-style-type: none; padding-left: 0;">
<li>๐Ÿ“ฝ๏ธ <strong>์˜์ƒ ID:</strong> {video_id}</li>
<li>๐Ÿ‘๏ธ <strong>์‹ค์ œ ์กฐํšŒ์ˆ˜:</strong> {info['์‹ค์ œ ์กฐํšŒ์ˆ˜']:,}ํšŒ</li>
<li>โฐ <strong>์‹œ๊ฐ„๋Œ€:</strong> {features['์‹œ๊ฐ„๋Œ€'].values[0]}์‹œ</li>
<li>๐Ÿ“… <strong>์š”์ผ:</strong> {์š”์ผ_ํ…์ŠคํŠธ}</li>
<li>๐Ÿ’ฌ <strong>์ž๋ง‰ ์ˆ˜:</strong> {features['์ž๋ง‰์ˆ˜'].values[0]}</li>
<li>๐ŸŽจ <strong>์ƒ‰์ƒ ๊ณ„์—ด:</strong> {info['Hue ๊ทธ๋ฃน']}</li>
<li>๐Ÿ˜ƒ <strong>์ธ๋„ค์ผ ์–ผ๊ตด ์ˆ˜:</strong> {features['์ธ๋„ค์ผ ์–ผ๊ตด ์ˆ˜'].values[0]}</li>
<li>๐Ÿง  <strong>๊ฐ์„ฑ ์ ์ˆ˜:</strong> {features['๊ฐ์„ฑ์ ์ˆ˜'].values[0]:.2f}</li>
</ul>
</div>
"""
return html