유튜브 업로드 주기 계산 로직 추가
parent
c0fa48ff75
commit
5ff1d68a27
|
|
@ -48,7 +48,7 @@ class YouTubeClient:
|
|||
resp = await http_request(
|
||||
HTTPMethod.GET,
|
||||
url=f"{YT}/channels",
|
||||
params={"part": "snippet,statistics", "id": channel_id, "key": self.api_key},
|
||||
params={"part": "snippet,statistics,contentDetails", "id": channel_id, "key": self.api_key},
|
||||
label="yt-channel",
|
||||
)
|
||||
if not resp or not resp.is_success:
|
||||
|
|
@ -58,26 +58,52 @@ class YouTubeClient:
|
|||
return None
|
||||
channel = items[0]
|
||||
|
||||
video_ids: list[str] = []
|
||||
resp = await http_request(
|
||||
HTTPMethod.GET,
|
||||
url=f"{YT}/search",
|
||||
params={"part": "snippet", "channelId": channel_id, "order": "viewCount", "type": "video", "maxResults": 10, "key": self.api_key},
|
||||
label="yt-search",
|
||||
)
|
||||
if resp and resp.is_success:
|
||||
video_ids = [i["id"]["videoId"] for i in resp.json().get("items", []) if i.get("id", {}).get("videoId")]
|
||||
|
||||
videos: list[dict] = []
|
||||
if video_ids:
|
||||
async def _video_details(video_ids: list[str]) -> list[dict]:
|
||||
"""video_ids 순서를 보존한 채 snippet/statistics/contentDetails 채워서 반환."""
|
||||
if not video_ids:
|
||||
return []
|
||||
resp = await http_request(
|
||||
HTTPMethod.GET,
|
||||
url=f"{YT}/videos",
|
||||
params={"part": "snippet,statistics,contentDetails", "id": ",".join(video_ids), "key": self.api_key},
|
||||
label="yt-videos",
|
||||
)
|
||||
if not resp or not resp.is_success:
|
||||
return []
|
||||
video_map = {v["id"]: v for v in resp.json().get("items", [])}
|
||||
return [video_map[vid] for vid in video_ids if vid in video_map]
|
||||
|
||||
# 인기 영상 top 10 (조회수순) — search index 기반, 정확한 정렬 보장.
|
||||
resp = await http_request(
|
||||
HTTPMethod.GET,
|
||||
url=f"{YT}/search",
|
||||
params={"part": "snippet", "channelId": channel_id, "order": "viewCount", "type": "video", "maxResults": 10, "key": self.api_key},
|
||||
label="yt-search-top",
|
||||
)
|
||||
top_ids: list[str] = []
|
||||
if resp and resp.is_success:
|
||||
videos = resp.json().get("items", [])[:10]
|
||||
top_ids = [i["id"]["videoId"] for i in resp.json().get("items", []) if i.get("id", {}).get("videoId")]
|
||||
videos = await _video_details(top_ids)
|
||||
|
||||
# 최근 영상 10개 — search index는 누락이 흔해 채널의 실제 uploads 재생목록에서 직접 읽고
|
||||
# publishedAt 기준으로 코드에서 직접 정렬 (재생목록 순서 자체는 보장되지 않으므로).
|
||||
recents: list[dict] = []
|
||||
uploads_id = (channel.get("contentDetails") or {}).get("relatedPlaylists", {}).get("uploads")
|
||||
if uploads_id:
|
||||
resp = await http_request(
|
||||
HTTPMethod.GET,
|
||||
url=f"{YT}/playlistItems",
|
||||
params={"part": "snippet", "playlistId": uploads_id, "maxResults": 10, "key": self.api_key},
|
||||
label="yt-uploads",
|
||||
)
|
||||
if resp and resp.is_success:
|
||||
entries = resp.json().get("items", [])
|
||||
entries.sort(key=lambda i: i.get("snippet", {}).get("publishedAt") or "", reverse=True)
|
||||
recent_ids = [
|
||||
vid for e in entries
|
||||
if (vid := e.get("snippet", {}).get("resourceId", {}).get("videoId"))
|
||||
]
|
||||
recents = await _video_details(recent_ids)
|
||||
|
||||
playlists: list[dict] = []
|
||||
resp = await http_request(
|
||||
|
|
@ -89,7 +115,19 @@ class YouTubeClient:
|
|||
if resp and resp.is_success:
|
||||
playlists = resp.json().get("items", [])
|
||||
|
||||
return {"channelId": channel_id, "channel": channel, "videos": videos, "playlists": playlists}
|
||||
return {"channelId": channel_id, "channel": channel, "videos": videos, "recents": recents, "playlists": playlists}
|
||||
|
||||
@staticmethod
|
||||
def _video_item(v: dict) -> dict:
|
||||
return {
|
||||
"title": v.get("snippet", {}).get("title"),
|
||||
"views": int(v.get("statistics", {}).get("viewCount", 0)),
|
||||
"likes": int(v.get("statistics", {}).get("likeCount", 0)),
|
||||
"comments": int(v.get("statistics", {}).get("commentCount", 0)),
|
||||
"date": v.get("snippet", {}).get("publishedAt"),
|
||||
"duration": v.get("contentDetails", {}).get("duration"),
|
||||
"url": f"https://www.youtube.com/watch?v={v['id']}",
|
||||
}
|
||||
|
||||
async def get_channel(self, url: str) -> dict | None:
|
||||
raw = await self.fetch_channel(url)
|
||||
|
|
@ -110,18 +148,8 @@ class YouTubeClient:
|
|||
"subscribers": int(stats.get("subscriberCount", 0)),
|
||||
"totalViews": int(stats.get("viewCount", 0)),
|
||||
"totalVideos": int(stats.get("videoCount", 0)),
|
||||
"videos": [
|
||||
{
|
||||
"title": v.get("snippet", {}).get("title"),
|
||||
"views": int(v.get("statistics", {}).get("viewCount", 0)),
|
||||
"likes": int(v.get("statistics", {}).get("likeCount", 0)),
|
||||
"comments": int(v.get("statistics", {}).get("commentCount", 0)),
|
||||
"date": v.get("snippet", {}).get("publishedAt"),
|
||||
"duration": v.get("contentDetails", {}).get("duration"),
|
||||
"url": f"https://www.youtube.com/watch?v={v['id']}",
|
||||
}
|
||||
for v in raw["videos"]
|
||||
],
|
||||
"videos": [self._video_item(v) for v in raw["videos"]],
|
||||
"recents": [self._video_item(v) for v in raw["recents"]],
|
||||
"playlists": [
|
||||
p.get("snippet", {}).get("title")
|
||||
for p in raw["playlists"]
|
||||
|
|
|
|||
|
|
@ -114,12 +114,13 @@ def _naver_blog_summary(blog: dict | None) -> dict | None:
|
|||
|
||||
async def _build_youtube_audit(youtube: dict) -> dict: # 기획상 1개의 input channel, 다중 채널은 기획에 없음.
|
||||
videos = youtube.get("videos", [])
|
||||
recents = youtube.get("recents", [])
|
||||
yt_patch: dict = {
|
||||
"weekly_view_growth": {"absolute": 0, "percentage": 0.0},
|
||||
"estimated_monthly_revenue": {"min": 0, "max": 0},
|
||||
"linked_urls": [],
|
||||
"avg_video_length": calc_avg_video_length(videos),
|
||||
"upload_frequency": calc_upload_frequency(videos),
|
||||
"upload_frequency": calc_upload_frequency(recents),
|
||||
}
|
||||
if youtube.get("channelName"): yt_patch["channel_name"] = youtube["channelName"]
|
||||
if youtube.get("handle"): yt_patch["handle"] = youtube["handle"]
|
||||
|
|
|
|||
Loading…
Reference in New Issue