@@ -53,7 +53,14 @@ def load_email_list_config(config_path: str = "email_list.json") -> list[dict]:
5353 Returns:
5454 list[dict]: A list of validated configuration entries, each containing:
5555 - email (str): Recipient email address
56- - search_url (str): YouTube search URL
56+ - search_url (str, optional): YouTube search URL
57+ - channel_id (str, optional): YouTube channel ID
58+ - channel_url (str, optional): YouTube channel URL
59+ - channel_username (str, optional): YouTube channel username
60+
61+ Note:
62+ Each entry must have either a search_url OR at least one channel field
63+ (channel_id, channel_url, or channel_username).
5764
5865 Raises:
5966 FileNotFoundError: If the configuration file doesn't exist.
@@ -78,20 +85,45 @@ def load_email_list_config(config_path: str = "email_list.json") -> list[dict]:
7885
7986 email = entry .get ("email" )
8087 search_url = entry .get ("search_url" )
88+ channel_id = entry .get ("channel_id" )
89+ channel_url = entry .get ("channel_url" )
90+ channel_username = entry .get ("channel_username" )
8191
8292 if not email or not isinstance (email , str ) or not email .strip ():
8393 logging .warning (f"Entry at index { idx } )
8494 continue
8595
86- if not search_url or not isinstance (search_url , str ) or not search_url .strip ():
87- logging .warning (f"Entry at index { idx } )
88- continue
89-
9096 # Basic email format validation
9197 if "@" not in email :
9298 logging .warning (f"Entry at index { idx } )
9399 continue
94- validated_entries .append ({"email" : email .strip (), "search_url" : search_url .strip ()})
100+
101+ # Validate that at least one source is provided (search_url OR channel)
102+ has_search_url = search_url and isinstance (search_url , str ) and search_url .strip ()
103+ has_channel_id = channel_id and isinstance (channel_id , str ) and channel_id .strip ()
104+ has_channel_url = channel_url and isinstance (channel_url , str ) and channel_url .strip ()
105+ has_channel_username = channel_username and isinstance (channel_username , str ) and channel_username .strip ()
106+
107+ if not (has_search_url or has_channel_id or has_channel_url or has_channel_username ):
108+ logging .warning (
109+ f"Entry at index { idx }
110+ "(channel_id, channel_url, or channel_username)"
111+ )
112+ continue
113+
114+ # Build validated entry
115+ validated_entry = {"email" : email .strip ()}
116+ # Additional checks help mypy understand these are not None
117+ if has_search_url and search_url :
118+ validated_entry ["search_url" ] = search_url .strip ()
119+ if has_channel_id and channel_id :
120+ validated_entry ["channel_id" ] = channel_id .strip ()
121+ if has_channel_url and channel_url :
122+ validated_entry ["channel_url" ] = channel_url .strip ()
123+ if has_channel_username and channel_username :
124+ validated_entry ["channel_username" ] = channel_username .strip ()
125+
126+ validated_entries .append (validated_entry )
95127
96128 if len (validated_entries ) == 0 :
97129 logging .warning ("Configuration file contains no valid entries" )
@@ -101,6 +133,141 @@ def load_email_list_config(config_path: str = "email_list.json") -> list[dict]:
101133 return validated_entries
102134
103135
136+ def is_video_within_last_day (video : dict ) -> bool :
137+ """
138+ Checks if a video was published within the last 24 hours.
139+
140+ Args:
141+ video (dict): A video dictionary from scrapetube containing publishedTimeText.
142+
143+ Returns:
144+ bool: True if the video was published within the last 24 hours, False otherwise.
145+ """
146+ try :
147+ published_text = video .get ("publishedTimeText" , {}).get ("simpleText" , "" )
148+ if not published_text :
149+ return False
150+
151+ # Parse relative time strings like "2 hours ago", "1 day ago", etc.
152+ published_text_lower = published_text .lower ()
153+
154+ # Check for videos from within the last day
155+ if "minute" in published_text_lower or "hour" in published_text_lower :
156+ return True
157+ elif "day" in published_text_lower :
158+ # Extract the number of days
159+ parts = published_text_lower .split ()
160+ try :
161+ num_days = int (parts [0 ])
162+ return num_days <= 1
163+ except (ValueError , IndexError ):
164+ return False
165+ else :
166+ # Anything else (weeks, months, years) is not within the last day
167+ return False
168+ except Exception as e :
169+ logging .warning (f"Error parsing video publish date: { e } )
170+ return False
171+
172+
173+ def get_channel_videos_last_day (
174+ channel_id : str | None = None ,
175+ channel_url : str | None = None ,
176+ channel_username : str | None = None ,
177+ api_client : YouTubeTranscriptApi | None = None ,
178+ ) -> list [dict ]:
179+ """
180+ Retrieves videos from a channel that were published within the last 24 hours.
181+
182+ Args:
183+ channel_id (str, optional): The YouTube channel ID.
184+ channel_url (str, optional): The YouTube channel URL.
185+ channel_username (str, optional): The YouTube channel username (without @).
186+ api_client (YouTubeTranscriptApi, optional): An instance of YouTubeTranscriptApi.
187+
188+ Returns:
189+ list[dict]: List of dictionaries containing video_id, title, and transcript.
190+ """
191+ channel_identifier = channel_id or channel_url or channel_username
192+ logging .info (f"Fetching videos from channel: { channel_identifier } )
193+
194+ try :
195+ # Get videos from the channel, sorted by newest first
196+ # Note: Using scrapetube.get_channel() (top-level API) instead of
197+ # scrapetube.scrapetube.get_videos() (lower-level function) for channel queries
198+ channel_videos = scrapetube .get_channel (
199+ channel_id = channel_id ,
200+ channel_url = channel_url ,
201+ channel_username = channel_username ,
202+ sort_by = "newest" ,
203+ sleep = YOUTUBE_SEARCH_SLEEP_SECONDS ,
204+ )
205+
206+ # Filter videos from the last 24 hours
207+ recent_videos = []
208+ for video in channel_videos :
209+ if is_video_within_last_day (video ):
210+ recent_videos .append (video )
211+ else :
212+ # Since videos are sorted by newest, we can stop once we hit an older video
213+ break
214+
215+ logging .info (f"Found { len (recent_videos )} { channel_identifier } )
216+
217+ # Process transcripts for the recent videos
218+ results_data = []
219+ transcript_api = api_client or get_transcript_api ()
220+
221+ for video in recent_videos :
222+ video_id = video .get ("videoId" )
223+ try :
224+ title = video ["title" ]["runs" ][0 ]["text" ]
225+ except (KeyError , IndexError ):
226+ title = "Unknown Title"
227+
228+ logging .info (f"Processing channel video: { title } { video_id } )
229+
230+ try :
231+ transcript_list_obj = transcript_api .list (video_id )
232+
233+ # Try to find English variants first
234+ try :
235+ transcript_obj = transcript_list_obj .find_transcript (["en" , "en-US" , "en-GB" ])
236+ logging .info (
237+ f"Found English transcript for video ID: { video_id } { transcript_obj .language_code }
238+ )
239+ except :
240+ # Fallback: If no English, just take the first available one
241+ transcript_obj = next (iter (transcript_list_obj ))
242+ logging .info (
243+ f"No English transcript found. Using available transcript with language code: { transcript_obj .language_code } { video_id }
244+ )
245+
246+ # fetch() returns a list of dictionaries with 'text', 'start', and 'duration'
247+ fetched_transcript = transcript_obj .fetch ()
248+
249+ # Preserve transcript items with timestamps
250+ transcript_items = [{"text" : item .text , "start" : item .start } for item in fetched_transcript ]
251+
252+ except TranscriptsDisabled :
253+ logging .info (f"Transcripts are disabled for video ID: { video_id } )
254+ continue
255+ except NoTranscriptFound :
256+ logging .info (f"No transcript found for video ID: { video_id } )
257+ continue
258+ except Exception as e :
259+ logging .info (f"Error retrieving transcript for video ID: { video_id } { str (e )} )
260+ continue
261+
262+ results_data .append ({"video_id" : video_id , "title" : title , "transcript" : transcript_items })
263+
264+ return results_data
265+
266+ except Exception as e :
267+ logging .error (f"Error fetching channel videos: { e } )
268+ return []
269+
270+
104271def get_recent_transcripts (url : str , limit : int = 10 , api_client : YouTubeTranscriptApi | None = None ) -> list [dict ]:
105272 """
106273 Searches for the most recent videos by URL and retrieves their transcripts.
@@ -438,17 +605,37 @@ def send_newsletter_resend(subject: str, body: str, recipients: list):
438605 # Process each configuration entry
439606 for idx , entry in enumerate (config_entries ):
440607 recipient_email = entry ["email" ]
441- search_url = entry ["search_url" ]
608+ search_url = entry .get ("search_url" )
609+ channel_id = entry .get ("channel_id" )
610+ channel_url = entry .get ("channel_url" )
611+ channel_username = entry .get ("channel_username" )
442612
443613 logging .info (f"\n { '=' * 60 } )
444614 logging .info (f"Processing entry { idx + 1 } { len (config_entries )} )
445615 logging .info (f"Recipient: { recipient_email } )
446- logging .info (f"Search URL: { search_url } )
616+ if search_url :
617+ logging .info (f"Search URL: { search_url } )
618+ if channel_id :
619+ logging .info (f"Channel ID: { channel_id } )
620+ if channel_url :
621+ logging .info (f"Channel URL: { channel_url } )
622+ if channel_username :
623+ logging .info (f"Channel Username: { channel_username } )
447624 logging .info (f"{ '=' * 60 } \n " )
448625
449626 try :
450- # Fetch transcripts
451- data = get_recent_transcripts (search_url , limit = 2 )
627+ # Fetch transcripts from channels (if specified)
628+ data = []
629+ if channel_id or channel_url or channel_username :
630+ channel_data = get_channel_videos_last_day (
631+ channel_id = channel_id , channel_url = channel_url , channel_username = channel_username
632+ )
633+ data .extend (channel_data )
634+
635+ # Also fetch from search URL if specified
636+ if search_url :
637+ search_data = get_recent_transcripts (search_url , limit = 2 )
638+ data .extend (search_data )
452639
453640 if not data :
454641 logging .warning (f"No transcripts found for { recipient_email } )
0 commit comments