From 5ff1d68a27aee25779d732b8769a2dc3bb417b26 Mon Sep 17 00:00:00 2001 From: jaehwang Date: Mon, 8 Jun 2026 13:58:07 +0900 Subject: [PATCH] =?UTF-8?q?=EC=9C=A0=ED=8A=9C=EB=B8=8C=20=EC=97=85?= =?UTF-8?q?=EB=A1=9C=EB=93=9C=20=EC=A3=BC=EA=B8=B0=20=EA=B3=84=EC=82=B0=20?= =?UTF-8?q?=EB=A1=9C=EC=A7=81=20=EC=B6=94=EA=B0=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/integrations/youtube.py | 82 +++++++++++++++++++++++++------------ app/services/analysis.py | 5 ++- 2 files changed, 58 insertions(+), 29 deletions(-) diff --git a/app/integrations/youtube.py b/app/integrations/youtube.py index f72ccc3..56309f7 100644 --- a/app/integrations/youtube.py +++ b/app/integrations/youtube.py @@ -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: + 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: - videos = resp.json().get("items", [])[:10] + 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"] diff --git a/app/services/analysis.py b/app/services/analysis.py index 0f2603f..e640658 100644 --- a/app/services/analysis.py +++ b/app/services/analysis.py @@ -113,13 +113,14 @@ def _naver_blog_summary(blog: dict | None) -> dict | None: } async def _build_youtube_audit(youtube: dict) -> dict: # 기획상 1개의 input channel, 다중 채널은 기획에 없음. - videos = youtube.get("videos", []) + 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"]