1
2
3

GET /video/play?vid=abc123&t=1702800000&sign=e3b0c44298fc
Host: video.example.com
Referer: https://video.example.com/watch/abc123

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

function generateSign(vid, timestamp) {
const key = "secret_key_2024";
const str = `${vid}|${timestamp}|${key}`;
return md5(str).substr(0, 12);
}

function getPlayUrl(vid) {
const timestamp = Math.floor(Date.now() / 1000);
const sign = generateSign(vid, timestamp);

return fetch(`/video/play?vid=${vid}&t=${timestamp}&sign=${sign}`, {
headers: {
Referer: "https://video.example.com/watch/" + vid,
},
}).then((r) => r.json());
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

// 混淆后
var _0x1a2b = [
"s",
"e",
"c",
"r",
"e",
"t",
"_",
"k",
"e",
"y",
"_",
"2",
"0",
"2",
"4",
];
var key = _0x1a2b.join("");

 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

import hashlib
import time
import requests

class VideoDownloader:
def __init__(self):
self.base_url = "https://video.example.com"
self.secret_key = "secret_key_2024"

def generate_sign(self, vid, timestamp):
"""生成视频签名"""
str_to_sign = f"{vid}|{timestamp}|{self.secret_key}"
return hashlib.md5(str_to_sign.encode()).hexdigest()[:12]

def get_play_url(self, vid):
"""获取播放地址"""
timestamp = int(time.time())
sign = self.generate_sign(vid, timestamp)

url = f"{self.base_url}/video/play"
params = {
'vid': vid,
't': timestamp,
'sign': sign
}
headers = {
'Referer': f'{self.base_url}/watch/{vid}',
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
}

response = requests.get(url, params=params, headers=headers)
data = response.json()

if data['code'] == 0:
return data['data']['url']
else:
raise Exception(f"Failed to get play url: {data}")

def download_video(self, vid, output_path):
"""下载视频"""
play_url = self.get_play_url(vid)
print(f"Play URL: {play_url}")

# 下载 M3U8 或直接视频
response = requests.get(play_url, stream=True)
with open(output_path, 'wb') as f:
for chunk in response.iter_content(chunk_size=8192):
f.write(chunk)

# 使用示例
downloader = VideoDownloader()
downloader.download_video('abc123', 'video.mp4')

1
2
3
4
5

#EXTM3U
#EXT-X-STREAM-INF:BANDWIDTH=1280000,RESOLUTION=1280x720
https://cdn.example.com/video/720p.m3u8
#EXT-X-STREAM-INF:BANDWIDTH=2560000,RESOLUTION=1920x1080
https://cdn.example.com/video/1080p.m3u8

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

#EXTM3U
#EXT-X-VERSION:3
#EXT-X-TARGETDURATION:10
#EXT-X-KEY:METHOD=AES-128,URI="https://cdn.example.com/key/abc123",IV=0x12345678901234567890123456789012
#EXTINF:10.0,
segment0.ts
#EXTINF:10.0,
segment1.ts
#EXTINF:10.0,
segment2.ts
#EXT-X-ENDLIST

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

// 在播放器代码中找到密钥获取逻辑
function getDecryptionKey(keyUri) {
return fetch(keyUri, {
headers: {
"X-Key-Token": generateKeyToken(),
},
}).then((r) => r.arrayBuffer());
}

function generateKeyToken() {
const timestamp = Date.now();
const nonce = Math.random().toString(36);
return btoa(timestamp + ":" + nonce);
}

  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
132
133
134
135
136
137
138

import re
import os
import requests
from Crypto.Cipher import AES
from concurrent.futures import ThreadPoolExecutor
from urllib.parse import urljoin

class M3U8Downloader:
def __init__(self, m3u8_url):
self.m3u8_url = m3u8_url
self.session = requests.Session()
self.session.headers.update({
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
})

def parse_m3u8(self, content, base_url):
"""解析 M3U8 文件"""
lines = content.strip().split('\n')
segments = []
key_info = None

for i, line in enumerate(lines):
line = line.strip()

# 解析加密密钥
if line.startswith('#EXT-X-KEY'):
match = re.search(r'METHOD=([^,]+),URI="([^"]+)"(?:,IV=(.+))?', line)
if match:
method, uri, iv = match.groups()
key_url = urljoin(base_url, uri)
key_info = {
'method': method,
'uri': key_url,
'iv': iv
}

# 解析 TS 片段
elif line.startswith('#EXTINF'):
if i + 1 < len(lines):
segment_url = urljoin(base_url, lines[i + 1].strip())
segments.append(segment_url)

return segments, key_info

def get_decryption_key(self, key_url):
"""获取解密密钥"""
# 添加必要的请求头
import time
import random
import base64

timestamp = int(time.time() * 1000)
nonce = ''.join(random.choices('abcdefghijklmnopqrstuvwxyz0123456789', k=8))
token = base64.b64encode(f"{timestamp}:{nonce}".encode()).decode()

headers = {'X-Key-Token': token}
response = self.session.get(key_url, headers=headers)
return response.content

def decrypt_segment(self, encrypted_data, key, iv):
"""解密 TS 片段"""
if iv.startswith('0x'):
iv_bytes = bytes.fromhex(iv[2:])
else:
iv_bytes = iv.encode()[:16].ljust(16, b'\0')

cipher = AES.new(key, AES.MODE_CBC, iv_bytes)
decrypted = cipher.decrypt(encrypted_data)

# 移除 PKCS7 填充
padding_length = decrypted[-1]
return decrypted[:-padding_length]

def download_segment(self, url, index, key_info=None):
"""下载并解密单个片段"""
try:
response = self.session.get(url, timeout=30)
data = response.content

# 如果有加密，进行解密
if key_info and key_info['method'] == 'AES-128':
key = self.get_decryption_key(key_info['uri'])
iv = key_info.get('iv', f'0x{index:032x}')
data = self.decrypt_segment(data, key, iv)

return index, data
except Exception as e:
print(f"Failed to download segment {index}: {e}")
return index, None

def download(self, output_path, max_workers=10):
"""下载完整视频"""
# 获取 M3U8 内容
response = self.session.get(self.m3u8_url)
m3u8_content = response.text

# 检查是否是 Master Playlist
if '#EXT-X-STREAM-INF' in m3u8_content:
# 选择最高质量
lines = m3u8_content.strip().split('\n')
for i, line in enumerate(lines):
if line.startswith('#EXT-X-STREAM-INF'):
if i + 1 < len(lines):
best_quality_url = urljoin(self.m3u8_url, lines[i + 1].strip())
response = self.session.get(best_quality_url)
m3u8_content = response.text
break

# 解析 M3U8
base_url = self.m3u8_url.rsplit('/', 1)[0] + '/'
segments, key_info = self.parse_m3u8(m3u8_content, base_url)

print(f"Found {len(segments)} segments")
if key_info:
print(f"Encryption: {key_info['method']}")

# 并发下载所有片段
with ThreadPoolExecutor(max_workers=max_workers) as executor:
futures = [
executor.submit(self.download_segment, url, i, key_info)
for i, url in enumerate(segments)
]

results = [future.result() for future in futures]

# 按顺序合并片段
results.sort(key=lambda x: x[0])

with open(output_path, 'wb') as f:
for index, data in results:
if data:
f.write(data)

print(f"Video saved to: {output_path}")

# 使用示例
downloader = M3U8Downloader('https://cdn.example.com/video/1080p.m3u8')
downloader.download('output.mp4', max_workers=20)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

def download_live_stream(self, output_path, duration=3600):
"""下载直播流（指定时长）"""
downloaded_segments = set()
start_time = time.time()

with open(output_path, 'wb') as f:
while time.time() - start_time < duration:
response = self.session.get(self.m3u8_url)
segments, key_info = self.parse_m3u8(response.text, self.m3u8_url)

for i, url in enumerate(segments):
if url not in downloaded_segments:
_, data = self.download_segment(url, i, key_info)
if data:
f.write(data)
downloaded_segments.add(url)

time.sleep(5) # 等待新片段

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

from xml.etree import ElementTree as ET

def parse_mpd(mpd_url):
"""解析 MPEG-DASH MPD 文件"""
response = requests.get(mpd_url)
root = ET.fromstring(response.content)

# 提取视频和音频流
video_urls = []
audio_urls = []

for adaptation_set in root.findall('.//{urn:mpeg:dash:schema:mpd:2011}AdaptationSet'):
content_type = adaptation_set.get('contentType')

for representation in adaptation_set.findall('.//{urn:mpeg:dash:schema:mpd:2011}Representation'):
base_url = representation.find('.//{urn:mpeg:dash:schema:mpd:2011}BaseURL').text

if content_type == 'video':
video_urls.append(base_url)
elif content_type == 'audio':
audio_urls.append(base_url)

return video_urls, audio_urls

1
2
3
4
5
6

# License Request
POST /widevine/license HTTP/1.1
Host: license.example.com
Content-Type: application/octet-stream

[Binary Protobuf Data: License Request]

1
2
3
4
5

message LicenseRequest {
ClientIdentification client_id = 1;
ContentIdentification content_id = 2;
bytes encrypted_client_id = 3;
}

 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

// Hook MediaKeySession
(function () {
const originalGenerateRequest = MediaKeySession.prototype.generateRequest;

MediaKeySession.prototype.generateRequest = function (
initDataType,
initData
) {
console.log("=== License Request ===");
console.log("Init Data Type:", initDataType);
console.log("Init Data:", new Uint8Array(initData));

// 拦截 PSSH (Protection System Specific Header)
const pssh = new Uint8Array(initData);
console.log(
"PSSH:",
Array.from(pssh)
.map((b) => b.toString(16).padStart(2, "0"))
.join(" ")
);

return originalGenerateRequest.call(this, initDataType, initData);
};

const originalUpdate = MediaKeySession.prototype.update;

MediaKeySession.prototype.update = function (response) {
console.log("=== License Response ===");
console.log("Response:", new Uint8Array(response));

return originalUpdate.call(this, response);
};
})();

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

// Frida script to hook Widevine CDM
Interceptor.attach(
Module.findExportByName("widevinecdm.dll", "DecryptAndDecode"),
{
onEnter: function (args) {
console.log("DecryptAndDecode called");
console.log("Encrypted buffer:", args[0]);
console.log("Buffer size:", args[1].toInt32());
},
onLeave: function (retval) {
console.log("Decrypted successfully:", retval);
},
}
);

 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

import frida
import sys

class WidevineL3Extractor:
def __init__(self):
self.keys = []

def on_message(self, message, data):
"""处理 Frida 消息"""
if message['type'] == 'send':
payload = message['payload']
if 'key' in payload:
print(f"Found key: {payload['key']}")
self.keys.append(payload)

def extract_keys(self, process_name='chrome.exe'):
"""从浏览器进程提取密钥"""
session = frida.attach(process_name)

script_code = """
// Hook Widevine OEMCrypto functions
var oemcrypto = Process.findModuleByName('widevinecdm.dll');

if (oemcrypto) {
var decrypt_func = oemcrypto.findExportByName('OEMCrypto_DecryptCTR');

Interceptor.attach(decrypt_func, {
onEnter: function(args) {
// args[2] is the key
var key = args[2];
var keyBuffer = Memory.readByteArray(key, 16);

send({
'type': 'key',
'key': Array.from(new Uint8Array(keyBuffer)).map(b =>
b.toString(16).padStart(2, '0')
).join('')
});
}
});
}
"""

script = session.create_script(script_code)
script.on('message', self.on_message)
script.load()

print("Waiting for keys... (play some DRM content)")
sys.stdin.read()

return self.keys

# 使用示例
extractor = WidevineL3Extractor()
keys = extractor.extract_keys('chrome.exe')

# 使用提取的密钥解密视频
for key_info in keys:
print(f"KID: {key_info.get('kid', 'N/A')}")
print(f"Key: {key_info['key']}")

 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

from Crypto.Cipher import AES

def decrypt_video_with_key(encrypted_file, output_file, key_hex, iv_hex=None):
"""使用提取的密钥解密视频"""
key = bytes.fromhex(key_hex)

if iv_hex:
iv = bytes.fromhex(iv_hex)
else:
# 默认 IV (全 0)
iv = b'\x00' * 16

cipher = AES.new(key, AES.MODE_CTR, nonce=iv[:8], initial_value=int.from_bytes(iv[8:], 'big'))

with open(encrypted_file, 'rb') as f_in:
with open(output_file, 'wb') as f_out:
while True:
chunk = f_in.read(64 * 1024)
if not chunk:
break
decrypted_chunk = cipher.decrypt(chunk)
f_out.write(decrypted_chunk)

# 使用示例
decrypt_video_with_key(
'encrypted_video.mp4',
'decrypted_video.mp4',
key_hex='0123456789abcdef0123456789abcdef'
)

 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

class AdaptiveBitrateController {
constructor() {
this.currentLevel = 0;
this.levels = [
{ width: 640, height: 360, bitrate: 800000 },
{ width: 1280, height: 720, bitrate: 2500000 },
{ width: 1920, height: 1080, bitrate: 5000000 },
{ width: 3840, height: 2160, bitrate: 15000000 },
];
}

selectLevel(bandwidth, bufferLevel) {
/**
* 选择算法:
* 1. 如果缓冲区低于 5 秒，降低质量
* 2. 如果带宽 > 比特率 * 1.5，可以升级
* 3. 考虑切换成本（避免频繁切换）
*/

if (bufferLevel < 5) {
// 缓冲区不足，降级
return Math.max(0, this.currentLevel - 1);
}

// 找到最高可用质量
for (let i = this.levels.length - 1; i >= 0; i--) {
if (bandwidth >= this.levels[i].bitrate * 1.5) {
// 避免频繁切换
if (Math.abs(i - this.currentLevel) <= 1) {
return i;
}
}
}

return this.currentLevel;
}

measureBandwidth() {
/**
* 带宽测量:
* 使用最近下载片段的速度估算
*/
const recentDownloads = this.downloadHistory.slice(-5);
const totalBytes = recentDownloads.reduce((sum, d) => sum + d.bytes, 0);
const totalTime = recentDownloads.reduce((sum, d) => sum + d.duration, 0);

return (totalBytes * 8) / totalTime; // bps
}
}

 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

// 方法 1: 覆盖 selectLevel 方法
(function () {
const originalSelectLevel = AdaptiveBitrateController.prototype.selectLevel;

AdaptiveBitrateController.prototype.selectLevel = function (
bandwidth,
bufferLevel
) {
// 总是返回最高质量索引
return this.levels.length - 1;
};

console.log("Forced highest quality");
})();

// 方法 2: 修改带宽测量结果
(function () {
const originalMeasureBandwidth =
AdaptiveBitrateController.prototype.measureBandwidth;

AdaptiveBitrateController.prototype.measureBandwidth = function () {
// 返回一个很大的带宽值
return 100000000; // 100 Mbps
};
})();

 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

// ==UserScript==
// @name Force Highest Video Quality
// @namespace http://tampermonkey.net/
// @version 1.0
// @description 强制视频播放最高质量
// @match https://video.example.com/*
// @grant none
// ==/UserScript==

(function () {
"use strict";

// 等待播放器加载
const observer = new MutationObserver((mutations, obs) => {
const player = document.querySelector("video");
if (player) {
obs.disconnect();
forceHighestQuality();
}
});

observer.observe(document.body, {
childList: true,
subtree: true,
});

function forceHighestQuality() {
// 方法 1: 修改 HTMLMediaElement
const video = document.querySelector("video");

Object.defineProperty(video, "playbackQuality", {
get: function () {
return {
totalVideoFrames: 10000,
droppedVideoFrames: 0,
corruptedVideoFrames: 0,
};
},
});

// 方法 2: Hook hls.js
if (window.Hls) {
const originalLoadSource = Hls.prototype.loadSource;

Hls.prototype.loadSource = function (url) {
// 强制加载最高质量的 m3u8
url = url.replace(/\d+p\.m3u8/, "2160p.m3u8");
return originalLoadSource.call(this, url);
};

// 禁用自动质量切换
const player = new Hls({
autoStartLoad: true,
startLevel: -1, // -1 表示最高质量
capLevelToPlayerSize: false, // 不限制质量
});
}

console.log("Forced to highest quality");
}
})();

 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

class VideoQualitySelector:
def __init__(self):
self.qualities = [
{'name': '360p', 'width': 640, 'height': 360, 'bitrate': 800000},
{'name': '720p', 'width': 1280, 'height': 720, 'bitrate': 2500000},
{'name': '1080p', 'width': 1920, 'height': 1080, 'bitrate': 5000000},
{'name': '4K', 'width': 3840, 'height': 2160, 'bitrate': 15000000},
]
self.current_quality = 1 # 默认 720p

def estimate_bandwidth(self, download_history):
"""估算当前带宽"""
if not download_history:
return 5000000 # 默认 5 Mbps

# 使用最近 5 次下载的平均速度
recent = download_history[-5:]
total_bytes = sum(d['bytes'] for d in recent)
total_time = sum(d['time'] for d in recent)

if total_time == 0:
return 5000000

return (total_bytes * 8) / total_time # bps

def select_quality(self, bandwidth, buffer_level, force_highest=False):
"""选择合适的视频质量"""
if force_highest:
return len(self.qualities) - 1

# 缓冲区不足，降级
if buffer_level < 5:
return max(0, self.current_quality - 1)

# 选择最高可用质量（带宽 > 比特率 * 1.5）
for i in range(len(self.qualities) - 1, -1, -1):
if bandwidth >= self.qualities[i]['bitrate'] * 1.5:
# 避免频繁切换（最多一次变化一级）
if abs(i - self.current_quality) <= 1:
self.current_quality = i
return i

return self.current_quality

def get_quality_url(self, base_url, quality_index):
"""获取指定质量的播放地址"""
quality = self.qualities[quality_index]
return f"{base_url}/{quality['name']}.m3u8"

# 使用示例
selector = VideoQualitySelector()

# 模拟下载历史
download_history = [
{'bytes': 1000000, 'time': 1.5}, # ~5.3 Mbps
{'bytes': 1200000, 'time': 1.8}, # ~5.3 Mbps
{'bytes': 1500000, 'time': 2.0}, # 6 Mbps
]

bandwidth = selector.estimate_bandwidth(download_history)
quality_index = selector.select_quality(
bandwidth=bandwidth,
buffer_level=10, # 10 秒缓冲
force_highest=True # 强制最高质量
)

print(f"Selected quality: {selector.qualities[quality_index]['name']}")

1
2
3
4
5

cdn.example.com -> CNAME -> geo.cdn.example.com
geo.cdn.example.com -> A 记录:
- 1.2.3.4 (北京节点)
- 5.6.7.8 (上海节点)
- 9.10.11.12 (广州节点)

1
2
3

GET /video/abc123.mp4
HTTP/1.1 302 Found
Location: https://bj-cdn01.example.com/video/abc123.mp4

 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

class CDNSelector {
constructor() {
this.nodes = [
{ id: "bj-01", url: "https://bj-cdn01.example.com", region: "beijing" },
{ id: "sh-01", url: "https://sh-cdn01.example.com", region: "shanghai" },
{ id: "gz-01", url: "https://gz-cdn01.example.com", region: "guangzhou" },
];
this.selectedNode = null;
}

async selectBestNode(testFile = "/ping.txt") {
/**
* 节点选择策略:
* 1. 并发测试所有节点的延迟
* 2. 选择延迟最低的节点
* 3. 缓存结果 5 分钟
*/

const results = await Promise.all(
this.nodes.map((node) => this.testNode(node, testFile))
);

// 按延迟排序
results.sort((a, b) => a.latency - b.latency);

this.selectedNode = results[0].node;
console.log(
`Selected CDN node: ${this.selectedNode.id} (${results[0].latency}ms)`
);

return this.selectedNode;
}

async testNode(node, testFile) {
const start = performance.now();

try {
await fetch(node.url + testFile, {
method: "HEAD",
cache: "no-cache",
});

const latency = performance.now() - start;
return { node, latency };
} catch (error) {
return { node, latency: Infinity };
}
}

getVideoUrl(path) {
if (!this.selectedNode) {
throw new Error("No CDN node selected");
}

return this.selectedNode.url + path;
}
}

 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

async function discoverCDNSources(videoId) {
/**
* 探测多个可能的 CDN 源
*/
const cdnProviders = [
"akamai.example.com",
"cloudflare.example.com",
"cloudfront.example.com",
"fastly.example.com",
];

const sources = [];

for (const provider of cdnProviders) {
const url = `https://${provider}/video/${videoId}.mp4`;

try {
const response = await fetch(url, { method: "HEAD" });
if (response.ok) {
sources.push({
provider: provider,
url: url,
size: response.headers.get("Content-Length"),
server: response.headers.get("Server"),
});
}
} catch (e) {
// 该 CDN 不可用
}
}

return sources;
}

  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
132
133

import time
import asyncio
import aiohttp
from typing import List, Dict

class CDNOptimizer:
def __init__(self, cdn_nodes: List[Dict]):
"""
cdn_nodes: [
{'id': 'bj-01', 'url': 'https://bj-cdn.example.com'},
{'id': 'sh-01', 'url': 'https://sh-cdn.example.com'},
...
]
"""
self.cdn_nodes = cdn_nodes
self.best_node = None

async def test_node_latency(self, session, node, test_file='/ping.txt'):
"""测试单个节点的延迟"""
url = node['url'] + test_file

try:
start = time.time()
async with session.head(url, timeout=5) as response:
latency = (time.time() - start) * 1000 # ms

return {
'node': node,
'latency': latency,
'status': response.status,
'server': response.headers.get('Server', 'Unknown')
}
except Exception as e:
return {
'node': node,
'latency': float('inf'),
'status': 0,
'error': str(e)
}

async def test_node_bandwidth(self, session, node, test_size_mb=1):
"""测试节点带宽"""
# 下载测试文件（例如 1MB）
test_url = f"{node['url']}/speedtest/{test_size_mb}mb.bin"

try:
start = time.time()
async with session.get(test_url, timeout=30) as response:
data = await response.read()
duration = time.time() - start

# 计算带宽 (Mbps)
bandwidth = (len(data) * 8) / (duration * 1_000_000)

return {
'node': node,
'bandwidth': bandwidth,
'duration': duration
}
except Exception as e:
return {
'node': node,
'bandwidth': 0,
'error': str(e)
}

async def select_best_node(self, mode='latency'):
"""选择最佳节点"""
async with aiohttp.ClientSession() as session:
if mode == 'latency':
# 基于延迟选择
tasks = [
self.test_node_latency(session, node)
for node in self.cdn_nodes
]
results = await asyncio.gather(*tasks)

# 排序并选择延迟最低的
results.sort(key=lambda x: x['latency'])
self.best_node = results[0]['node']

print(f"Best node (latency): {self.best_node['id']}")
for r in results[:3]: # 显示前 3 个
print(f" {r['node']['id']}: {r['latency']:.2f}ms")

elif mode == 'bandwidth':
# 基于带宽选择
tasks = [
self.test_node_bandwidth(session, node)
for node in self.cdn_nodes
]
results = await asyncio.gather(*tasks)

# 排序并选择带宽最高的
results.sort(key=lambda x: x['bandwidth'], reverse=True)
self.best_node = results[0]['node']

print(f"Best node (bandwidth): {self.best_node['id']}")
for r in results[:3]:
print(f" {r['node']['id']}: {r['bandwidth']:.2f} Mbps")

return self.best_node

def get_video_url(self, video_path):
"""获取使用最佳节点的视频 URL"""
if not self.best_node:
raise Exception("Please select best node first")

return self.best_node['url'] + video_path

# 使用示例
async def main():
cdn_nodes = [
{'id': 'bj-01', 'url': 'https://bj-cdn.example.com'},
{'id': 'sh-01', 'url': 'https://sh-cdn.example.com'},
{'id': 'gz-01', 'url': 'https://gz-cdn.example.com'},
{'id': 'sz-01', 'url': 'https://sz-cdn.example.com'},
]

optimizer = CDNOptimizer(cdn_nodes)

# 方法 1: 基于延迟选择
await optimizer.select_best_node(mode='latency')

# 方法 2: 基于带宽选择
# await optimizer.select_best_node(mode='bandwidth')

# 获取视频 URL
video_url = optimizer.get_video_url('/video/abc123.mp4')
print(f"Video URL: {video_url}")

# 运行
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
36
37
38

import dns.resolver
import socket

def discover_all_cdn_nodes(domain):
"""通过 DNS 查询发现所有 CDN 节点"""
nodes = []

try:
# 解析 CNAME
answers = dns.resolver.resolve(domain, 'CNAME')
for rdata in answers:
print(f"CNAME: {rdata.target}")

# 解析 A 记录
a_records = dns.resolver.resolve(str(rdata.target), 'A')
for a_rdata in a_records:
ip = str(a_rdata)

# 反向 DNS 查询获取节点名称
try:
hostname = socket.gethostbyaddr(ip)[0]
except:
hostname = ip

nodes.append({
'ip': ip,
'hostname': hostname
})

print(f" -> {ip} ({hostname})")

except Exception as e:
print(f"Error: {e}")

return nodes

# 使用示例
nodes = discover_all_cdn_nodes('cdn.example.com')