2019-12-22

DataCollection / Twitter

트위터에서 많은 팔로워를 크롤링하는 방법 [2]: HTML 웹 크롤링을 해보자

앞서 쓴 글(1편)에서는 몇가지 조건들과 나쁜 상황을 걸어 단순한 수집이 아니라 뭔가 꼼수가 필요한 상황을 가정해 보았다.

API를 유료로 못쓴다? ➡️ 못쓸정도로 느리다! 😨

따라서 다른 방법인 직접 데이터 크롤링이 필요했고, 이번에는 Twint와 HTML 웹 크롤링을 통해서 수집을 해 본 과정을 적어보았다.

HTML 웹 크롤링을 해보자!

인증이 필요해서 ➡️ 느린 속도(Quota limit)가 걸린다면,

인증 없이 수집할 수 있다면 ➡️ 빠른 속도로 수집할 수 있겠다!

1) Twint: 내가 짠 것보다는 3k 스타 오픈소스가 낫겠지?

Twint on Github

처음에는 기존에 존재하는 트위터 수집 라이브러리를 사용해보자!는 생각을 했다.

내가 생각했다면, 분명 다른 누군가는 만들어뒀을거라는 나름 타당한 생각에 검색을 했고, 가장 적합한 라이브러리라고 생각했던 것이 바로 Twint였다.

Spoiler: Twint는 쓸만하지 않았다.

Twint를 이용해 로컬에서 적당한 사이즈(1k 팔로잉 이하)를 크롤링 하는 것에 성공했지만 MultiProcess를 통해 돌릴 경우 생각보다 굉장히 낮은 성능을 보이는 면을 보여, AWS Lambda에 올려 병렬로 수집하고자 했다. 수집에 성공시 Boto3를 통해 DynamoDB에 적재하는 방법을 취했다. (아래 코드가 AWS Lambda에 올린 코드.)

Sqlite를 사용하기까지 해서 이상한 코드(8-9번째 줄)까지 추가로 지정해줘야 했다.

import imp
import sys
import json
import boto3

db = boto3.client('dynamodb')

sys.modules["sqlite"] = imp.new_module("sqlite")
sys.modules["sqlite3.dbapi2"] = imp.new_module("sqlite.dbapi2")

def get_following(username):
    import twint
    twint.output.follows_list = []
    c = twint.Config()
    c.Username = username
    c.Store_object = True
    
    twint.run.Following(c)
    following = twint.output.follows_list
    return list(set(following))

def lambda_handler(event, context):
    if event.get('body'):
        event = json.loads(event['body'])
        print("HTTPAPI /", event['username'])
    username = event['username']
    followings = get_following(username)
    
    if not followings == ['message']:
        db.put_item(
        	TableName='tweet-followings',
        	Item={
        		'username': {'S': username},
        		'following': {'SS': followings},
        	}
        )
    
    return {
        'statusCode': 200,
        'body': json.dumps(followings),
        'headers': {'Content-Type': 'application/json'},
    }

한편, Twint 자체가 Async하게 동작하고있음에도 굉장히 느린 속도를 보이기도 하고 동시에 일부 대형 유저에서는 데이터 유실이 발생하기도 해, 해당 부분에 대한 신뢰도가 굉장히 낮아지게 되어, 다른 방법을 찾아보기로 했다.

2) 직접 짠 크롤링 코드: 차라리 내가 짜고 말지!

앞서 말한 것과 같이, Twint에 대한 신뢰도가 팍팍 떨어지고 나서 데이터를 싹 버리고 새로 수집하는데, 기왕 이렇게 된 것 내가 직접 코드를 짜고 말지! 하는 생각을 했다.

간단하게, requests 그리고 beautifulsoup4 로 어떻게 수집, 안될까?

1st try: 어라? 생각보다 잘 되는데?

정말로 심플하게 웹 페이지에 들어가서 수집하는 코드를 돌렸다.

유저의 아이디를 입력하면 해당 유저의 팔로잉을 모아 s3에 저장하는 코드를 구성했다.

import json
import requests as r
from bs4 import BeautifulSoup as bs
import boto3

BASE_URL = 'https://mobile.twitter.com'
s3 = boto3.client('s3')

def lambda_handler(event, context):
    if event.get('Records'):
        event = json.loads(event['Records'][0]['Sns']['Message'])    
    username = event['username']
    users = []
    url = f'/{username}/following?lang=en'
    
    while True:
        try:
            s = bs(r.get(BASE_URL + url).text, 'html.parser')
            f_count = int(s.select_one('span.count').text.replace(',', ''))
            users += [i['name'] for i in s.select('div.user-list td.screenname a[name]')]
            url = f"{s.select_one('div.w-button-more > a')['href']}&lang=en"
        except TypeError as e:
            print("Cursor Ended")
            print(len(set(users)))
            
            s3.put_object(
              Bucket='datas3asdfasdf',
              Key=f'twitter/following/{username}.json',
              Body=json.dumps({
                'username': username,
                'following': users,
                'count': f_count,
                'len': len(users),
              })
            )

한번에 20개의 데이터를 보여주고, 각 항목에 나오는 HTML을 파싱해 유저의 screen_name (트위터의 @id)을 수집했다.

그리고 ‘더 보기’ 버튼이 있으면 해당 Cursor가 존재하기 때문에 해당 커서로 넘어가고, 만약 커서가 없다 = 종료되었다고 보고 크롤링을 종료했다.

하지만, 문제가 발생했다.

2nd try: 그럼, 한번에 잘 될리가 없지. 😅

유저들 중에서 ‘정상적 페이지’가 나오지 않는 유저가 자주 발생했다.

NOTE : 크롤링 진행시 Edge case를 미리 목록으로 만들고 진행하는 것이 낫다.

또한, 크롤링한 데이터는 메모리가 아닌 Json 등의 파일로 적재하고 수집한 것은 다시 수집하지 않도록 만드는 것이 필수다!

이유는 다양했다.

Protected 계정: 유저가 자신의 정보를 보지 못하도록 비공개로 돌린 계정으로, 팔로우/팔로워 숫자는 볼 수 있지만 누구인지는 알 수 없다.
“Sorry, that page doesn’t exist” 계정: 유저가 계정명을 변경했거나 탈퇴한 경우.
‘suspended’ 계정: 유저가 트위터 약관을 위반해 트위터에서 차단한 경우. (Spam 등)

특히 3번의 경우 정말로 예상치 못한 경우여서 데이터 수집을 중간에 놓치기도 했다.

그래서 몇번의 시행착오 끝에(이틀 넘게 걸렸다) 코드를 완성했다.

아래 코드는 위의 모든 엣지 케이스를 통과하는 방향으로 진행되었고, 결과를 확인했을때 99%정도의 확실한 결과를 보여주었다.

import json
import requests as r
from bs4 import BeautifulSoup as bs
import boto3

BASE_URL = 'https://mobile.twitter.com'
s3 = boto3.client('s3')

def lambda_handler(event, context):
    if event.get('Records'):
        event = json.loads(event['Records'][0]['Sns']['Message'])    
    username = event['username']
    
    user_html = r.get(f'{BASE_URL}/{username}?lang=en').text
    # Proteced User 제거하기 
    if 'protected' in user_html:
        print(f"User [{username}] is protected")
        s3.put_object(
            Bucket='datas3asdfasdf',
            Key=f'twitter/following-protected/{username}.json',
            Body=json.dumps({
                'username': username,
            })
        )
        return False
    elif "Sorry, that page doesn't exist" in user_html:
        print(f"User [{username}] does not exist(deleted)")
        s3.put_object(
            Bucket='datas3asdfasdf',
            Key=f'twitter/following-deleted/{username}.json',
            Body=json.dumps({
                'username': username,
            })
        )
        return False
    
    users = []
    
    url = f'/{username}/following?lang=en'
    
    while True:
        try:
            s = bs(r.get(BASE_URL + url).text, 'html.parser')
            try:
                f_count = int(s.select_one('span.count').text.replace(',', ''))
            except AttributeError:
                try:
                    f_count = int(s.select('div.statnum')[-1].text.replace(',', ''))
                except IndexError as e:
                    if "Sorry, that page doesn't exist" in s.select_one('body').text:
                        print(f"User [{username}] is deleted")
                        s3.put_object(
                            Bucket='datas3asdfasdf',
                            Key=f'twitter/following-deleted/{username}.json',
                            Body=json.dumps({
                                'username': username,
                            })
                        )
                        return False
                    if 'suspended' in r.get(f'{BASE_URL}/{username}?lang=en').text:
                        print(f"User [{username}] is suspended")
                        s3.put_object(
                            Bucket='datas3asdfasdf',
                            Key=f'twitter/following-suspended/{username}.json',
                            Body=json.dumps({
                                'username': username,
                            })
                        )
                        return False
            users += [i['name'] for i in s.select('div.user-list td.screenname a[name]')]
            url = f"{s.select_one('div.w-button-more > a')['href']}&lang=en"
        except TypeError as e:
            print("Cursor Ended")
            print(len(set(users)))
            
            try:
                is_complete = f_count == len(users)
            except UnboundLocalError as e:
                print(e)
                print(f"[UnboundLocalError] username: {username}")
                s3.put_object(
                    Bucket='datas3asdfasdf',
                    Key=f'twitter/following-no-fcount/{username}.json',
                    Body=json.dumps({
                        'username': username,
                        'following': users,
                        'count': -1,
                        'len': len(users),
                    })
                )
                return False
            if is_complete:
                s3.put_object(
                    Bucket='datas3asdfasdf',
                    Key=f'twitter/following/{username}.json',
                    Body=json.dumps({
                        'username': username,
                        'following': users,
                        'count': f_count,
                        'len': len(users),
                    })
                )
            else:
                print(f"Unexpected End on crawling, counnt:{f_count}, len:{len(users)}")
                s3.put_object(
                    Bucket='datas3asdfasdf',
                    Key=f'twitter/following-needmore/{username}.json',
                    Body=json.dumps({
                        'username': username,
                        'following': users,
                        'count': f_count,
                        'len': len(users),
                    })
                )
            return True

와! 잘 된다!

그렇다면, 전부 끝난걸까? ➡️ 그럴리가요…

앞서 작성한 코드를 AWS Lambda에 태워 S3에 적재하는 방향으로 진행했다.

해당 Lambda 함수는 AWS SNS를 통해 트리거되도록 만들어, Rate Limit exceed (Boto3을 통해 Lambda를 많이 호출시 발생하는 오류)를 회피하도록 구성했다.

한편, 여전히 문제는 남아있었다.

AWS Lambda는 15분이 최대 😢

항상 Lambda를 쓸 때 마다 투덜거리는 투로 올리는 말이지만, 서버리스가 정말 좋긴 하지만 Lambda에는 여전히 15분 max 제약이 걸려있다.

실제로 수천개 (2k~5k) 이내의 유저는 손쉽게 & 안정적으로 데이터 수집을 완료했지만, 10k+ 계정들에서는 시간 부족의 문제로 수집되지 않는 등 이슈가 생겼다.

대체 어떻게 해야 다 모을 수 있을까?

트위터에서 많은 팔로워를 크롤링하는 방법 3편: 1초에 5천개 데이터 가져오기으로 이어집니다.

# crawling, python, twitter