Requests 自定义请求头设置

import requests

def debug_headers(url, headers=None):
    """调试请求头的工具函数"""

    # 准备请求头
    if headers is None:
        headers = {}

    # 添加调试头
    headers['X-Debug-Mode'] = 'true'
    headers['X-Request-Timestamp'] = '2023-01-01T12:00:00Z'

    # 发送请求
    response = requests.get(url, headers=headers)

    print(f"请求URL: {response.url}")
    print(f"状态码: {response.status_code}")
    print(f"响应头数量: {len(response.headers)}")

    if response.status_code == 200:
        # 获取服务器接收到的请求头
        data = response.json()

        print("\n=== 发送的请求头 ===")
        for key, value in headers.items():
            print(f"{key}: {value}")

        print("\n=== 服务器接收到的请求头 ===")
        server_headers = data.get('headers', {})
        for key in sorted(server_headers.keys()):
            if key.lower().startswith('x-') or key.lower() in ['user-agent', 'accept', 'authorization']:
                value = server_headers[key]
                if key.lower() == 'authorization' and len(value) > 30:
                    value = value[:30] + '...'
                print(f"{key}: {value}")

    return response

# 使用示例
url = "https://httpbin.org/headers"
test_headers = {
    'User-Agent': 'DebugClient/1.0',
    'Accept': 'application/json',
    'X-Test-Case': 'HeaderDebugging'
}

debug_headers(url, test_headers)

import requests
import base64

url = "https://httpbin.org/basic-auth/user/passwd"

# 方法1：使用requests.auth
from requests.auth import HTTPBasicAuth

response = requests.get(
    url,
    auth=HTTPBasicAuth('user', 'passwd')
)
print(f"Basic Auth 状态码: {response.status_code}")

# 方法2：手动设置Authorization头
username = "user"
password = "passwd"

# 编码用户名和密码
credentials = f"{username}:{password}"
encoded_credentials = base64.b64encode(
    credentials.encode('utf-8')
).decode('utf-8')

headers = {
    'Authorization': f'Basic {encoded_credentials}'
}

response = requests.get(url, headers=headers)
print(f"手动设置Basic头状态码: {response.status_code}")

# 方法3：使用元组（Requests自动处理）
response = requests.get(url, auth=('user', 'passwd'))
print(f"使用元组认证状态码: {response.status_code}")

import requests
import jwt  # 需要安装: pip install PyJWT
import datetime

def create_jwt_token(secret_key, payload, expires_in_hours=1):
    """创建JWT令牌"""
    payload['exp'] = datetime.datetime.utcnow() + datetime.timedelta(hours=expires_in_hours)
    token = jwt.encode(payload, secret_key, algorithm='HS256')
    return token

# 使用Bearer Token认证
url = "https://httpbin.org/bearer"

# 方法1：直接设置Authorization头
token = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIiwiaWF0IjoxNTE2MjM5MDIyfQ.SflKxwRJSMeKKF2QT4fwpMeJf36POk6yJV_adQssw5c"
headers = {
    'Authorization': f'Bearer {token}'
}

response = requests.get(url, headers=headers)
print(f"Bearer Token认证状态码: {response.status_code}")

# 方法2：使用自定义认证类
class BearerAuth(requests.auth.AuthBase):
    """Bearer Token认证类"""
    def __init__(self, token):
        self.token = token

    def __call__(self, r):
        r.headers['Authorization'] = f'Bearer {self.token}'
        return r

# 使用自定义认证类
response = requests.get(url, auth=BearerAuth(token))
print(f"使用BearerAuth类状态码: {response.status_code}")

# 动态令牌示例
class DynamicTokenAuth(requests.auth.AuthBase):
    """动态令牌认证（如刷新令牌）"""
    def __init__(self, token_getter):
        self.token_getter = token_getter

    def __call__(self, r):
        token = self.token_getter()
        r.headers['Authorization'] = f'Bearer {token}'
        return r

# 模拟令牌获取函数
def get_token():
    # 这里可以从缓存、数据库或API获取令牌
    return "dynamic_token_123"

response = requests.get(url, auth=DynamicTokenAuth(get_token))
print(f"动态令牌认证状态码: {response.status_code}")

import requests
import os
from dotenv import load_dotenv  # 需要安装: pip install python-dotenv

# 从环境变量加载API密钥
load_dotenv()

url = "https://httpbin.org/headers"

# 方法1：使用自定义请求头
api_key = os.getenv('API_KEY', 'default_api_key_123')

headers = {
    'X-API-Key': api_key,
    'X-API-Secret': 'your_secret_here'  # 不建议，仅示例
}

response = requests.get(url, headers=headers)
print(f"API Key认证状态码: {response.status_code}")

# 方法2：API Key放在URL参数中（不推荐用于敏感数据）
params = {
    'api_key': api_key,
    'action': 'get_data'
}

response = requests.get('https://httpbin.org/get', params=params)
print(f"URL参数API Key状态码: {response.status_code}")

# 方法3：复合认证（API Key + 签名）
import hashlib
import hmac
import time

def generate_api_signature(api_key, api_secret, timestamp, path):
    """生成API签名"""
    message = f"{api_key}{timestamp}{path}"
    signature = hmac.new(
        api_secret.encode('utf-8'),
        message.encode('utf-8'),
        hashlib.sha256
    ).hexdigest()
    return signature

# 使用签名认证
api_key = "your_api_key"
api_secret = "your_api_secret"
timestamp = str(int(time.time()))
path = "/api/v1/data"

signature = generate_api_signature(api_key, api_secret, timestamp, path)

headers = {
    'X-API-Key': api_key,
    'X-API-Timestamp': timestamp,
    'X-API-Signature': signature
}

print(f"签名认证头示例:")
for key, value in headers.items():
    print(f"  {key}: {value}")

import requests
import platform

url = "https://httpbin.org/headers"

# 1. 简单的User-Agent
headers = {
    'User-Agent': 'MyPythonApp/1.0.0'
}
response = requests.get(url, headers=headers)

# 2. 详细的User-Agent（模拟浏览器）
browser_ua = (
    'Mozilla/5.0 (Windows NT 10.0; Win64; x64) '
    'AppleWebKit/537.36 (KHTML, like Gecko) '
    'Chrome/91.0.4472.124 Safari/537.36'
)

headers['User-Agent'] = browser_ua
response = requests.get(url, headers=headers)

# 3. 动态生成User-Agent
def generate_user_agent(app_name, app_version, os_info=None):
    """生成User-Agent字符串"""
    if os_info is None:
        os_info = platform.platform()

    python_version = platform.python_version()
    requests_version = requests.__version__

    user_agent = (
        f"{app_name}/{app_version} "
        f"({os_info}) "
        f"Python/{python_version} "
        f"Requests/{requests_version}"
    )
    return user_agent

# 生成自定义User-Agent
custom_ua = generate_user_agent(
    app_name="DataCollector",
    app_version="2.1.0",
    os_info="Linux-5.4.0-80-generic-x86_64"
)

headers['User-Agent'] = custom_ua
response = requests.get(url, headers=headers)

print(f"自定义User-Agent: {custom_ua}")
print(f"状态码: {response.status_code}")

# 4. User-Agent轮换（用于爬虫）
user_agents = [
    'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
    'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15',
    'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36',
    'MyApp/1.0 (Custom Client)'
]

import random
selected_ua = random.choice(user_agents)
headers['User-Agent'] = selected_ua

response = requests.get(url, headers=headers)
print(f"随机User-Agent: {selected_ua}")
print(f"请求成功: {response.status_code == 200}")

import requests
import uuid

url = "https://httpbin.org/headers"

# Referer头（来源页面）
headers = {
    'Referer': 'https://example.com/dashboard',
    'Referrer-Policy': 'strict-origin-when-cross-origin'
}

# From头（用户邮箱 - 谨慎使用，涉及隐私）
# headers['From'] = 'user@example.com'  # 通常不设置

# DNT（Do Not Track）头
headers['DNT'] = '1'  # 1表示不希望被追踪

# 客户端信息头
headers['X-Client-ID'] = str(uuid.uuid4())
headers['X-Client-Version'] = '2.3.1'
headers['X-Client-Platform'] = 'Web'
headers['X-Client-Language'] = 'zh-CN'

# 设备信息
headers['X-Device-ID'] = 'device_123456'
headers['X-Device-Model'] = 'Custom Device'
headers['X-Device-OS'] = 'Linux 5.4.0'
headers['X-Device-OS-Version'] = '5.4.0-80-generic'

response = requests.get(url, headers=headers)

if response.status_code == 200:
    data = response.json()
    print("服务器接收到的客户端标识头:")

    client_headers = {}
    for key, value in data['headers'].items():
        key_lower = key.lower()
        if any(x in key_lower for x in ['user-agent', 'referer', 'client', 'device', 'dnt']):
            client_headers[key] = value

    for key, value in sorted(client_headers.items()):
        print(f"  {key}: {value[:50]}{'...' if len(value) > 50 else ''}")

import requests

url = "https://httpbin.org/headers"

# 1. 基本的Accept头设置
headers = {
    'Accept': 'application/json'
}
response = requests.get(url, headers=headers)
print(f"只接受JSON: {response.headers.get('content-type')}")

# 2. 多内容类型协商（带权重）
headers['Accept'] = (
    'application/json;q=1.0, '
    'text/html;q=0.9, '
    'application/xml;q=0.8, '
    '*/*;q=0.1'
)
response = requests.get(url, headers=headers)
print(f"多类型协商: {response.headers.get('content-type')}")

# 3. 针对不同API端点的Accept策略
class ContentNegotiator:
    """内容协商器"""

    ACCEPT_MAP = {
        'json_api': 'application/json',
        'xml_api': 'application/xml',
        'html_page': 'text/html',
        'any': '*/*'
    }

    @staticmethod
    def get_accept_header(api_type='json_api', include_fallback=True):
        """获取Accept头"""
        accept = ContentNegotiator.ACCEPT_MAP.get(api_type, 'application/json')

        if include_fallback:
            if api_type == 'json_api':
                accept += ', text/plain;q=0.5'
            elif api_type == 'xml_api':
                accept += ', text/plain;q=0.5'

        return accept

# 使用示例
for api_type in ['json_api', 'xml_api', 'html_page']:
    accept_header = ContentNegotiator.get_accept_header(api_type)
    headers = {'Accept': accept_header}

    response = requests.get(url, headers=headers)
    print(f"{api_type} - Accept: {accept_header[:30]}...")
    print(f"  响应类型: {response.headers.get('content-type')}")

import requests

url = "https://httpbin.org/post"

# 1. JSON数据（自动设置Content-Type）
data = {"name": "张三", "age": 25}
response = requests.post(url, json=data)
print(f"使用json参数 - Content-Type: {response.json()['headers'].get('Content-Type')}")

# 2. 表单数据（自动设置Content-Type）
form_data = {"username": "test", "password": "secret"}
response = requests.post(url, data=form_data)
print(f"使用data参数 - Content-Type: {response.json()['headers'].get('Content-Type')}")

# 3. 手动设置Content-Type
headers = {'Content-Type': 'application/json'}
response = requests.post(url, data='{"manual": "json"}', headers=headers)
print(f"手动设置JSON - Content-Type: {response.json()['headers'].get('Content-Type')}")

# 4. 其他Content-Type
content_types = {
    'xml': 'application/xml',
    'form_urlencoded': 'application/x-www-form-urlencoded',
    'multipart': 'multipart/form-data',
    'text': 'text/plain',
    'html': 'text/html',
    'binary': 'application/octet-stream'
}

for name, content_type in content_types.items():
    headers = {'Content-Type': content_type}
    response = requests.post(url, data=f'test {name}', headers=headers)
    ct = response.json()['headers'].get('Content-Type', '')
    print(f"{name}: {ct[:40]}...")

import requests

url = "https://httpbin.org/headers"

# 完整的语言和编码协商头
headers = {
    'Accept-Language': 'zh-CN,zh;q=0.9,en-US;q=0.8,en;q=0.7,ja;q=0.6',
    'Accept-Encoding': 'gzip, deflate, br, zstd',
    'Accept-Charset': 'utf-8, iso-8859-1;q=0.5',

    # 国际化相关
    'X-Timezone': 'Asia/Shanghai',
    'X-Currency': 'CNY',
    'X-Locale': 'zh_CN'
}

# 智能语言协商
def get_accept_language(user_languages=None):
    """根据用户偏好生成Accept-Language头"""
    if user_languages is None:
        user_languages = ['zh-CN', 'en-US']

    # 添加权重
    language_with_q = []
    for i, lang in enumerate(user_languages):
        q = 1.0 - (i * 0.1)  # 递减权重
        language_with_q.append(f"{lang};q={q:.1f}")

    # 添加通配符作为备选
    language_with_q.append('*;q=0.1')

    return ', '.join(language_with_q)

# 使用智能语言协商
headers['Accept-Language'] = get_accept_language(['zh-CN', 'en-US', 'ja-JP'])
headers['Accept-Encoding'] = 'gzip, deflate'  # 精简版，兼容性更好

response = requests.get(url, headers=headers)

if response.status_code == 200:
    data = response.json()
    print("内容协商头:")
    for header in ['Accept-Language', 'Accept-Encoding', 'Accept-Charset']:
        value = data['headers'].get(header, '未设置')
        print(f"  {header}: {value}")

import requests

url = "https://httpbin.org/cache"

# 1. 禁止缓存
headers = {
    'Cache-Control': 'no-cache, no-store, must-revalidate',
    'Pragma': 'no-cache',
    'Expires': '0'
}

response = requests.get(f"{url}/0", headers=headers)
print(f"禁止缓存 - 状态码: {response.status_code}")

# 2. 允许缓存但有条件
headers['Cache-Control'] = 'max-age=300'  # 缓存5分钟
response = requests.get(f"{url}/300", headers=headers)
print(f"缓存5分钟 - 状态码: {response.status_code}")

# 3. 条件请求（If-Modified-Since）
import datetime

# 假设我们之前获取过资源，知道最后修改时间
last_modified = datetime.datetime(2023, 1, 1, 12, 0, 0)
headers['If-Modified-Since'] = last_modified.strftime('%a, %d %b %Y %H:%M:%S GMT')

response = requests.get('https://httpbin.org/etag/etag-123', headers=headers)
print(f"条件请求 - 状态码: {response.status_code} (304表示未修改)")

# 4. ETag条件请求
headers['If-None-Match'] = '"etag-123"'
response = requests.get('https://httpbin.org/etag/etag-123', headers=headers)
print(f"ETag条件请求 - 状态码: {response.status_code}")

# 5. 缓存策略生成器
class CachePolicy:
    """缓存策略生成器"""

    @staticmethod
    def no_cache():
        """不缓存"""
        return {
            'Cache-Control': 'no-cache, no-store, must-revalidate',
            'Pragma': 'no-cache'
        }

    @staticmethod
    def cache_public(max_age=3600):
        """公共缓存（CDN、代理）"""
        return {
            'Cache-Control': f'public, max-age={max_age}, s-maxage={max_age}'
        }

    @staticmethod
    def cache_private(max_age=3600):
        """私有缓存（仅客户端）"""
        return {
            'Cache-Control': f'private, max-age={max_age}'
        }

    @staticmethod
    def stale_while_revalidate(max_age=3600, stale_while_revalidate=86400):
        """在重新验证期间使用过期缓存"""
        return {
            'Cache-Control': f'max-age={max_age}, stale-while-revalidate={stale_while_revalidate}'
        }

# 使用缓存策略
for policy_name in ['no_cache', 'cache_public', 'cache_private']:
    policy_method = getattr(CachePolicy, policy_name)
    headers = policy_method()

    print(f"\n{policy_name}策略:")
    for key, value in headers.items():
        print(f"  {key}: {value}")

import requests
import uuid
import time

url = "https://httpbin.org/headers"

# 完整的自定义请求头示例
headers = {
    # 请求追踪
    'X-Request-ID': str(uuid.uuid4()),
    'X-Correlation-ID': str(uuid.uuid4()),
    'X-Trace-ID': 'trace-123456789',

    # API版本控制
    'X-API-Version': 'v2.0',
    'X-API-Deprecation': 'Sun, 01 Jan 2024 00:00:00 GMT',

    # 调试信息
    'X-Debug-Mode': 'true',
    'X-Debug-Level': 'verbose',
    'X-Request-Timestamp': str(int(time.time())),

    # 应用特定信息
    'X-Application-ID': 'myapp-123',
    'X-Application-Version': '2.3.1',
    'X-Feature-Flags': 'new-ui:true,dark-mode:false',

    # 业务上下文
    'X-User-ID': 'user_12345',
    'X-Organization-ID': 'org_67890',
    'X-Tenant-ID': 'tenant_abc',

    # 性能监控
    'X-Response-Time-Threshold': '1000',  # 毫秒
    'X-Monitoring-Sample-Rate': '0.1',

    # 安全相关
    'X-Content-Type-Options': 'nosniff',
    'X-Frame-Options': 'DENY',
    'X-XSS-Protection': '1; mode=block'
}

response = requests.get(url, headers=headers)

if response.status_code == 200:
    data = response.json()
    print("服务器接收到的自定义请求头:")

    custom_headers = {}
    for key, value in data['headers'].items():
        if key.startswith('X-'):
            custom_headers[key] = value

    for key, value in sorted(custom_headers.items()):
        print(f"  {key}: {value}")

import requests
import uuid
import time
from typing import Dict, Any, Optional

class CustomHeaderBuilder:
    """自定义请求头生成器"""

    def __init__(self):
        self.headers = {}
        self.request_id = str(uuid.uuid4())

    def add_tracing(self, correlation_id: Optional[str] = None):
        """添加请求追踪头"""
        self.headers['X-Request-ID'] = self.request_id
        self.headers['X-Correlation-ID'] = correlation_id or str(uuid.uuid4())
        self.headers['X-Trace-ID'] = f"trace-{int(time.time())}"
        return self

    def add_api_info(self, version: str = 'v1.0', deprecated: Optional[str] = None):
        """添加API信息头"""
        self.headers['X-API-Version'] = version
        if deprecated:
            self.headers['X-API-Deprecation'] = deprecated
        return self

    def add_debug_info(self, debug_mode: bool = False, level: str = 'info'):
        """添加调试信息头"""
        if debug_mode:
            self.headers['X-Debug-Mode'] = 'true'
            self.headers['X-Debug-Level'] = level
            self.headers['X-Request-Timestamp'] = str(int(time.time()))
        return self

    def add_business_context(self, user_id: Optional[str] = None,
                            org_id: Optional[str] = None):
        """添加上下文信息"""
        if user_id:
            self.headers['X-User-ID'] = user_id
        if org_id:
            self.headers['X-Organization-ID'] = org_id
        return self

    def add_monitoring(self, sample_rate: float = 0.1):
        """添加监控信息"""
        self.headers['X-Monitoring-Sample-Rate'] = str(sample_rate)
        return self

    def add_security(self):
        """添加安全头"""
        self.headers['X-Content-Type-Options'] = 'nosniff'
        self.headers['X-Frame-Options'] = 'DENY'
        return self

    def build(self) -> Dict[str, str]:
        """构建请求头字典"""
        return self.headers.copy()

    @classmethod
    def create_default(cls) -> 'CustomHeaderBuilder':
        """创建默认配置的生成器"""
        return (cls()
                .add_tracing()
                .add_api_info('v1.0')
                .add_debug_info(debug_mode=True)
                .add_security())

# 使用示例
builder = CustomHeaderBuilder()
headers = (builder
          .add_tracing()
          .add_api_info('v2.1', 'Sun, 01 Jan 2024 00:00:00 GMT')
          .add_debug_info(debug_mode=True, level='verbose')
          .add_business_context(user_id='user_123', org_id='org_456')
          .build())

print("生成的请求头:")
for key, value in headers.items():
    print(f"  {key}: {value}")

# 发送请求
url = "https://httpbin.org/headers"
response = requests.get(url, headers=headers)

if response.status_code == 200:
    print(f"\n请求成功，Request ID: {headers['X-Request-ID']}")

import requests

# 创建Session对象
session = requests.Session()

# 设置Session级别的请求头
session.headers.update({
    'User-Agent': 'MyApp/2.0.0',
    'Accept': 'application/json',
    'Accept-Language': 'zh-CN',
    'X-Client-Version': '2.0.0',
    'X-Session-ID': str(uuid.uuid4())
})

print("Session默认请求头:")
for key, value in session.headers.items():
    print(f"  {key}: {value}")

# 所有通过这个Session发送的请求都会自动包含这些请求头
url = "https://httpbin.org/headers"
response1 = session.get(url)
print(f"\n第一次请求状态码: {response1.status_code}")

# 可以临时覆盖Session级别的请求头
custom_headers = {'User-Agent': 'CustomAgent/1.0', 'X-Custom': 'value'}
response2 = session.get(url, headers=custom_headers)
print(f"第二次请求（临时覆盖）状态码: {response2.status_code}")

# 临时添加请求头（不覆盖Session级别）
from requests.structures import CaseInsensitiveDict

def add_headers(session, additional_headers):
    """向Session添加额外的请求头（不覆盖现有值）"""
    merged_headers = CaseInsensitiveDict(session.headers)
    merged_headers.update(additional_headers)
    return merged_headers

# 使用
additional = {'X-Additional': 'extra_value'}
all_headers = add_headers(session, additional)
response3 = session.get(url, headers=all_headers)
print(f"第三次请求（添加额外头）状态码: {response3.status_code}")

# 关闭Session
session.close()

import requests
import uuid
from typing import Dict, Optional

class ManagedSession:
    """带请求头管理的Session类"""

    def __init__(self, base_headers: Optional[Dict] = None):
        self.session = requests.Session()
        self.base_headers = base_headers or {}
        self.request_counter = 0

        # 初始化基本请求头
        self._init_base_headers()

    def _init_base_headers(self):
        """初始化基本请求头"""
        default_headers = {
            'User-Agent': f'ManagedSession/1.0',
            'Accept': 'application/json',
            'Accept-Encoding': 'gzip, deflate',
            'Connection': 'keep-alive',
            'X-Session-ID': str(uuid.uuid4())
        }

        # 合并用户提供的请求头
        default_headers.update(self.base_headers)
        self.session.headers.update(default_headers)

    def request(self, method: str, url: str,
                headers: Optional[Dict] = None, **kwargs):
        """发送请求，自动管理请求头"""
        self.request_counter += 1

        # 准备请求头
        request_headers = {}

        # 1. 添加请求计数头
        request_headers['X-Request-Count'] = str(self.request_counter)
        request_headers['X-Request-Sequence'] = str(uuid.uuid4())[:8]

        # 2. 合并传入的请求头
        if headers:
            request_headers.update(headers)

        # 3. 发送请求
        response = self.session.request(
            method=method.upper(),
            url=url,
            headers=request_headers,
            **kwargs
        )

        # 4. 记录请求头信息
        self._log_request_headers(request_headers, response)

        return response

    def _log_request_headers(self, sent_headers: Dict, response):
        """记录请求头信息"""
        if response.status_code == 200:
            try:
                data = response.json()
                received_headers = data.get('headers', {})

                # 找出我们发送的自定义头
                custom_sent = {k: v for k, v in sent_headers.items()
                              if k.startswith('X-')}

                print(f"\n请求 #{self.request_counter} 头信息:")
                for key, value in custom_sent.items():
                    received = received_headers.get(key, '未收到')
                    print(f"  {key}: 发送={value}, 接收={received}")

            except ValueError:
                pass

    def add_permanent_header(self, key: str, value: str):
        """添加永久请求头（Session级别）"""
        self.session.headers[key] = value

    def remove_header(self, key: str):
        """移除请求头"""
        if key in self.session.headers:
            del self.session.headers[key]

    def get_current_headers(self) -> Dict:
        """获取当前Session的请求头"""
        return dict(self.session.headers)

    def close(self):
        """关闭Session"""
        self.session.close()

# 使用示例
print("=== ManagedSession 使用示例 ===")

# 创建带基本请求头的Session
base_headers = {
    'X-Application': 'TestApp',
    'X-Environment': 'development'
}

managed_session = ManagedSession(base_headers)

# 发送请求
url = "https://httpbin.org/headers"

# 请求1：使用默认请求头
response1 = managed_session.request('GET', url)
print(f"请求1状态码: {response1.status_code}")

# 请求2：添加临时请求头
temp_headers = {'X-Test-Case': 'HeaderTest', 'X-Timestamp': '1234567890'}
response2 = managed_session.request('GET', url, headers=temp_headers)
print(f"请求2状态码: {response2.status_code}")

# 添加永久请求头
managed_session.add_permanent_header('X-Permanent', 'always_send')
print(f"\n当前永久请求头: {managed_session.get_current_headers()}")

# 请求3：验证永久请求头
response3 = managed_session.request('GET', url)
print(f"请求3状态码: {response3.status_code}")

# 关闭Session
managed_session.close()

import requests
import os
import uuid
import time
from typing import Dict, Any

class ProductionRequestHeaders:
    """生产环境请求头配置"""

    @staticmethod
    def create_headers(api_endpoint: str,
                      include_auth: bool = True,
                      include_tracing: bool = True) -> Dict[str, str]:
        """创建生产环境请求头"""

        headers = {}

        # 1. 基础头
        headers.update(ProductionRequestHeaders._get_basic_headers())

        # 2. 认证头
        if include_auth:
            auth_headers = ProductionRequestHeaders._get_auth_headers(api_endpoint)
            if auth_headers:
                headers.update(auth_headers)

        # 3. 追踪头
        if include_tracing:
            headers.update(ProductionRequestHeaders._get_tracing_headers())

        # 4. 环境特定头
        env = os.getenv('ENVIRONMENT', 'development')
        headers.update(ProductionRequestHeaders._get_environment_headers(env))

        return headers

    @staticmethod
    def _get_basic_headers() -> Dict[str, str]:
        """获取基础请求头"""
        return {
            'User-Agent': f'ProductionApp/{os.getenv("APP_VERSION", "1.0.0")}',
            'Accept': 'application/json',
            'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
            'Accept-Encoding': 'gzip, deflate',
            'Cache-Control': 'no-cache',
            'Connection': 'keep-alive'
        }

    @staticmethod
    def _get_auth_headers(api_endpoint: str) -> Dict[str, str]:
        """获取认证请求头"""
        headers = {}

        # 根据端点类型选择认证方式
        if 'auth' in api_endpoint:
            # OAuth 2.0 Bearer Token
            token = os.getenv('ACCESS_TOKEN')
            if token:
                headers['Authorization'] = f'Bearer {token}'
        else:
            # API Key认证
            api_key = os.getenv('API_KEY')
            if api_key:
                headers['X-API-Key'] = api_key

        return headers

    @staticmethod
    def _get_tracing_headers() -> Dict[str, str]:
        """获取请求追踪头"""
        return {
            'X-Request-ID': str(uuid.uuid4()),
            'X-Correlation-ID': str(uuid.uuid4()),
            'X-Trace-ID': f"trace-{int(time.time())}",
            'X-Request-Timestamp': str(int(time.time() * 1000))
        }

    @staticmethod
    def _get_environment_headers(env: str) -> Dict[str, str]:
        """获取环境特定请求头"""
        headers = {
            'X-Environment': env,
            'X-Deployment-ID': os.getenv('DEPLOYMENT_ID', 'unknown')
        }

        # 开发环境添加调试头
        if env == 'development':
            headers['X-Debug-Mode'] = 'true'
            headers['X-Debug-Level'] = 'info'

        return headers

# 使用示例
def make_production_request(url: str, method: str = 'GET', **kwargs):
    """生产环境请求函数"""

    # 创建生产环境请求头
    headers = ProductionRequestHeaders.create_headers(
        api_endpoint=url,
        include_auth=True,
        include_tracing=True
    )

    # 合并用户提供的请求头
    if 'headers' in kwargs:
        headers.update(kwargs['headers'])

    # 更新请求头参数
    kwargs['headers'] = headers

    # 设置超时
    if 'timeout' not in kwargs:
        kwargs['timeout'] = 30.0

    # 发送请求
    try:
        response = requests.request(method, url, **kwargs)

        # 记录请求ID用于追踪
        request_id = headers.get('X-Request-ID', 'unknown')
        print(f"[{request_id}] 请求 {method} {url} - 状态码: {response.status_code}")

        return response

    except Exception as e:
        request_id = headers.get('X-Request-ID', 'unknown')
        print(f"[{request_id}] 请求失败: {e}")
        raise

# 模拟环境变量
os.environ['ENVIRONMENT'] = 'production'
os.environ['APP_VERSION'] = '2.3.1'
os.environ['API_KEY'] = 'prod_api_key_123'
os.environ['DEPLOYMENT_ID'] = 'deploy-2023-01-01'

# 发送生产环境请求
url = "https://httpbin.org/headers"
response = make_production_request(url)

if response.status_code == 200:
    print("\n生产环境请求头配置测试成功!")

    # 显示发送的请求头
    data = response.json()
    sent_headers = {k: v for k, v in data['headers'].items()
                   if k.startswith('X-') or k in ['User-Agent', 'Accept', 'Authorization']}

    print("\n发送的请求头:")
    for key, value in sorted(sent_headers.items()):
        print(f"  {key}: {value[:50]}{'...' if len(value) > 50 else ''}")