[트위터 크롤링] GetOldTweets3

 강의 중간 프로젝트로 트위터 데이터를 크롤링했다. 트위터 크롤링은 API(공식 Standard API를 활용하더라도 수집할 수 있는 기간에 제한이 있다), Request 에러 핸들링 등으로 인해 골치가 아프다.

위의 문제에 부딪히지 않기 위해 Selenium 및 네트워크 요청 분석을 통해 코드를 짜서 스크레이핑을 진행하려 했으나, 미니 프로젝트였기 때문에, 시간이 많지 않아 성공하지는 못했다.

 하루 이틀 고생했는데, 결론적으로 내가 구현하고 싶은 방법을 잘 구현해 놓은 라이브러리를 발견했다. API 없이도 이전 트윗을 수집할 수 있었다. 데이터 분석이 목적이 아니라 단순히 여러 사이트에서 데이터를 수집하는 방법을 연습하는 프로젝트였기 때문에, 어떻게 라이브러리를 활용했는지를 위주로 크롤링 방법만 기록한다.

 

 

---

 

# 예제 코드

 

 2020년 5월에 마지막으로 확인한 코드로, 작동 확인 후 업데이트할 예정이다.

 

 

https://github.com/sirzzang/crawl-tweets/blob/master/notebook/crawl_tweets.ipynb

GitHub - sirzzang/crawl-tweets: 트위터 크롤링 예제 코드

 

 

---

 

# 사용한 라이브러리

 

• GetOldTweets3

Mottl/GetOldTweets3

• BeautifulSoup
• urllib, sys, json, datetime, time, os, tqdm, pandas, csv

 

 

 

---

 

# 트위터에서 검색 결과 JSON Response 반환

 

 GetOldTweets의 static method인 getJsonResponse 함수를 재정의했다. 주로 수정한 부분은 다음과 같다.

 

• HTTP 429 Request 에러 방지 위해 코드를 중지하는 time.sleep 부분을 추가했다. 경험적으로 판단한 결과, 1초 정도로 주어도 충분히 크롤링을 진행하는 데 문제가 없다.
• URL 에러, HTTP Response 에러, JSON 파싱 에러 발생 시 코드를 중지하는 부분을 수정했다. 디버그를 위해 에러가 발생하면 url을 출력하도록 하고, try, except 문에서 except 부분에 pass 옵션을 주었다.

 

 그 외에, request URL을 설정하는 부분에서 내가 필요하지 않은 부분들의 코드는 삭제했다. 검색 옵션으로 설정하지 않을 부분들이 그것이다. 

 

# GOT 스태틱 메소드 수정
def getJsonResponse(tweetCriteria, refreshCursor, cookieJar, proxy, useragent=None, debug=False):
    url = "https://twitter.com/i/search/timeline?"

    if not tweetCriteria.topTweets:
        url += "f=tweets&"

    url += ("vertical=news&q=%s&src=typd&%s"
            "&include_available_features=1&include_entities=1&max_position=%s"
            "&reset_error_state=false")

    urlGetData = ''
    
    # url + query search, since, since, until, lang    
    urlGetData += tweetCriteria.querySearch
    urlGetData += ' since:' + tweetCriteria.since
    urlGetData += ' until:' + tweetCriteria.until
    urlLang = 'l=' + tweetCriteria.lang + '&'
    
    # url 설정
    url = url % (urllib.parse.quote(urlGetData.strip()), urlLang, urllib.parse.quote(refreshCursor))
    useragent = useragent or TweetManager.user_agents[0]

    headers = [
        ('Host', "twitter.com"),
        ('User-Agent', useragent),
        ('Accept', "application/json, text/javascript, */*; q=0.01"),
        ('Accept-Language', "en-US,en;q=0.5"),
        ('X-Requested-With', "XMLHttpRequest"),
        ('Referer', url),
        ('Connection', "keep-alive")
    ]

    if proxy:
        opener = urllib.request.build_opener(urllib.request.ProxyHandler({'http': proxy, 'https': proxy}), urllib.request.HTTPCookieProcessor(cookieJar))
    else:
        opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(cookieJar))
    opener.addheaders = headers        
    
    time.sleep(1) # HTTP Request 429 에러 방지.

    ##################### 에러 핸들링 수정 #####################
    
    # TimeOut 에러
    try:
        response = opener.open(url)
        jsonResponse = response.read()
    except TimeoutError as e:
        if debug: # 디버그 옵션
            print("에러 url 주소:", url)            
        print("Timeout error")
        print("30초 정지")
        time.sleep(30)

        # 한 번 더 시도
        try:
            response = opener.open(url)
            jsonResponse = response.read()
        except TimeoutError as e:
            print("Timeout error again. 패스.")
            pass

    # UrlParse 에러: 다시 시도 X
    except Exception as e:
        if debug: # 디버그 옵션
            print("에러 url 주소:", url)
            
        print("HTTP 요청 오류", str(e))        
        print("브라우저 오픈: https://twitter.com/search?q=%s&src=typd" % urllib.parse.quote(urlGetData))
        print("30초 정지")
        
        pass

    # Json 데이터 오류
    try:
        s_json = jsonResponse.decode()
    except:
        print("올바르지 못한 응답")
        if debug: # 디버그 옵션
            print("에러 url 주소:", url)
        pass
    else:
        try:
            dataJson = json.loads(s_json)
        except: # json 데이터 파싱 오류
            if debug: # 디버그 옵션
                print("에러 url 주소:", url)                
            print("JSON: %s" % s_json)
        pass

    return dataJson

 

 

---

 

# 크롤링 날짜 설정

 

 트위터 검색량이 워낙 방대하기 때문에, 일단 2015년부터 2020년까지 월 초에만 검색을 진행하도록 했다. pandas 라이브러리의 date_range 함수를 사용해 2015년 1월부터 2020년 6월까지 1일을 설정하고, GetOldTweets3 라이브러리 사용법에 맞게(since와 until에 들어갈 수 있도록) 날짜를 반환하는 함수를 작성했다.

 

 

dateRange = pd.date_range(start='20150101', end='20200601', freq='MS')

# setUntil : 마지막 날짜 배제되므로 주의.
def set_crawl_date(start_date, freq=1):    
    end_date = start_date + datetime.timedelta(days=freq)
    
    # timestamp to string format
    start_date = start_date.strftime("%Y-%m-%d")
    end_date = end_date.strftime("%Y-%m-%d")
    
    # check
    print("트윗 수집 날짜 설정: {0}부터 {1}까지".format(start_date, end_date)) 
    
    return start_date, end_date

# 체크용
for date in dateRange[:3]:
    print(set_crawl_date(date))

 

 

---

 

# 트윗 크롤링

 

 설정된 옵션을 바탕으로 GetOldTweets의 getTweets 메소드를 사용한다. 

 

 

def crawl_tweets(start_date, end_date, query, lang='en', debug=True):    
    '''
    query: 검색할 트윗 검색어
    lang: 검색할 트윗 언어
    debug: 설정 시 에러 url 표시
    '''
    
    print("========== 트윗 수집 시작: {0} ~ {1} ==========".format(start_date, end_date))
    start_time = time.time()
    tweet_criteria = got.manager.TweetCriteria().setQuerySearch(query)\
                                                .setSince(start_date)\
                                                .setUntil(end_date)\
                                                .setLang(lang)
    tweets = got.manager.TweetManager.getTweets(tweet_criteria, debug=debug)
    
    elapsed_time = time.time()-start_time
    
    print("수집 완료 : {}".format(time.strftime("%H:%M:%S", time.gmtime(elapsed_time))))
    print("총 수집 트윗 개수 : {0}".format(len(tweets)))
    
    return tweets

 

 

---

 

# 결과 추출 및 저장

 

 위의 함수를 실행하면 각 날짜별로 검색어에 맞게 스크레이핑된 GetOldTweets3 라이브러리의 Tweet 객체가 반환된다. 해당 객체에서 필요한 부분만 뽑아 저장했다.

 

# got tweet 객체로부터 결과 추출
def get_results(tweet_data):
    results = []
    for tweet in tqdm(tweet_data):
        results.append({'url': tweet.permalink,
                        'date': tweet.date,
                        'text': tweet.text,
                        'user': tweet.username,
                        'mentions': tweet.mentions,
                        'retweets': tweet.retweets,
                        'favorites': tweet.favorites,
                        'hashtags': tweet.hashtags})
    return results
    
# 추출한 결과 저장
def save_tweets(tweet_lists, base_file_dir="tweets"):
    
    if not os.path.exists(base_file_dir):
        os.makedirs(base_file_dir)
        
    with open(f"{base_file_dir}/tweets_{crawl_start}_{crawl_end}.csv", "a", -1, encoding="utf-8") as f:    
        writer = csv.writer(f)
        writer.writerow(['url', 'date', 'text', 'user', 'mentions', 'retweets', 'favorites', 'hashtags'])        
        for tweet_list in tqdm(tweet_lists):
            writer.writerow(list(tweet_list.values()))
            
    return 

 

 

---

 

 

 

# 크롤링 진행

 

 

dateRange = pd.date_range(start='20150101', end='20200601', freq='MS')

for date in dateRange:
    crawl_start, crawl_end = set_crawl_date(date) # freq 변경 가능
    tweet_results = crawl_tweets(crawl_start, crawl_end, query='SomeString')
    tweet_results_lists = get_results(tweet_results)
    save_tweets(tweet_results_lists)

 

 

 작업을 진행하는 중간 중간 다음과 같은 상태가 콘솔에 출력된다.

 

 

 

 

중간 결과 확인

 

 

 

 

 에러가 몇 군데서 나긴 했지만, 에러가 나는 트윗을 제외하고도 약 55만 개의 트윗을 수집했기 때문에(;;) 굳이 에러 나는 트윗까지 모두 수집하지는 않기로 했다.

 

 

 

---