API de datos de Youtube: obtenga la identificación de video más reciente del canal excluyendo transmisiones en vivo

Tendrá que iterar su llamada al PlaylistItems.listpunto final de la API (usando la paginación) para filtrar manualmente los videos que son transmisiones en vivo.

def get_non_livestream_videos(youtube, video_ids):
    assert len(video_ids) <= 50

    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

Tenga en cuenta que anteriormente utilicé el fieldsparámetro de solicitud para obtener de las API solo la información que realmente se necesita.

También tenga en cuenta que es posible que deba elaborar un poco la función get_non_livestream_videos, ya que el Videos.listpunto final de la API consultado con su idparámetro como una lista separada por comas de ID de video puede alterar el orden de los elementos que devuelve en el orden dado de las ID video_ids.

Sin embargo, una nota importante: si está ejecutando el código anterior en Python 3 (su pregunta no menciona esto), asegúrese de tener el siguiente código de configuración insertado en la parte superior de su secuencia de comandos:

if sys.version_info[0] >= 3:
    from builtins import map as builtin_map
    map = lambda *args: list(builtin_map(*args))

Esto es necesario ya que, en Python 3, la función incorporada mapdevuelve un iterador , mientras que en Python 2, mapdevuelve una lista .

Aquí está el código que resuelve el problema que mencioné anteriormente en el caso de Videos.listalterar el orden de los elementos devueltos en relación con el orden de los ID dados por el argumento video_idsde la función 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

El código anterior es un programa independiente (que se ejecuta tanto en Python v2 como v3) que implementa una función de fusiónmerge_video_lists .

Tendrá que usar esta función dentro de la función get_non_livestream_videosreemplazando la línea:

return map(video_id, filter(not_live, items))

con:

return map(video_id, merge_video_lists(
    video_ids, filter(not_live, items)))

para Python 2. Para Python 3, el reemplazo sería:

return map(video_id, merge_video_lists(
    video_ids, list(filter(not_live, items))))

En lugar de reemplazar la returndeclaración, simplemente haga que esa declaración esté precedida por esta:

items = merge_video_lists(video_ids, items)

Esta última variante es mejor, ya que también valida las ID de video devueltas por la API: si hay una ID que no está video_ids, entonces merge_video_listslanza una MergeVideoListsErrorexcepción que indica la ID del culpable.