You'll have to iterate your call to PlaylistItems.list API endpoint (using pagination) for to filter out manually the videos that are live streams.
def get_non_livestream_videos(youtube, video_ids):
assert len(video_ids) <= 50
if not video_ids: return []
response = youtube.videos().list(
fields = 'items(id,liveStreamingDetails)',
part = 'id,liveStreamingDetails',
maxResults = len(video_ids),
id = ','.join(video_ids),
).execute()
items = response.get('items', [])
assert len(items) <= len(video_ids)
not_live = lambda video: \
not video.get('liveStreamingDetails')
video_id = lambda video: video['id']
return map(video_id, filter(not_live, items))
def get_latest_video_id(youtube, playlistId):
request = youtube.playlistItems().list(
fields = 'nextPageToken,items/snippet/resourceId',
playlistId = playlistId,
maxResults = 50,
part = 'snippet'
)
is_video = lambda item: \
item['snippet']['resourceId']['kind'] == 'youtube#video'
video_id = lambda item: \
item['snippet']['resourceId']['videoId']
while request:
response = request.execute()
items = response.get('items', [])
assert len(items) <= 50
videos = map(video_id, filter(is_video, items))
if videos:
videos = get_non_livestream_videos(youtube, videos)
if videos: return videos[0]
request = youtube.playlistItems().list_next(
request, response)
return None
Note that above I used the fields request parameter for to get from the APIs only the info that's actually needed.
Also note that you may have to elaborate a bit the function get_non_livestream_videos, since the Videos.list API endpoint queried with its id parameter as a comma-separated list of video IDs may well alter the order of the items it returns w.r.t. the given order of the IDs in video_ids.
Yet an important note: if you're running the code above under Python 3 (your question does not mention this), then make sure you have the following configuration code inserted at the top of your script:
if sys.version_info[0] >= 3:
from builtins import map as builtin_map
map = lambda *args: list(builtin_map(*args))
This is needed since, under Python 3, the builtin function map returns an iterator, whereas under Python 2, map returns a list.
Here is the code that solves the issue I mentioned above w.r.t. the case of Videos.list altering the order of items returned relative to the order of the IDs given by the argument video_ids of function get_non_livestream_videos:
import sys
if sys.version_info[0] >= 3:
from builtins import map as builtin_map
map = lambda *args: list(builtin_map(*args))
class MergeVideoListsError(Exception): pass
def merge_video_lists(video_ids, video_res):
pair0 = lambda pair: pair[0]
pair1 = lambda pair: pair[1]
video_ids = sorted(
enumerate(video_ids), key = pair1)
video_res.sort(
key = lambda video: video['id'])
def error(video_id):
raise MergeVideoListsError(
"unexpected video resource of ID '%s'" % video_id)
def do_merge():
N = len(video_ids)
R = len(video_res)
assert R <= N
l = []
i, j = 0, 0
while i < N and j < R:
v = video_ids[i]
r = video_res[j]
s = v[1]
d = r['id']
if s == d:
l.append((v[0], r))
i += 1
j += 1
elif s < d:
i += 1
else:
error(d)
if j < R:
error(video_res[j]['id'])
return l
video_res = do_merge()
video_res.sort(key = pair0)
return map(pair1, video_res)
def println(*args):
for a in args:
sys.stdout.write(str(a))
sys.stdout.write('\n')
def test_merge_video_lists(ids, res, val):
try:
println("ids: ", ids)
println("res: ", res)
r = merge_video_lists(ids, res)
println("merge: ", r)
except MergeVideoListsError as e:
println("error: ", e)
r = str(e)
finally:
println("test: ", "OK" \
if val == r \
else "failed")
TESTS = ((
['c', 'b', 'a'],
[{'id': 'c'}, {'id': 'a'}, {'id': 'b'}],
[{'id': 'c'}, {'id': 'b'}, {'id': 'a'}]
),(
['c', 'b', 'a'],
[{'id': 'b'}, {'id': 'c'}],
[{'id': 'c'}, {'id': 'b'}]
),(
['c', 'b', 'a'],
[{'id': 'a'}, {'id': 'c'}],
[{'id': 'c'}, {'id': 'a'}]
),(
['c', 'b', 'a'],
[{'id': 'a'}, {'id': 'b'}],
[{'id': 'b'}, {'id': 'a'}]
),(
['c', 'b', 'a'],
[{'id': 'z'}, {'id': 'b'}, {'id': 'c'}],
"unexpected video resource of ID 'z'"
),(
['c', 'b', 'a'],
[{'id': 'a'}, {'id': 'z'}, {'id': 'c'}],
"unexpected video resource of ID 'z'"
),(
['c', 'b', 'a'],
[{'id': 'a'}, {'id': 'b'}, {'id': 'z'}],
"unexpected video resource of ID 'z'"
))
def main():
for i, t in enumerate(TESTS):
if i: println()
test_merge_video_lists(*t)
if __name__ == '__main__':
main()
# $ python merge-video-lists.py
# ids: ['c', 'b', 'a']
# res: [{'id': 'c'}, {'id': 'a'}, {'id': 'b'}]
# merge: [{'id': 'c'}, {'id': 'b'}, {'id': 'a'}]
# test: OK
#
# ids: ['c', 'b', 'a']
# res: [{'id': 'b'}, {'id': 'c'}]
# merge: [{'id': 'c'}, {'id': 'b'}]
# test: OK
#
# ids: ['c', 'b', 'a']
# res: [{'id': 'a'}, {'id': 'c'}]
# merge: [{'id': 'c'}, {'id': 'a'}]
# test: OK
#
# ids: ['c', 'b', 'a']
# res: [{'id': 'a'}, {'id': 'b'}]
# merge: [{'id': 'b'}, {'id': 'a'}]
# test: OK
#
# ids: ['c', 'b', 'a']
# res: [{'id': 'z'}, {'id': 'b'}, {'id': 'c'}]
# error: unexpected video resource of ID 'z'
# test: OK
#
# ids: ['c', 'b', 'a']
# res: [{'id': 'a'}, {'id': 'z'}, {'id': 'c'}]
# error: unexpected video resource of ID 'z'
# test: OK
#
# ids: ['c', 'b', 'a']
# res: [{'id': 'a'}, {'id': 'b'}, {'id': 'z'}]
# error: unexpected video resource of ID 'z'
# test: OK
The code above is a standalone program (running both under Python v2 and v3) that implements a merging function merge_video_lists.
You'll have to use this function within the function get_non_livestream_videos by replacing the line:
return map(video_id, filter(not_live, items))
with:
return map(video_id, merge_video_lists(
video_ids, filter(not_live, items)))
for Python 2. For Python 3 the replacement would be:
return map(video_id, merge_video_lists(
video_ids, list(filter(not_live, items))))
Instead of replacing the return statement, just have that statement preceded by this one:
items = merge_video_lists(video_ids, items)
This latter variant is better, since it also validates the video IDs returned by the API: if there is an ID that is not in video_ids, then merge_video_lists throws a MergeVideoListsError exception indicating the culprit ID.
For obtaining all videos that are exactly N days old, excluding live streams, use the function below:
def get_days_old_video_ids(youtube, playlistId, days = 7):
from datetime import date, datetime, timedelta
n_days = date.today() - timedelta(days = days)
request = youtube.playlistItems().list(
fields = 'nextPageToken,items(snippet/resourceId,contentDetails/videoPublishedAt)',
part = 'snippet,contentDetails',
playlistId = playlistId,
maxResults = 50
)
def parse_published_at(item):
details = item['contentDetails']
details['videoPublishedAt'] = datetime.strptime(
details['videoPublishedAt'],
'%Y-%m-%dT%H:%M:%SZ') \
.date()
return item
def find_if(cond, items):
for item in items:
if cond(item):
return True
return False
n_days_eq = lambda item: \
item['contentDetails']['videoPublishedAt'] == n_days
n_days_lt = lambda item: \
item['contentDetails']['videoPublishedAt'] < n_days
is_video = lambda item: \
item['snippet']['resourceId']['kind'] == 'youtube#video'
video_id = lambda item: \
item['snippet']['resourceId']['videoId']
videos = []
while request:
response = request.execute()
items = response.get('items', [])
assert len(items) <= 50
# remove the non-video entries in 'items'
items = filter(is_video, items)
# replace each 'videoPublishedAt' with
# its corresponding parsed date object
items = map(parse_published_at, items)
# terminate loop when found a 'videoPublishedAt' < n_days
done = find_if(n_days_lt, items)
# retain only the items with 'videoPublishedAt' == n_days
items = filter(n_days_eq, items)
# add to 'videos' the IDs of videos in 'items' that are not live streams
videos.extend(get_non_livestream_videos(youtube, map(video_id, items)))
if done: break
request = youtube.playlistItems().list_next(
request, response)
return videos
The function get_days_old_video_ids above needs filter and map to return lists, therefore the configuration code above has to be updated to:
if sys.version_info[0] >= 3:
from builtins import map as builtin_map
map = lambda *args: list(builtin_map(*args))
from builtins import filter as builtin_filter
filter = lambda *args: list(builtin_filter(*args))
Do note that get_days_old_video_ids is relying on the following undocumented property of the result set produced by PlaylistItems.list: for the uploads playlist of a channel, the items returned by PlaylistItems.list are ordered in reverse chronological order (newest first) by contentDetails.videoPublishedAt.
Therefore you have to make sure the argument playlistId of get_days_old_video_ids is the ID of the uploads playlist of your channel. Usually, a channel ID and its corresponding uploads playlist ID are related by s/^UC([0-9a-zA-Z_-]{22})$/UU\1/.
Also note that get_days_old_video_ids is returning the IDs of videos that are exactly days old. If needing to obtain the IDs of videos that are at most days old, then have defined:
n_days_ge = lambda item: \
item['contentDetails']['videoPublishedAt'] >= n_days
and have n_days_eq replaced with n_days_ge.
Yet something to note: at the top of function get_non_livestream_videos above, I added the statement:
if not video_ids: return []
such that to avoid processing an empty video_ids list.
