Jmaa/personal-data
1
0
Fork 0
Code Actions Packages

refactor: Simplify fetch_data and to_csv methods in YoutubeFavoritesScraper

Browse Source
This commit is contained in:
Jon Michael Aanes (aider) 2025-03-15 21:54:19 +01:00
parent 965689df7a
commit 2a4aec9d33
1 changed files with 7 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
personal_data/fetchers/youtube.py
View File

@ -15,10 +15,7 @@ class YoutubeFavoritesScraper(Scraper):
deduplicate_mode: DeduplicateMode = DeduplicateMode.BY_ALL_COLUMNS deduplicate_mode: DeduplicateMode = DeduplicateMode.BY_ALL_COLUMNS
def fetch_data(self) -> list[dict]: def fetch_data(self) -> list[dict]:
""" """Use yt-dlp to fetch the list of favorited videos. This is a placeholder for invoking yt-dlp and parsing its output."""
Use yt-dlp to fetch the list of favorited videos.
This is a placeholder for invoking yt-dlp and parsing its output.
"""
try: try:
# Replace 'YOUR_FAVORITES_ID' with your actual favorites playlist ID. # Replace 'YOUR_FAVORITES_ID' with your actual favorites playlist ID.
result = subprocess.run( result = subprocess.run(
@ -32,33 +29,18 @@ class YoutubeFavoritesScraper(Scraper):
check=True, check=True,
text=True, text=True,
) )
videos = [json.loads(line) for line in result.stdout.splitlines()] return [json.loads(line) for line in result.stdout.splitlines()]
return videos except Exception:
except Exception as e: logger.exception('Failed to fetch YouTube favorites')
logger.error('Failed to fetch YouTube favorites: %s', e)
raise raise
def to_csv(self, videos: list[dict]) -> str: def to_csv(self, videos: list[dict]) -> str:
""" """Convert the list of videos to CSV format."""
Convert the list of videos to CSV format.
"""
output = []
headers = ['id', 'title', 'url', 'upload_date'] headers = ['id', 'title', 'url', 'upload_date']
output.append(headers) rows = [headers] + [[video.get('id'), video.get('title'), video.get('url'), video.get('upload_date')] for video in videos]
for video in videos:
output.append(
[
video.get('id'),
video.get('title'),
video.get('url'),
video.get('upload_date'),
],
)
from io import StringIO from io import StringIO
sio = StringIO() sio = StringIO()
csv_writer = csv.writer(sio) csv.writer(sio).writerows(rows)
csv_writer.writerows(output)
return sio.getvalue() return sio.getvalue()
def run(self) -> None: def run(self) -> None: