HTTP 状态码：客户端与服务器的通信语言——第二部分：成功类状态码（2xx）深度解析（二）

第8章：206 Partial Content

8.1 206 Partial Content的语义与使用场景

RFC标准定义深度解析

根据RFC 7233第4.1节，206 Partial Content状态码的定义如下：

"206 (Partial Content)状态码表示服务器正在成功执行一个或多个字节范围请求，因为请求包含Range请求头部字段（第3.1节）。响应必须包含Content-Range头部字段（第4.2节），说明响应表示中包含的范围，或者使用multipart/byteranges媒体类型和每个部分都包含自己的Content-Range字段。"

核心语义特征

适用场景矩阵

python

# 206 Partial Content适用场景分析器
class PartialContentUseCaseAnalyzer:
"""分析206状态码的适用场景"""

SCENARIOS = {
"resumable_downloads": {
"description": "可恢复的文件下载",
"examples": [
"大文件下载中断后恢复",
"网络不稳定时的分片下载",
"浏览器下载管理器",
"移动应用中的媒体下载"
],
"range_type": "单范围或连续范围",
"typical_usage": "Range: bytes=1000-2000"
},
"streaming_media": {
"description": "流媒体播放",
"examples": [
"视频播放器跳转时间点",
"音频播放器随机访问",
"在线视频预览",
"流媒体服务器"
],
"range_type": "单范围",
"typical_usage": "Range: bytes=0-1023 (初始片段)"
},
"parallel_downloads": {
"description": "并行下载加速",
"examples": [
"下载管理器分块下载",
"BitTorrent客户端",
"多线程下载工具",
"CDN边缘节点同步"
],
"range_type": "多范围",
"typical_usage": "Range: bytes=0-999,2000-2999,5000-5999"
},
"data_sampling": {
"description": "大数据采样",
"examples": [
"大型日志文件查看",
"数据库备份文件检查",
"媒体文件元数据读取",
"文件格式验证"
],
"range_type": "单范围",
"typical_usage": "Range: bytes=0-511 (读取文件头)"
},
"efficient_updates": {
"description": "高效增量更新",
"examples": [
"游戏补丁下载",
"软件增量更新",
"数据集增量同步",
"文档版本差异"
],
"range_type": "单范围或多个单范围",
"typical_usage": "Range: bytes=5000-"
}
}

def analyze_request(self, request_headers, resource_size):
"""
分析范围请求

Args:
request_headers: 请求头部字典
resource_size: 资源总大小（字节）

Returns:
分析结果字典
"""
range_header = request_headers.get('Range')

if not range_header:
return {
'is_range_request': False,
'recommendation': '使用常规请求'
}

try:
parsed_ranges = self._parse_range_header(range_header, resource_size)

analysis = {
'is_range_request': True,
'range_type': 'single' if len(parsed_ranges) == 1 else 'multiple',
'ranges': parsed_ranges,
'total_size': resource_size,
'recommended_response': self._recommend_response_type(parsed_ranges)
}

# 检查场景适配性
analysis['scenarios'] = self._match_scenarios(parsed_ranges, resource_size)

return analysis

except ValueError as e:
return {
'is_range_request': True,
'error': str(e),
'recommendation': '返回416 Range Not Satisfiable'
}

def _parse_range_header(self, range_header, resource_size):
"""解析Range头部"""
if not range_header.startswith('bytes='):
raise ValueError('Only byte ranges are supported')

ranges_str = range_header[6:] # 移除'bytes='
range_specs = ranges_str.split(',')

parsed_ranges = []

for spec in range_specs:
spec = spec.strip()

if '-' not in spec:
raise ValueError(f'Invalid range spec: {spec}')

start_str, end_str = spec.split('-', 1)

if start_str == '' and end_str == '':
raise ValueError('Empty range spec')

# 处理不同类型的范围
if start_str == '':
# suffix-byte-range-spec: 最后N字节
suffix_length = int(end_str)
if suffix_length <= 0:
raise ValueError('Suffix length must be positive')

start = max(0, resource_size – suffix_length)
end = resource_size – 1

elif end_str == '':
# 从start到结尾
start = int(start_str)
if start >= resource_size:
raise ValueError(f'Start {start} beyond resource size {resource_size}')
end = resource_size – 1

else:
# 完整的范围
start = int(start_str)
end = int(end_str)

if start > end:
raise ValueError(f'Start {start} greater than end {end}')

if start >= resource_size:
raise ValueError(f'Start {start} beyond resource size {resource_size}')

# 调整结束位置不超过资源大小
end = min(end, resource_size – 1)

# 确保范围有效
if start < 0 or end < 0 or start > end:
raise ValueError(f'Invalid range: {start}-{end}')

parsed_ranges.append({
'start': start,
'end': end,
'length': end – start + 1,
'spec': spec
})

return parsed_ranges

def _recommend_response_type(self, ranges):
"""推荐响应类型"""
if len(ranges) == 1:
return {
'status': 206,
'content_type': 'application/octet-stream',
'headers': ['Content-Range'],
'body_type': 'single_part'
}
else:
return {
'status': 206,
'content_type': 'multipart/byteranges; boundary=',
'headers': ['Content-Type', 'Content-Length'],
'body_type': 'multipart'
}

def _match_scenarios(self, ranges, resource_size):
"""匹配适用场景"""
matched = []

# 检查是否适合流媒体
if len(ranges) == 1:
range_length = ranges[0]['length']
if range_length <= 1024 * 1024: # 1MB以下
matched.append('streaming_media')

# 检查是否适合并行下载
if len(ranges) > 1:
matched.append('parallel_downloads')

# 检查是否适合恢复下载
for r in ranges:
if r['start'] > 0:
matched.append('resumable_downloads')
break

# 检查是否适合数据采样
if len(ranges) == 1 and ranges[0]['length'] < 1024: # 小于1KB
matched.append('data_sampling')

return list(set(matched)) # 去重

8.2 Range和Content-Range头部详解

Range头部格式规范

python

# Range头部解析器
class RangeHeaderParser:
"""解析和验证Range请求头部"""

def __init__(self, max_ranges=10, max_range_length=100*1024*1024): # 100MB
self.max_ranges = max_ranges
self.max_range_length = max_range_length

def parse(self, range_header, content_length):
"""
解析Range头部

Args:
range_header: Range头部值
content_length: 内容长度

Returns:
{
'valid': bool,
'ranges': list,
'error': str,
'content_range': str (用于416响应)
}
"""
if not range_header:
return {
'valid': False,
'error': 'No Range header provided',
'content_range': f'bytes */{content_length}'
}

if not range_header.startswith('bytes='):
return {
'valid': False,
'error': 'Only byte ranges are supported',
'content_range': f'bytes */{content_length}'
}

# 提取范围规格
ranges_str = range_header[6:] # 移除'bytes='
range_specs = [s.strip() for s in ranges_str.split(',')]

# 检查数量限制
if len(range_specs) > self.max_ranges:
return {
'valid': False,
'error': f'Too many ranges (max {self.max_ranges})',
'content_range': f'bytes */{content_length}'
}

parsed_ranges = []
valid_ranges = []

for i, spec in enumerate(range_specs):
result = self._parse_single_range(spec, content_length)

if not result['valid']:
# 单个范围无效，整个请求无效
return {
'valid': False,
'error': f'Invalid range spec {i+1}: {result["error"]}',
'content_range': f'bytes */{content_length}'
}

parsed_ranges.append(result)

# 检查范围长度限制
if result['length'] > self.max_range_length:
return {
'valid': False,
'error': f'Range {i+1} too large (max {self.max_range_length} bytes)',
'content_range': f'bytes */{content_length}'
}

# 添加到有效范围列表
valid_ranges.append({
'start': result['start'],
'end': result['end'],
'length': result['length']
})

# 检查范围重叠和排序
sorted_ranges = sorted(valid_ranges, key=lambda x: x['start'])
for i in range(1, len(sorted_ranges)):
if sorted_ranges[i]['start'] <= sorted_ranges[i-1]['end']:
return {
'valid': False,
'error': f'Overlapping ranges: {sorted_ranges[i-1]} and {sorted_ranges[i]}',
'content_range': f'bytes */{content_length}'
}

return {
'valid': True,
'ranges': sorted_ranges,
'total_length': content_length,
'requested_ranges': len(range_specs)
}

def _parse_single_range(self, spec, content_length):
"""解析单个范围规格"""

if '-' not in spec:
return {
'valid': False,
'error': f'Invalid format: {spec}'
}

start_str, end_str = spec.split('-', 1)

try:
# 后缀字节范围规格 (例如: -500)
if start_str == '':
if not end_str.isdigit():
return {
'valid': False,
'error': f'Invalid suffix: {end_str}'
}

suffix_length = int(end_str)
if suffix_length <= 0:
return {
'valid': False,
'error': f'Suffix must be positive: {suffix_length}'
}

start = max(0, content_length – suffix_length)
end = content_length – 1

# 开始到结束的范围 (例如: 1000-2000)
elif end_str == '':
if not start_str.isdigit():
return {
'valid': False,
'error': f'Invalid start: {start_str}'
}

start = int(start_str)
if start >= content_length:
return {
'valid': False,
'error': f'Start {start} beyond content length {content_length}'
}
end = content_length – 1

# 完整的范围规格
else:
if not (start_str.isdigit() and end_str.isdigit()):
return {
'valid': False,
'error': f'Invalid start or end: {start_str}-{end_str}'
}

start = int(start_str)
end = int(end_str)

if start > end:
return {
'valid': False,
'error': f'Start {start} greater than end {end}'
}

if start >= content_length:
return {
'valid': False,
'error': f'Start {start} beyond content length {content_length}'
}

# 调整结束位置
end = min(end, content_length – 1)

# 最终验证
if start < 0 or end < 0 or start > end:
return {
'valid': False,
'error': f'Invalid range: {start}-{end}'
}

return {
'valid': True,
'start': start,
'end': end,
'length': end – start + 1,
'spec': spec
}

except ValueError:
return {
'valid': False,
'error': f'Invalid range spec: {spec}'
}

def generate_content_range(self, start, end, total_length):
"""生成Content-Range头部值"""
return f'bytes {start}-{end}/{total_length}'

def generate_416_content_range(self, total_length):
"""生成416响应的Content-Range头部值"""
return f'bytes */{total_length}'

# 使用示例
parser = RangeHeaderParser()

# 测试各种Range头部
test_cases = [
("bytes=0-499", 1000),
("bytes=500-999", 1000),
("bytes=-500", 1000), # 最后500字节
("bytes=950-", 1000), # 从950到结尾
("bytes=0-0", 1000), # 第一个字节
("bytes=2-5,10-15", 1000), # 多个范围
("bytes=1000-2000", 500), # 无效：开始超出范围
("bytes=200-100", 1000), # 无效：开始>结束
("bytes=abc-def", 1000), # 无效：非数字
]

for range_header, content_length in test_cases:
print(f"\\nRange: {range_header}, Content-Length: {content_length}")
result = parser.parse(range_header, content_length)

if result['valid']:
print(f" 有效范围: {result['ranges']}")
else:
print(f" 无效: {result['error']}")
print(f" 416 Content-Range: {result.get('content_range', 'N/A')}")

Content-Range头部格式

python

# Content-Range头部生成器
class ContentRangeHeaderBuilder:
"""生成Content-Range响应头部"""

@staticmethod
def for_single_range(start, end, total_length):
"""单个范围的Content-Range"""
if start < 0 or end < 0 or start > end or end >= total_length:
raise ValueError(f"Invalid range: {start}-{end}/{total_length}")

return f"bytes {start}-{end}/{total_length}"

@staticmethod
def for_multipart_boundary(boundary):
"""多部分响应的Content-Type"""
return f"multipart/byteranges; boundary={boundary}"

@staticmethod
def for_part_in_multipart(start, end, total_length, content_type="application/octet-stream"):
"""多部分响应中每个部分的头部"""
headers = [
f"Content-Range: bytes {start}-{end}/{total_length}",
f"Content-Type: {content_type}"
]
return "\\r\\n".join(headers)

@staticmethod
def for_unsatisfiable_range(total_length):
"""416响应的Content-Range"""
return f"bytes */{total_length}"

# 完整206响应生成器
class PartialContentResponseBuilder:
"""构建206 Partial Content响应"""

def __init__(self, resource_path, resource_size, mime_type="application/octet-stream"):
self.resource_path = resource_path
self.resource_size = resource_size
self.mime_type = mime_type
self.range_parser = RangeHeaderParser()

def build_response(self, range_header, request_headers=None):
"""
构建206响应

Args:
range_header: Range请求头部值
request_headers: 完整的请求头部字典

Returns:
(status_code, headers, body_generator)
"""
# 解析Range头部
parse_result = self.range_parser.parse(range_header, self.resource_size)

if not parse_result['valid']:
# 返回416 Range Not Satisfiable
return 416, {
'Content-Range': parse_result.get('content_range', f'bytes */{self.resource_size}'),
'Content-Type': 'text/plain'
}, b'Requested range not satisfiable'

ranges = parse_result['ranges']

# 如果是单个范围
if len(ranges) == 1:
return self._build_single_range_response(ranges[0], request_headers)
else:
return self._build_multipart_response(ranges, request_headers)

def _build_single_range_response(self, range_info, request_headers):
"""构建单个范围响应"""
start = range_info['start']
end = range_info['end']
length = range_info['length']

headers = {
'Content-Range': ContentRangeHeaderBuilder.for_single_range(
start, end, self.resource_size
),
'Content-Length': str(length),
'Content-Type': self.mime_type,
'Accept-Ranges': 'bytes',
'ETag': self._generate_etag(),
'Last-Modified': self._get_last_modified(),
'Cache-Control': 'public, max-age=31536000', # 可缓存1年
}

# 添加条件请求支持
if request_headers:
self._add_conditional_headers(headers, request_headers)

# 创建生成器来读取文件部分
def body_generator():
with open(self.resource_path, 'rb') as f:
f.seek(start)
remaining = length
chunk_size = 64 * 1024 # 64KB块

while remaining > 0:
read_size = min(chunk_size, remaining)
chunk = f.read(read_size)
if not chunk:
break
yield chunk
remaining -= len(chunk)

return 206, headers, body_generator()

def _build_multipart_response(self, ranges, request_headers):
"""构建多部分范围响应"""
boundary = f"boundary_{hash(str(ranges))}_{int(time.time())}"
content_type = f"multipart/byteranges; boundary={boundary}"

# 首先计算总长度
total_length = self._calculate_multipart_length(ranges, boundary)

headers = {
'Content-Type': content_type,
'Content-Length': str(total_length),
'Accept-Ranges': 'bytes',
'ETag': self._generate_etag(),
'Cache-Control': 'no-store', # 多部分响应通常不缓存
}

def body_generator():
with open(self.resource_path, 'rb') as f:
for i, range_info in enumerate(ranges):
# 每个部分的边界
yield f"\\r\\n–{boundary}\\r\\n".encode()

# 部分头部
part_headers = ContentRangeHeaderBuilder.for_part_in_multipart(
range_info['start'], range_info['end'],
self.resource_size, self.mime_type
)
yield f"{part_headers}\\r\\n\\r\\n".encode()

# 部分内容
f.seek(range_info['start'])
remaining = range_info['length']
chunk_size = 64 * 1024

while remaining > 0:
read_size = min(chunk_size, remaining)
chunk = f.read(read_size)
if not chunk:
break
yield chunk
remaining -= len(chunk)

# 结束边界
yield f"\\r\\n–{boundary}–\\r\\n".encode()

return 206, headers, body_generator()

def _calculate_multipart_length(self, ranges, boundary):
"""计算多部分响应的总长度"""
total = 0

for range_info in ranges:
# 边界行
total += len(f"\\r\\n–{boundary}\\r\\n")

# 头部
part_headers = ContentRangeHeaderBuilder.for_part_in_multipart(
range_info['start'], range_info['end'],
self.resource_size, self.mime_type
)
total += len(f"{part_headers}\\r\\n\\r\\n")

# 内容
total += range_info['length']

# 结束边界
total += len(f"\\r\\n–{boundary}–\\r\\n")

return total

def _generate_etag(self):
"""生成ETag（示例实现）"""
import hashlib
import os

stat = os.stat(self.resource_path)
content = f"{self.resource_path}:{stat.st_size}:{stat.st_mtime}"
return f'"{hashlib.md5(content.encode()).hexdigest()}"'

def _get_last_modified(self):
"""获取最后修改时间"""
import os
from datetime import datetime

mtime = os.path.getmtime(self.resource_path)
return datetime.fromtimestamp(mtime).strftime('%a, %d %b %Y %H:%M:%S GMT')

def _add_conditional_headers(self, response_headers, request_headers):
"""添加条件请求相关头部"""
# 检查If-Range
if 'If-Range' in request_headers:
# 这里应该验证If-Range条件
# 如果条件不满足，应该返回整个资源而不是部分内容
pass

# 检查If-Match/If-None-Match
if 'If-Match' in request_headers or 'If-None-Match' in request_headers:
response_headers['Vary'] = 'If-Match, If-None-Match'

8.3 服务器端实现

Nginx配置示例

nginx

# Nginx部分内容响应配置
http {
# 启用字节范围支持
slice 1m; # 切片大小，用于视频流
proxy_cache_path /var/cache/nginx levels=1:2 keys_zone=video_cache:10m
max_size=10g inactive=60m use_temp_path=off;

# 视频流服务器配置
server {
listen 80;
server_name video.example.com;

location /videos/ {
# 启用范围请求支持
mp4;
mp4_buffer_size 1m;
mp4_max_buffer_size 5m;

# 限制请求速率
limit_rate 100m; # 最大100MB/s
limit_rate_after 10m; # 10MB后开始限速

# 缓存控制
add_header Cache-Control "public, max-age=31536000";
add_header Accept-Ranges bytes;
add_header X-Content-Type-Options nosniff;

# 访问控制
satisfy any;
allow 192.168.1.0/24;
deny all;

# 代理设置
proxy_pass http://video_backend;
proxy_set_header Range $http_range;
proxy_set_header If-Range $http_if_range;

# 缓冲和超时
proxy_buffering on;
proxy_buffer_size 128k;
proxy_buffers 4 256k;
proxy_busy_buffers_size 256k;
proxy_read_timeout 300;
}

# 大文件下载配置
location /downloads/ {
# 基础设置
root /var/www;
autoindex off;

# 范围请求支持
add_header Accept-Ranges bytes;

# 安全头部
add_header X-Content-Type-Options nosniff;
add_header X-Frame-Options DENY;
add_header X-XSS-Protection "1; mode=block";

# 下载限制
limit_rate 50m; # 限制下载速度
limit_rate_after 100m; # 100MB后限速

# 防盗链
valid_referers none blocked server_names *.example.com;
if ($invalid_referer) {
return 403;
}

# 文件类型处理
location ~* \\.(mp4|avi|mkv|mov)$ {
# 视频文件特殊处理
add_header Cache-Control "public, max-age=604800"; # 7天
}

location ~* \\.(zip|rar|7z|tar\\.gz)$ {
# 压缩文件
add_header Content-Disposition "attachment";
add_header Cache-Control "no-cache";
}
}

# API端点处理范围请求
location /api/files/ {
# 代理到应用服务器
proxy_pass http://app_backend;

# 传递范围相关头部
proxy_set_header Range $http_range;
proxy_set_header If-Range $http_if_range;
proxy_set_header X-Original-Range $http_range;

# 处理应用服务器的206响应
proxy_intercept_errors on;
error_page 416 = @handle_416;

# 缓冲设置
proxy_buffering on;
proxy_buffer_size 4k;
proxy_buffers 8 4k;

# 超时设置
proxy_connect_timeout 60;
proxy_send_timeout 300;
proxy_read_timeout 300;
}

# 416错误处理
location @handle_416 {
default_type application/json;
return 416 '{"error": "Requested range not satisfiable"}';
}
}

# 上游服务器配置
upstream video_backend {
server 192.168.1.100:8080;
server 192.168.1.101:8080;
keepalive 32;
}

upstream app_backend {
server 192.168.1.200:8000;
server 192.168.1.201:8000;
keepalive 16;
}
}

# 特殊情况处理
events {
worker_connections 1024;
use epoll;
}

Python Flask实现

python

# 完整的206 Partial Content服务器实现
from flask import Flask, request, Response, send_file, jsonify
import os
import mimetypes
from datetime import datetime
import hashlib
import time
from functools import wraps
import re

app = Flask(__name__)

# 配置
app.config['MAX_CONTENT_LENGTH'] = 100 * 1024 * 1024 * 1024 # 100GB最大文件
app.config['RANGE_REQUEST_MAX_RANGES'] = 10
app.config['RANGE_REQUEST_MAX_SIZE'] = 100 * 1024 * 1024 # 100MB最大范围
app.config['CHUNK_SIZE'] = 64 * 1024 # 64KB块大小

class RangeRequestHandler:
"""处理范围请求的处理器"""

def __init__(self, file_path, mime_type=None):
self.file_path = file_path
self.mime_type = mime_type or self._guess_mime_type(file_path)
self.file_size = os.path.getsize(file_path)
self.last_modified = datetime.fromtimestamp(
os.path.getmtime(file_path)
)
self.etag = self._generate_etag()

def _guess_mime_type(self, file_path):
"""猜测MIME类型"""
mime_type, _ = mimetypes.guess_type(file_path)
return mime_type or 'application/octet-stream'

def _generate_etag(self):
"""生成ETag"""
stat = os.stat(self.file_path)
content = f"{self.file_path}:{stat.st_size}:{stat.st_mtime}"
return f'"{hashlib.md5(content.encode()).hexdigest()}"'

def handle_request(self, range_header=None):
"""处理HTTP请求"""

# 检查条件请求
if not self._check_conditions():
return self._build_conditional_failure_response()

# 如果没有Range头部，返回整个文件
if not range_header:
return self._serve_full_file()

# 解析Range头部
ranges = self._parse_range_header(range_header)
if not ranges:
return self._build_range_not_satisfiable_response()

# 处理范围请求
if len(ranges) == 1:
return self._serve_single_range(ranges[0])
else:
return self._serve_multipart_ranges(ranges)

def _check_conditions(self):
"""检查条件请求头部"""
# 检查If-Match
if_match = request.headers.get('If-Match')
if if_match and self.etag not in self._parse_etag_list(if_match):
return False

# 检查If-None-Match
if_none_match = request.headers.get('If-None-Match')
if if_none_match and self.etag in self._parse_etag_list(if_none_match):
# 对于GET/HEAD，返回304；对于其他方法，返回412
if request.method in ('GET', 'HEAD'):
return None # 特殊标记，表示304
return False

# 检查If-Modified-Since
if_modified_since = request.headers.get('If-Modified-Since')
if if_modified_since:
try:
since = datetime.strptime(
if_modified_since, '%a, %d %b %Y %H:%M:%S GMT'
)
if self.last_modified <= since:
return None # 特殊标记，表示304
except ValueError:
pass

# 检查If-Unmodified-Since
if_unmodified_since = request.headers.get('If-Unmodified-Since')
if if_unmodified_since:
try:
since = datetime.strptime(
if_unmodified_since, '%a, %d %b %Y %H:%M:%S GMT'
)
if self.last_modified > since:
return False
except ValueError:
pass

# 检查If-Range（仅在部分请求时检查）
if_range = request.headers.get('If-Range')
if if_range and request.headers.get('Range'):
# If-Range可以是ETag或日期
if if_range.startswith('"'):
# ETag格式
if if_range != self.etag:
# 返回整个资源
return self._serve_full_file()
else:
# 日期格式
try:
since = datetime.strptime(
if_range, '%a, %d %b %Y %H:%M:%S GMT'
)
if self.last_modified > since:
# 返回整个资源
return self._serve_full_file()
except ValueError:
pass

return True

def _parse_etag_list(self, etag_header):
"""解析ETag列表"""
etags = []
for etag in etag_header.split(','):
etag = etag.strip()
if etag:
etags.append(etag)
return etags

def _parse_range_header(self, range_header):
"""解析Range头部"""
if not range_header.startswith('bytes='):
return None

range_str = range_header[6:] # 移除'bytes='
range_specs = [s.strip() for s in range_str.split(',')]

# 限制范围数量
if len(range_specs) > app.config['RANGE_REQUEST_MAX_RANGES']:
return None

ranges = []

for spec in range_specs:
range_info = self._parse_single_range(spec)
if not range_info:
return None

# 检查范围大小限制
range_length = range_info['end'] – range_info['start'] + 1
if range_length > app.config['RANGE_REQUEST_MAX_SIZE']:
return None

ranges.append(range_info)

# 检查范围重叠和排序
ranges.sort(key=lambda x: x['start'])
for i in range(1, len(ranges)):
if ranges[i]['start'] <= ranges[i-1]['end']:
return None

return ranges

def _parse_single_range(self, range_spec):
"""解析单个范围规格"""
if '-' not in range_spec:
return None

start_str, end_str = range_spec.split('-', 1)

try:
# 后缀范围规格
if start_str == '':
suffix_length = int(end_str)
if suffix_length <= 0:
return None

start = max(0, self.file_size – suffix_length)
end = self.file_size – 1

# 开始到结尾
elif end_str == '':
start = int(start_str)
if start >= self.file_size:
return None
end = self.file_size – 1

# 完整范围
else:
start = int(start_str)
end = int(end_str)

if start > end:
return None

if start >= self.file_size:
return None

# 调整结束位置
end = min(end, self.file_size – 1)

# 验证最终范围
if start < 0 or end < 0 or start > end:
return None

return {
'start': start,
'end': end,
'length': end – start + 1
}

except ValueError:
return None

def _build_conditional_failure_response(self):
"""构建条件请求失败响应"""
if request.method in ('GET', 'HEAD'):
# 304 Not Modified
return Response(status=304, headers={
'ETag': self.etag,
'Last-Modified': self.last_modified.strftime('%a, %d %b %Y %H:%M:%S GMT')
})
else:
# 412 Precondition Failed
return Response('Precondition Failed', status=412)

def _build_range_not_satisfiable_response(self):
"""构建416响应"""
return Response(
'Requested range not satisfiable',
status=416,
headers={
'Content-Range': f'bytes */{self.file_size}',
'Content-Type': 'text/plain'
}
)

def _serve_full_file(self):
"""提供整个文件"""
headers = {
'Content-Type': self.mime_type,
'Content-Length': str(self.file_size),
'Accept-Ranges': 'bytes',
'ETag': self.etag,
'Last-Modified': self.last_modified.strftime('%a, %d %b %Y %H:%M:%S GMT'),
'Cache-Control': 'public, max-age=31536000' # 1年
}

if request.method == 'HEAD':
return Response(headers=headers)

return send_file(
self.file_path,
mimetype=self.mime_type,
as_attachment=False,
conditional=True
)

def _serve_single_range(self, range_info):
"""提供单个范围"""
start = range_info['start']
end = range_info['end']
length = range_info['length']

headers = {
'Content-Range': f'bytes {start}-{end}/{self.file_size}',
'Content-Length': str(length),
'Content-Type': self.mime_type,
'Accept-Ranges': 'bytes',
'ETag': self.etag,
'Cache-Control': 'public, max-age=31536000'
}

if request.method == 'HEAD':
return Response(headers=headers)

def generate():
with open(self.file_path, 'rb') as f:
f.seek(start)
remaining = length
chunk_size = app.config['CHUNK_SIZE']

while remaining > 0:
read_size = min(chunk_size, remaining)
chunk = f.read(read_size)
if not chunk:
break
yield chunk
remaining -= len(chunk)

return Response(
generate(),
status=206,
headers=headers
)

def _serve_multipart_ranges(self, ranges):
"""提供多部分范围"""
boundary = f"boundary_{hash(str(ranges))}_{int(time.time())}"
content_type = f"multipart/byteranges; boundary={boundary}"

# 计算总长度
total_length = self._calculate_multipart_length(ranges, boundary)

headers = {
'Content-Type': content_type,
'Content-Length': str(total_length),
'Accept-Ranges': 'bytes',
'ETag': self.etag,
'Cache-Control': 'no-store' # 多部分响应不缓存
}

if request.method == 'HEAD':
return Response(headers=headers)

def generate():
with open(self.file_path, 'rb') as f:
for i, range_info in enumerate(ranges):
# 边界
yield f"\\r\\n–{boundary}\\r\\n".encode()

# 部分头部
part_headers = [
f"Content-Range: bytes {range_info['start']}-{range_info['end']}/{self.file_size}",
f"Content-Type: {self.mime_type}"
]
yield "\\r\\n".join(part_headers).encode()
yield "\\r\\n\\r\\n".encode()

# 部分内容
f.seek(range_info['start'])
remaining = range_info['length']
chunk_size = app.config['CHUNK_SIZE']

while remaining > 0:
read_size = min(chunk_size, remaining)
chunk = f.read(read_size)
if not chunk:
break
yield chunk
remaining -= len(chunk)

# 结束边界
yield f"\\r\\n–{boundary}–\\r\\n".encode()

return Response(
generate(),
status=206,
headers=headers
)

def _calculate_multipart_length(self, ranges, boundary):
"""计算多部分响应长度"""
total = 0

for range_info in ranges:
# 边界行
total += len(f"\\r\\n–{boundary}\\r\\n")

# 头部
headers = [
f"Content-Range: bytes {range_info['start']}-{range_info['end']}/{self.file_size}",
f"Content-Type: {self.mime_type}"
]
total += len("\\r\\n".join(headers)) + len("\\r\\n\\r\\n")

# 内容
total += range_info['length']

# 结束边界
total += len(f"\\r\\n–{boundary}–\\r\\n")

return total

# API端点
@app.route('/api/files/<path:filename>')
def serve_file(filename):
"""提供文件下载（支持范围请求）"""

# 安全验证：防止目录遍历
safe_filename = os.path.basename(filename)
file_path = os.path.join(app.config['UPLOAD_FOLDER'], safe_filename)

if not os.path.exists(file_path):
return jsonify({'error': 'File not found'}), 404

# 创建处理器
handler = RangeRequestHandler(file_path)

# 获取Range头部
range_header = request.headers.get('Range')

# 处理请求
return handler.handle_request(range_header)

@app.route('/api/stream/video/<video_id>')
def stream_video(video_id):
"""流式传输视频（支持范围请求）"""

# 获取视频文件路径
video_path = get_video_path(video_id) # 假设的函数
if not video_path or not os.path.exists(video_path):
return jsonify({'error': 'Video not found'}), 404

# 视频文件的MIME类型
mime_type = 'video/mp4' # 根据实际格式调整

# 创建处理器
handler = RangeRequestHandler(video_path, mime_type)

# 获取Range头部
range_header = request.headers.get('Range')

# 处理请求
return handler.handle_request(range_header)

# 中间件：记录范围请求统计
@app.after_request
def log_range_requests(response):
"""记录范围请求统计"""
if request.headers.get('Range') and response.status_code == 206:
# 记录到日志
app.logger.info(f"Range request: {request.headers.get('Range')} -> {response.status}")

# 添加自定义头部
response.headers['X-Range-Request'] = 'processed'

# 如果是范围响应，添加统计信息
content_range = response.headers.get('Content-Range')
if content_range:
# 解析范围信息
match = re.match(r'bytes (\\d+)-(\\d+)/(\\d+)', content_range)
if match:
start, end, total = match.groups()
bytes_served = int(end) – int(start) + 1
response.headers['X-Bytes-Served'] = str(bytes_served)
response.headers['X-Total-Bytes'] = total

return response

# 辅助函数
def get_video_path(video_id):
"""获取视频文件路径（示例实现）"""
# 实际中可能会从数据库或文件系统中查找
video_dir = '/var/www/videos'
for ext in ['.mp4', '.avi', '.mkv', '.mov']:
path = os.path.join(video_dir, f"{video_id}{ext}")
if os.path.exists(path):
return path
return None

if __name__ == '__main__':
app.config['UPLOAD_FOLDER'] = '/var/www/uploads'
os.makedirs(app.config['UPLOAD_FOLDER'], exist_ok=True)

app.run(host='0.0.0.0', port=8000, threaded=True)

8.4 客户端处理实现

JavaScript Fetch API实现

javascript

// 浏览器端范围请求处理器
class RangeRequestClient {
/**
* 浏览器端范围请求处理器
*/

constructor(options = {}) {
this.options = {
chunkSize: 1024 * 1024, // 1MB块大小
maxRetries: 3,
retryDelay: 1000,
parallelDownloads: 3,
…options
};

this.downloadQueue = [];
this.activeDownloads = 0;
this.downloadStats = new Map();
}

/**
* 分块下载文件
* @param {string} url – 文件URL
* @param {Object} options – 下载选项
*/
async downloadFile(url, options = {}) {
const {
onProgress,
onComplete,
onError,
resume = false
} = options;

try {
// 1. 获取文件信息
const fileInfo = await this.getFileInfo(url);
if (!fileInfo) {
throw new Error('无法获取文件信息');
}

// 2. 检查是否支持范围请求
if (!fileInfo.acceptRanges) {
// 回退到普通下载
return await this.downloadFullFile(url, options);
}

// 3. 检查是否有部分下载的文件
const downloadedParts = resume ? await this.getDownloadedParts(url, fileInfo) : [];

// 4. 计算需要下载的范围
const ranges = this.calculateRanges(fileInfo.size, downloadedParts);

// 5. 并行下载所有范围
const chunks = await this.downloadRanges(url, ranges, {
onProgress: (chunkIndex, progress) => {
if (onProgress) {
const overallProgress = this.calculateOverallProgress(
ranges, chunkIndex, progress
);
onProgress(overallProgress);
}
}
});

// 6. 合并所有块
const blob = this.mergeChunks(chunks, fileInfo.size);

// 7. 清理临时数据
await this.cleanupTempData(url);

if (onComplete) {
onComplete(blob, fileInfo);
}

return blob;

} catch (error) {
if (onError) {
onError(error);
}
throw error;
}
}

/**
* 获取文件信息
*/
async getFileInfo(url) {
try {
const response = await fetch(url, {
method: 'HEAD',
headers: {
'Accept': '*/*'
}
});

if (!response.ok) {
throw new Error(`HTTP ${response.status}`);
}

const contentLength = response.headers.get('Content-Length');
const acceptRanges = response.headers.get('Accept-Ranges') === 'bytes';
const contentType = response.headers.get('Content-Type');
const lastModified = response.headers.get('Last-Modified');
const etag = response.headers.get('ETag');

return {
size: parseInt(contentLength, 10) || 0,
acceptRanges,
contentType,
lastModified,
etag,
url
};

} catch (error) {
console.error('Failed to get file info:', error);
return null;
}
}

/**
* 获取已下载的部分
*/
async getDownloadedParts(url, fileInfo) {
const storageKey = `download:${btoa(url)}`;

try {
const saved = localStorage.getItem(storageKey);
if (!saved) {
return [];
}

const data = JSON.parse(saved);

// 验证文件是否已更改
if (data.etag !== fileInfo.etag ||
data.size !== fileInfo.size) {
localStorage.removeItem(storageKey);
return [];
}

return data.parts || [];

} catch (error) {
console.error('Failed to get downloaded parts:', error);
return [];
}
}

/**
* 计算需要下载的范围
*/
calculateRanges(fileSize, downloadedParts) {
const ranges = [];

if (downloadedParts.length === 0) {
// 全新下载：将文件分成多个块
for (let start = 0; start < fileSize; start += this.options.chunkSize) {
const end = Math.min(start + this.options.chunkSize – 1, fileSize – 1);
ranges.push({ start, end, downloaded: false });
}
} else {
// 恢复下载：只下载缺失的部分
let currentPos = 0;

// 按开始位置排序
downloadedParts.sort((a, b) => a.start – b.start);

for (const part of downloadedParts) {
if (currentPos < part.start) {
// 缺失的部分
ranges.push({
start: currentPos,
end: part.start – 1,
downloaded: false
});
}
// 已下载的部分
ranges.push({
…part,
downloaded: true
});
currentPos = part.end + 1;
}

// 检查最后的部分
if (currentPos < fileSize) {
ranges.push({
start: currentPos,
end: fileSize – 1,
downloaded: false
});
}
}

return ranges;
}

/**
* 并行下载范围
*/
async downloadRanges(url, ranges, callbacks) {
const chunks = new Array(ranges.length);
const errors = [];

// 只下载未完成的部分
const rangesToDownload = ranges.filter(r => !r.downloaded);

// 创建下载任务
const downloadTasks = rangesToDownload.map((range, index) => ({
index: ranges.indexOf(range),
range,
retries: 0
}));

// 并行下载
await this.parallelDownload(url, downloadTasks, chunks, errors, callbacks);

// 处理已下载的部分
ranges.forEach((range, index) => {
if (range.downloaded) {
// 从localStorage加载已下载的数据
const chunk = this.loadChunkFromStorage(url, range);
if (chunk) {
chunks[index] = chunk;
} else {
errors.push(`Failed to load chunk ${index} from storage`);
}
}
});

if (errors.length > 0) {
throw new Error(`Download failed: ${errors.join(', ')}`);
}

return chunks;
}

/**
* 并行下载实现
*/
async parallelDownload(url, tasks, chunks, errors, callbacks) {
return new Promise((resolve, reject) => {
let completed = 0;

const processNext = () => {
// 找到下一个未开始的任务
const task = tasks.find(t => !t.started && t.retries < this.options.maxRetries);

if (!task) {
if (completed === tasks.length) {
resolve();
}
return;
}

task.started = true;
this.activeDownloads++;

this.downloadRange(url, task.range, task.retries)
.then(async (chunk) => {
chunks[task.index] = chunk;

// 保存到本地存储
await this.saveChunkToStorage(url, task.range, chunk);

if (callbacks?.onProgress) {
callbacks.onProgress(task.index, 100);
}

completed++;
this.activeDownloads–;

// 处理下一个任务
setTimeout(processNext, 0);
})
.catch(async (error) => {
console.error(`Failed to download range ${task.index}:`, error);

task.retries++;
task.started = false;
this.activeDownloads–;

if (task.retries >= this.options.maxRetries) {
errors.push(`Range ${task.range.start}-${task.range.end}: ${error.message}`);
}

// 延迟后重试
setTimeout(processNext, this.options.retryDelay);
});
};

// 启动初始下载
for (let i = 0; i < Math.min(this.options.parallelDownloads, tasks.length); i++) {
setTimeout(processNext, 0);
}
});
}

/**
* 下载单个范围
*/
async downloadRange(url, range, retryCount = 0) {
const rangeHeader = `bytes=${range.start}-${range.end}`;

try {
const response = await fetch(url, {
headers: {
'Range': rangeHeader,
'Accept': '*/*'
},
// 添加超时和重试逻辑
signal: AbortSignal.timeout(30000) // 30秒超时
});

if (response.status === 206) {
const blob = await response.blob();

// 验证下载的大小
if (blob.size !== range.end – range.start + 1) {
throw new Error(`Size mismatch: expected ${range.end – range.start + 1}, got ${blob.size}`);
}

return blob;

} else if (response.status === 416) {
// 范围不可满足，可能是文件已更改
throw new Error('Range not satisfiable – file may have changed');

} else {
throw new Error(`HTTP ${response.status}`);
}

} catch (error) {
if (retryCount < this.options.maxRetries – 1) {
console.log(`Retrying range ${range.start}-${range.end} (attempt ${retryCount + 2})`);
}
throw error;
}
}

/**
* 合并块
*/
mergeChunks(chunks, totalSize) {
// 创建一个足够大的ArrayBuffer
const buffer = new ArrayBuffer(totalSize);
const view = new Uint8Array(buffer);

let offset = 0;

for (const chunk of chunks) {
if (!chunk) {
throw new Error('Missing chunk');
}

const chunkArray = new Uint8Array(chunk);
view.set(chunkArray, offset);
offset += chunkArray.length;
}

return new Blob([buffer]);
}

/**
* 保存块到本地存储
*/
async saveChunkToStorage(url, range, chunk) {
const storageKey = `download:${btoa(url)}`;

try {
// 将块转换为base64
const base64 = await this.blobToBase64(chunk);

// 获取现有数据
const existing = localStorage.getItem(storageKey);
const data = existing ? JSON.parse(existing) : {
parts: [],
etag: null,
size: 0
};

// 更新部分信息
const existingPartIndex = data.parts.findIndex(p =>
p.start === range.start && p.end === range.end
);

if (existingPartIndex >= 0) {
data.parts[existingPartIndex].data = base64;
} else {
data.parts.push({
start: range.start,
end: range.end,
data: base64
});
}

// 保存到localStorage
localStorage.setItem(storageKey, JSON.stringify(data));

} catch (error) {
console.error('Failed to save chunk:', error);
}
}

/**
* 从本地存储加载块
*/
loadChunkFromStorage(url, range) {
const storageKey = `download:${btoa(url)}`;

try {
const saved = localStorage.getItem(storageKey);
if (!saved) return null;

const data = JSON.parse(saved);
const part = data.parts.find(p =>
p.start === range.start && p.end === range.end
);

if (!part || !part.data) return null;

// 将base64转换回Blob
return this.base64ToBlob(part.data, 'application/octet-stream');

} catch (error) {
console.error('Failed to load chunk:', error);
return null;
}
}

/**
* Blob转base64
*/
blobToBase64(blob) {
return new Promise((resolve, reject) => {
const reader = new FileReader();
reader.onloadend = () => {
const base64 = reader.result.split(',')[1];
resolve(base64);
};
reader.onerror = reject;
reader.readAsDataURL(blob);
});
}

/**
* base64转Blob
*/
base64ToBlob(base64, contentType) {
const byteCharacters = atob(base64);
const byteArrays = [];

for (let offset = 0; offset < byteCharacters.length; offset += 512) {
const slice = byteCharacters.slice(offset, offset + 512);
const byteNumbers = new Array(slice.length);

for (let i = 0; i < slice.length; i++) {
byteNumbers[i] = slice.charCodeAt(i);
}

const byteArray = new Uint8Array(byteNumbers);
byteArrays.push(byteArray);
}

return new Blob(byteArrays, { type: contentType });
}

/**
* 计算整体进度
*/
calculateOverallProgress(ranges, chunkIndex, chunkProgress) {
let totalDownloaded = 0;

for (let i = 0; i < ranges.length; i++) {
const range = ranges[i];
if (i < chunkIndex) {
// 已完成的范围
totalDownloaded += range.end – range.start + 1;
} else if (i === chunkIndex) {
// 当前正在下载的范围
const rangeSize = range.end – range.start + 1;
totalDownloaded += (rangeSize * chunkProgress) / 100;
}
}

const totalSize = ranges[ranges.length – 1].end + 1;
return (totalDownloaded / totalSize) * 100;
}

/**
* 清理临时数据
*/
async cleanupTempData(url) {
const storageKey = `download:${btoa(url)}`;
localStorage.removeItem(storageKey);
}

/**
* 完整文件下载（不支持范围请求时的回退）
*/
async downloadFullFile(url, options) {
const { onProgress, onComplete, onError } = options;

try {
const response = await fetch(url);

if (!response.ok) {
throw new Error(`HTTP ${response.status}`);
}

const reader = response.body.getReader();
const contentLength = parseInt(response.headers.get('Content-Length'), 10) || 0;

let receivedLength = 0;
const chunks = [];

while (true) {
const { done, value } = await reader.read();

if (done) {
break;
}

chunks.push(value);
receivedLength += value.length;

if (onProgress && contentLength > 0) {
onProgress((receivedLength / contentLength) * 100);
}
}

const blob = new Blob(chunks);

if (onComplete) {
onComplete(blob, {
size: contentLength,
contentType: response.headers.get('Content-Type')
});
}

return blob;

} catch (error) {
if (onError) {
onError(error);
}
throw error;
}
}
}

// 使用示例
const downloadManager = new RangeRequestClient({
chunkSize: 1024 * 1024 * 5, // 5MB块
parallelDownloads: 4, // 4个并行下载
maxRetries: 3 // 最多重试3次
});

// 开始下载
document.getElementById('download-btn').addEventListener('click', async () => {
const url = document.getElementById('file-url').value;
const progressBar = document.getElementById('progress-bar');
const statusText = document.getElementById('status-text');

try {
statusText.textContent = '正在准备下载…';

await downloadManager.downloadFile(url, {
resume: true, // 启用断点续传

onProgress: (progress) => {
progressBar.style.width = `${progress}%`;
progressBar.textContent = `${progress.toFixed(1)}%`;
statusText.textContent = `下载中: ${progress.toFixed(1)}%`;
},

onComplete: (blob, fileInfo) => {
statusText.textContent = '下载完成!';

// 创建下载链接
const downloadUrl = URL.createObjectURL(blob);
const a = document.createElement('a');
a.href = downloadUrl;
a.download = getFileNameFromUrl(url);
a.click();

// 清理
URL.revokeObjectURL(downloadUrl);
},

onError: (error) => {
console.error('下载失败:', error);
statusText.textContent = `下载失败: ${error.message}`;
progressBar.style.backgroundColor = '#dc3545';
}
});

} catch (error) {
console.error('下载失败:', error);
statusText.textContent = `错误: ${error.message}`;
}
});

// 暂停下载（示例）
document.getElementById('pause-btn').addEventListener('click', () => {
// 在实际实现中，这里需要存储当前状态
localStorage.setItem('downloadPaused', 'true');
alert('下载已暂停，下次可以恢复下载');
});

function getFileNameFromUrl(url) {
return url.split('/').pop() || 'download.bin';
}

8.5 最佳实践与性能优化

性能优化策略

python

# 范围请求性能优化器
class RangeRequestOptimizer:
"""范围请求性能优化器"""

def __init__(self):
self.cache = {}
self.statistics = {
'total_requests': 0,
'range_requests': 0,
'bytes_served': 0,
'cache_hits': 0,
'cache_misses': 0
}

def optimize_response(self, file_path, range_info, use_cache=True):
"""
优化范围响应

Args:
file_path: 文件路径
range_info: 范围信息
use_cache: 是否使用缓存

Returns:
(data, from_cache)
"""
self.statistics['total_requests'] += 1
self.statistics['range_requests'] += 1

cache_key = self._generate_cache_key(file_path, range_info)

# 检查缓存
if use_cache and cache_key in self.cache:
self.statistics['cache_hits'] += 1
return self.cache[cache_key], True

# 从文件读取
data = self._read_file_range(file_path, range_info)

# 缓存结果（如果值得缓存）
if self._should_cache(range_info):
self.cache[cache_key] = data
self.statistics['cache_misses'] += 1

return data, False

def _generate_cache_key(self, file_path, range_info):
"""生成缓存键"""
import hashlib
import os

stat = os.stat(file_path)
key_content = f"{file_path}:{stat.st_mtime}:{range_info['start']}:{range_info['end']}"
return hashlib.md5(key_content.encode()).hexdigest()

def _read_file_range(self, file_path, range_info):
"""从文件读取范围"""
start = range_info['start']
end = range_info['end']
length = end – start + 1

with open(file_path, 'rb') as f:
f.seek(start)
return f.read(length)

def _should_cache(self, range_info):
"""判断是否应该缓存"""
# 基于范围大小决定是否缓存
range_size = range_info['end'] – range_info['start'] + 1

# 小范围请求更可能被重复请求（例如视频流的开始部分）
if range_size <= 1024 * 1024: # 1MB以下
return True

# 特定模式的范围请求（例如视频的固定时间点）
if range_info['start'] % (1024 * 1024) == 0: # 对齐到1MB边界
return True

return False

def preload_cache(self, file_path, common_ranges):
"""
预加载常用范围到缓存

Args:
file_path: 文件路径
common_ranges: 常用范围列表
"""
for range_info in common_ranges:
cache_key = self._generate_cache_key(file_path, range_info)
if cache_key not in self.cache:
data = self._read_file_range(file_path, range_info)
self.cache[cache_key] = data

def cleanup_cache(self, max_size_mb=100):
"""清理缓存，保持最大大小"""
import sys

cache_size = sum(sys.getsizeof(v) for v in self.cache.values())
max_size = max_size_mb * 1024 * 1024

if cache_size > max_size:
# 使用LRU策略清理缓存
# 这里简化实现：直接清理一半缓存
keys = list(self.cache.keys())
for key in keys[:len(keys)//2]:
del self.cache[key]

def get_optimization_suggestions(self, request_patterns):
"""
基于请求模式提供优化建议

Args:
request_patterns: 请求模式列表

Returns:
优化建议字典
"""
suggestions = []

# 分析请求模式
sequential_requests = self._analyze_sequential_requests(request_patterns)
repeated_ranges = self._find_repeated_ranges(request_patterns)

# 建议1：预加载常用范围
if repeated_ranges:
suggestions.append({
'type': 'preload',
'priority': 'high',
'message': f'发现{len(repeated_ranges)}个重复请求的范围，建议预加载到缓存',
'ranges': repeated_ranges[:5] # 前5个最常请求的范围
})

# 建议2：调整块大小
avg_range_size = self._calculate_average_range_size(request_patterns)
if avg_range_size > 1024 * 1024 * 10: # 10MB以上
suggestions.append({
'type': 'chunk_size',
'priority': 'medium',
'message': f'平均请求范围较大({avg_range_size/1024/1024:.1f}MB)，建议增加块大小',
'recommended_size': min(avg_range_size * 2, 1024 * 1024 * 100) # 最大100MB
})

# 建议3：顺序读取优化
if sequential_requests['is_sequential']:
suggestions.append({
'type': 'sequential_read',
'priority': 'low',
'message': '检测到顺序读取模式，可以启用预读优化',
'read_ahead_size': sequential_requests['avg_gap'] * 2
})

return suggestions

def _analyze_sequential_requests(self, patterns):
"""分析顺序请求模式"""
if len(patterns) < 3:
return {'is_sequential': False}

# 检查是否大致连续
gaps = []
for i in range(1, len(patterns)):
prev_end = patterns[i-1]['end']
curr_start = patterns[i]['start']

if curr_start >= prev_end:
gaps.append(curr_start – prev_end)

if len(gaps) == 0:
return {'is_sequential': False}

avg_gap = sum(gaps) / len(gaps)

# 如果平均间隔小于块大小的10%，认为是顺序读取
avg_range_size = self._calculate_average_range_size(patterns)

return {
'is_sequential': avg_gap < avg_range_size * 0.1,
'avg_gap': avg_gap,
'gap_count': len(gaps)
}

def _find_repeated_ranges(self, patterns):
"""查找重复请求的范围"""
from collections import Counter

range_strings = [f"{p['start']}-{p['end']}" for p in patterns]
counter = Counter(range_strings)

# 返回请求次数大于1的范围
repeated = []
for range_str, count in counter.items():
if count > 1:
start, end = map(int, range_str.split('-'))
repeated.append({
'start': start,
'end': end,
'count': count
})

# 按请求次数排序
repeated.sort(key=lambda x: x['count'], reverse=True)
return repeated

def _calculate_average_range_size(self, patterns):
"""计算平均范围大小"""
if not patterns:
return 0

total_size = sum(p['end'] – p['start'] + 1 for p in patterns)
return total_size / len(patterns)

# 完整的HTTP范围请求服务器优化版
class OptimizedRangeServer:
"""优化的范围请求服务器"""

def __init__(self, base_dir, cache_size_mb=100):
self.base_dir = base_dir
self.optimizer = RangeRequestOptimizer()
self.cache_size_mb = cache_size_mb

# 预定义常用范围（例如视频文件的开始部分）
self.common_ranges = [
{'start': 0, 'end': 1024 * 1024 – 1}, # 第一个1MB
{'start': 0, 'end': 1024 * 10 – 1}, # 前10KB（文件头）
{'start': 0, 'end': 1024 * 512 – 1}, # 前512KB
]

def serve_request(self, file_path, range_header):
"""
提供范围请求服务

Args:
file_path: 相对文件路径
range_header: Range头部值

Returns:
HTTP响应
"""
full_path = os.path.join(self.base_dir, file_path)

if not os.path.exists(full_path):
return self._error_response(404, 'File not found')

file_size = os.path.getsize(full_path)

# 解析Range头部
parser = RangeHeaderParser()
parse_result = parser.parse(range_header, file_size)

if not parse_result['valid']:
return self._range_not_satisfiable_response(file_size)

ranges = parse_result['ranges']

# 单个范围
if len(ranges) == 1:
return self._serve_single_range(full_path, ranges[0], file_size)

# 多个范围
else:
return self._serve_multipart_ranges(full_path, ranges, file_size)

def _serve_single_range(self, file_path, range_info, file_size):
"""提供单个范围"""
# 使用优化器获取数据
data, from_cache = self.optimizer.optimize_response(
file_path, range_info, use_cache=True
)

headers = {
'Content-Range': f'bytes {range_info["start"]}-{range_info["end"]}/{file_size}',
'Content-Length': str(len(data)),
'Content-Type': self._get_content_type(file_path),
'Accept-Ranges': 'bytes',
'Cache-Control': 'public, max-age=31536000',
'X-Cache': 'HIT' if from_cache else 'MISS',
'X-Optimized': 'true'
}

return Response(data, status=206, headers=headers)

def _serve_multipart_ranges(self, file_path, ranges, file_size):
"""提供多个范围"""
boundary = f"boundary_{hash(str(ranges))}_{int(time.time())}"
content_type = f"multipart/byteranges; boundary={boundary}"

# 生成多部分响应
parts = []
total_length = 0

for range_info in ranges:
# 获取范围数据
data, from_cache = self.optimizer.optimize_response(
file_path, range_info, use_cache=True
)

# 构建部分
part = self._build_multipart_part(
boundary, range_info, file_size, data, from_cache
)
parts.append(part)
total_length += len(part)

# 添加结束边界
end_boundary = f"\\r\\n–{boundary}–\\r\\n"
parts.append(end_boundary.encode())
total_length += len(end_boundary)

headers = {
'Content-Type': content_type,
'Content-Length': str(total_length),
'Accept-Ranges': 'bytes',
'Cache-Control': 'no-store',
'X-Optimized': 'true'
}

# 组合所有部分
def generate():
for part in parts:
if isinstance(part, bytes):
yield part
else:
yield part

return Response(generate(), status=206, headers=headers)

def _build_multipart_part(self, boundary, range_info, file_size, data, from_cache):
"""构建多部分响应的单个部分"""
part_headers = [
f'–{boundary}',
f'Content-Range: bytes {range_info["start"]}-{range_info["end"]}/{file_size}',
f'Content-Type: {self._get_content_type(None)}',
f'X-Cache: {"HIT" if from_cache else "MISS"}',
''
]

header_bytes = '\\r\\n'.join(part_headers).encode()
return header_bytes + data

def _get_content_type(self, file_path):
"""获取内容类型"""
if not file_path:
return 'application/octet-stream'

mime_types = {
'.mp4': 'video/mp4',
'.avi': 'video/x-msvideo',
'.mkv': 'video/x-matroska',
'.mov': 'video/quicktime',
'.mp3': 'audio/mpeg',
'.jpg': 'image/jpeg',
'.png': 'image/png',
'.pdf': 'application/pdf',
'.zip': 'application/zip'
}

ext = os.path.splitext(file_path)[1].lower()
return mime_types.get(ext, 'application/octet-stream')

def _error_response(self, status, message):
"""错误响应"""
return Response(message, status=status)

def _range_not_satisfiable_response(self, file_size):
"""416响应"""
return Response(
'Requested range not satisfiable',
status=416,
headers={
'Content-Range': f'bytes */{file_size}',
'Content-Type': 'text/plain'
}
)

def preload_common_ranges(self, file_path):
"""预加载常用范围"""
full_path = os.path.join(self.base_dir, file_path)
if os.path.exists(full_path):
self.optimizer.preload_cache(full_path, self.common_ranges)

def cleanup(self):
"""清理资源"""
self.optimizer.cleanup_cache(self.cache_size_mb)

def get_stats(self):
"""获取统计信息"""
return self.optimizer.statistics

# 使用示例
if __name__ == '__main__':
import argparse

parser = argparse.ArgumentParser(description='Optimized Range Request Server')
parser.add_argument('–port', type=int, default=8080, help='Port to listen on')
parser.add_argument('–dir', default='./files', help='Directory to serve files from')
parser.add_argument('–cache', type=int, default=100, help='Cache size in MB')

args = parser.parse_args()

# 创建服务器
server = OptimizedRangeServer(args.dir, args.cache)

# 创建Flask应用
app = Flask(__name__)

@app.route('/files/<path:filename>')
def serve_file(filename):
"""提供文件服务"""
range_header = request.headers.get('Range')

if range_header:
return server.serve_request(filename, range_header)
else:
# 完整文件请求
full_path = os.path.join(server.base_dir, filename)
if os.path.exists(full_path):
return send_file(full_path)
else:
return jsonify({'error': 'File not found'}), 404

@app.route('/preload/<path:filename>')
def preload_file(filename):
"""预加载文件常用范围"""
try:
server.preload_common_ranges(filename)
return jsonify({
'success': True,
'message': f'Preloaded common ranges for {filename}'
})
except Exception as e:
return jsonify({
'success': False,
'error': str(e)
}), 500

@app.route('/stats')
def get_stats():
"""获取服务器统计"""
return jsonify(server.get_stats())

@app.before_request
def log_request():
"""记录请求日志"""
app.logger.info(f"{request.method} {request.path} – Range: {request.headers.get('Range')}")

# 定期清理缓存
import atexit
atexit.register(server.cleanup)

# 启动服务器
app.run(host='0.0.0.0', port=args.port, threaded=True)

最佳实践总结

yaml

# 206 Partial Content最佳实践清单
best_practices_206:

# 服务器实现
server_implementation:
– 始终设置Accept-Ranges: bytes头部
– 正确解析和处理Range请求头部
– 对于无效范围返回416 Range Not Satisfiable
– 支持单范围和多范围请求
– 为多范围请求生成正确的multipart/byteranges响应

# 性能优化
performance_optimization:
– 为小范围请求实现缓存
– 预加载常用范围（如文件开头）
– 使用合适的块大小（通常64KB-1MB）
– 支持并行范围请求
– 实现条件请求（If-Range, If-Match等）

# 安全考虑
security:
– 验证范围请求的合法性
– 限制最大范围大小和数量
– 防止资源耗尽攻击
– 记录异常范围请求模式
– 实施速率限制

# 客户端处理
client_handling:
– 检查Accept-Ranges头部以了解服务器支持
– 正确处理206和416状态码
– 实现断点续传功能
– 对大型文件使用并行下载
– 处理网络错误和重试

# 缓存策略
caching:
– 206响应通常可缓存
– 设置适当的Cache-Control头部
– 对于视频流使用public缓存
– 考虑范围特定的缓存策略

# 特殊情况处理
special_cases:
– 处理后缀范围（如bytes=-100）
– 处理开区间范围（如bytes=100-）
– 处理重叠范围的请求
– 处理超出文件大小的范围

# 监控和日志
monitoring:
– 记录范围请求统计
– 监控范围请求的成功率
– 跟踪字节服务量
– 检测异常请求模式

# 兼容性考虑
compatibility:
– 支持HTTP/1.1 Range规范
– 考虑不支持范围请求的客户端
– 提供回退机制
– 测试各种Range格式

通过本章的详细解析，我们深入了解了206 Partial Content状态码的实现细节、优化策略和最佳实践。206状态码在现代Web应用中非常重要，特别是在大文件传输、流媒体和断点续传等场景中。

正确实现206响应可以显著提高数据传输效率，改善用户体验，同时减少服务器负载。

第9章：其他2xx状态码（203、205、207、208、226）

9.1 203 Non-Authoritative Information

9.1.1 语义深度解析

根据RFC 7231第6.3.4节，203 Non-Authoritative Information状态码的定义如下：

203 (Non-Authoritative Information)状态码表示请求已成功，但包含的实体头部（entity-header）信息并非来自原始服务器，而是来自本地或第三方副本。这可能包括对表示形式的注解（如添加"via"头部），这些注解由中间节点（如代理服务器）应用。

核心语义特征

9.1.2 适用场景与实现

python

# 203 Non-Authoritative Information服务器实现
from flask import Flask, jsonify, Response
from datetime import datetime
import hashlib

app = Flask(__name__)

class NonAuthoritativeResponse:
"""203响应生成器"""

@staticmethod
def create_response(data, original_headers=None, transformed_by=None):
"""
创建203响应

Args:
data: 响应数据
original_headers: 原始头部（如果有）
transformed_by: 转换服务的标识

Returns:
Flask Response对象
"""

# 构建响应
response = jsonify(data)
response.status_code = 203

# 设置标准头部
response.headers['Date'] = datetime.utcnow().strftime('%a, %d %b %Y %H:%M:%S GMT')
response.headers['Server'] = 'ProxyServer/1.0'

# 添加Warning头部（RFC 7234定义）
response.headers['Warning'] = '214 Transformation applied'

# 添加Via头部（如果有中间代理）
if transformed_by:
response.headers['Via'] = f'1.1 {transformed_by}'

# 添加X-头部表示非权威信息
response.headers['X-Information-Source'] = 'Non-authoritative'
response.headers['X-Transformed-By'] = transformed_by or 'proxy-server'

# 保留原始头部信息（如果提供）
if original_headers:
response.headers['X-Original-Headers'] = str(original_headers)

# 添加缓存控制
response.headers['Cache-Control'] = 'public, max-age=300, s-maxage=600'

# 添加实体标签
etag = NonAuthoritativeResponse._generate_etag(data)
response.headers['ETag'] = etag

return response

@staticmethod
def _generate_etag(data):
"""生成ETag"""
content = str(data).encode('utf-8')
return f'"{hashlib.md5(content).hexdigest()}"'

# 代理服务器端点
@app.route('/proxy/api/<path:endpoint>')
def proxy_endpoint(endpoint):
"""
代理端点示例：从上游获取数据并添加非权威信息

模拟场景：
1. 从原始服务器获取数据
2. 修改或添加头部信息
3. 返回203响应
"""

# 模拟从上游服务器获取数据
upstream_data = {
'id': 123,
'name': 'Example Resource',
'timestamp': datetime.utcnow().isoformat(),
'source': 'upstream-server'
}

# 模拟添加的转换或注解
upstream_data['annotations'] = {
'cached': True,
'cache_ttl': 300,
'proxy_version': '1.2.3',
'processed_at': datetime.utcnow().isoformat()
}

# 模拟原始头部
original_headers = {
'Server': 'UpstreamServer/1.0',
'X-Upstream-ID': 'server-abc123'
}

# 创建203响应
return NonAuthoritativeResponse.create_response(
data=upstream_data,
original_headers=original_headers,
transformed_by='edge-proxy.example.com'
)

# 内容转换服务示例
class ContentTransformer:
"""内容转换服务，返回203响应"""

def __init__(self):
self.transformations = {
'compress': self._compress_content,
'filter': self._filter_content,
'annotate': self._annotate_content,
'translate': self._translate_content
}

def transform_response(self, original_response, transformation_type, **kwargs):
"""
转换响应并返回203

Args:
original_response: 原始响应数据
transformation_type: 转换类型
**kwargs: 转换参数

Returns:
203响应
"""

# 应用转换
transformer = self.transformations.get(transformation_type)
if not transformer:
return original_response

transformed_data = transformer(original_response, **kwargs)

# 创建203响应
response_data = {
'data': transformed_data,
'metadata': {
'original_size': len(str(original_response)),
'transformed_size': len(str(transformed_data)),
'transformation': transformation_type,
'applied_at': datetime.utcnow().isoformat(),
'reduction_percentage': self._calculate_reduction(
original_response, transformed_data
)
},
'original_data': original_response if kwargs.get('include_original') else None
}

return NonAuthoritativeResponse.create_response(
data=response_data,
transformed_by=f'content-transformer-{transformation_type}'
)

def _compress_content(self, data, **kwargs):
"""压缩内容（模拟）"""
# 实际中会使用gzip或其他压缩
import json
return json.dumps(data, separators=(',', ':')) # 最小化JSON

def _filter_content(self, data, fields=None, **kwargs):
"""过滤内容字段"""
if not fields or not isinstance(data, dict):
return data

filtered = {}
for field in fields:
if field in data:
filtered[field] = data[field]

return filtered

def _annotate_content(self, data, **kwargs):
"""添加注释"""
annotated = dict(data)
annotated['_annotations'] = {
'processed': True,
'processor': 'annotation-service',
'version': '1.0',
'timestamp': datetime.utcnow().isoformat()
}
return annotated

def _translate_content(self, data, target_lang='en', **kwargs):
"""翻译内容（模拟）"""
# 实际中会调用翻译服务
translated = dict(data)
if 'name' in translated:
translated['name'] = f"[{target_lang}] {translated['name']}"
return translated

def _calculate_reduction(self, original, transformed):
"""计算大小减少百分比"""
orig_size = len(str(original))
trans_size = len(str(transformed))

if orig_size == 0:
return 0

reduction = (orig_size – trans_size) / orig_size * 100
return round(reduction, 2)

# 使用示例
transformer = ContentTransformer()

@app.route('/api/transform/<transformation>')
def transform_content(transformation):
"""内容转换端点"""

# 原始数据
original_data = {
'id': 456,
'name': '示例资源',
'description': '这是一个需要转换的示例资源',
'content': '这里是详细的内容信息，可能会很长…',
'metadata': {
'created': '2023-01-01',
'updated': '2023-01-23',
'tags': ['example', 'test', 'data']
}
}

# 应用转换
transformed_response = transformer.transform_response(
original_data,
transformation,
include_original=False
)

return transformed_response

# 缓存代理实现
class CachingProxy:
"""缓存代理返回203响应"""

def __init__(self, cache_duration=300):
self.cache = {}
self.cache_duration = cache_duration

def get_cached_response(self, request_url, force_refresh=False):
"""
获取缓存响应

Args:
request_url: 请求URL
force_refresh: 是否强制刷新缓存

Returns:
缓存响应（203）或新响应
"""

cache_key = hashlib.md5(request_url.encode()).hexdigest()
now = datetime.utcnow()

# 检查缓存
if not force_refresh and cache_key in self.cache:
cache_entry = self.cache[cache_key]

# 检查是否过期
age = (now – cache_entry['cached_at']).total_seconds()
if age < self.cache_duration:
# 返回缓存响应（203）
return self._create_cached_response(
cache_entry['data'],
cache_entry['original_headers'],
age
)

# 获取新数据
new_data = self._fetch_from_upstream(request_url)
original_headers = self._get_upstream_headers(request_url)

# 更新缓存
self.cache[cache_key] = {
'data': new_data,
'original_headers': original_headers,
'cached_at': now
}

# 如果是强制刷新或第一次获取，返回200
return jsonify(new_data), 200

def _fetch_from_upstream(self, url):
"""从上游获取数据（模拟）"""
# 实际中会进行HTTP请求
return {
'url': url,
'data': 'Fresh data from upstream',
'timestamp': datetime.utcnow().isoformat(),
'source': 'upstream-server'
}

def _get_upstream_headers(self, url):
"""获取上游头部（模拟）"""
return {
'Server': 'Upstream/1.0',
'X-Request-ID': 'req_123456'
}

def _create_cached_response(self, data, original_headers, cache_age):
"""创建缓存响应（203）"""

response_data = {
**data,
'_cache_info': {
'cached': True,
'cache_age_seconds': cache_age,
'max_age': self.cache_duration,
'stale': cache_age > self.cache_duration * 0.8 # 80%过期
}
}

return NonAuthoritativeResponse.create_response(
data=response_data,
original_headers=original_headers,
transformed_by='caching-proxy'
)

# 缓存代理端点
caching_proxy = CachingProxy(cache_duration=300)

@app.route('/cached/<path:url>')
def cached_proxy(url):
"""缓存代理端点"""
full_url = f'https://example.com/{url}'
force_refresh = request.args.get('refresh', 'false').lower() == 'true'

return caching_proxy.get_cached_response(full_url, force_refresh)

if __name__ == '__main__':
app.run(debug=True, port=5001)

9.1.3 客户端处理

javascript

// 客户端处理203响应的示例
class NonAuthoritativeResponseHandler {
/**
* 处理203 Non-Authoritative Information响应
*/

constructor() {
this.warningHandlers = {
'110': this.handleResponseIsStale,
'111': this.handleRevalidationFailed,
'112': this.handleDisconnectedOperation,
'113': this.handleHeuristicExpiration,
'199': this.handleMiscellaneousWarning,
'214': this.handleTransformationApplied,
'299': this.handleMiscellaneousPersistentWarning
};
}

async handle203Response(response) {
console.log('Received 203 Non-Authoritative Information response');

// 解析响应
const data = await response.json();

// 检查Warning头部
const warningHeader = response.headers.get('Warning');
if (warningHeader) {
await this.handleWarnings(warningHeader, response);
}

// 检查Via头部
const viaHeader = response.headers.get('Via');
if (viaHeader) {
console.log(`Response came via: ${viaHeader}`);
}

// 检查X-头部
const infoSource = response.headers.get('X-Information-Source');
if (infoSource === 'Non-authoritative') {
console.warn('This response contains non-authoritative information');

// 根据业务逻辑决定是否使用
if (this.requiresAuthoritativeData()) {
throw new Error('Authoritative data required, but received non-authoritative');
}
}

// 检查缓存信息
if (data._cache_info) {
await this.handleCacheInfo(data._cache_info);
}

return {
data: data,
isAuthoritative: false,
warnings: this.parseWarnings(warningHeader),
via: viaHeader,
metadata: {
etag: response.headers.get('ETag'),
lastModified: response.headers.get('Last-Modified'),
transformedBy: response.headers.get('X-Transformed-By')
}
};
}

parseWarnings(warningHeader) {
if (!warningHeader) return [];

// Warning: 214 Transformation applied, 315 Response is stale
const warnings = [];
const warningParts = warningHeader.split(',');

for (const part of warningParts) {
const match = part.trim().match(/^(\\d{3})\\s+(.+)$/);
if (match) {
const [_, code, text] = match;
warnings.push({
code: parseInt(code, 10),
text: text.trim(),
agent: this.extractAgent(part)
});
}
}

return warnings;
}

extractAgent(warningText) {
// 提取代理信息，例如：214 Transformation applied "proxy.example.com"
const match = warningText.match(/"([^"]+)"/);
return match ? match[1] : null;
}

async handleWarnings(warningHeader, response) {
const warnings = this.parseWarnings(warningHeader);

for (const warning of warnings) {
const handler = this.warningHandlers[warning.code];
if (handler) {
await handler.call(this, warning, response);
} else {
console.warn(`Unhandled warning ${warning.code}: ${warning.text}`);
}
}
}

handleResponseIsStale(warning) {
console.warn('Response is stale:', warning.text);
// 可以考虑重新验证
}

handleRevalidationFailed(warning) {
console.warn('Revalidation failed:', warning.text);
// 无法重新验证，使用可能过期的数据
}

handleTransformationApplied(warning) {
console.log('Transformation applied by:', warning.agent);
// 响应已被转换
}

requiresAuthoritativeData() {
// 根据应用需求决定是否需要权威数据
// 例如：金融交易需要权威数据，而新闻文章可能不需要
return false; // 默认不需要
}

async handleCacheInfo(cacheInfo) {
console.log('Cache information:', cacheInfo);

if (cacheInfo.stale) {
console.warn('Cache is stale, consider refreshing');

// 如果数据过于陈旧，可以触发后台刷新
if (cacheInfo.cache_age_seconds > 600) { // 10分钟
this.triggerBackgroundRefresh();
}
}
}

triggerBackgroundRefresh() {
// 在后台刷新数据
console.log('Triggering background refresh…');
// 实际实现会发送请求并在更新后通知UI
}

// 使用示例
async fetchWith203Handling(url, options = {}) {
try {
const response = await fetch(url, options);

if (response.status === 203) {
const handler = new NonAuthoritativeResponseHandler();
return await handler.handle203Response(response);
} else if (response.ok) {
// 正常响应
const data = await response.json();
return {
data: data,
isAuthoritative: true
};
} else {
throw new Error(`HTTP ${response.status}`);
}
} catch (error) {
console.error('Fetch failed:', error);
throw error;
}
}
}

// 使用示例
document.getElementById('load-data').addEventListener('click', async () => {
const handler = new NonAuthoritativeResponseHandler();

try {
const result = await handler.fetchWith203Handling('/proxy/api/data');

if (!result.isAuthoritative) {
// 显示非权威数据的警告
document.getElementById('warning-banner').style.display = 'block';
document.getElementById('warning-banner').textContent =
'显示的数据可能不是最新的，来自缓存或转换服务';
}

// 使用数据
displayData(result.data);

} catch (error) {
console.error('Failed to load data:', error);
alert('加载数据失败');
}
});

function displayData(data) {
// 显示数据的逻辑
const container = document.getElementById('data-container');
container.innerHTML = JSON.stringify(data, null, 2);
}

9.2 205 Reset Content

9.2.1 语义深度解析

根据RFC 7231第6.3.6节，205 Reset Content状态码的定义如下：

205 (Reset Content)状态码表示服务器已成功处理请求，并且用户代理应该重置发送此请求的文档视图。这个响应主要是为了支持数据输入（如表单）的输入清理，以便用户可以轻松地开始另一次输入。

核心语义特征

9.2.2 适用场景与实现

python

# 205 Reset Content服务器实现
from flask import Flask, request, jsonify, Response, render_template_string
from flask_wtf import FlaskForm
from wtforms import StringField, TextAreaField, SubmitField
from wtforms.validators import DataRequired
import datetime

app = Flask(__name__)
app.config['SECRET_KEY'] = 'your-secret-key-here'

# 表单定义
class ContactForm(FlaskForm):
name = StringField('Name', validators=[DataRequired()])
email = StringField('Email', validators=[DataRequired()])
message = TextAreaField('Message', validators=[DataRequired()])
submit = SubmitField('Submit')

# HTML模板
FORM_TEMPLATE = '''
<!DOCTYPE html>
<html>
<head>
<title>Contact Form</title>
<style>
body { font-family: Arial, sans-serif; max-width: 600px; margin: 0 auto; padding: 20px; }
.form-group { margin-bottom: 15px; }
label { display: block; margin-bottom: 5px; font-weight: bold; }
input, textarea { width: 100%; padding: 8px; border: 1px solid #ddd; border-radius: 4px; }
.error { color: #dc3545; font-size: 0.9em; margin-top: 5px; }
.success { background: #d4edda; color: #155724; padding: 10px; border-radius: 4px; margin-bottom: 20px; }
.btn { background: #007bff; color: white; padding: 10px 20px; border: none; border-radius: 4px; cursor: pointer; }
.btn:hover { background: #0056b3; }
</style>
</head>
<body>
<h1>Contact Us</h1>

{% if success %}
<div class="success" id="success-message">
Thank you for your message! The form has been reset for your next message.
</div>
{% endif %}

<form method="POST" id="contact-form">
{{ form.hidden_tag() }}

<div class="form-group">
{{ form.name.label }}
{{ form.name(size=32) }}
{% for error in form.name.errors %}
<div class="error">{{ error }}</div>
{% endfor %}
</div>

<div class="form-group">
{{ form.email.label }}
{{ form.email(size=32) }}
{% for error in form.email.errors %}
<div class="error">{{ error }}</div>
{% endfor %}
</div>

<div class="form-group">
{{ form.message.label }}
{{ form.message(rows=5) }}
{% for error in form.message.errors %}
<div class="error">{{ error }}</div>
{% endfor %}
</div>

<div class="form-group">
{{ form.submit(class="btn") }}
</div>
</form>

<script>
// 如果收到205响应，重置表单
document.getElementById('contact-form').addEventListener('submit', async function(e) {
e.preventDefault();

const formData = new FormData(this);

try {
const response = await fetch(this.action, {
method: this.method,
body: formData,
headers: {
'X-Requested-With': 'XMLHttpRequest'
}
});

if (response.status === 205) {
// 重置表单
this.reset();

// 显示成功消息
const successDiv = document.createElement('div');
successDiv.className = 'success';
successDiv.id = 'success-message';
successDiv.textContent = 'Thank you for your message! The form has been reset for your next message.';

const existingSuccess = document.getElementById('success-message');
if (existingSuccess) {
existingSuccess.replaceWith(successDiv);
} else {
this.parentNode.insertBefore(successDiv, this);
}

// 5秒后隐藏成功消息
setTimeout(() => {
successDiv.style.transition = 'opacity 0.5s';
successDiv.style.opacity = '0';
setTimeout(() => successDiv.remove(), 500);
}, 5000);

} else {
// 处理其他响应
const data = await response.json();
if (data.errors) {
// 显示错误
alert('Error: ' + JSON.stringify(data.errors));
}
}

} catch (error) {
console.error('Submission error:', error);
alert('An error occurred. Please try again.');
}
});
</script>
</body>
</html>
'''

@app.route('/contact', methods=['GET', 'POST'])
def contact():
form = ContactForm()

if request.method == 'POST':
# 检查是否是AJAX请求
is_ajax = request.headers.get('X-Requested-With') == 'XMLHttpRequest'

if form.validate():
# 处理表单数据（例如保存到数据库）
form_data = {
'name': form.name.data,
'email': form.email.data,
'message': form.message.data,
'timestamp': datetime.datetime.utcnow().isoformat(),
'ip_address': request.remote_addr
}

# 模拟保存到数据库
save_contact_message(form_data)

if is_ajax:
# 对于AJAX请求，返回205 Reset Content
response = Response(status=205)
response.headers['X-Form-Processed'] = 'true'
response.headers['X-Message-ID'] = 'msg_' + str(hash(str(form_data)))
return response
else:
# 对于传统表单提交，重定向到相同页面并显示成功消息
return render_template_string(FORM_TEMPLATE, form=ContactForm(), success=True)
else:
# 验证失败
if is_ajax:
return jsonify({
'success': False,
'errors': form.errors
}), 400
else:
# 重新显示带有错误信息的表单
return render_template_string(FORM_TEMPLATE, form=form, success=False)

# GET请求：显示表单
return render_template_string(FORM_TEMPLATE, form=form, success=False)

def save_contact_message(data):
"""保存联系消息（模拟）"""
# 实际中会保存到数据库
print(f"Saving contact message: {data}")

# 重置游戏状态示例
@app.route('/api/game/reset', methods=['POST'])
def reset_game():
"""
重置游戏状态

典型场景：
1. 游戏结束后重置
2. 用户想要重新开始
3. 清除所有进度
"""

# 验证用户身份
user_id = request.headers.get('X-User-ID')
if not user_id:
return jsonify({'error': 'Authentication required'}), 401

# 重置游戏状态（模拟）
reset_user_game_state(user_id)

# 返回205 Reset Content
response = Response(status=205)
response.headers['X-Game-Reset'] = 'true'
response.headers['X-Reset-Timestamp'] = datetime.datetime.utcnow().isoformat()
response.headers['X-User-ID'] = user_id

return response

def reset_user_game_state(user_id):
"""重置用户游戏状态（模拟）"""
# 实际中会更新数据库
print(f"Resetting game state for user: {user_id}")

# 画布重置示例
@app.route('/api/canvas/reset', methods=['POST'])
def reset_canvas():
"""
重置画布

典型场景：
1. 绘图应用中的清空画布
2. 白板应用中的重置
3. 图表编辑器中的清除
"""

# 获取画布ID
canvas_id = request.json.get('canvas_id')
if not canvas_id:
return jsonify({'error': 'Canvas ID required'}), 400

# 重置画布（模拟）
clear_canvas_data(canvas_id)

# 返回205 Reset Content
response = Response(status=205)
response.headers['X-Canvas-Reset'] = 'true'
response.headers['X-Canvas-ID'] = canvas_id
response.headers['X-Reset-At'] = datetime.datetime.utcnow().isoformat()

return response

def clear_canvas_data(canvas_id):
"""清除画布数据（模拟）"""
# 实际中会清除数据库中的画布数据
print(f"Clearing canvas data for: {canvas_id}")

# 批量表单重置
@app.route('/api/forms/batch-reset', methods=['POST'])
def batch_reset_forms():
"""
批量重置多个表单

典型场景：
1. 多步骤向导中的重置
2. 包含多个表单的页面
3. 问卷应用中的清除所有答案
"""

# 获取要重置的表单ID列表
form_ids = request.json.get('form_ids', [])

if not form_ids:
return jsonify({'error': 'No form IDs provided'}), 400

# 重置每个表单（模拟）
reset_results = []
for form_id in form_ids:
try:
reset_form_data(form_id)
reset_results.append({
'form_id': form_id,
'status': 'reset',
'timestamp': datetime.datetime.utcnow().isoformat()
})
except Exception as e:
reset_results.append({
'form_id': form_id,
'status': 'failed',
'error': str(e)
})

# 返回205 Reset Content（即使部分失败也重置客户端视图）
response = Response(status=205)
response.headers['X-Batch-Reset'] = 'true'
response.headers['X-Reset-Count'] = str(len([r for r in reset_results if r['status'] == 'reset']))
response.headers['X-Failed-Count'] = str(len([r for r in reset_results if r['status'] == 'failed']))

return response

def reset_form_data(form_id):
"""重置表单数据（模拟）"""
# 实际中会清除表单数据
print(f"Resetting form data for: {form_id}")

if __name__ == '__main__':
app.run(debug=True, port=5002)

9.2.3 客户端处理

javascript

// 客户端处理205响应的示例
class ResetContentHandler {
/**
* 处理205 Reset Content响应
*/

constructor() {
this.formSelectors = [
'form[data-reset-on-205]',
'form.reset-on-success',
'.resettable-form'
];

this.initializeEventListeners();
}

initializeEventListeners() {
// 监听所有表单的提交事件
document.querySelectorAll('form').forEach(form => {
form.addEventListener('submit', (e) => this.handleFormSubmit(e));
});

// 监听自定义重置事件
document.addEventListener('resetContentRequired', (e) => {
this.resetContent(e.detail.element, e.detail.options);
});
}

async handleFormSubmit(event) {
const form = event.target;

// 检查是否应该拦截提交
if (!this.shouldInterceptForm(form)) {
return; // 让表单正常提交
}

event.preventDefault();

try {
const response = await this.submitForm(form);

if (response.status === 205) {
// 重置表单
this.resetForm(form, response);

// 显示成功消息
this.showSuccessMessage(form, response);

// 触发自定义事件
this.triggerResetComplete(form, response);

} else if (response.ok) {
// 其他成功响应
const data = await response.json();
this.handleSuccessResponse(form, data);
} else {
// 错误响应
const error = await response.json();
this.handleErrorResponse(form, error);
}

} catch (error) {
console.error('Form submission failed:', error);
this.showError(form, 'Network error. Please try again.');
}
}

shouldInterceptForm(form) {
// 检查表单是否应该被拦截
return form.hasAttribute('data-ajax-submit') ||
form.classList.contains('ajax-form') ||
this.formSelectors.some(selector => form.matches(selector));
}

async submitForm(form) {
const formData = new FormData(form);
const method = form.method.toUpperCase();
const url = form.action;

// 对于GET请求，构建查询字符串
if (method === 'GET') {
const params = new URLSearchParams(formData);
const fullUrl = url + (url.includes('?') ? '&' : '?') + params.toString();
return fetch(fullUrl, { method: 'GET' });
}

// 对于POST/PUT/PATCH请求
return fetch(url, {
method: method,
body: method === 'POST' ? formData : JSON.stringify(Object.fromEntries(formData)),
headers: {
'X-Requested-With': 'XMLHttpRequest',
…(method !== 'POST' && { 'Content-Type': 'application/json' })
}
});
}

resetForm(form, response) {
// 重置表单字段
form.reset();

// 清除自定义字段状态
this.clearCustomFieldStates(form);

// 重置验证状态
this.resetValidation(form);

// 如果有文件输入，清除文件
this.clearFileInputs(form);

// 触发表单重置事件
form.dispatchEvent(new Event('reset', { bubbles: true }));

// 添加视觉反馈
this.addResetAnimation(form);
}

clearCustomFieldStates(form) {
// 清除自定义字段（如富文本编辑器、日期选择器等）
form.querySelectorAll('[data-custom-field]').forEach(field => {
if (field.type === 'textarea' && field.dataset.richText) {
// 重置富文本编辑器
if (window.tinymce && window.tinymce.get(field.id)) {
window.tinymce.get(field.id).setContent('');
}
}

if (field.type === 'select-one' && field.dataset.select2) {
// 重置Select2
if ($ && $(field).data('select2')) {
$(field).val(null).trigger('change');
}
}
});
}

resetValidation(form) {
// 清除验证错误
form.querySelectorAll('.error, .is-invalid').forEach(el => {
el.classList.remove('error', 'is-invalid');
});

// 清除错误消息
form.querySelectorAll('.error-message, .invalid-feedback').forEach(el => {
el.remove();
});
}

clearFileInputs(form) {
// 清除文件输入
form.querySelectorAll('input[type="file"]').forEach(input => {
input.value = '';

// 触发更改事件
input.dispatchEvent(new Event('change', { bubbles: true }));
});
}

addResetAnimation(form) {
// 添加重置动画
form.style.transition = 'all 0.3s';
form.style.backgroundColor = '#f8f9fa';

setTimeout(() => {
form.style.backgroundColor = '';
}, 300);
}

showSuccessMessage(form, response) {
// 显示成功消息
const successDiv = document.createElement('div');
successDiv.className = 'alert alert-success success-message';
successDiv.setAttribute('role', 'alert');

// 从响应头获取消息
const message = response.headers.get('X-Success-Message') ||
'Form submitted successfully! The form has been reset.';

successDiv.textContent = message;

// 插入到表单前
form.parentNode.insertBefore(successDiv, form);

// 5秒后自动隐藏
setTimeout(() => {
successDiv.style.opacity = '0';
successDiv.style.transition = 'opacity 0.5s';

setTimeout(() => {
successDiv.remove();
}, 500);
}, 5000);
}

triggerResetComplete(form, response) {
// 触发自定义事件
const event = new CustomEvent('formResetComplete', {
detail: {
form: form,
response: response,
timestamp: new Date().toISOString(),
headers: Object.fromEntries(response.headers.entries())
},
bubbles: true
});

form.dispatchEvent(event);
}

handleSuccessResponse(form, data) {
// 处理其他成功响应
console.log('Form submitted successfully:', data);

// 显示消息
this.showSuccessMessage(form, {
headers: new Headers({ 'X-Success-Message': data.message || 'Success!' })
});
}

handleErrorResponse(form, error) {
// 显示错误
console.error('Form submission error:', error);

// 显示错误消息
const errorDiv = document.createElement('div');
errorDiv.className = 'alert alert-danger';
errorDiv.textContent = error.message || 'An error occurred. Please check your input.';

form.parentNode.insertBefore(errorDiv, form);

// 3秒后自动隐藏
setTimeout(() => {
errorDiv.remove();
}, 3000);
}

showError(form, message) {
// 显示通用错误
alert(message); // 在实际应用中替换为更好的UI
}

// 通用内容重置方法
resetContent(element, options = {}) {
const defaults = {
clearInputs: true,
resetSelects: true,
clearTextareas: true,
removeDynamicContent: true,
animation: true
};

const settings = { …defaults, …options };

// 重置表单元素
if (element.tagName === 'FORM') {
element.reset();

if (settings.clearInputs) {
element.querySelectorAll('input:not([type="hidden"])').forEach(input => {
if (input.type !== 'checkbox' && input.type !== 'radio') {
input.value = '';
} else {
input.checked = false;
}
});
}

if (settings.resetSelects) {
element.querySelectorAll('select').forEach(select => {
select.selectedIndex = 0;
});
}

if (settings.clearTextareas) {
element.querySelectorAll('textarea').forEach(textarea => {
textarea.value = '';
});
}
}

// 重置动态内容
if (settings.removeDynamicContent) {
element.querySelectorAll('[data-dynamic]').forEach(el => {
el.remove();
});
}

// 添加动画效果
if (settings.animation) {
element.style.transition = 'all 0.3s';
element.style.opacity = '0.8';

setTimeout(() => {
element.style.opacity = '1';
}, 300);
}

// 触发重置事件
element.dispatchEvent(new CustomEvent('contentReset', {
detail: { settings, timestamp: new Date().toISOString() }
}));
}
}

// 使用示例
document.addEventListener('DOMContentLoaded', () => {
// 初始化处理器
const resetHandler = new ResetContentHandler();

// 重置画布示例
document.getElementById('reset-canvas-btn').addEventListener('click', async () => {
try {
const response = await fetch('/api/canvas/reset', {
method: 'POST',
headers: {
'Content-Type': 'application/json'
},
body: JSON.stringify({
canvas_id: 'canvas-123'
})
});

if (response.status === 205) {
// 清空画布
const canvas = document.getElementById('drawing-canvas');
const ctx = canvas.getContext('2d');
ctx.clearRect(0, 0, canvas.width, canvas.height);

// 显示消息
alert('Canvas cleared successfully!');
}
} catch (error) {
console.error('Failed to reset canvas:', error);
}
});

// 重置游戏示例
document.getElementById('reset-game-btn').addEventListener('click', async () => {
if (!confirm('Are you sure you want to reset the game? All progress will be lost.')) {
return;
}

try {
const response = await fetch('/api/game/reset', {
method: 'POST',
headers: {
'X-User-ID': 'user-123'
}
});

if (response.status === 205) {
// 重置游戏界面
resetHandler.resetContent(document.getElementById('game-container'), {
clearInputs: true,
animation: true
});

// 显示重置消息
document.getElementById('game-message').textContent = 'Game reset! Ready to start new game.';
}
} catch (error) {
console.error('Failed to reset game:', error);
}
});

// 批量重置示例
document.getElementById('reset-all-forms').addEventListener('click', () => {
// 触发自定义重置事件
document.dispatchEvent(new CustomEvent('resetContentRequired', {
detail: {
element: document.body,
options: {
clearInputs: true,
resetSelects: true,
animation: true
}
}
}));
});
});

9.3 207 Multi-Status

9.3.1 语义深度解析

根据RFC 4918（WebDAV）第11.1节，207 Multi-Status状态码的定义如下：

207 (Multi-Status)状态码提供有关多个资源的状态信息，适用于需要多个独立状态码的情况。响应体是一个XML文档，包含多个"response"元素，每个元素描述一个独立资源请求的结果。

核心语义特征

9.3.2 适用场景与实现

python

# 207 Multi-Status服务器实现
from flask import Flask, request, Response
import xml.etree.ElementTree as ET
from xml.dom import minidom
from datetime import datetime
from typing import List, Dict, Any
import json

app = Flask(__name__)

class MultiStatusResponse:
"""207 Multi-Status响应生成器"""

DAV_NAMESPACE = 'DAV:'
DAV_PREFIX = 'd'

def __init__(self):
# 注册命名空间
ET.register_namespace(self.DAV_PREFIX, self.DAV_NAMESPACE)

def create_response(self, responses: List[Dict[str, Any]]) -> Response:
"""
创建207 Multi-Status响应

Args:
responses: 响应列表，每个元素包含:
– href: 资源URL
– status: HTTP状态码
– message: 状态消息
– props: 属性字典（可选）
– error: 错误信息（可选）

Returns:
Flask Response对象
"""

# 创建根元素
root = ET.Element(f'{{{self.DAV_NAMESPACE}}}multistatus')

for resp in responses:
response_elem = self._create_response_element(resp)
root.append(response_elem)

# 转换为格式化的XML
xml_str = self._prettify_xml(root)

# 创建响应
return Response(
xml_str,
status=207,
content_type='application/xml; charset="utf-8"',
headers={
'Date': datetime.utcnow().strftime('%a, %d %b %Y %H:%M:%S GMT'),
'Server': 'WebDAV-Server/1.0'
}
)

def _create_response_element(self, response_info: Dict[str, Any]) -> ET.Element:
"""创建单个response元素"""
response_elem = ET.Element(f'{{{self.DAV_NAMESPACE}}}response')

# href元素
href_elem = ET.SubElement(response_elem, f'{{{self.DAV_NAMESPACE}}}href')
href_elem.text = response_info.get('href', '')

# 状态元素
if 'status' in response_info:
status_elem = ET.SubElement(response_elem, f'{{{self.DAV_NAMESPACE}}}status')
status_code = response_info['status']
status_msg = response_info.get('message', 'OK')
status_elem.text = f'HTTP/1.1 {status_code} {status_msg}'

# 属性元素（如果有）
if 'props' in response_info:
propstat_elem = ET.SubElement(response_elem, f'{{{self.DAV_NAMESPACE}}}propstat')

# prop元素
prop_elem = ET.SubElement(propstat_elem, f'{{{self.DAV_NAMESPACE}}}prop')
for prop_name, prop_value in response_info['props'].items():
prop_item = ET.SubElement(prop_elem, f'{{{self.DAV_NAMESPACE}}}{prop_name}')
prop_item.text = str(prop_value)

# 属性状态
prop_status = ET.SubElement(propstat_elem, f'{{{self.DAV_NAMESPACE}}}status')
prop_status.text = f'HTTP/1.1 {response_info.get("prop_status", 200)} OK'

# 错误元素（如果有）
if 'error' in response_info:
error_elem = ET.SubElement(response_elem, f'{{{self.DAV_NAMESPACE}}}error')
error_desc = ET.SubElement(error_elem, f'{{{self.DAV_NAMESPACE}}}description')
error_desc.text = response_info['error']

return response_elem

def _prettify_xml(self, elem: ET.Element) -> str:
"""格式化XML"""
rough_string = ET.tostring(elem, 'utf-8')
parsed = minidom.parseString(rough_string)
return parsed.toprettyxml(indent=' ')

def parse_request(self, xml_body: str) -> List[Dict[str, Any]]:
"""
解析Multi-Status请求体

Args:
xml_body: XML请求体

Returns:
解析后的请求列表
"""
try:
root = ET.fromstring(xml_body)
requests = []

# 查找所有href元素
for href_elem in root.findall(f'.//{{{self.DAV_NAMESPACE}}}href'):
request_info = {
'href': href_elem.text,
'method': self._extract_method(href_elem),
'properties': self._extract_properties(href_elem)
}
requests.append(request_info)

return requests

except ET.ParseError as e:
raise ValueError(f'Invalid XML: {e}')

def _extract_method(self, href_elem: ET.Element) -> str:
"""提取请求方法"""
# 在实际WebDAV中，这可能来自不同的位置
parent = href_elem.getparent()
if parent is not None:
method_elem = parent.find(f'{{{self.DAV_NAMESPACE}}}method')
if method_elem is not None:
return method_elem.text
return 'GET' # 默认

def _extract_properties(self, href_elem: ET.Element) -> Dict[str, str]:
"""提取请求属性"""
properties = {}
parent = href_elem.getparent()

if parent is not None:
prop_elem = parent.find(f'{{{self.DAV_NAMESPACE}}}prop')
if prop_elem is not None:
for child in prop_elem:
# 移除命名空间前缀
tag = child.tag
if '}' in tag:
tag = tag.split('}', 1)[1]
properties[tag] = child.text

return properties

# 创建响应生成器实例
multistatus = MultiStatusResponse()

# WebDAV PROPFIND方法实现
@app.route('/dav/<path:path>', methods=['PROPFIND'])
def webdav_propfind(path):
"""
WebDAV PROPFIND方法

用于检索资源的属性
"""

# 解析请求深度
depth = request.headers.get('Depth', '1')

# 构建响应列表
responses = []

# 添加请求的资源
base_url = f'/dav/{path}'
responses.append({
'href': base_url,
'status': 200,
'message': 'OK',
'props': {
'displayname': path.split('/')[-1] or 'Root',
'resourcetype': '<collection/>' if path.endswith('/') else '',
'getcontentlength': '1024', # 示例大小
'getlastmodified': datetime.utcnow().strftime('%a, %d %b %Y %H:%M:%S GMT'),
'creationdate': '2023-01-01T00:00:00Z'
}
})

# 如果深度大于0，添加子资源
if depth != '0':
# 模拟子资源
for i in range(3):
child_path = f'{base_url}child{i}'
responses.append({
'href': child_path,
'status': 200,
'message': 'OK',
'props': {
'displayname': f'child{i}',
'resourcetype': '',
'getcontentlength': '512',
'getlastmodified': datetime.utcnow().strftime('%a, %d %b %Y %H:%M:%S GMT')
}
})

# 返回207响应
return multistatus.create_response(responses)

# 批量删除实现
@app.route('/api/resources/batch-delete', methods=['POST'])
def batch_delete():
"""
批量删除资源

返回207 Multi-Status响应，包含每个资源的删除结果
"""

# 获取要删除的资源ID列表
resource_ids = request.json.get('resources', [])

if not resource_ids:
return jsonify({'error': 'No resources specified'}), 400

responses = []

for resource_id in resource_ids:
try:
# 尝试删除资源
success = delete_resource(resource_id)

if success:
responses.append({
'href': f'/api/resources/{resource_id}',
'status': 200,
'message': 'OK'
})
else:
responses.append({
'href': f'/api/resources/{resource_id}',
'status': 404,
'message': 'Not Found',
'error': f'Resource {resource_id} not found'
})

except Exception as e:
responses.append({
'href': f'/api/resources/{resource_id}',
'status': 500,
'message': 'Internal Server Error',
'error': str(e)
})

# 返回207响应
return multistatus.create_response(responses)

def delete_resource(resource_id):
"""删除资源（模拟）"""
# 实际中会从数据库删除
print(f"Deleting resource: {resource_id}")
return True # 模拟成功

# 批量创建实现
@app.route('/api/resources/batch-create', methods=['POST'])
def batch_create():
"""
批量创建资源

返回207 Multi-Status响应，包含每个资源的创建结果
"""

resources = request.json.get('resources', [])

if not resources:
return jsonify({'error': 'No resources specified'}), 400

responses = []

for resource_data in resources:
try:
# 验证资源数据
validation_result = validate_resource(resource_data)

if not validation_result['valid']:
responses.append({
'href': '/api/resources',
'status': 400,
'message': 'Bad Request',
'error': validation_result['errors']
})
continue

# 创建资源
created_resource = create_resource(resource_data)

responses.append({
'href': f'/api/resources/{created_resource["id"]}',
'status': 201,
'message': 'Created',
'props': created_resource
})

except Exception as e:
responses.append({
'href': '/api/resources',
'status': 500,
'message': 'Internal Server Error',
'error': str(e)
})

# 返回207响应
return multistatus.create_response(responses)

def validate_resource(data):
"""验证资源数据（模拟）"""
errors = []

if not data.get('name'):
errors.append('Name is required')

return {
'valid': len(errors) == 0,
'errors': errors
}

def create_resource(data):
"""创建资源（模拟）"""
import uuid
resource_id = str(uuid.uuid4())[:8]

return {
'id': resource_id,
'name': data.get('name'),
'created_at': datetime.utcnow().isoformat(),
'status': 'active'
}

# 批量更新实现
@app.route('/api/resources/batch-update', methods=['PATCH'])
def batch_update():
"""
批量更新资源

使用JSON Patch格式更新多个资源
"""

updates = request.json.get('updates', [])

if not updates:
return jsonify({'error': 'No updates specified'}), 400

responses = []

for update in updates:
resource_id = update.get('id')
patch_operations = update.get('patch', [])

if not resource_id or not patch_operations:
responses.append({
'href': '/api/resources',
'status': 400,
'message': 'Bad Request',
'error': 'Missing id or patch operations'
})
continue

try:
# 应用JSON Patch
updated_resource = apply_patch(resource_id, patch_operations)

responses.append({
'href': f'/api/resources/{resource_id}',
'status': 200,
'message': 'OK',
'props': {
'id': resource_id,
'updated': True,
'updated_at': datetime.utcnow().isoformat()
}
})

except ResourceNotFoundError:
responses.append({
'href': f'/api/resources/{resource_id}',
'status': 404,
'message': 'Not Found',
'error': f'Resource {resource_id} not found'
})

except PatchError as e:
responses.append({
'href': f'/api/resources/{resource_id}',
'status': 422,
'message': 'Unprocessable Entity',
'error': str(e)
})

except Exception as e:
responses.append({
'href': f'/api/resources/{resource_id}',
'status': 500,
'message': 'Internal Server Error',
'error': str(e)
})

return multistatus.create_response(responses)

class ResourceNotFoundError(Exception):
pass

class PatchError(Exception):
pass

def apply_patch(resource_id, patch_operations):
"""应用JSON Patch（模拟）"""
# 实际中会更新数据库
print(f"Applying patch to {resource_id}: {patch_operations}")

# 模拟验证
for op in patch_operations:
if op.get('op') not in ['add', 'remove', 'replace', 'move', 'copy']:
raise PatchError(f'Invalid operation: {op.get("op")}')

return True

if __name__ == '__main__':
app.run(debug=True, port=5003)

9.3.3 客户端处理

javascript

// 客户端处理207响应的示例
class MultiStatusResponseHandler {
/**
* 处理207 Multi-Status响应
*/

constructor() {
this.parser = new DOMParser();
}

async handle207Response(response) {
// 检查内容类型
const contentType = response.headers.get('Content-Type');
if (!contentType.includes('application/xml')) {
throw new Error('Expected XML response for 207 status');
}

// 获取XML文本
const xmlText = await response.text();

// 解析XML
const xmlDoc = this.parser.parseFromString(xmlText, 'application/xml');

// 检查解析错误
const parserError = xmlDoc.querySelector('parsererror');
if (parserError) {
throw new Error('Failed to parse XML response');
}

// 提取所有响应
const responses = this.extractResponses(xmlDoc);

// 分析响应结果
const analysis = this.analyzeResponses(responses);

return {
responses: responses,
analysis: analysis,
originalXml: xmlText,
success: analysis.successCount > 0 || analysis.partialSuccess
};
}

extractResponses(xmlDoc) {
const responses = [];

// 查找所有response元素
const responseElements = xmlDoc.getElementsByTagNameNS('DAV:', 'response');

for (const respElem of responseElements) {
const response = {
href: this.extractHref(respElem),
status: this.extractStatus(respElem),
properties: this.extractProperties(respElem),
error: this.extractError(respElem)
};

responses.push(response);
}

return responses;
}

extractHref(responseElem) {
const hrefElem = responseElem.getElementsByTagNameNS('DAV:', 'href')[0];
return hrefElem ? hrefElem.textContent : null;
}

extractStatus(responseElem) {
const statusElem = responseElem.getElementsByTagNameNS('DAV:', 'status')[0];
if (!statusElem) return null;

const statusText = statusElem.textContent;
// 格式: "HTTP/1.1 200 OK"
const match = statusText.match(/HTTP\\/\\d\\.\\d\\s+(\\d+)\\s+(.+)/);

if (match) {
return {
code: parseInt(match[1], 10),
message: match[2],
raw: statusText
};
}

return null;
}

extractProperties(responseElem) {
const properties = {};

// 查找propstat元素
const propstatElems = responseElem.getElementsByTagNameNS('DAV:', 'propstat');

for (const propstatElem of propstatElems) {
const propElem = propstatElem.getElementsByTagNameNS('DAV:', 'prop')[0];
if (!propElem) continue;

// 提取所有属性
for (const child of propElem.children) {
const tagName = child.tagName;
// 移除命名空间
const simpleName = tagName.includes(':') ?
tagName.split(':')[1] : tagName;

properties[simpleName] = child.textContent;
}
}

return properties;
}

extractError(responseElem) {
const errorElem = responseElem.getElementsByTagNameNS('DAV:', 'error')[0];
if (!errorElem) return null;

const descElem = errorElem.getElementsByTagNameNS('DAV:', 'description')[0];
return descElem ? descElem.textContent : 'Unknown error';
}

analyzeResponses(responses) {
const analysis = {
total: responses.length,
successCount: 0,
errorCount: 0,
partialSuccess: false,
statusCodes: {},
errors: []
};

for (const resp of responses) {
if (resp.status) {
const code = resp.status.code;

// 统计状态码
analysis.statusCodes[code] = (analysis.statusCodes[code] || 0) + 1;

// 判断成功/失败
if (code >= 200 && code < 300) {
analysis.successCount++;
} else {
analysis.errorCount++;
if (resp.error) {
analysis.errors.push({
href: resp.href,
error: resp.error,
status: resp.status
});
}
}
}
}

// 检查是否部分成功
analysis.partialSuccess = analysis.successCount > 0 &&
analysis.errorCount > 0;

return analysis;
}

// 批量操作示例
async batchDelete(resourceIds) {
try {
const response = await fetch('/api/resources/batch-delete', {
method: 'POST',
headers: {
'Content-Type': 'application/json'
},
body: JSON.stringify({
resources: resourceIds
})
});

if (response.status === 207) {
const result = await this.handle207Response(response);

// 更新UI
this.updateUIAfterBatchDelete(result);

return result;
} else {
throw new Error(`Unexpected status: ${response.status}`);
}

} catch (error) {
console.error('Batch delete failed:', error);
throw error;
}
}

updateUIAfterBatchDelete(result) {
const { analysis, responses } = result;

// 显示结果摘要
const summary = `
Batch delete completed:
Total: ${analysis.total}
Successful: ${analysis.successCount}
Failed: ${analysis.errorCount}
${analysis.partialSuccess ? '(Partial success)' : ''}
`;

console.log(summary);

// 更新UI：移除成功的资源
for (const resp of responses) {
if (resp.status && resp.status.code === 200) {
// 从UI中移除该资源
this.removeResourceFromUI(resp.href);
} else if (resp.error) {
// 显示错误
this.showResourceError(resp.href, resp.error);
}
}

// 显示通知
if (analysis.partialSuccess) {
this.showPartialSuccessNotification(analysis);
} else if (analysis.successCount === analysis.total) {
this.showSuccessNotification('All resources deleted successfully');
} else {
this.showErrorNotification('Some resources failed to delete');
}
}

removeResourceFromUI(resourceUrl) {
// 从URL中提取ID
const resourceId = resourceUrl.split('/').pop();

// 从列表中移除
const element = document.querySelector(`[data-resource-id="${resourceId}"]`);
if (element) {
element.style.transition = 'opacity 0.3s';
element.style.opacity = '0';

setTimeout(() => {
element.remove();

// 检查是否为空
this.checkEmptyState();
}, 300);
}
}

showResourceError(resourceUrl, error) {
const resourceId = resourceUrl.split('/').pop();
const element = document.querySelector(`[data-resource-id="${resourceId}"]`);

if (element) {
const errorDiv = document.createElement('div');
errorDiv.className = 'error-message';
errorDiv.textContent = `Delete failed: ${error}`;

element.appendChild(errorDiv);
}
}

showPartialSuccessNotification(analysis) {
// 显示部分成功通知
const notification = document.createElement('div');
notification.className = 'notification warning';
notification.innerHTML = `
<strong>Partial Success</strong>
<p>${analysis.successCount} of ${analysis.total} resources deleted successfully.</p>
<p>${analysis.errorCount} resources failed.</p>
`;

this.showNotification(notification, 5000);
}

showSuccessNotification(message) {
const notification = document.createElement('div');
notification.className = 'notification success';
notification.textContent = message;

this.showNotification(notification, 3000);
}

showErrorNotification(message) {
const notification = document.createElement('div');
notification.className = 'notification error';
notification.textContent = message;

this.showNotification(notification, 5000);
}

showNotification(element, duration) {
const container = document.getElementById('notification-container') ||
this.createNotificationContainer();

container.appendChild(element);

setTimeout(() => {
element.style.opacity = '0';
element.style.transition = 'opacity 0.5s';

setTimeout(() => {
element.remove();
}, 500);
}, duration);
}

createNotificationContainer() {
const container = document.createElement('div');
container.id = 'notification-container';
container.style.cssText = `
position: fixed;
top: 20px;
right: 20px;
z-index: 1000;
`;

document.body.appendChild(container);
return container;
}

checkEmptyState() {
const resourceList = document.getElementById('resource-list');
if (resourceList && resourceList.children.length === 0) {
resourceList.innerHTML = `
<div class="empty-state">
<h3>No resources found</h3>
<p>All resources have been deleted.</p>
</div>
`;
}
}

// WebDAV PROPFIND示例
async webdavPropfind(url, depth = '1') {
try {
const response = await fetch(url, {
method: 'PROPFIND',
headers: {
'Depth': depth,
'Content-Type': 'application/xml'
}
});

if (response.status === 207) {
const result = await this.handle207Response(response);

// 处理WebDAV响应
this.processWebDAVResponse(result);

return result;
} else {
throw new Error(`Unexpected status: ${response.status}`);
}

} catch (error) {
console.error('PROPFIND failed:', error);
throw error;
}
}

processWebDAVResponse(result) {
// 构建文件/文件夹树
const tree = this.buildResourceTree(result.responses);

// 显示资源列表
this.displayResourceTree(tree);

return tree;
}

buildResourceTree(responses) {
const tree = {
'/': { name: 'Root', type: 'collection', children: [] }
};

for (const resp of responses) {
const href = resp.href;
const props = resp.properties;

// 解析路径
const path = href.replace(/^\\/dav\\//, '/');
const parts = path.split('/').filter(p => p);

// 构建树节点
let current = tree['/'];

for (let i = 0; i < parts.length; i++) {
const part = parts[i];
const isLast = i === parts.length – 1;
const fullPath = '/' + parts.slice(0, i + 1).join('/');

if (!tree[fullPath]) {
tree[fullPath] = {
name: part,
type: props.resourcetype ? 'collection' : 'resource',
children: [],
properties: isLast ? props : {}
};

current.children.push(tree[fullPath]);
}

current = tree[fullPath];
}
}

return tree;
}

displayResourceTree(tree) {
const container = document.getElementById('dav-explorer');
if (!container) return;

container.innerHTML = this.renderTree(tree['/']);
}

renderTree(node, depth = 0) {
const indent = ' '.repeat(depth);
let html = `${indent}<div class="dav-item" data-type="${node.type}">`;

// 图标
const icon = node.type === 'collection' ? '📁' : '📄';
html += `${icon} ${node.name}`;

// 显示属性（如果展开）
if (node.properties && Object.keys(node.properties).length > 0) {
html += `<div class="dav-properties" style="margin-left: 20px; font-size: 0.9em; color: #666;">`;
for (const [key, value] of Object.entries(node.properties)) {
if (value && value.trim()) {
html += `<div>${key}: ${value}</div>`;
}
}
html += `</div>`;
}

// 子节点
if (node.children.length > 0) {
html += `<div class="dav-children">`;
for (const child of node.children) {
html += this.renderTree(child, depth + 1);
}
html += `</div>`;
}

html += `</div>`;
return html;
}
}

// 使用示例
document.addEventListener('DOMContentLoaded', () => {
const handler = new MultiStatusResponseHandler();

// 批量删除按钮
document.getElementById('batch-delete-btn').addEventListener('click', async () => {
// 获取选中的资源ID
const selectedIds = Array.from(
document.querySelectorAll('input[name="resource"]:checked')
).map(input => input.value);

if (selectedIds.length === 0) {
alert('Please select at least one resource');
return;
}

if (!confirm(`Delete ${selectedIds.length} resources?`)) {
return;
}

try {
await handler.batchDelete(selectedIds);
} catch (error) {
console.error('Batch delete failed:', error);
alert('Failed to delete resources');
}
});

// WebDAV浏览器
document.getElementById('load-dav-btn').addEventListener('click', async () => {
const url = document.getElementById('dav-url').value;

if (!url) {
alert('Please enter a WebDAV URL');
return;
}

try {
await handler.webdavPropfind(url);
} catch (error) {
console.error('Failed to load WebDAV:', error);
alert('Failed to load WebDAV contents');
}
});

// 批量创建示例
document.getElementById('batch-create-form').addEventListener('submit', async (e) => {
e.preventDefault();

const resources = [];
const resourceInputs = document.querySelectorAll('.resource-input');

resourceInputs.forEach(input => {
if (input.value.trim()) {
resources.push({
name: input.value.trim(),
type: input.dataset.type || 'default'
});
}
});

if (resources.length === 0) {
alert('Please enter at least one resource');
return;
}

try {
const response = await fetch('/api/resources/batch-create', {
method: 'POST',
headers: {
'Content-Type': 'application/json'
},
body: JSON.stringify({ resources })
});

if (response.status === 207) {
const result = await handler.handle207Response(response);

// 显示结果
const successCount = result.analysis.successCount;
const errorCount = result.analysis.errorCount;

alert(`Created ${successCount} resources successfully. ${errorCount} failed.`);

// 刷新列表
location.reload();
} else {
throw new Error(`Unexpected status: ${response.status}`);
}

} catch (error) {
console.error('Batch create failed:', error);
alert('Failed to create resources');
}
});
});

9.4 208 Already Reported

9.4.1 语义深度解析

根据RFC 5842（WebDAV绑定扩展）第7.1节，208 Already Reported状态码的定义如下：

208 (Already Reported)状态码在响应WebDAV PROPFIND请求时使用，用于避免重复枚举绑定集合的内部成员。当客户端请求一个绑定的集合（即一个资源有多个URI）时，服务器可以使用208响应来指示该资源已经在前面的响应中报告过，从而避免在响应中多次返回同一个资源。

核心语义特征

9.4.2 适用场景与实现

python

# 208 Already Reported服务器实现
from flask import Flask, request, Response
import xml.etree.ElementTree as ET
from xml.dom import minidom

app = Flask(__name__)

class AlreadyReportedResponse:
"""208 Already Reported响应生成器"""

DAV_NAMESPACE = 'DAV:'
DAV_PREFIX = 'd'

def __init__(self):
ET.register_namespace(self.DAV_PREFIX, self.DAV_NAMESPACE)

def create_propfind_response(self, bindings):
"""
创建PROPFIND响应，使用208避免重复

Args:
bindings: 绑定列表，每个元素是一个字典，包含：
– href: 资源URI
– bindings: 该资源的所有绑定URI列表
– props: 资源属性

Returns:
207 Multi-Status响应，内部可能包含208响应
"""
root = ET.Element(f'{{{self.DAV_NAMESPACE}}}multistatus')

# 用于跟踪已经报告过的资源
reported_resources = set()

for binding in bindings:
# 获取资源的规范URI（例如第一个绑定）
canonical_href = binding['bindings'][0]

# 如果已经报告过，则跳过
if canonical_href in reported_resources:
continue

# 标记为已报告
reported_resources.add(canonical_href)

# 为规范URI创建响应
self._add_response(root, canonical_href, binding['props'], 200)

# 为其他绑定URI创建208响应
for alt_href in binding['bindings'][1:]:
self._add_response(root, alt_href, {}, 208)

# 转换为XML
xml_str = self._prettify_xml(root)

return Response(
xml_str,
status=207,
content_type='application/xml; charset="utf-8"'
)

def _add_response(self, parent, href, props, status_code):
"""添加单个response元素到父元素"""
response_elem = ET.SubElement(parent, f'{{{self.DAV_NAMESPACE}}}response')

# href
href_elem = ET.SubElement(response_elem, f'{{{self.DAV_NAMESPACE}}}href')
href_elem.text = href

# propstat
if props or status_code != 208:
propstat_elem = ET.SubElement(response_elem, f'{{{self.DAV_NAMESPACE}}}propstat')

# prop
prop_elem = ET.SubElement(propstat_elem, f'{{{self.DAV_NAMESPACE}}}prop')
for key, value in props.items():
prop_item = ET.SubElement(prop_elem, f'{{{self.DAV_NAMESPACE}}}{key}')
prop_item.text = value

# status
status_elem = ET.SubElement(propstat_elem, f'{{{self.DAV_NAMESPACE}}}status')
if status_code == 200:
status_elem.text = 'HTTP/1.1 200 OK'
elif status_code == 208:
status_elem.text = 'HTTP/1.1 208 Already Reported'

# 对于208，也可以直接使用status元素（不在propstat内）
else:
status_elem = ET.SubElement(response_elem, f'{{{self.DAV_NAMESPACE}}}status')
status_elem.text = 'HTTP/1.1 208 Already Reported'

def _prettify_xml(self, elem):
"""格式化XML"""
rough_string = ET.tostring(elem, 'utf-8')
parsed = minidom.parseString(rough_string)
return parsed.toprettyxml(indent=' ')

# 模拟绑定集合的数据
def get_binding_data():
"""获取绑定集合数据（模拟）"""

# 假设有两个资源，每个资源有多个绑定（URI）
return [
{
'bindings': [
'/dav/files/doc1', # 规范URI
'/dav/aliases/doc1-alias1', # 别名1
'/dav/aliases/doc1-alias2' # 别名2
],
'props': {
'displayname': 'document1.pdf',
'getcontentlength': '1048576',
'getcontenttype': 'application/pdf',
'getlastmodified': 'Wed, 25 Jan 2023 10:30:00 GMT'
}
},
{
'bindings': [
'/dav/files/doc2',
'/dav/aliases/doc2-alias1'
],
'props': {
'displayname': 'document2.txt',
'getcontentlength': '2048',
'getcontenttype': 'text/plain',
'getlastmodified': 'Wed, 25 Jan 2023 11:45:00 GMT'
}
},
{
'bindings': [
'/dav/files/doc3'
],
'props': {
'displayname': 'document3.jpg',
'getcontentlength': '524288',
'getcontenttype': 'image/jpeg',
'getlastmodified': 'Wed, 25 Jan 2023 12:00:00 GMT'
}
}
]

# WebDAV PROPFIND端点
@app.route('/dav/<path:path>', methods=['PROPFIND'])
def webdav_propfind_with_bindings(path):
"""处理PROPFIND请求，支持绑定集合"""

# 获取绑定数据
bindings = get_binding_data()

# 创建响应
response_builder = AlreadyReportedResponse()
response = response_builder.create_propfind_response(bindings)

return response

if __name__ == '__main__':
app.run(debug=True, port=5004)

9.4.3 客户端处理

javascript

// 客户端处理208响应的示例
class AlreadyReportedHandler {
/**
* 处理WebDAV响应中的208 Already Reported状态
*/

constructor() {
this.parser = new DOMParser();
}

async processPropfindResponse(url) {
try {
const response = await fetch(url, {
method: 'PROPFIND',
headers: {
'Depth': '1',
'Content-Type': 'application/xml'
}
});

if (response.status === 207) {
const xmlText = await response.text();
const xmlDoc = this.parser.parseFromString(xmlText, 'application/xml');

// 提取所有响应
const responses = this.extractResponses(xmlDoc);

// 过滤掉208响应（它们只是指示重复，不包含新信息）
const uniqueResources = this.filterDuplicateResources(responses);

// 显示资源
this.displayResources(uniqueResources);

return uniqueResources;
} else {
throw new Error(`Unexpected status: ${response.status}`);
}
} catch (error) {
console.error('PROPFIND failed:', error);
throw error;
}
}

extractResponses(xmlDoc) {
const responses = [];
const responseElements = xmlDoc.getElementsByTagNameNS('DAV:', 'response');

for (const respElem of responseElements) {
const href = this.getHref(respElem);
const status = this.getStatus(respElem);
const props = this.getProperties(respElem);

responses.push({
href: href,
status: status,
properties: props
});
}

return responses;
}

getHref(responseElem) {
const hrefElem = responseElem.getElementsByTagNameNS('DAV:', 'href')[0];
return hrefElem ? hrefElem.textContent : null;
}

getStatus(responseElem) {
// 查找status元素
const statusElems = responseElem.getElementsByTagNameNS('DAV:', 'status');
if (statusElems.length === 0) return null;

const statusText = statusElems[0].textContent;
const match = statusText.match(/HTTP\\/\\d\\.\\d\\s+(\\d+)\\s+(.+)/);

if (match) {
return {
code: parseInt(match[1], 10),
message: match[2]
};
}

return null;
}

getProperties(responseElem) {
const properties = {};
const propstatElems = responseElem.getElementsByTagNameNS('DAV:', 'propstat');

for (const propstatElem of propstatElems) {
const propElem = propstatElem.getElementsByTagNameNS('DAV:', 'prop')[0];
if (!propElem) continue;

for (const child of propElem.children) {
const tagName = child.tagName;
const simpleName = tagName.includes(':') ?
tagName.split(':')[1] : tagName;

properties[simpleName] = child.textContent;
}
}

return properties;
}

filterDuplicateResources(responses) {
// 用于跟踪已经看到的资源（基于属性）
const seenResources = new Map();
const uniqueResources = [];

for (const resp of responses) {
// 跳过208响应
if (resp.status && resp.status.code === 208) {
console.log(`Skipping already reported resource: ${resp.href}`);
continue;
}

// 生成资源标识符（基于属性，而不是href）
const resourceId = this.generateResourceId(resp.properties);

if (resourceId && seenResources.has(resourceId)) {
console.log(`Duplicate resource detected: ${resp.href}`);
// 可以选择保留一个（例如，规范URI）
continue;
}

if (resourceId) {
seenResources.set(resourceId, resp);
}

uniqueResources.push(resp);
}

console.log(`Found ${uniqueResources.length} unique resources (${responses.length – uniqueResources.length} duplicates skipped)`);
return uniqueResources;
}

generateResourceId(properties) {
// 使用关键属性生成资源ID
// 在实际应用中，可能需要根据服务器行为调整
const keys = ['displayname', 'getcontentlength', 'getcontenttype', 'getlastmodified'];
const values = keys.map(key => properties[key] || '').join('|');

if (values.trim()) {
return btoa(values); // 简单编码
}

return null;
}

displayResources(resources) {
const container = document.getElementById('dav-resources');
if (!container) return;

container.innerHTML = '';

resources.forEach(resource => {
const resourceElem = document.createElement('div');
resourceElem.className = 'dav-resource';

const name = resource.properties.displayname || resource.href.split('/').pop();
const type = resource.properties.getcontenttype || 'unknown';
const size = resource.properties.getcontentlength || '0';

resourceElem.innerHTML = `
<h3>${name}</h3>
<p><strong>URI:</strong> ${resource.href}</p>
<p><strong>Type:</strong> ${type}</p>
<p><strong>Size:</strong> ${this.formatBytes(size)}</p>
<p><strong>Modified:</strong> ${resource.properties.getlastmodified || 'unknown'}</p>
`;

container.appendChild(resourceElem);
});
}

formatBytes(bytes) {
bytes = parseInt(bytes, 10);
if (bytes === 0) return '0 Bytes';

const k = 1024;
const sizes = ['Bytes', 'KB', 'MB', 'GB'];
const i = Math.floor(Math.log(bytes) / Math.log(k));

return parseFloat((bytes / Math.pow(k, i)).toFixed(2)) + ' ' + sizes[i];
}
}

// 使用示例
document.addEventListener('DOMContentLoaded', () => {
const handler = new AlreadyReportedHandler();

document.getElementById('load-dav-bindings').addEventListener('click', async () => {
const url = document.getElementById('dav-url').value;

if (!url) {
alert('Please enter a WebDAV URL');
return;
}

try {
await handler.processPropfindResponse(url);
} catch (error) {
console.error('Failed to load DAV bindings:', error);
alert('Failed to load resources');
}
});
});

9.5 226 IM Used

9.5.1 语义深度解析

根据RFC 3229（增量编码）第10.4.1节，226 IM Used状态码的定义如下：

226 (IM Used)状态码表示服务器已成功完成对资源的请求，并且响应是实例操作（instance manipulation）应用于当前实例后的表示。实例操作的示例包括差分编码（delta encoding）、压缩或转换。

核心语义特征

9.5.2 适用场景与实现

python

# 226 IM Used服务器实现
from flask import Flask, request, Response
import json
import hashlib
from datetime import datetime
from typing import Dict, Any, Optional

app = Flask(__name__)

class DeltaEncoder:
"""增量编码器，用于生成226响应"""

def __init__(self):
self.resource_versions = {}

def get_delta_response(self, resource_id: str, client_etag: Optional[str]) -> Dict[str, Any]:
"""
获取增量响应

Args:
resource_id: 资源ID
client_etag: 客户端当前版本的ETag

Returns:
响应数据和状态码
"""
# 获取当前资源状态
current_version = self.get_current_version(resource_id)
current_etag = self.generate_etag(current_version)

# 如果客户端没有提供ETag，返回完整资源
if not client_etag:
return {
'status': 200,
'data': current_version,
'headers': {
'ETag': current_etag,
'IM': None # 没有使用实例操作
}
}

# 如果客户端ETag与当前ETag相同，返回304
if client_etag == current_etag:
return {
'status': 304,
'data': None,
'headers': {'ETag': current_etag}
}

# 查找客户端版本
client_version = self.find_version_by_etag(resource_id, client_etag)

if client_version:
# 计算增量
delta = self.compute_delta(client_version, current_version)

# 如果增量比完整资源小，返回226
if self.is_delta_smaller(delta, current_version):
return {
'status': 226,
'data': delta,
'headers': {
'ETag': current_etag,
'IM': 'delta', # 使用了增量编码
'Delta-Base': client_etag # 基于哪个版本
}
}

# 回退到完整资源
return {
'status': 200,
'data': current_version,
'headers': {
'ETag': current_etag,
'IM': None
}
}

def get_current_version(self, resource_id: str) -> Dict[str, Any]:
"""获取资源的当前版本"""
# 模拟数据 – 实际中从数据库获取
if resource_id not in self.resource_versions:
self.resource_versions[resource_id] = {
'versions': [],
'current_index': -1
}

resource = self.resource_versions[resource_id]

# 如果没有版本，创建初始版本
if resource['current_index'] < 0:
initial_version = self.create_initial_version(resource_id)
resource['versions'].append({
'data': initial_version,
'etag': self.generate_etag(initial_version),
'timestamp': datetime.utcnow().isoformat()
})
resource['current_index'] = 0

return resource['versions'][resource['current_index']]['data']

def create_initial_version(self, resource_id: str) -> Dict[str, Any]:
"""创建初始版本"""
return {
'id': resource_id,
'title': f'Resource {resource_id}',
'content': 'Initial content',
'version': 1,
'created_at': datetime.utcnow().isoformat(),
'updated_at': datetime.utcnow().isoformat()
}

def update_resource(self, resource_id: str, updates: Dict[str, Any]) -> Dict[str, Any]:
"""更新资源，创建新版本"""
if resource_id not in self.resource_versions:
self.get_current_version(resource_id) # 初始化

resource = self.resource_versions[resource_id]
current = resource['versions'][resource['current_index']]['data'].copy()

# 应用更新
current.update(updates)
current['version'] += 1
current['updated_at'] = datetime.utcnow().isoformat()

# 保存新版本
new_version = {
'data': current,
'etag': self.generate_etag(current),
'timestamp': datetime.utcnow().isoformat()
}

resource['versions'].append(new_version)
resource['current_index'] += 1

# 限制版本历史长度
if len(resource['versions']) > 10:
resource['versions'] = resource['versions'][-10:]
resource['current_index'] = min(resource['current_index'], 9)

return current

def generate_etag(self, data: Dict[str, Any]) -> str:
"""生成ETag"""
content = json.dumps(data, sort_keys=True)
return f'"{hashlib.md5(content.encode()).hexdigest()}"'

def find_version_by_etag(self, resource_id: str, etag: str) -> Optional[Dict[str, Any]]:
"""根据ETag查找版本"""
if resource_id not in self.resource_versions:
return None

for version in self.resource_versions[resource_id]['versions']:
if version['etag'] == etag:
return version['data']

return None

def compute_delta(self, old_version: Dict[str, Any], new_version: Dict[str, Any]) -> Dict[str, Any]:
"""计算两个版本之间的差异"""
delta = {
'operation': 'patch',
'from_version': old_version.get('version'),
'to_version': new_version.get('version'),
'changes': []
}

# 简单比较：找出变化的字段
all_keys = set(old_version.keys()) | set(new_version.keys())

for key in all_keys:
old_value = old_version.get(key)
new_value = new_version.get(key)

if old_value != new_value:
delta['changes'].append({
'field': key,
'old': old_value,
'new': new_value,
'operation': 'replace' if old_value and new_value else ('add' if new_value else 'remove')
})

return delta

def is_delta_smaller(self, delta: Dict[str, Any], full_data: Dict[str, Any]) -> bool:
"""检查增量是否比完整数据小"""
delta_size = len(json.dumps(delta))
full_size = len(json.dumps(full_data))

# 如果增量小于完整数据的75%，则认为它更小
return delta_size < full_size * 0.75

# 创建编码器实例
delta_encoder = DeltaEncoder()

# 支持增量编码的资源端点
@app.route('/api/delta/<resource_id>', methods=['GET'])
def get_resource_with_delta(resource_id):
"""获取资源，支持增量编码"""

# 获取客户端的ETag（可能来自If-None-Match）
client_etag = request.headers.get('If-None-Match')

# 获取响应
response_info = delta_encoder.get_delta_response(resource_id, client_etag)

# 构建响应
if response_info['status'] == 226:
# 226 IM Used
response = jsonify({
'data': response_info['data'],
'_delta': True,
'_base_version': request.headers.get('If-None-Match'),
'_current_etag': response_info['headers']['ETag']
})
response.status_code = 226

# 设置头部
for key, value in response_info['headers'].items():
if value is not None:
response.headers[key] = value

# 添加增量编码特定的头部
response.headers['Vary'] = 'If-None-Match'
response.headers['Cache-Control'] = 'private, must-revalidate'

return response

elif response_info['status'] == 304:
# 304 Not Modified
response = Response(status=304)
response.headers['ETag'] = response_info['headers']['ETag']
return response

else:
# 200 OK – 完整资源
response = jsonify(response_info['data'])
response.headers['ETag'] = response_info['headers']['ETag']
response.headers['Cache-Control'] = 'private, must-revalidate'
return response

# 更新资源端点
@app.route('/api/delta/<resource_id>', methods=['PATCH'])
def update_resource(resource_id):
"""更新资源，创建新版本"""

updates = request.json
if not updates:
return jsonify({'error': 'No updates provided'}), 400

# 更新资源
new_version = delta_encoder.update_resource(resource_id, updates)

# 返回新版本
response = jsonify(new_version)
response.headers['ETag'] = delta_encoder.generate_etag(new_version)
response.headers['Cache-Control'] = 'private, no-cache'

return response

# 获取资源版本历史
@app.route('/api/delta/<resource_id>/history', methods=['GET'])
def get_resource_history(resource_id):
"""获取资源的版本历史"""

if resource_id not in delta_encoder.resource_versions:
return jsonify({'error': 'Resource not found'}), 404

resource = delta_encoder.resource_versions[resource_id]
history = []

for i, version in enumerate(resource['versions']):
history.append({
'version': version['data'].get('version'),
'etag': version['etag'],
'timestamp': version['timestamp'],
'is_current': i == resource['current_index']
})

return jsonify({
'resource_id': resource_id,
'current_version': resource['current_index'] + 1,
'history': history
})

if __name__ == '__main__':
app.run(debug=True, port=5005)

9.5.3 客户端处理

javascript

// 客户端处理226响应的示例
class DeltaEncodingClient {
/**
* 处理增量编码（226 IM Used）的客户端
*/

constructor(resourceUrl) {
this.resourceUrl = resourceUrl;
this.currentData = null;
this.currentEtag = null;
this.useDeltaEncoding = true;
}

async fetchResource(forceFull = false) {
try {
const headers = {};

// 如果已有ETag，并且不强制获取完整资源，则发送If-None-Match
if (this.currentEtag && !forceFull && this.useDeltaEncoding) {
headers['If-None-Match'] = this.currentEtag;
}

const response = await fetch(this.resourceUrl, { headers });

if (response.status === 226) {
// 增量响应
return await this.handle226Response(response);
} else if (response.status === 304) {
// 未修改
console.log('Resource not modified');
return this.currentData;
} else if (response.ok) {
// 完整响应
return await this.handle200Response(response);
} else {
throw new Error(`HTTP ${response.status}`);
}
} catch (error) {
console.error('Failed to fetch resource:', error);
throw error;
}
}

async handle226Response(response) {
console.log('Received 226 IM Used (delta encoded) response');

// 获取增量数据
const deltaData = await response.json();
const delta = deltaData.data;
const newEtag = response.headers.get('ETag');
const imHeader = response.headers.get('IM');

// 验证这是否真的是增量
if (!delta || !delta._delta) {
console.warn('Expected delta response but got full data');
this.currentData = deltaData;
this.currentEtag = newEtag;
return this.currentData;
}

// 应用增量
if (delta.operation === 'patch' && Array.isArray(delta.changes)) {
this.applyPatch(delta);
} else {
console.warn('Unknown delta format:', delta);
// 回退到完整请求
return await this.fetchResource(true);
}

// 更新ETag
this.currentEtag = newEtag;

// 记录统计信息
const deltaSize = JSON.stringify(delta).length;
console.log(`Applied delta: ${delta.changes.length} changes, delta size: ${deltaSize} bytes`);

// 触发更新事件
this.triggerUpdateEvent('delta-applied', {
changes: delta.changes.length,
deltaSize,
newEtag
});

return this.currentData;
}

applyPatch(delta) {
if (!this.currentData) {
console.error('Cannot apply patch without base data');
return;
}

// 应用每个变更
delta.changes.forEach(change => {
switch (change.operation) {
case 'replace':
this.currentData[change.field] = change.new;
break;
case 'add':
this.currentData[change.field] = change.new;
break;
case 'remove':
delete this.currentData[change.field];
break;
default:
console.warn(`Unknown operation: ${change.operation}`);
}
});

// 更新版本信息
if (delta.to_version !== undefined) {
this.currentData.version = delta.to_version;
}

this.currentData.updated_at = new Date().toISOString();
}

async handle200Response(response) {
console.log('Received 200 OK (full resource) response');

this.currentData = await response.json();
this.currentEtag = response.headers.get('ETag');

// 触发更新事件
this.triggerUpdateEvent('full-update', {
dataSize: JSON.stringify(this.currentData).length,
etag: this.currentEtag
});

return this.currentData;
}

triggerUpdateEvent(type, details) {
const event = new CustomEvent('resource-updated', {
detail: {
type: type,
data: this.currentData,
etag: this.currentEtag,
timestamp: new Date().toISOString(),
…details
}
});

// 在document上触发事件，以便其他组件可以监听
document.dispatchEvent(event);
}

async updateResource(updates) {
try {
const response = await fetch(this.resourceUrl, {
method: 'PATCH',
headers: {
'Content-Type': 'application/json'
},
body: JSON.stringify(updates)
});

if (response.ok) {
const newData = await response.json();
const newEtag = response.headers.get('ETag');

this.currentData = newData;
this.currentEtag = newEtag;

// 触发更新事件
this.triggerUpdateEvent('local-update', {
updates,
newEtag
});

return newData;
} else {
throw new Error(`Update failed: HTTP ${response.status}`);
}
} catch (error) {
console.error('Failed to update resource:', error);
throw error;
}
}

// 手动设置数据（例如从缓存加载）
setData(data, etag) {
this.currentData = data;
this.currentEtag = etag;
}

// 启用/禁用增量编码
setDeltaEncoding(enabled) {
this.useDeltaEncoding = enabled;
console.log(`Delta encoding ${enabled ? 'enabled' : 'disabled'}`);
}

// 获取当前状态
getStatus() {
return {
hasData: !!this.currentData,
etag: this.currentEtag,
usingDelta: this.useDeltaEncoding,
dataSize: this.currentData ? JSON.stringify(this.currentData).length : 0
};
}
}

// 使用示例
document.addEventListener('DOMContentLoaded', () => {
const client = new DeltaEncodingClient('/api/delta/document-123');

// 监听资源更新事件
document.addEventListener('resource-updated', (event) => {
console.log('Resource updated:', event.detail.type);

// 更新UI
updateResourceDisplay(event.detail.data);

// 显示通知
showUpdateNotification(event.detail);
});

// 初始加载
client.fetchResource().then(data => {
console.log('Initial load complete:', data);
updateResourceDisplay(data);
}).catch(error => {
console.error('Initial load failed:', error);
});

// 定期刷新（例如每30秒）
setInterval(() => {
client.fetchResource().catch(error => {
console.error('Background refresh failed:', error);
});
}, 30000);

// 更新按钮
document.getElementById('update-resource').addEventListener('click', async () => {
const newTitle = prompt('Enter new title:');
if (newTitle) {
try {
await client.updateResource({ title: newTitle });
} catch (error) {
alert('Update failed: ' + error.message);
}
}
});

// 切换增量编码
document.getElementById('toggle-delta').addEventListener('click', () => {
const enabled = document.getElementById('toggle-delta').checked;
client.setDeltaEncoding(enabled);
});
});

function updateResourceDisplay(data) {
const container = document.getElementById('resource-container');
if (!container) return;

container.innerHTML = `
<h2>${data.title || 'Untitled'}</h2>
<p><strong>ID:</strong> ${data.id}</p>
<p><strong>Version:</strong> ${data.version || 1}</p>
<p><strong>Content:</strong> ${data.content || 'No content'}</p>
<p><strong>Updated:</strong> ${new Date(data.updated_at).toLocaleString()}</p>
`;
}

function showUpdateNotification(detail) {
const notification = document.createElement('div');
notification.className = 'notification';

let message = '';
switch (detail.type) {
case 'delta-applied':
message = `Updated with ${detail.changes} changes (delta: ${detail.deltaSize} bytes)`;
notification.classList.add('info');
break;
case 'full-update':
message = `Full update received (${detail.dataSize} bytes)`;
notification.classList.add('success');
break;
case 'local-update':
message = 'Local update successful';
notification.classList.add('success');
break;
default:
message = 'Resource updated';
}

notification.textContent = message;

// 添加到页面
const container = document.getElementById('notification-container') || createNotificationContainer();
container.appendChild(notification);

// 3秒后自动移除
setTimeout(() => {
notification.style.opacity = '0';
notification.style.transition = 'opacity 0.5s';

setTimeout(() => {
notification.remove();
}, 500);
}, 3000);
}

function createNotificationContainer() {
const container = document.createElement('div');
container.id = 'notification-container';
container.style.cssText = `
position: fixed;
top: 20px;
right: 20px;
z-index: 1000;
`;

document.body.appendChild(container);
return container;
}

9.6 总结与对比分析

2xx状态码对比矩阵

选择指南

python

# 2xx状态码选择决策树
class StatusCodeSelector:
"""2xx状态码选择器"""

@staticmethod
def select_status_code(scenario):
"""
根据场景选择最合适的2xx状态码

Args:
scenario: 包含以下字段的字典
– method: HTTP方法
– action: 操作类型（create, read, update, delete, batch, etc.）
– needs_response_body: 是否需要响应体
– is_async: 是否异步处理
– is_partial: 是否部分内容
– is_batch: 是否批量操作
– transformed: 是否经过转换
– client_should_reset: 客户端是否应重置

Returns:
推荐的状态码
"""

method = scenario.get('method', 'GET')
action = scenario.get('action')

# 特殊情况：范围请求
if scenario.get('is_partial'):
return 206

# 特殊情况：批量操作（WebDAV风格）
if scenario.get('is_batch') and scenario.get('format') == 'xml':
return 207

# 特殊情况：代理转换的响应
if scenario.get('transformed'):
return 203

# 特殊情况：客户端需要重置
if scenario.get('client_should_reset'):
return 205

# 根据方法和操作选择
if action == 'create':
if scenario.get('is_async'):
return 202
return 201

elif action == 'delete':
if scenario.get('needs_response_body'):
return 200
return 204

elif action == 'update':
if scenario.get('needs_response_body'):
return 200
return 204

elif action == 'read':
return 200

else:
# 默认
return 200

@staticmethod
def get_best_practices(status_code):
"""获取特定状态码的最佳实践"""

practices = {
200: {
'body': '应该包含请求的资源表示或操作结果',
'headers': ['Content-Type', 'Content-Length', 'ETag', 'Last-Modified'],
'cache': '默认可缓存，设置适当的Cache-Control',
'clients': '应该处理响应体并更新UI'
},
201: {
'body': '应该包含创建的资源表示',
'headers': ['Location', 'Content-Type', 'Content-Length'],
'cache': '通常不缓存或短时间缓存',
'clients': '应该跟随Location头部获取新资源'
},
202: {
'body': '应该包含任务状态和跟踪信息',
'headers': ['Location', 'Retry-After', 'X-Task-ID'],
'cache': '不缓存',
'clients': '应该轮询状态或等待回调'
},
203: {
'body': '可选，但通常包含数据',
'headers': ['Warning', 'Via', 'X-Transformed-By'],
'cache': '可缓存，但注意非权威性',
'clients': '应该注意数据可能不是最新的'
},
204: {
'body': '禁止包含响应体',
'headers': ['ETag', 'Last-Modified', 'X-Operation-Type'],
'cache': '通常不缓存',
'clients': '应该更新本地状态和UI'
},
205: {
'body': '禁止包含响应体',
'headers': ['X-Reset-Type', 'X-Form-ID'],
'cache': '不缓存',
'clients': '应该重置表单或视图状态'
},
206: {
'body': '必须包含请求的范围内容',
'headers': ['Content-Range', 'Accept-Ranges', 'Content-Length'],
'cache': '可缓存',
'clients': '应该合并多个范围或处理部分内容'
},
207: {
'body': '必须是XML格式，包含多个response元素',
'headers': ['Content-Type: application/xml'],
'cache': '不缓存',
'clients': '应该解析XML并处理每个独立状态'
},
208: {
'body': 'XML格式，在207响应内部使用',
'headers': [],
'cache': '不适用',
'clients': '应该跳过已报告的资源'
},
226: {
'body': '必须包含增量数据',
'headers': ['IM', 'Delta-Base', 'ETag'],
'cache': '可缓存，但注意增量特性',
'clients': '应该应用增量到现有数据'
}
}

return practices.get(status_code, {})

9.7 实际应用案例

案例1：电子商务订单系统

python

# 电子商务系统中的2xx状态码应用
class ECommerceOrderSystem:
"""电子商务订单系统"""

def process_order(self, order_data):
"""处理订单 – 使用202 Accepted表示异步处理"""
# 验证订单
if not self.validate_order(order_data):
return {'status': 400, 'error': 'Invalid order'}

# 创建异步任务
task_id = self.create_processing_task(order_data)

# 返回202 Accepted
return {
'status': 202,
'body': {
'task_id': task_id,
'message': 'Order processing started',
'status_url': f'/api/orders/status/{task_id}',
'estimated_completion': '10 minutes'
},
'headers': {
'Location': f'/api/orders/status/{task_id}',
'Retry-After': '30'
}
}

def cancel_order(self, order_id):
"""取消订单 – 使用204 No Content"""
success = self.cancel_order_in_db(order_id)

if success:
return {
'status': 204,
'headers': {
'X-Order-Cancelled': order_id,
'X-Cancelled-At': datetime.now().isoformat()
}
}
else:
return {'status': 404, 'error': 'Order not found'}

def update_order_status(self, order_id, status):
"""更新订单状态 – 使用200 OK或204 No Content"""
updated_order = self.update_status_in_db(order_id, status)

if updated_order:
# 如果需要返回更新后的订单，使用200
return {
'status': 200,
'body': updated_order,
'headers': {'ETag': self.generate_etag(updated_order)}
}
else:
# 如果只是状态更新，使用204
return {
'status': 204,
'headers': {
'X-Order-Updated': order_id,
'X-New-Status': status
}
}

案例2：内容管理系统

python

# CMS系统中的2xx状态码应用
class ContentManagementSystem:
"""内容管理系统"""

def publish_article(self, article_id):
"""发布文章 – 使用200 OK或204 No Content"""
article = self.get_article(article_id)

if not article:
return {'status': 404, 'error': 'Article not found'}

# 如果已经发布，返回204（幂等）
if article['published']:
return {
'status': 204,
'headers': {
'X-Article-Already-Published': article_id,
'X-Published-At': article['published_at']
}
}

# 发布文章
published_article = self.publish_in_db(article_id)

# 返回200 OK包含完整文章
return {
'status': 200,
'body': published_article,
'headers': {
'ETag': self.generate_etag(published_article),
'X-Published-At': published_article['published_at']
}
}

def batch_delete_comments(self, comment_ids):
"""批量删除评论 – 使用207 Multi-Status"""
results = []

for comment_id in comment_ids:
try:
success = self.delete_comment(comment_id)

results.append({
'href': f'/api/comments/{comment_id}',
'status': 200 if success else 404,
'message': 'Deleted' if success else 'Not found'
})
except Exception as e:
results.append({
'href': f'/api/comments/{comment_id}',
'status': 500,
'error': str(e)
})

# 创建207响应
return {
'status': 207,
'body': self.create_multistatus_xml(results),
'headers': {'Content-Type': 'application/xml'}
}

9.8 监控和调试

监控仪表板

python

# 2xx状态码监控
class StatusCodeMonitor:
"""状态码监控器"""

def __init__(self):
self.stats = {
200: 0, 201: 0, 202: 0, 203: 0, 204: 0,
205: 0, 206: 0, 207: 0, 208: 0, 226: 0
}
self.timestamps = []
self.errors = []

def record_response(self, status_code, endpoint, duration):
"""记录响应"""
if status_code in self.stats:
self.stats[status_code] += 1

self.timestamps.append({
'time': datetime.now(),
'status': status_code,
'endpoint': endpoint,
'duration': duration
})

# 保留最近1000条记录
if len(self.timestamps) > 1000:
self.timestamps = self.timestamps[-1000:]

def get_statistics(self, time_window_minutes=60):
"""获取统计信息"""
cutoff = datetime.now() – timedelta(minutes=time_window_minutes)
recent = [t for t in self.timestamps if t['time'] > cutoff]

stats = {
'total': len(recent),
'by_status': {},
'by_endpoint': {},
'avg_duration': 0,
'success_rate': 0
}

# 按状态码统计
for status, count in self.stats.items():
stats['by_status'][status] = count

# 按端点统计
for record in recent:
endpoint = record['endpoint']
stats['by_endpoint'][endpoint] = stats['by_endpoint'].get(endpoint, 0) + 1

# 平均持续时间
if recent:
avg_duration = sum(r['duration'] for r in recent) / len(recent)
stats['avg_duration'] = avg_duration

# 成功率（2xx状态码比例）
if recent:
success_count = sum(1 for r in recent if 200 <= r['status'] < 300)
stats['success_rate'] = (success_count / len(recent)) * 100

return stats

def generate_alerts(self):
"""生成告警"""
alerts = []
stats = self.get_statistics(5) # 最近5分钟

# 检查成功率
if stats['success_rate'] < 95:
alerts.append({
'level': 'warning',
'message': f'Low success rate: {stats["success_rate"]:.1f}%',
'metric': 'success_rate',
'value': stats['success_rate']
})

# 检查异常状态码比例
unexpected_codes = [203, 205, 207, 208, 226]
for code in unexpected_codes:
count = stats['by_status'].get(code, 0)
if count > 10: # 短时间内出现多次
alerts.append({
'level': 'info',
'message': f'Unusual number of {code} responses: {count}',
'metric': f'status_{code}',
'value': count
})

return alerts

调试工具

javascript

// 2xx状态码调试工具
class StatusCodeDebugger {
/**
* 状态码调试工具
*/

constructor() {
this.interceptRequests();
this.setupUI();
}

interceptRequests() {
// 拦截fetch请求
const originalFetch = window.fetch;
window.fetch = async function(…args) {
const startTime = Date.now();
const response = await originalFetch.apply(this, args);
const duration = Date.now() – startTime;

// 记录2xx响应
if (response.status >= 200 && response.status < 300) {
StatusCodeDebugger.logResponse({
url: args[0],
method: args[1]?.method || 'GET',
status: response.status,
statusText: response.statusText,
duration: duration,
headers: Object.fromEntries(response.headers.entries()),
timestamp: new Date().toISOString()
});
}

return response;
};

// 拦截XMLHttpRequest
const originalOpen = XMLHttpRequest.prototype.open;
XMLHttpRequest.prototype.open = function(…args) {
this._url = args[1];
this._method = args[0];
return originalOpen.apply(this, args);
};

const originalSend = XMLHttpRequest.prototype.send;
XMLHttpRequest.prototype.send = function(…args) {
const startTime = Date.now();

this.addEventListener('load', function() {
const duration = Date.now() – startTime;

if (this.status >= 200 && this.status < 300) {
StatusCodeDebugger.logResponse({
url: this._url,
method: this._method,
status: this.status,
statusText: this.statusText,
duration: duration,
headers: this.getAllResponseHeaders(),
timestamp: new Date().toISOString()
});
}
});

return originalSend.apply(this, args);
};
}

static logResponse(info) {
// 存储到localStorage
const key = `debug_${Date.now()}`;
localStorage.setItem(key, JSON.stringify(info));

// 触发事件
const event = new CustomEvent('statusCodeResponse', { detail: info });
document.dispatchEvent(event);
}

setupUI() {
// 创建调试面板
const panel = document.createElement('div');
panel.id = 'status-code-debug-panel';
panel.style.cssText = `
position: fixed;
bottom: 10px;
right: 10px;
width: 400px;
max-height: 500px;
background: white;
border: 1px solid #ccc;
box-shadow: 0 2px 10px rgba(0,0,0,0.1);
z-index: 9999;
font-family: monospace;
font-size: 12px;
overflow: hidden;
`;

panel.innerHTML = `
<div style="background: #f5f5f5; padding: 10px; border-bottom: 1px solid #ccc;">
<strong>2xx Status Code Debugger</strong>
<button id="debug-clear" style="float: right;">Clear</button>
</div>
<div id="debug-content" style="padding: 10px; overflow-y: auto; max-height: 450px;"></div>
`;

document.body.appendChild(panel);

// 监听响应事件
document.addEventListener('statusCodeResponse', (e) => {
this.updateDebugPanel(e.detail);
});

// 清空按钮
document.getElementById('debug-clear').addEventListener('click', () => {
document.getElementById('debug-content').innerHTML = '';
});
}

updateDebugPanel(info) {
const content = document.getElementById('debug-content');

const entry = document.createElement('div');
entry.style.cssText = `
margin-bottom: 10px;
padding: 8px;
border-left: 4px solid ${this.getStatusColor(info.status)};
background: #f9f9f9;
`;

entry.innerHTML = `
<div><strong>${info.method} ${info.url}</strong></div>
<div>${info.status} ${info.statusText} – ${info.duration}ms</div>
<div style="color: #666; font-size: 11px;">${info.timestamp}</div>
<button onclick="this.nextElementSibling.style.display='block'">Show Headers</button>
<div style="display: none; margin-top: 5px; font-size: 11px;">
${this.formatHeaders(info.headers)}
</div>
`;

content.prepend(entry);

// 限制条目数量
const entries = content.children;
if (entries.length > 20) {
entries[entries.length – 1].remove();
}
}

getStatusColor(status) {
const colors = {
200: '#28a745', // 绿色
201: '#20c997', // 青绿色
202: '#17a2b8', // 蓝色
203: '#6c757d', // 灰色
204: '#ffc107', // 黄色
205: '#fd7e14', // 橙色
206: '#007bff', // 深蓝色
207: '#6610f2', // 紫色
208: '#e83e8c', // 粉色
226: '#20c997' // 青绿色
};

return colors[status] || '#6c757d';
}

formatHeaders(headers) {
if (typeof headers === 'string') {
return headers.split('\\r\\n').map(line =>
`<div>${line}</div>`
).join('');
} else if (typeof headers === 'object') {
return Object.entries(headers).map(([key, value]) =>
`<div><strong>${key}:</strong> ${value}</div>`
).join('');
}
return '';
}
}

// 使用示例
if (process.env.NODE_ENV === 'development') {
new StatusCodeDebugger();
}

9.9 兼容性和后备方案

向后兼容策略

python

# 2xx状态码兼容性处理器
class StatusCodeCompatibility:
"""处理状态码兼容性问题"""

def __init__(self):
self.problematic_clients = [
'old-browser-1.0',
'legacy-api-client',
'custom-http-client'
]

def adapt_response(self, response, user_agent):
"""
根据客户端适配响应

Args:
response: 原始响应
user_agent: 用户代理字符串

Returns:
适配后的响应
"""

# 检查是否是问题客户端
if self.is_problematic_client(user_agent):
return self.fallback_response(response)

return response

def is_problematic_client(self, user_agent):
"""检查是否是问题客户端"""
if not user_agent:
return False

ua_lower = user_agent.lower()
return any(client in ua_lower for client in self.problematic_clients)

def fallback_response(self, original_response):
"""提供后备响应"""
status = original_response.get('status', 200)

# 将非常见状态码映射到常见状态码
mapping = {
202: 200, # 异步处理 -> 标准成功
203: 200, # 非权威信息 -> 标准成功
205: 200, # 重置内容 -> 标准成功
206: 200, # 部分内容 -> 标准成功（完整内容）
207: 200, # 多状态 -> 标准成功（第一个结果）
208: 200, # 已报告 -> 标准成功
226: 200 # IM已使用 -> 标准成功
}

if status in mapping:
new_status = mapping[status]

# 创建兼容响应
compatible_response = {
'status': new_status,
'body': self.adapt_body(original_response.get('body'), status),
'headers': self.adapt_headers(original_response.get('headers', {}), status)
}

# 添加兼容性头部
compatible_response['headers']['X-Original-Status'] = str(status)
compatible_response['headers']['X-Compatible-Response'] = 'true'

return compatible_response

return original_response

def adapt_body(self, original_body, original_status):
"""适配响应体"""
if original_status == 202:
# 异步任务信息
return {
'message': 'Request accepted and will be processed',
'original_response': original_body
}

elif original_status == 206:
# 部分内容 -> 完整内容（如果可能）
return original_body # 注意：这里应该返回完整内容

elif original_status == 207:
# 多状态 -> 第一个成功的结果
if isinstance(original_body, dict) and 'responses' in original_body:
for resp in original_body['responses']:
if resp.get('status', 200) < 300:
return resp
return original_body

else:
return original_body

def adapt_headers(self, original_headers, original_status):
"""适配响应头部"""
headers = dict(original_headers)

# 移除特定状态码的专用头部
if original_status == 206:
headers.pop('Content-Range', None)
headers.pop('Accept-Ranges', None)

elif original_status == 226:
headers.pop('IM', None)
headers.pop('Delta-Base', None)

# 更新Content-Length（如果需要）
return headers

客户端兼容性检测

javascript

// 客户端兼容性检测
class ClientCompatibilityChecker {
/**
* 检测客户端对2xx状态码的支持
*/

constructor() {
this.supportedCodes = new Set([200, 201, 204]); // 基本支持
this.testResults = {};
}

async testStatusCodeSupport() {
console.log('Testing HTTP status code support…');

const codesToTest = [202, 203, 205, 206, 207, 208, 226];

for (const code of codesToTest) {
try {
const supported = await this.testSingleCode(code);
this.testResults[code] = supported;

if (supported) {
this.supportedCodes.add(code);
}

console.log(`Status ${code}: ${supported ? 'Supported' : 'Not supported'}`);
} catch (error) {
console.error(`Test failed for ${code}:`, error);
this.testResults[code] = false;
}
}

this.saveResults();
return this.supportedCodes;
}

async testSingleCode(statusCode) {
return new Promise((resolve, reject) => {
// 创建一个测试端点URL
const testUrl = `/api/test/status/${statusCode}`;

// 使用fetch测试
if (window.fetch) {
fetch(testUrl)
.then(response => {
// 检查是否收到了正确的状态码
const supported = response.status === statusCode;
resolve(supported);
})
.catch(reject);
}
// 使用XMLHttpRequest测试
else if (window.XMLHttpRequest) {
const xhr = new XMLHttpRequest();
xhr.open('GET', testUrl);
xhr.onload = () => {
const supported = xhr.status === statusCode;
resolve(supported);
};
xhr.onerror = reject;
xhr.send();
}
// 都不支持
else {
resolve(false);
}
});
}

saveResults() {
localStorage.setItem('http_support', JSON.stringify({
supportedCodes: Array.from(this.supportedCodes),
testResults: this.testResults,
testDate: new Date().toISOString(),
userAgent: navigator.userAgent
}));
}

loadResults() {
const saved = localStorage.getItem('http_support');
if (saved) {
const data = JSON.parse(saved);
this.supportedCodes = new Set(data.supportedCodes);
this.testResults = data.testResults;
return true;
}
return false;
}

isSupported(statusCode) {
return this.supportedCodes.has(statusCode);
}

getRecommendation(statusCode) {
if (this.isSupported(statusCode)) {
return `Status ${statusCode} is supported by this client`;
} else {
return `Status ${statusCode} is not supported. Consider using alternative status ${this.getAlternative(statusCode)}`;
}
}

getAlternative(statusCode) {
const alternatives = {
202: 200, // 异步 -> 标准成功
203: 200, // 非权威 -> 标准成功
205: 200, // 重置 -> 标准成功
206: 200, // 部分 -> 完整
207: 200, // 多状态 -> 单个状态
208: 200, // 已报告 -> 标准成功
226: 200 // IM已使用 -> 标准成功
};

return alternatives[statusCode] || 200;
}
}

// 使用示例
document.addEventListener('DOMContentLoaded', async () => {
const checker = new ClientCompatibilityChecker();

// 加载之前的结果
if (!checker.loadResults()) {
// 首次运行，进行测试
await checker.testStatusCodeSupport();
}

// 检查特定状态码
console.log(checker.getRecommendation(206)); // 部分内容
console.log(checker.getRecommendation(202)); // 异步接受

// 根据支持情况调整应用行为
if (!checker.isSupported(206)) {
console.warn('Range requests not supported, using full downloads');
// 禁用范围请求功能
disableRangeRequests();
}
});

function disableRangeRequests() {
// 禁用范围请求的实现
window.RangeRequestClient = class {
downloadFile(url) {
// 总是下载完整文件
return fetch(url).then(response => response.blob());
}
};
}

9.10 最佳实践总结

服务器端最佳实践

客户端最佳实践

监控和运维最佳实践

结语

2xx状态码系列提供了丰富的语义来表达不同类型的成功响应。正确使用这些状态码可以：

虽然200 OK足以表示任何成功，但使用更具体的状态码可以让HTTP API更加优雅和强大。在选择状态码时，始终考虑：

• 操作的语义

• 客户端的期望

• 性能影响

• 兼容性要求

通过本章的详细解析，您现在应该能够：