아프리카TV 실시간채팅 크롤링

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131

import certifi
import ssl
import asyncio
import websockets
import requests
from api import get_player_live

# 유니코드 및 기타 상수
F = "\x0c"
ESC = "\x1b\t"
SEPARATOR = "+" + "-" * 70 + "+"

# 아프리카TV에서 제공하는 API로 채팅 정보를 받습니다.
def get_player_live(bno, bid):
    url = 'https://live.afreecatv.com/afreeca/player_live_api.php'
    data = {
        'bid': bid,
        'bno': bno,
        'type': 'live',
        'confirm_adult': 'false',
        'player_type': 'html5',
        'mode': 'landing',
        'from_api': '0',
        'pwd': '',
        'stream_type': 'common',
        'quality': 'HD'
    }

    try:
        response = requests.post(f'{url}?bjid={bid}', data=data)
        response.raise_for_status()  # HTTP 요청 에러를 확인하고, 에러가 있을 경우 예외를 발생시킵니다.
        res = response.json()

        CHDOMAIN = res["CHANNEL"]["CHDOMAIN"].lower()
        CHATNO = res["CHANNEL"]["CHATNO"]
        FTK = res["CHANNEL"]["FTK"]
        TITLE = res["CHANNEL"]["TITLE"]
        BJID = res["CHANNEL"]["BJID"]
        CHPT = str(int(res["CHANNEL"]["CHPT"]) + 1)

        return CHDOMAIN, CHATNO, FTK, TITLE, BJID, CHPT

    except requests.RequestException as e:
        print(f"  ERROR: API 요청 중 오류 발생: {e}")
        return None
    except KeyError as e:
        print(f"  ERROR: 응답에서 필요한 데이터를 찾을 수 없습니다: {e}")
        return None

# SSL 컨텍스트 생성
def create_ssl_context():
    ssl_context = ssl.create_default_context()
    ssl_context.load_verify_locations(certifi.where())
    ssl_context.check_hostname = False
    ssl_context.verify_mode = ssl.CERT_NONE
    return ssl_context

# 메시지 디코드 및 출력
def decode_message(bytes):
    parts = bytes.split(b'\x0c')
    messages = [part.decode('utf-8') for part in parts]
    if len(messages) > 5 and messages[1] not in ['-1', '1'] and '|' not in messages[1]:
        user_id, comment, user_nickname = messages[2], messages[1], messages[6]
        print(SEPARATOR)
        print(f"| {user_nickname}[{user_id}] - {comment}")
    else:
        # 채팅 뿐만 아니라 다른 메세지도 동시에 내려옵니다.
        pass

# 바이트 크기 계산
def calculate_byte_size(string):
    return len(string.encode('utf-8')) + 6

# 채팅에 연결
async def connect_to_chat(url, ssl_context):
    try:
        BNO, BID = url.split('/')[-1], url.split('/')[-2]
        CHDOMAIN, CHATNO, FTK, TITLE, BJID, CHPT = get_player_live(BNO, BID)
        print(f"{SEPARATOR}\n"
              f"  CHDOMAIN: {CHDOMAIN}\n  CHATNO: {CHATNO}\n  FTK: {FTK}\n"
              f"  TITLE: {TITLE}\n  BJID: {BJID}\n  CHPT: {CHPT}\n"
              f"{SEPARATOR}")
    except Exception as e:
        print(f"  ERROR: API 호출 실패 - {e}")
        return

    try:
        async with websockets.connect(
            f"wss://{CHDOMAIN}:{CHPT}/Websocket/{BID}",
            subprotocols=['chat'],
            ssl=ssl_context,
            ping_interval=None
        ) as websocket:
            # 최초 연결시 전달하는 패킷
            CONNECT_PACKET = f'{ESC}000100000600{F*3}16{F}'
            # 메세지를 내려받기 위해 보내는 패킷
            JOIN_PACKET = f'{ESC}0002{calculate_byte_size(CHATNO):06}00{F}{CHATNO}{F*5}'
            # 주기적으로 핑을 보내서 메세지를 계속 수신하는 패킷
            PING_PACKET = f'{ESC}000000000100{F}'

            await websocket.send(CONNECT_PACKET)
            print(f"  연결 성공, 채팅방 정보 수신 대기중...")
            await asyncio.sleep(2)
            await websocket.send(JOIN_PACKET)

            async def ping():
                while True:
                    # 5분동안 핑이 보내지지 않으면 소켓은 끊어집니다.
                    await asyncio.sleep(60)  # 1분 = 60초
                    await websocket.send(PING_PACKET)

            async def receive_messages():
                while True:
                    data = await websocket.recv()
                    decode_message(data)

            await asyncio.gather(
                receive_messages(),
                ping(),
            )

    except Exception as e:
        print(f"  ERROR: 웹소켓 연결 오류 - {e}")

async def main():
    url = input("아프리카TV URL을 입력해주세요: ")
    ssl_context = create_ssl_context()
    await connect_to_chat(url, ssl_context)

if __name__ == "__main__":
    asyncio.run(main())

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

import certifi
import ssl
import base64
import asyncio
import websockets

# WSS 연결 위한 SSL 설정
ssl_context = ssl.create_default_context()
ssl_context.load_verify_locations(certifi.where())
ssl_context.check_hostname = False
ssl_context.verify_mode = ssl.CERT_NONE

# 아래 3개 변수값은 어딘가의 API로 부터 받은 값들의 조합으로 우선은 그대로 사용
Base64PdboxTicket = 'GwkwMDAxMDAwNTg3MDAMLkEzMi43YmJUNTZ2eUhNOWZLWmsuQ1V0N0dxNXdzSW95ZGdUejRvT0JkUGJEVTRPTEJjbDNEUkFUenJUYkhrQzlYRmtiZnl1YXMwUzE3SFlyZDFjbFU1VzV4U3hpTUJYS2lmV25fVXFDQThWWUYxczVidE04QzJuenhWN3NEVzlTcHFPM2V4VHNHQUMxU3ZmNTlCNUE1bDVrUFJsZHpleHB4OGxKbkJFYXE5UXNLbW1KQ21TTExQRm1JNkc3Q0FHZ1R0UDFWcGhzaTYzTDZXUHRDdkRPalZDNTg1LXVJQV9hRGRxcTlWX1pOWUlBQ0N3d3Yxb0NDckRLeE9icldKS0xpQlNrbGs3bW1TMkkwWVdtNnhiMDB2TG53OGJuQXVyZzRyNkpHeWVma0ZDWmZaM0V6c29LTEFwRU9xeFlkX0JXMVNxRWVNM1FuNkoyc0E5Y0d5WGFZLWhySm5wblJGNmN4RzFPTWJBSi1KVGVXc2JTLTNaNUNULS1fUW1vTWJCb2stSmJwUmt1Y1oxMURJSkFpY25NZmwxaWtzU1Y4aHh6YUVqWVExb19pamI1OXVCWUNYMlNsMFdLSEwydjk4WkhSLXZGNjNRRDY5VEtqSEpjaHBIaDh0RFNzUUxJekc3WUFZbmpKYjl3cDlvV2JfVi0wSFgyRFRYVnVUSEtKRTFPMWNtbHE1bC1qaXZ6bW1HdnJ0WnVmQVVfRG1NQzA1bUpPelNxZTl0bzNKVFZmMjBJWldfWXFpW...=='
Base64ChannelInfo = 'GwkwMDAyMDAwMjk3MDAMOTYxOAw0NTY1MzFjMDg0YjUxMGJkOTAyYWViZDcyMjFiMjUyNl92bGZ2bGY3ODlfMjQ1NTkwMzY1X2lvcwwwDAxsb2cRBiYGc2V0X2JwcwY9BjgwMDAGJgZ2aWV3X2JwcwY9BjEwMDAGJgZxdWFsaXR5Bj0Gbm9ybWFsBiYGdXVpZAY9BjFlNDNjZjZkMzc5MTNjMzZiMzVkNTgwZTBiNTY1NmVjBiYGZ2VvX2NjBj0GS1IGJgZnZW9fcmMGPQYxMQYmBmFjcHRfbGFuZwY9BmtvX0tSBiYGc3ZjX2xhbmcGPQZrb19LUgYmBmpvaW5fY2MGPQY0MTAScHdkERJhdXRoX2luZm8RTlVMTBJwdmVyETESYWNjZXNzX3N5c3R...='
WSSUrl = 'wss://chat-76dbfccf.afreecatv.com:8001/Websocket/vlfvlf789'

async def connect():
    # 웹 소켓에 접속을 합니다.
    async with websockets.connect(WSSUrl, subprotocols=['chat'], ssl=ssl_context, ping_interval=None) as websocket:

        # 핸드쉐이크
        await websocket.send(base64.b64decode(Base64PdboxTicket))
        await websocket.recv()
        await websocket.send(base64.b64decode(Base64ChannelInfo))

        # 이후부터 채팅내용 받아와짐
        while True:
            try:
                data = await websocket.recv()
                print(data)
            except Exception as e:
                print("ERROR:", e)

asyncio.get_event_loop().run_until_complete(connect())

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80

import certifi
import json
import ssl
import base64
import asyncio
import requests
import websockets

# WSS 연결 위한 SSL 설정
ssl_context = ssl.create_default_context()
ssl_context.load_verify_locations(certifi.where())
ssl_context.check_hostname = False
ssl_context.verify_mode = ssl.CERT_NONE

def decode(bytes):
    test = bytes.split(b'\x0c')
    res = []
    for i in test:
        res.append(str(i, 'utf-8'))
    if(res[1] != '-1' and res[1] != '1' and '|' not in res[1]):
        if(len(res) > 5):
            print(res[1], '\t| ', res[2], '|', res[6])
        else:
            print(res)
            pass

def player_live_api(bno, bid):
    # type=aid 일 경우 aid(.A32~~~)불러옴
    data = {'bid': bid, 'bno':bno, 'type':'live', 'confirm_adult':'false', 'player_type':'html5', 'mode':'landing', 'from_api':'0', 'pwd':'', 'stream_type':'common', 'quality':'HD'}
    res = requests.post(f'https://live.afreecatv.com/afreeca/player_live_api.php?bjid={bid}', data=data).json()
    CHDOMAIN = res["CHANNEL"]["CHDOMAIN"].lower()
    CHATNO = res["CHANNEL"]["CHATNO"]
    FTK = res["CHANNEL"]["FTK"]
    TITLE = res["CHANNEL"]["TITLE"]
    BJID = res["CHANNEL"]["BJID"]
    CHPT = str(int(res["CHANNEL"]["CHPT"]) + 1)
    return CHDOMAIN, CHATNO, FTK, TITLE, BJID, CHPT

async def connect(url):
    BNO = str(url.split('/')[-1])
    BID = str(url.split('/')[-2])
    CHDOMAIN, CHATNO, FTK, TITLE, BJID, CHPT = player_live_api(BNO, BID)

    KEY = ''
    if(len(BJID) == 5):
        KEY = '80'
    elif(len(BJID) == 6):
        KEY = '81'
    elif(len(BJID) == 7):
        KEY = '82'
    elif(len(BJID) == 8):
        KEY = '83'
    elif(len(BJID) == 9):
        KEY = '84'
    elif(len(BJID) == 10):
        KEY = '85'
    elif(len(BJID) == 11):
        KEY = '86'
    elif(len(BJID) == 12):
        KEY = '87'

    handshake = f'	00020002{KEY}00{CHATNO}{FTK}0log&set_bps=8000&view_bps=1000&quality=normal&uuid=1e43cf6d37913c36b35d580e0b5656ec&geo_cc=KR&geo_rc=11&acpt_lang=ko_KR&svc_lang=ko_KRpwdauth_infoNULLpver1access_systemhtml5'

    async with websockets.connect(f"wss://{CHDOMAIN}:{CHPT}/Websocket/{BID}", subprotocols=['chat'],ssl=ssl_context, ping_interval=None) as websocket:
        # 핸드쉐이크
        # await websocket.send(secret_1)
        await websocket.send('	00010000060016')
        data = await websocket.recv()
        await websocket.send(handshake)
        # 이후부터 채팅내용 받아와짐
        while True:
            try:
                data = await websocket.recv()
                decode(data)
            except Exception as e:
                print("ERROR:", e)
                # break

url = input("아프리카TV URL을 입력해주세요 : ")
asyncio.get_event_loop().run_until_complete(connect(url))

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79

import time
import re
import threading as th
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.service import Service
from webdriver_manager.chrome import ChromeDriverManager

# 상수, 아프리카도우미 채팅 URL, 컴퓨터 성능에 따라 STRESS 조절
URL = input('아프리카 도우미 채팅 url')
STRESS = 3
DEBUG = False
CAPTURING = True

# SELENIUM 초기화
DRIVER = webdriver.Chrome(service=Service(ChromeDriverManager().install()))

# global 변수
USERS = []      # 결과값
CONTENTS = []   # DEBUG일때 채팅 내용 터미널에 표시
CNT = 0


def stop_catpure():
    global CAPTURING
    input()
    CAPTURING = False

def start_capture():
    global USERS
    global CONTENTS
    global CNT

    DRIVER.get(URL)
    time.sleep(STRESS)

    th.Thread(target=stop_catpure, args=(), name='stop_catpure', daemon=True).start()
    while CAPTURING:
        try:
            DRIVER.implicitly_wait(STRESS)
            chat_list = DRIVER.find_element(By.CLASS_NAME, 'chat_list')
            chats = chat_list.find_elements(By.TAG_NAME, 'li')
            # print("chat len > ", len(chats))
            for chat in chats:
                name = chat.get_attribute('data-name')
                id = chat.get_attribute('data-id')
                classname = chat.get_attribute('class')
                platform = 'afreeca' if 'afreeca' in classname else 'youtube'
                user =  f'{name} | {platform}' if platform == 'afreeca' else f'{name} | @{id} | {platform}'
                secrets = str(re.sub(r'[^0-9]', '', classname)) + user
                if (user not in USERS) and (len(name) * len(id) != 0):
                    USERS.append(user)
                    content = chat.find_element(By.CLASS_NAME, 'text').text
                if secrets not in CONTENTS:
                    CONTENTS.append(secrets)
                    if DEBUG:
                        print(f'{platform} - {content} - {name} - {secrets}')
                    else:
                        print(f'------- \t 아이디 : {user}\n\t\t 내용 : {content}')
                    CNT += 1
            time.sleep(STRESS / 3)
        except KeyboardInterrupt:
            break
        except Exception as e:
            print("!", e)

def quit_capture():
    DRIVER.close()

def print_results():
    global USERS
    global CONTENTS
    users = (list(set(USERS)))
    print(users)


start_capture()
quit_capture()
print_results()