import struct
import json
import argparse
from collections import namedtuple
from typing import List, Tuple, Dict, Optional, Any
# 定义数据结构
FLVHeader = namedtuple('FLVHeader', ['signature', 'version', 'flags', 'header_size'])
FLVTagHeader = namedtuple('FLVTagHeader', [
'prev_tag_size', 'tag_type', 'data_size', 'timestamp', 'timestamp_ext', 'stream_id'
])
MetadataHeader = namedtuple('MetadataHeader', [
'AMF1_type', 'size', 'onMetadata', "AMF2_type", 'arr_size'
])
MetadataArray = namedtuple('MetadataArray', [
'duration', 'width', 'height', 'videodatarate', 'framerate', 'videocodecid', 'audiodatarate',
'audiosamplerate', 'audiosamplesize', 'stereo', 'audiocodecid', 'encoder', 'filesize', "custom_fields"
])
AudioHeader = namedtuple('AudioHeader', [
'format', 'rate', 'size', 'type', 'aac_packet_type', 'audio_object_type',
'sampling_index', 'channel_config'
])
VideoHeader = namedtuple('VideoHeader', [
'frame_type', 'codec_id', 'avc_packet_type', 'composition_time'
])
NALUnit = namedtuple('NALUnit', ['type', 'ref_idc', 'size', 'data'])
# 常量定义
AUDIO_FORMATS = {
0: "Linear PCM, platform endian",
1: "ADPCM",
2: "MP3",
3: "Linear PCM, little endian",
4: "Nellymoser 16kHz mono",
5: "Nellymoser 8kHz mono",
6: "Nellymoser",
7: "G.711 A-law",
8: "G.711 mu-law",
10: "AAC",
11: "Speex",
14: "MP3 8kHz",
15: "Device-specific sound"
}
SAMPLE_RATES = {
0: "5.5kHz",
1: "11kHz",
2: "22kHz",
3: "44kHz"
}
FRAME_TYPES = {
1: "Keyframe",
2: "Inter frame",
3: "Disposable inter frame",
4: "Generated keyframe",
5: "Video info/command frame"
}
CODEC_IDS = {
1: "JPEG",
2: "Sorenson H.263",
3: "Screen video",
4: "On2 VP6",
5: "On2 VP6 with alpha",
6: "Screen video v2",
7: "AVC"
}
AVC_PACKET_TYPES = {
0: "AVC sequence header",
1: "AVC NALU",
2: "AVC end of sequence"
}
NALU_TYPES = {
1: "Coded slice of a non-IDR picture",
5: "Coded slice of an IDR picture",
6: "Supplemental enhancement information (SEI)",
7: "Sequence parameter set",
8: "Picture parameter set",
9: "Access unit delimiter",
10: "End of sequence",
11: "End of stream",
12: "Filler data",
13: "Sequence parameter set extension",
14: "Prefix NAL unit",
15: "Subset sequence parameter set",
19: "Coded slice of an auxiliary coded picture without partitioning",
20: "Coded slice extension"
}
NALU_REF_IDC = {
0: "Disposable",
1: "Lowest",
2: "Low",
3: "High"
}
SCRIPT_DATA_VALUE = {
0: "Number",
1: "Boolean",
2: "String",
3: "Object",
4: "MovieClip (reserved, not supported)",
5: "Null",
6: "Undefined",
7: "Reference",
8: "ECMA array",
9: "Object end marker",
10: "Strict array",
11: "Date",
12: "Long string"
}
class FLVProcessor:
def __init__(self, input_file: str):
self.input_file = input_file
self.header: Optional[FLVHeader] = None
self.tags: List[Tuple] = []
self.raw_data: bytes = b''
def read_file(self) -> None:
#读取文件内容
with open(self.input_file, 'rb') as f:
self.raw_data = f.read()
def _process_http_response(self, data: bytes) -> bytes:
#处理HTTP响应，自动判断是否为chunk编码
if not data.startswith(b'HTTP/1.1 200 OK'):
return data
# 查找HTTP头结束位置
http_header_end = data.find(b'\r\n\r\n') + 4
if http_header_end < 4: # 没有找到完整的HTTP头
return data
# 检查是否为chunk编码
headers = data[:http_header_end-4].split(b'\r\n')
is_chunked = any(b'Transfer-Encoding: chunked' in h for h in headers)
# 提取主体数据
body_data = data[http_header_end:]
# 如果是chunk编码，则解码
if is_chunked:
return self._decode_chunked_data(body_data)
return body_data
def _decode_chunked_data(self, data: bytes) -> bytes:
#解码HTTP chunk传输编码的数据
result = b''
pos = 0
while pos < len(data):
# 查找块大小行结束位置
chunk_size_end = data.find(b'\r\n', pos)
if chunk_size_end == -1:
break
# 解析块大小（16进制）
chunk_size_str = data[pos:chunk_size_end].decode('ascii').strip()
try:
chunk_size = int(chunk_size_str, 16)
except ValueError:
break
# 0大小的块表示结束
if chunk_size == 0:
break
# 移动到块数据开始位置
chunk_start = chunk_size_end + 2
chunk_end = chunk_start + chunk_size
# 检查是否有足够的空间
if chunk_end > len(data):
break
# 添加块数据到结果
result += data[chunk_start:chunk_end]
# 移动到下一个块开始位置（跳过CRLF）
pos = chunk_end + 2
return result
def parse(self) -> None:
#解析FLV文件
if not self.raw_data:
self.read_file()
# 处理可能的HTTP响应
flv_data = self._process_http_response(self.raw_data)
# 验证FLV签名
if not flv_data.startswith(b'FLV'):
raise ValueError("不是有效的FLV文件, 签名不匹配")
# 解析FLV头
self.header = self._parse_flv_header(flv_data[:9])
pos = 9
# 解析标签
while pos + 15 < len(flv_data):
# 解析前一个标签大小
prev_tag_size = struct.unpack('>I', flv_data[pos:pos+4])[0]
pos += 4
# 解析标签头
tag_type = flv_data[pos]
data_size = struct.unpack('>I', b'\x00' + flv_data[pos+1:pos+4])[0]
timestamp = struct.unpack('>I', flv_data[pos+4:pos+7] + b'\x00')[0] >> 8
timestamp_ext = flv_data[pos+7]
stream_id = struct.unpack('>I', b'\x00' + flv_data[pos+8:pos+11])[0]
tag_header = FLVTagHeader(
prev_tag_size=prev_tag_size,
tag_type=tag_type,
data_size=data_size,
timestamp=timestamp,
timestamp_ext=timestamp_ext,
stream_id=stream_id
)
pos += 11
# 检查数据大小是否有效
if pos + data_size > len(flv_data):
print(f"警告: 数据大小({data_size})超出文件范围，终止解析")
break
# 读取标签数据
tag_data = flv_data[pos:pos+data_size]
pos += data_size
# 根据标签类型解析
if tag_type == 8: # 音频
audio_info = self._parse_audio_tag(tag_data)
self.tags.append(('audio', tag_header, audio_info, tag_data))
elif tag_type == 9: # 视频
video_info, nal_units, video_body = self._parse_video_tag(tag_data)
self.tags.append(('video', tag_header, (video_info, nal_units, video_body), tag_data))
elif tag_type == 18: # 脚本数据/Metadata
metadata_info, metadata_array = self._parse_flv_metadate(tag_data)
self.tags.append(('metadata', tag_header, (metadata_info, metadata_array), tag_data))
# self.tags.append(('metadata', tag_header, tag_data, tag_data))
else:
self.tags.append(('unknown', tag_header, tag_data, tag_data))
def _parse_flv_header(self, header_data: bytes) -> FLVHeader:
#解析FLV头
signature = header_data[:3].decode('ascii')
version = header_data[3]
flags = header_data[4]
header_size = struct.unpack('>I', header_data[5:9])[0]
return FLVHeader(signature, version, flags, header_size)
def _parse_flv_metadate(self , data: bytes) -> Tuple[Optional[MetadataHeader], Optional[MetadataArray]]:
#解析Metadata
if not data:
return None, None
pos = 0
AMF1_type = data[pos]
pos += 1
# 字符串长度
name_size = struct.unpack('>H', data[pos:pos+2])[0]
pos += 2
# 检查剩余数据是否足够
if len(data) < pos + name_size + 1:
return None, None
# 提取字符串
onMetadata = data[pos:pos+name_size].decode('utf-8')
pos += name_size
# AMF2类型
AMF2_type = data[pos]
pos += 1
metadata_dict = {}
if AMF2_type == 0x08: # 0x08 ECMA数组
arr_size = struct.unpack('>I', data[pos:pos+4])[0]
pos += 4
metadata_dict, pos = self._parse_amf_object(data, pos)
elif AMF2_type == 0x03: # 0x03 对象
metadata_dict, pos = self._parse_amf_object(data, pos)
arr_size = len(metadata_dict)
elif AMF2_type == 0x0A: # 0x0A 严格数组
arr_size = struct.unpack('>I', data[pos:pos+4])[0]
pos += 4
metadata_dict, pos = self._parse_amf_strict_array(data, pos, arr_size)
elif AMF2_type == 0x05: # 0x05 Null
pass # metadata_dict保持为空
elif AMF2_type == 0x0B: # 0x0B Date
timestamp = struct.unpack('>d', data[pos:pos+8])[0]
pos += 8
timezone = struct.unpack('>h', data[pos:pos+2])[0]
pos += 2
metadata_dict["date"] = f"Date({timestamp}, tz={timezone})"
elif AMF2_type == 0x0C: # 0x0C 长字符串
str_size = struct.unpack('>I', data[pos:pos+4])[0]
pos += 4
metadata_dict["long_string"] = data[pos:pos+str_size].decode('utf-8', errors='replace')
pos += str_size
elif AMF2_type in [0x06, 0x0D]: # 0x06 或 0x0D
metadata_dict["unsupported"] = None
else: # 其他未实现类型（如0x04 MovieClip）
raise ValueError(f"不支持的AMF2类型: {hex(AMF2_type)}")
header = MetadataHeader(
AMF1_type=AMF1_type,
size=name_size,
onMetadata=onMetadata,
AMF2_type=AMF2_type,
arr_size=arr_size
)
standard_data = {}
custom_fields = {}
standard_fields = {
'duration': 'duration',
'width': 'width',
'height': 'height',
'videodatarate': 'videodatarate',
'framerate': 'framerate',
'videocodecid': 'videocodecid',
'audiodatarate': 'audiodatarate',
'audiosamplerate': 'audiosamplerate',
'audiosamplesize': 'audiosamplesize',
'stereo': 'stereo',
'audiocodecid': 'audiocodecid',
'encoder': 'encoder',
'filesize': 'filesize'
}
for key_in_data, mapped_key in standard_fields.items():
if key_in_data in metadata_dict:
standard_data[mapped_key] = metadata_dict[key_in_data]
for key, value in metadata_dict.items():
if key not in standard_fields:
custom_fields[key] = value
metadata = MetadataArray(
duration=standard_data.get('duration'),
width=standard_data.get('width'),
height=standard_data.get('height'),
videodatarate=standard_data.get('videodatarate'),
framerate=standard_data.get('framerate'),
videocodecid=standard_data.get('videocodecid'),
audiodatarate=standard_data.get('audiodatarate'),
audiosamplerate=standard_data.get('audiosamplerate'),
audiosamplesize=standard_data.get('audiosamplesize'),
stereo=standard_data.get('stereo'),
audiocodecid=standard_data.get('audiocodecid'),
encoder=standard_data.get('encoder'),
filesize=standard_data.get('filesize'),
custom_fields = custom_fields
)
return header, metadata
def _parse_amf_value(self, data: bytes, pos: int) -> Tuple[Any, int]:
#单值
value_type = data[pos]
pos += 1
if value_type == 0x00: # Double
value = struct.unpack('>d', data[pos:pos+8])[0]
pos += 8
elif value_type == 0x01: # Boolean
value = bool(data[pos])
pos += 1
elif value_type == 0x02: # String
str_size = struct.unpack('>H', data[pos:pos+2])[0]
pos += 2
value = data[pos:pos+str_size].decode('latin-1')
pos += str_size
elif value_type == 0x03: # Object
value, pos = self._parse_amf_object(data, pos)
elif value_type == AMF_NULL: # Null
value = None
elif value_type == 0x06: # Undefined
value = "undefined"
elif value_type == 0x08: # ECMA Array
arr_size = struct.unpack('>I', data[pos:pos+4])[0]
pos += 4
value, pos = self._parse_amf_object(data, pos)
elif value_type == 0x0A: # Strict Array
arr_size = struct.unpack('>I', data[pos:pos+4])[0]
pos += 4
value, pos = self._parse_amf_strict_array(data, pos, arr_size)
elif value_type == 0X0B: # Date
timestamp = struct.unpack('>d', data[pos:pos+8])[0]
pos += 8
timezone = struct.unpack('>h', data[pos:pos+2])[0]
pos += 2
value = f"Date({timestamp}, tz={timezone})"
elif value_type == 0X0C: # Long String
str_size = struct.unpack('>I', data[pos:pos+4])[0]
pos += 4
value = data[pos:pos+str_size].decode('latin-1')
pos += str_size
else:
raise ValueError(f"Unknown AMF type: {hex(value_type)}")
return value, pos
def _parse_amf_object(self, data: bytes, pos: int) -> Tuple[Dict[str, Any], int]:
#object
obj = {}
while pos < len(data):
# 检查对象结束标记（0x000009）
if data[pos:pos+3] == b'\x00\x00\x09':
pos += 3
break
# 解析键名
key_size = struct.unpack('>H', data[pos:pos+2])[0]
pos += 2
key = data[pos:pos+key_size].decode('latin-1')
pos += key_size
# 解析值
value, pos = self._parse_amf_value(data, pos)
obj[key] = value
return obj, pos
def _parse_amf_strict_array(self, data: bytes, pos: int, size: int) -> Tuple[Dict[str, Any], int]:
#strict array
arr = {}
for i in range(size):
value, pos = self._parse_amf_value(data, pos)
arr[str(i)] = value
return arr, pos
def _parse_audio_tag(self, data: bytes) -> Optional[AudioHeader]:
#解析音频标签
if not data:
return None
header_byte = data[0]
format = (header_byte & 0xF0) >> 4
rate = (header_byte & 0x0C) >> 2
size = (header_byte & 0x02) >> 1
audio_type = header_byte & 0x01
aac_packet_type = None
audio_object_type = None
sampling_index = None
channel_config = None
if format == 10: # AAC
if len(data) > 1:
aac_packet_type = data[1]
if aac_packet_type == 0 and len(data) > 2: # AAC sequence header
audio_specific_config = data[2:]
if audio_specific_config:
audio_object_type = (audio_specific_config[0] & 0xF8) >> 3
sampling_index = ((audio_specific_config[0] & 0x07) << 1) | ((audio_specific_config[1] & 0x80) >> 7)
channel_config = (audio_specific_config[1] & 0x78) >> 3
return AudioHeader(
format=format,
rate=rate,
size=size,
type=audio_type,
aac_packet_type=aac_packet_type,
audio_object_type=audio_object_type,
sampling_index=sampling_index,
channel_config=channel_config
)
def _parse_video_tag(self, data: bytes) -> Tuple[Optional[VideoHeader], List[NALUnit], Optional[bytes]]:
#解析视频标签
if not data:
return None, [], None
header_byte = data[0]
frame_type = (header_byte & 0xF0) >> 4
codec_id = header_byte & 0x0F
avc_packet_type = None
composition_time = None
nal_units = []
video_body = None
if codec_id == 7: # AVC
if len(data) > 4:
avc_packet_type = data[1]
composition_time = struct.unpack('>I', b'\x00' + data[2:5])[0]
if avc_packet_type == 0: # AVC sequence header
video_body = data[5:]
elif avc_packet_type == 1: # AVC NALU
pos = 5
while pos + 4 < len(data):
nalu_size = struct.unpack('>I', data[pos:pos+4])[0]
pos += 4
if pos + nalu_size > len(data):
break
nalu_data = data[pos:pos+nalu_size]
pos += nalu_size
if nalu_data:
nalu_header = nalu_data[0]
nalu_type = nalu_header & 0x1F
nalu_ref_idc = (nalu_header & 0x60) >> 5
nal_units.append(NALUnit(
type=nalu_type,
ref_idc=nalu_ref_idc,
size=nalu_size,
data=nalu_data
))
video_body = data[5:pos]
else:
video_body = data[5:]
else:
video_body = data[1:]
return VideoHeader(
frame_type=frame_type,
codec_id=codec_id,
avc_packet_type=avc_packet_type,
composition_time=composition_time
), nal_units, video_body
def save_as_json(self, output_path: str) -> None:
#将解析结果保存为JSON
if not self.header:
raise ValueError("未解析FLV文件, 请先调用parse()方法")
result = {
'file': self.input_file,
'header': {
'signature': self.header.signature,
'version': self.header.version,
'flags': self.header.flags,
'header_size': self.header.header_size
},
'statistics': {
'total_tags': len(self.tags),
'audio_tags': sum(1 for tag in self.tags if tag[0] == 'audio'),
'video_tags': sum(1 for tag in self.tags if tag[0] == 'video'),
'metadata_tags': sum(1 for tag in self.tags if tag[0] == 'metadata'),
'unknown_tags': sum(1 for tag in self.tags if tag[0] == 'unknown')
},
'tags': []
}
for tag in self.tags:
tag_type, tag_header, tag_data, raw_data = tag
tag_dict = {
'prev_tag_size': tag_header.prev_tag_size,
'type': tag_type,
'data_size': tag_header.data_size,
'timestamp': tag_header.timestamp,
'timestamp_extended': tag_header.timestamp_ext,
'stream_id': tag_header.stream_id,
'details': {}
}
if tag_type == 'audio':
audio_header = tag_data
tag_dict['details'] = {
'format': audio_header.format,
'format_description': AUDIO_FORMATS.get(audio_header.format, "Unknown"),
'sample_rate': audio_header.rate,
'sample_rate_description': SAMPLE_RATES.get(audio_header.rate, "Unknown"),
'sample_size': audio_header.size,
'channels': audio_header.type,
'aac_packet_type': audio_header.aac_packet_type,
'audio_object_type': audio_header.audio_object_type,
'sampling_index': audio_header.sampling_index,
'channel_config': audio_header.channel_config
}
elif tag_type == 'video':
video_header, nal_units, video_body = tag_data
tag_dict['details'] = {
'frame_type': video_header.frame_type,
'frame_type_description': FRAME_TYPES.get(video_header.frame_type, "Unknown"),
'codec_id': video_header.codec_id,
'codec_description': CODEC_IDS.get(video_header.codec_id, "Unknown"),
'avc_packet_type': video_header.avc_packet_type,
'avc_packet_type_description': AVC_PACKET_TYPES.get(video_header.avc_packet_type, "Unknown"),
'composition_time': video_header.composition_time,
'nal_units_count': len(nal_units),
'video_body_size': len(video_body) if video_body else 0,
'nal_units': [{
'type': unit.type,
'type_description': NALU_TYPES.get(unit.type, "Unknown"),
'ref_idc': unit.ref_idc,
'ref_idc_description': NALU_REF_IDC.get(unit.ref_idc, "Unknown"),
'size': unit.size
} for unit in nal_units]
}
elif tag_type == 'metadata':
metadata_info, metadata_array = tag_data
tag_dict['details'] = {
"AMF1_type": metadata_info.AMF1_type,
"AMF1_type_description": SCRIPT_DATA_VALUE[metadata_info.AMF1_type],
"size":metadata_info.size,
SCRIPT_DATA_VALUE[metadata_info.AMF1_type]: metadata_info.onMetadata,
"AMF2_type": metadata_info.AMF2_type,
"AMF2_type_description": SCRIPT_DATA_VALUE[metadata_info.AMF2_type],
SCRIPT_DATA_VALUE[metadata_info.AMF2_type] + " size": metadata_info.arr_size,
"Metadata_array_data": [{
"duration": metadata_array.duration,
"width": metadata_array.width,
"height": metadata_array.height,
"videodatarate": metadata_array.videodatarate,
"framerate": metadata_array.framerate,
"videocodecid": metadata_array.videocodecid,
"audiodatarate": metadata_array.audiodatarate,
"audiosamplerate": metadata_array.audiosamplerate,
"audiosamplesize": metadata_array.audiosamplesize,
"stereo": metadata_array.stereo,
"audiocodecid": metadata_array.audiocodecid,
"encoder": metadata_array.encoder,
"filesize": metadata_array.filesize,
**metadata_array.custom_fields
}]
}
# try:
# tag_dict['details'] = {'metadata': tag_data.decode('utf-8', errors='replace')}
# except UnicodeDecodeError:
# tag_dict['details'] = {'metadata': '二进制数据，无法解码为文本'}
result['tags'].append(tag_dict)
with open(output_path, 'w', encoding='utf-8') as f:
json.dump(result, f, indent=2, ensure_ascii=False)
print(f"已保存JSON解析结果到: {output_path}")
def main():
parser = argparse.ArgumentParser(
description='FLV文件解析与保存工具 - 自动处理HTTP和chunk编码 兼容AMF解析',
formatter_class=argparse.RawTextHelpFormatter
)
parser.add_argument('input', help='输入FLV文件路径')
parser.add_argument('--save-json', metavar='OUTPUT', help='将解析结果保存为JSON')
args = parser.parse_args()
try:
processor = FLVProcessor(args.input)
processor.parse()
if args.save_json:
processor.save_as_json(args.save_json)
else:
print("未指定输出操作, 使用python process.py input.raw --save-json out.json")
except Exception as e:
print(f"处理文件时出错: {str(e)}")
if __name__ == "__main__":
main()
运行：
python3 flv.py cap.raw --save-json flv-ana.json