dylan/crosspost
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

style: run python linter & prettifier on backend code

Browse Source
This commit is contained in:
Dylan De Faoite
2026-03-25 19:34:43 +00:00
parent aae10c4d9d
commit 376773a0cc
17 changed files with 408 additions and 315 deletions
Download Patch File Download Diff File Expand all files Collapse all files

104
server/connectors/youtube_api.py
View File

@@ -12,6 +12,7 @@ load_dotenv()
API_KEY = os.getenv("YOUTUBE_API_KEY")
class YouTubeAPI(BaseConnector):
source_name: str = "youtube"
display_name: str = "YouTube"
@@ -19,73 +20,72 @@ class YouTubeAPI(BaseConnector):
categories_enabled: bool = False
def __init__(self):
self.youtube = build('youtube', 'v3', developerKey=API_KEY)
self.youtube = build("youtube", "v3", developerKey=API_KEY)
def get_new_posts_by_search(self,
search: str,
category: str,
post_limit: int
) -> list[Post]:
videos = self._search_videos(search, post_limit)
posts = []
def get_new_posts_by_search(
self, search: str, category: str, post_limit: int
) -> list[Post]:
videos = self._search_videos(search, post_limit)
posts = []
for video in videos:
video_id = video['id']['videoId']
snippet = video['snippet']
title = snippet['title']
description = snippet['description']
published_at = datetime.datetime.strptime(snippet['publishedAt'], "%Y-%m-%dT%H:%M:%SZ").timestamp()
channel_title = snippet['channelTitle']
for video in videos:
video_id = video["id"]["videoId"]
snippet = video["snippet"]
title = snippet["title"]
description = snippet["description"]
published_at = datetime.datetime.strptime(
snippet["publishedAt"], "%Y-%m-%dT%H:%M:%SZ"
).timestamp()
channel_title = snippet["channelTitle"]
comments = []
comments_data = self._get_video_comments(video_id)
for comment_thread in comments_data:
comment_snippet = comment_thread['snippet']['topLevelComment']['snippet']
comment = Comment(
id=comment_thread['id'],
post_id=video_id,
content=comment_snippet['textDisplay'],
author=comment_snippet['authorDisplayName'],
timestamp=datetime.datetime.strptime(comment_snippet['publishedAt'], "%Y-%m-%dT%H:%M:%SZ").timestamp(),
reply_to=None,
source=self.source_name
)
comments.append(comment)
post = Post(
id=video_id,
content=f"{title}\n\n{description}",
author=channel_title,
timestamp=published_at,
url=f"https://www.youtube.com/watch?v={video_id}",
title=title,
comments = []
comments_data = self._get_video_comments(video_id)
for comment_thread in comments_data:
comment_snippet = comment_thread["snippet"]["topLevelComment"][
"snippet"
]
comment = Comment(
id=comment_thread["id"],
post_id=video_id,
content=comment_snippet["textDisplay"],
author=comment_snippet["authorDisplayName"],
timestamp=datetime.datetime.strptime(
comment_snippet["publishedAt"], "%Y-%m-%dT%H:%M:%SZ"
).timestamp(),
reply_to=None,
source=self.source_name,
comments=comments
)
posts.append(post)
comments.append(comment)
post = Post(
id=video_id,
content=f"{title}\n\n{description}",
author=channel_title,
timestamp=published_at,
url=f"https://www.youtube.com/watch?v={video_id}",
title=title,
source=self.source_name,
comments=comments,
)
posts.append(post)
return posts
return posts
def category_exists(self, category):
return True
def _search_videos(self, query, limit):
request = self.youtube.search().list(
q=query,
part='snippet',
type='video',
maxResults=limit
q=query, part="snippet", type="video", maxResults=limit
)
response = request.execute()
return response.get('items', [])
return response.get("items", [])
def _get_video_comments(self, video_id):
request = self.youtube.commentThreads().list(
part='snippet',
videoId=video_id,
textFormat='plainText'
part="snippet", videoId=video_id, textFormat="plainText"
)
try:
@@ -93,4 +93,4 @@ class YouTubeAPI(BaseConnector):
except HttpError as e:
print(f"Error fetching comments for video {video_id}: {e}")
return []
return response.get('items', [])
return response.get("items", [])