youtubeprovider.py 文件源码

def __search_video_id(self, query):
        params = {
            "key": self.key,
            "part": "id,snippet",
            "order": "relevance",
            "type": "video",
            "videoSyndicated": "true",
            "maxResults": 3,
            "q": query
        }

        buf = BytesIO()

        client = pycurl.Curl()
        client.setopt(pycurl.URL, self.SERVICE_HOST + "/youtube/v3/search?" +
                      urlencode(params))
        client.setopt(pycurl.WRITEFUNCTION, buf.write)
        client.perform()
        client.close()

        body = json.loads(buf.getvalue().decode("utf-8"))
        buf.close()

        if "error" in body:
            raise Exception("query error: {0}".format(body))

        if len(body["items"]) == 0:
            raise Exception("result not found")

        video_id = body["items"][0]["id"]["videoId"]

        return video_id