[Python] 동영상 수집 모듈 만들기

Youtube에서 계정이 보유중인 채널들의 데이터를 수집하는 모듈을 만들었다.

 

✅ 목적

• 채널 통계 수집 : DB에 보유하고 있는 채널들의 리스트들을 가지고와 일별 통계를 업데이트
• 비디오 통계 수집 :각 채널들이 보유하고 있는 비디오들을 가지고와 일별 통계 업데이트

 

 

✅ 실행환경

Pycham + Python 3.7 
MacOS(intel)
Youtube Data API
YouTube Analytics API
MySQL

 

 

 

✔️ 채널정보 업데이트

 

1. 관리중인 채널들을 DB에서 가져온다.

def getChannelList(self):
    channelList = self.mysql_.searchChannelList()
    return channelList

 

2. 그런다음 채널리스트마다 각각 API호출을 통해 정보를 가져온다.

def getChannelList(self, channelId):
    try:
        requestURL = urlparse(conf.API_URL + "/channels")

        requestParam = dict(parse_qsl(requestURL.query))
        requestParam['part'] = 'id,snippet,statistics'
        requestParam['id'] = channelId
        requestParam['key'] = conf.API_KEY

        requestURL = requestURL._replace(query=urlencode(requestParam))
        targetURL = urlunparse(requestURL)

        self.result_data = self.httpRequestTo(targetURL)
    except requests.exceptions.RequestException as err:
        LOGGER.error("[YOUTUBE DATA API] GET CHANNEL LIST EXCEPTION => %s\n%s" % (err, traceback.format_exc()))
        traceback.print_exc()
    return self.result_data

제목,설명,썸네일,구독자수,조회수,소유한동영상수등을 가져온다.

 

3. 최신정보로 업데이트

self.mysql_.updateChannelList(channelId, resultData)

 

 

 

채널 일별 통계 업데이트

def procChannelStatDaily(self, day, regNeed=True):
    try:
        # 조회기준날짜
        apiDate = dateUtil.getDayCalBefore(self, day, 3)  # 3일전부터 조회가능
        LOGGER.debug("[channel_stat_daily] 채널 일별통계 수집 프로세스 시작 ===> %s / API 조회 날짜 ===> %s" % (day, apiDate))

        # 수집중인 채널 리스트
        channel_list = self.getChannelList()

        # 일별통계 등록 및 업데이트
        if regNeed:
            LOGGER.debug("[channel_stat_daily] 일별 통계만 등록")
            self.regChannelFromYTDataAPI(day, channel_list)
        else:
            LOGGER.debug("[channel_stat_daily] Analytics 오디언스 통계만 업데이트")
            self.uptChannelFromYTAnalyAPI(apiDate, channel_list)
    except Exception as err:
        LOGGER.error("[채널일별통계 EXCEPTION] %s\n%s" % (err, traceback.format_exc()))
        traceback.print_exc()
    LOGGER.debug("[channel_stat_daily] 채널 일별통계 수집완료 ========")

def uptChannelFromYTAnalyAPI(self, day, channel_list):
    # API인증
    if not hasattr(self, 'youtube_analytics'):
        self.youtube_analytics = self.youtube_analyticsAPI.authenticatedServices()

    for idx, channels in enumerate(channel_list):
        resultData = dict()
        # 1.오디언스 통계 조회
        resultData = self.getChannelAudianceInfoFromAnalyAPI(day, channels)
        # 2.국가별 조회수 조회
        resultData['country_view_json'] = self.getChannelCountryViewFromAnalyAPI(day, channels)

        LOGGER.debug("[channel_stat_daily][%d] 채널 일별통계 오디언스 업데이트 : %s" % (idx+1, str(resultData)))

        # 3.api 조회 결과 업데이트
        self.mysql_.updateChannelStat(day, channels['channel_id'], resultData)

 

 

마찬가지로 수집중인 리스트를 가지고 와 전일자 날짜데이터를 비교하여 계산후,  일별통계를 따로 등록한다.

그런다음 오디언스 통계조회를 통해 오디언스정보를 업데이트시킨다.

 

 

4. 일별 통계 오디언스 업데이트

• YouTube Analytics API 인증

def authenticatedServices(self, retry=False):
    file_list_size = len(self.SECRET_FILE_LIST)

    ## 재인증일 경우
    if retry:
        file_list_size = len(self.SECRET_FILE_LIST)
        self.FILE_INDEX += 1
        LOGGER.info("[YOUTUBE_ANALYTIC_AUTH] 재인증 %d / %d" % (self.FILE_INDEX, file_list_size))

    if 0 < file_list_size <= self.FILE_INDEX:
        LOGGER.error("[YOUTUBE_ANALYTIC_AUTH] 모든 파일 인증실패!!")
        return None

    self.setAuthClientFile()
    LOGGER.info("[YOUTUBE_ANALYTIC_AUTH] CLIENT_SECRETS_FILE:%s" % self.CLIENT_SECRETS_FILE)
    LOGGER.info("[YOUTUBE_ANALYTIC_AUTH] STORAGE_FILE:%s" % self.STORAGE_FILE)

    try:
        flow = flow_from_clientsecrets(self.CLIENT_SECRETS_FILE,
                                       scope=" ".join(self.YOUTUBE_SCOPES),
                                       message=self.MISSING_CLIENT_SECRETS_MESSAGE)

        storage = Storage(self.STORAGE_FILE)
        credentials = storage.get()

        # 승인오류
        if credentials is None or credentials.invalid:
            LOGGER.error("[YOUTUBE_ANALYTIC_승인오류] Http 400 error, redirect_uri_mismatch. To update the authorized redirect URIs")
            self.authenticatedServices(True)

        http = credentials.authorize(httplib2.Http())
        #youtube = build(self.YOUTUBE_API_SERVICE_NAME, self.YOUTUBE_API_VERSION, http=http)
        youtubeAnalytics = build(self.YOUTUBE_ANALYTICS_API_SERVICE_NAME,
                                 self.YOUTUBE_ANALYTICS_API_VERSION, http=http)

        return youtubeAnalytics
    except HttpError as err:
        LOGGER.error("[YOUTUBE_ANALYTIC_AUTH EXCEPTION] Auth HttpError : %s" % (err))
        traceback.print_exc()
    except Exception as e:
        LOGGER.error("[YOUTUBE_ANALYTIC_AUTH EXCEPTION] Auth Error : %s" % (e))
        traceback.print_exc()

Storage 객체를 사용하여 인증된 사용자의 자격 증명을 로컬 파일에서 로드하여 확인한다.

인증이 성공하면 build 메서드를 사용하여 YouTube Analytics API의 서비스 객체를 생성한다.

 

 

• 통계데이터 조회 API

def requestQuery(self, youtube_analytics, param):
    response_data = list()
    result_data = dict()
    response_code = 999
    try:
        LOGGER.debug("[YOUTUBE ANALYTIC REQUEST] query param: %s" % str(param))
        if youtube_analytics:
            analytics_query_response = youtube_analytics.reports().query(
                ids=param.get('ids')
                , metrics=param.get('metrics')
                , dimensions=param.get('dimensions')
                , startDate=param.get('startDate')
                , endDate=param.get('endDate')
                , filters=param.get('filters')
                , sort=param.get('sort')
            ).execute()
            response_code = 200
            response_data = (analytics_query_response.get("rows", []))
    except HttpError as err:
        response_code = err.resp.status
        LOGGER.error("[YOUTUBE ANALYTIC HTTP ERROR] query error status code : %s\n%s" % (response_code, str(param)))
        traceback.print_exc()
    except HttpAccessTokenRefreshError as re:
        response_code = 400
        LOGGER.error("[YOUTUBE ANALYTIC ACCESS TOKEN ERROR] %s\n%s" % (re, str(param)))
        traceback.print_exc()

    result_data['resCode'] = response_code
    result_data['resData'] = response_data
    return result_data

 

인증파일을 통해 인증정보를 가지고온 후 

 youtube_analytics.reports().query()를 사용하여 YouTube Analytics API에 대한 요청을 생성하고 execute()를 호출하여 실제로 API를 호출한다.

 

 

---

 

 

✔️ 비디오 정보 수집

 

1. 각 채널이 소유하고 있는 비디오 정보 업데이트

def procContentsInfo(self, channelList, day):
    video_list = list()
    tmp = 0

    self.startDate = dateUtil().getDayCalBefore(day, conf.MAX_DAY)  # 최대 90일 기준
    self.endDate = day
    LOGGER.debug("[content] 각 채널별 콘텐츠 수집 일자기준 : %s ~ %s" % (self.startDate, self.endDate))

    try:
        # 1. 채널별 동영상 리스트 GET
        for channels in channelList:
            video_dict = dict()
            contentList = []
            channelId = channels['channel_id']
            contentOwner = channels['content_owner']

            # 재생리스트 가져오기
            contentList = self.getVideoInfoListFromDataAPI(channelId, self.startDate, contentList)

            # 결과 세팅
            video_dict['channel_id'] = channelId
            video_dict['content_owner'] = contentOwner
            video_dict['content_list'] = contentList

            video_list.append(video_dict)
            LOGGER.debug("[content] 수집채널 : %s / 동영상갯수 : %s" % (channelId, len(video_dict.get('content_list'))))
            tmp += len(video_dict.get('content_list'))

        if len(video_list) > 0:
            # 각 동영상 별 상세정보 조회하여 DB에 등록
            self.regVideoDetailInfo(video_list)
        LOGGER.debug("[content] 수집 완료 ===> 총 동영상 갯수 : %d / 처리된 동영상 갯수 : %d" % (self.total_count, self.proc_count))
        LOGGER.debug("[content] 채널별 처리완료된 콘텐츠 카운트 : %s" % str(self.log_info_buff))
    except Exception as err:
        LOGGER.error("[content EXCEPTION] %s\n%s" % (err, traceback.format_exc()))
        traceback.print_exc()

 

def getVideoList(self, data):
    try:
        requestURL = urlparse(conf.API_URL + "/search")

        requestParam = dict(parse_qsl(requestURL.query))
        requestParam['part'] = 'snippet'
        requestParam['key'] = conf.API_KEY
        requestParam['order'] = conf.API_ORDER_DATE
        requestParam['type'] = conf.API_TYPE_RESOURCE
        requestParam['maxResults'] = conf.API_MAX_RESULT
        requestParam['pageToken'] = data['nextPageToken']
        requestParam['channelId'] = data['channel_id']
        requestParam['publishedAfter'] = data['startDate']

        requestURL = requestURL._replace(query=urlencode(requestParam))
        targetURL = urlunparse(requestURL)
        self.result_data = self.httpRequestTo(targetURL)
    except requests.exceptions.RequestException as err:
        LOGGER.error("[YOUTUBE DATA API] GET VIDEO LIST EXCEPTION => %s\n%s" % (err, traceback.format_exc()))
        traceback.print_exc()
    return self.result_data

동영상 리스트를 가져온다.

 

2. 비디오 통계정보 업데이트

def procContentsDaliyStat(self, day, regNeed=True):
    try:
        backdays = dateUtil.getDayCalBefore(self, day, 3)
        LOGGER.debug("[content_stat_daily] 콘텐츠 일별통계 수집 프로세스 시작 ===> %s / API 조회 날짜 ===> %s" % (day, backdays))
        # DB
        mysql_dicon = dicon.dicon()

        # 수집중인 콘텐츠 리스트 조회
        #contents_list = self.mysql_dicon.searchContentsList()
        contents_list = mysql_dicon.searchContentsList()
        LOGGER.debug("[content_stat_daily] 현재 수집중인 콘텐츠 수 : %d" % len(contents_list))

        # 각 콘텐츠 별 데일리 통계 데이터 수집
        if regNeed:
            LOGGER.debug("[content_stat_daily] 콘텐츠 일별통계 등록")
            self.regContentFromYTDataAPI(day, contents_list)
            LOGGER.debug("[content_stat_daily] 콘텐츠 일별통계 등록 완료! (등록수:%d)" % len(contents_list))
        else:
            LOGGER.debug("[content_stat_daily] 콘텐츠 일별통계 오디언스 업데이트")
            self.uptContentFromYTAnalyAPI(backdays, contents_list)
            LOGGER.debug("[content_stat_daily] 콘텐츠 일별통계 오디언스 업데이트 완료! (완료수:%d)" % len(contents_list))

    except Exception as err:
        LOGGER.error("[content_stat_daily EXCEPTION] %s\n%s" % (err, traceback.format_exc()))
        traceback.print_exc()
    LOGGER.debug("[content_stat_daily] 콘텐츠 일별통계 수집 완료!! ============")

 

 

 

끝