1. m3u8 파일 요청 링크 탐지

Douyu에서 특정 영상의 m3u8 파일을 얻기 위해 개발자 도구(F12)를 사용하여 네트워크 요청을 분석해보자. XHR 필터를 적용하고 페이지를 새로고침하면 비디오 스트림 정보를 가져오는 API 요청을 확인할 수 있다.

getStreamUrl POST 요청의 상세 정보는 다음과 같다:

2. sign 파라미터 암호화 원리 분석 (JS 역분석)

개발자 도구의 호출 스택을 확인하면 여러 JavaScript 파일이 로드되는 것을 볼 수 있다. 특정 스크립트 파일을 추적하면 암호화 함수가 포함된 코드를 찾을 수 있다.

해당 함수를仔细 분석해보면, eval 함수를 사용하여 동적으로 JavaScript 코드를 실행하는 구조이다. strc 변수에 포함된 문자열 코드에는 sign 생성 로직이 들어있으며, CryptoJS库的 MD5 함수를 사용하고 있다.

핵심 암호화 로직은 다음과 같다:

var cb = xx0 + xx1 + xx2 + "220320220115";
var rb = CryptoJS.MD5(cb).toString();

여기서 xx0, xx1, xx2는 각각 비디오 ID, 디바이스 ID, 타임스탬프를 의미한다.

3. Python으로 암호화 시뮬레이션

JavaScript의 암호화 로직을 Python으로 구현하려면 다음과 같은 접근이 필요하다. 먼저 JavaScript 코드를 추출하고, 필요한 부분을 Python으로 재구현하거나 execjs库를 활용한다.

import execjs
import hashlib
import re
import time

# JavaScript 암호화 함수 추출
js_code = '''
(function(xx0, xx1, xx2) {
    var cb = xx0 + xx1 + xx2 + "220320220115";
    var rb = CryptoJS.MD5(cb).toString();
    
    // 추가 암호화 로직...
    var re = [];
    for(var i = 0; i < rb.length / 8; i++) {
        re[i] = (parseInt(rb.substr(i * 8, 2), 16) & 0xff) |
                ((parseInt(rb.substr(i * 8 + 2, 2), 16) << 8) & 0xff00) |
                ((parseInt(rb.substr(i * 8 + 4, 2), 16) << 24) >>> 8) |
                (parseInt(rb.substr(i * 8 + 6, 2), 16) << 24);
    }
    
    // 후처리 로직...
    var rt = "v=220320220115" + "&did=" + xx1 + "&tt=" + xx2 + "&sign=" + re;
    return rt;
});
'''

# 파라미터 설정
video_id = "25685173"
device_id = "10000000000000000000000000001501"
timestamp = int(time.time())

# MD5 해시 생성
combined_str = str(video_id) + device_id + str(timestamp) + "220320220115"
md5_hash = hashlib.md5(combined_str.encode('utf-8')).hexdigest()

# JavaScript 함수 호출
ctx = execjs.compile(js_code)
result = ctx.call('function_name', video_id, device_id, timestamp)
print(result)

이 방식을 통해 JavaScript에서 생성되는 sign 값을 Python으로 동일하게 생성할 수 있다.

4. 전체 구현 코드

import requests
import re
import demjson
import execjs
import hashlib
import time
import m3u8
from queue import Queue
import os
import threading

def get_video_info(url):
    """비디오 정보 추출"""
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
    }
    
    response = requests.get(url=url, headers=headers)
    response.encoding = 'utf-8'
    html_content = response.text
    
    # 비디오 ID 추출
    video_id = re.findall(r'https://v.douyu.com/show/(.*?)\?', url)[0]
    
    # 페이지 데이터 파싱
    json_data = re.findall(r'window.\$DATA=(.*?);</script>', html_content)[0]
    data_dict = demjson.decode(json_data)
    
    return video_id, data_dict

def generate_signature(video_id, device_id, timestamp):
    """서명 생성"""
    # JavaScript 코드에서 필요한 부분 추출 및 수정
    js_code = '''
    // 암호화 로직 구현
    '''
    
    combined = f"{video_id}{device_id}{timestamp}220320220115"
    md5_result = hashlib.md5(combined.encode('utf-8')).hexdigest()
    
    # 추가 처리 로직...
    return md5_result

def download_m3u8_stream(m3u8_url, output_dir):
    """m3u8 스트림 다운로드"""
    os.makedirs(output_dir, exist_ok=True)
    
    m3u8_obj = m3u8.load(uri=m3u8_url)
    queue = Queue()
    
    base_url = m3u8_url[:m3u8_url.rfind('/') + 1]
    
    for segment in m3u8_obj.files:
        ts_url = base_url + segment if not segment.startswith('http') else segment
        queue.put(ts_url)
    
    return queue

def worker(ts_queue, directory, thread_name):
    """워커 스레드 함수"""
    while not ts_queue.empty():
        url = ts_queue.get()
        response = requests.get(url, stream=True)
        
        filename = re.findall(r'[a-zA-Z0-9\-]+.ts', url)[0]
        
        with open(f"{directory}/{filename}", 'wb') as f:
            f.write(response.content)
        
        print(f"{thread_name}: {filename} 다운로드 완료")

def merge_ts_files(directory, output_name):
    """TS 파일 병합"""
    files = os.listdir(directory)
    files.sort(key=lambda x: int(x[:x.rfind('.ts')]))
    
    with open(f"{directory}/file_list.txt", 'w', encoding='utf-8') as f:
        for ts_file in files:
            f.write(f"file '{ts_file}'\n")
    
    os.system(f'ffmpeg -f concat -i {directory}/file_list.txt -c copy {output_name}')
    
    # 정리
    for ts_file in files:
        os.remove(f"{directory}/{ts_file}")
    os.remove(f"{directory}/file_list.txt")

# 메인 실행
if __name__ == "__main__":
    video_url = input("비디오 URL 입력: ")
    output_directory = input("저장 폴더 이름: ")
    
    vid, page_data = get_video_info(video_url)
    
    # 요청 데이터 구성
    device_id = "10000000000000000000000000001501"
    timestamp = int(time.time())
    
    # 서명 생성
    sign = generate_signature(vid, device_id, timestamp)
    
    # API 요청
    api_url = "https://v.douyu.com/api/stream/getStreamUrl"
    payload = {
        'v': '220320220115',
        'did': device_id,
        'tt': timestamp,
        'sign': sign,
        'vid': vid
    }
    
    api_response = requests.post(api_url, data=payload)
    stream_data = demjson.decode(api_response.text)
    
    # 스트림 URL 확보
    video_streams = stream_data['data']['thumb_video']
    m3u8_link = video_streams['high']['url']
    
    # 다운로드 실행
    ts_queue = download_m3u8_stream(m3u8_link, output_directory)
    
    threads = []
    for i in range(15):
        thread = threading.Thread(target=worker, args=(ts_queue, output_directory, f"스레드-{i}"))
        threads.append(thread)
        thread.start()
    
    for thread in threads:
        thread.join()
    
    # 병합
    merge_ts_files(output_directory, "output.mp4")
    print("다운로드 및 병합 완료!")

5. TS 파일 병합

import os

# ffmpeg를 사용한 병합
os.system('ffmpeg -f concat -i file_list.txt -c copy output.mp4')

# 임시 파일 정리
os.system('del *.ts')
os.system('del file_list.txt')

참고: ffmpeg가 설치되어 있고 환경 변수가 설정되어 있어야 한다.

이 구현의 핵심은 요청 파라미터 중 sign 값의 암호화 로직을 정확하게 분석하고 재현하는 것이다. Douyu의 암호화 방식은 비디오마다 다를 수 있으므로, 실제 적용 시 해당 비디오의 JavaScript 코드를 분석하여 적절히 수정해야 한다.