Spaces:

Tianluhua
/

api-proxy

Runtime error

Upload 5 files

bc7ad40 verified 6 months ago

9.62 kB

from flask import Flask, request, Response, jsonify
import requests
import logging
from datetime import datetime
from config import PROXY_CONFIGS, GLOBAL_CONFIG
from functools import wraps
import json

app = Flask(__name__)

# 配置日志
if GLOBAL_CONFIG.get("log_requests", True):
    logging.basicConfig(level=logging.INFO)
    logger = logging.getLogger(__name__)

def require_auth(f):
    """访问密钥验证装饰器"""
    @wraps(f)
    def decorated_function(*args, **kwargs):
        if GLOBAL_CONFIG.get("require_auth", False):
            auth_header = request.headers.get('X-API-Key')
            if not auth_header or auth_header != GLOBAL_CONFIG.get("auth_token"):
                return jsonify({"error": "Unauthorized", "message": "Invalid or missing API key"}), 401
        return f(*args, **kwargs)
    return decorated_function

def add_cors_headers(response):
    """添加CORS头"""
    origin = request.headers.get('Origin', '*')
    allowed_origins = GLOBAL_CONFIG.get("allowed_origins", ["*"])
    if '*' in allowed_origins or origin in allowed_origins:
        response.headers['Access-Control-Allow-Origin'] = origin
        response.headers['Access-Control-Allow-Methods'] = 'GET, POST, PUT, DELETE, OPTIONS, HEAD'
        response.headers['Access-Control-Allow-Headers'] = 'Content-Type, X-API-Key'
    return response

@app.after_request
def after_request(response):
    return add_cors_headers(response)

@app.route('/health', methods=['GET', 'HEAD'])
def health_check():
    """健康检查端点 - 支持 GET 和 HEAD 方法"""
    if request.method == 'HEAD':
        # HEAD 请求：极轻量，只返回状态码和必要的响应头
        return Response('', status=200, headers={
            'Content-Type': 'application/json',
            'X-Service-Status': 'healthy',
            'X-Proxies-Count': str(len(PROXY_CONFIGS)),
            'X-Timestamp': datetime.now().isoformat()
        })
    
    # GET 请求：返回详细信息
    return jsonify({
        "status": "healthy",
        "timestamp": datetime.now().isoformat(),
        "proxies": list(PROXY_CONFIGS.keys()),
        "proxies_detail": {name: config.get("name", name) for name, config in PROXY_CONFIGS.items()}
    })

@app.route('/api/proxies', methods=['GET', 'HEAD'])
def list_proxies():
    """列出所有可用的代理"""
    if request.method == 'HEAD':
        return Response('', status=200, headers={
            'X-Proxies-Count': str(len(PROXY_CONFIGS))
        })
    
    proxies_info = []
    for name, config in PROXY_CONFIGS.items():
        proxies_info.append({
            "name": name,
            "display_name": config.get("name", name),
            "endpoints": list(config["endpoints"].keys())
        })
    return jsonify({"proxies": proxies_info})

@app.route('/api/proxy/<proxy_name>/info', methods=['GET', 'HEAD'])
def proxy_info(proxy_name):
    """获取代理信息"""
    if proxy_name not in PROXY_CONFIGS:
        return jsonify({"error": "Proxy not found"}), 404
    
    if request.method == 'HEAD':
        return Response('', status=200)
    
    proxy_config = PROXY_CONFIGS[proxy_name]
    return jsonify({
        "name": proxy_config.get("name", proxy_name),
        "base_url": proxy_config["base_url"],
        "endpoints": list(proxy_config["endpoints"].keys()),
        "endpoints_detail": proxy_config["endpoints"]
    })

@app.route('/api/proxy/<proxy_name>', defaults={'sub_path': ''}, methods=['GET', 'POST', 'PUT', 'DELETE', 'HEAD'])
@app.route('/api/proxy/<proxy_name>/<path:sub_path>', methods=['GET', 'POST', 'PUT', 'DELETE', 'HEAD'])
@require_auth
def proxy_endpoint(proxy_name, sub_path):
    """
    动态代理端点
    使用方式: /api/proxy/hotlist/api/hot/list?type=0
    """
    # 获取代理配置
    if proxy_name not in PROXY_CONFIGS:
        return jsonify({"error": "Proxy not found", "available": list(PROXY_CONFIGS.keys())}), 404
    
    proxy_config = PROXY_CONFIGS[proxy_name]
    
    # 构建完整路径
    if not sub_path:
        sub_path = '/'
    else:
        sub_path = '/' + sub_path
    
    # 查找匹配的端点配置（仅用于验证）
    matched_config = None
    for endpoint, config in proxy_config["endpoints"].items():
        if sub_path.startswith(endpoint) or sub_path == endpoint:
            matched_config = config
            break
    
    # HEAD 请求的特殊处理
    if request.method == 'HEAD':
        # 只验证路径是否存在，不实际转发请求
        if matched_config or len(proxy_config["endpoints"]) > 0:
            return Response('', status=200, headers={
                'X-Proxy-Name': proxy_name,
                'X-Path': sub_path
            })
        else:
            return Response('', status=404)
    
    # 检查方法是否允许
    if matched_config and request.method not in matched_config.get("methods", ["GET"]):
        return jsonify({
            "error": "Method not allowed",
            "allowed_methods": matched_config.get("methods")
        }), 405
    
    # 构建目标URL
    target_url = f"{proxy_config['base_url']}{sub_path}"
    
    # 合并参数（请求参数 > 默认参数）
    params = {}
    if matched_config:
        params.update(matched_config.get("params", {}))
    params.update(request.args.to_dict())
    
    # 合并请求头
    headers = {}
    headers.update(proxy_config.get("headers", {}))
    # 移除客户端可能发送的冲突头
    for key, value in request.headers.items():
        if key.lower() not in ['host', 'content-length', 'content-type']:
            headers[key] = value
    
    # 处理请求体
    data = request.get_data()
    json_data = None
    
    if request.is_json:
        json_data = request.get_json()
        data = None
    
    # 记录请求日志
    if GLOBAL_CONFIG.get("log_requests", True):
        logger.info(f"PROXY: {proxy_name} -> {request.method} {target_url}")
    
    try:
        # 发送请求
        resp = requests.request(
            method=request.method,
            url=target_url,
            headers=headers,
            params=params,
            data=data,
            json=json_data,
            timeout=GLOBAL_CONFIG.get("timeout", 30),
            verify=True
        )
        
        # 处理响应
        excluded_headers = ['content-encoding', 'content-length', 'transfer-encoding', 'connection']
        response_headers = [(name, value) for name, value in resp.raw.headers.items()
                          if name.lower() not in excluded_headers]
        
        return Response(resp.content, status=resp.status_code, headers=response_headers)
    
    except requests.Timeout:
        return jsonify({"error": "Request timeout", "message": f"Target server did not respond within {GLOBAL_CONFIG.get('timeout', 30)} seconds"}), 504
    except requests.RequestException as e:
        logger.error(f"Proxy error for {proxy_name}: {str(e)}")
        return jsonify({"error": "Proxy error", "message": str(e)}), 500

@app.route('/', defaults={'path': ''}, methods=['GET', 'POST', 'PUT', 'DELETE', 'HEAD'])
@app.route('/<path:path>', methods=['GET', 'POST', 'PUT', 'DELETE', 'HEAD'])
def legacy_proxy(path):
    """兼容模式：如果不使用/api/proxy格式，默认使用第一个代理"""
    
    # 保活专用：HEAD 请求到根路径
    if request.method == 'HEAD' and path == '':
        return Response('', status=200, headers={
            'Content-Type': 'text/plain',
            'X-Keep-Alive': 'true',
            'X-Service': 'api-proxy',
            'X-Status': 'running',
            'X-Timestamp': datetime.now().isoformat()
        })
    
    # GET 请求到根路径返回帮助信息
    if request.method == 'GET' and path == '':
        return jsonify({
            "service": "API Proxy Service",
            "version": "1.0.0",
            "endpoints": {
                "health": "/health",
                "proxies": "/api/proxies",
                "proxy_example": "/api/proxy/hotlist/api/hot/list",
                "proxy_info": "/api/proxy/hotlist/info"
            },
            "keep_alive": "Use HEAD request to / for keep-alive"
        })
    
    if not PROXY_CONFIGS:
        return jsonify({"error": "No proxy configured"}), 500
    
    # 使用第一个配置的代理
    default_proxy = list(PROXY_CONFIGS.keys())[0]
    
    # 重新构造请求到新的端点
    if path:
        new_path = f"/api/proxy/{default_proxy}/{path}"
    else:
        new_path = f"/api/proxy/{default_proxy}"
    
    # 直接调用代理函数（避免重复认证）
    from flask import current_app
    with app.test_request_context(new_path, method=request.method, headers=dict(request.headers), query_string=request.query_string.decode() if request.query_string else None):
        if request.data:
            request._get_data_for_json = lambda: request.data
        return proxy_endpoint(default_proxy, path)

@app.route('/api/proxy/<proxy_name>/health', methods=['GET', 'HEAD'])
def proxy_health(proxy_name):
    """检查特定代理的健康状态"""
    if proxy_name not in PROXY_CONFIGS:
        return jsonify({"error": "Proxy not found"}), 404
    
    if request.method == 'HEAD':
        return Response('', status=200)
    
    proxy_config = PROXY_CONFIGS[proxy_name]
    return jsonify({
        "proxy": proxy_name,
        "name": proxy_config.get("name"),
        "base_url": proxy_config["base_url"],
        "status": "configured",
        "endpoints_count": len(proxy_config["endpoints"])
    })

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=7860, debug=False)

	from flask import Flask, request, Response, jsonify
	import requests
	import logging
	from datetime import datetime
	from config import PROXY_CONFIGS, GLOBAL_CONFIG
	from functools import wraps
	import json

	app = Flask(__name__)

	# 配置日志
	if GLOBAL_CONFIG.get("log_requests", True):
	logging.basicConfig(level=logging.INFO)
	logger = logging.getLogger(__name__)

	def require_auth(f):
	"""访问密钥验证装饰器"""
	@wraps(f)
	def decorated_function(args, *kwargs):
	if GLOBAL_CONFIG.get("require_auth", False):
	auth_header = request.headers.get('X-API-Key')
	if not auth_header or auth_header != GLOBAL_CONFIG.get("auth_token"):
	return jsonify({"error": "Unauthorized", "message": "Invalid or missing API key"}), 401
	return f(args, *kwargs)
	return decorated_function

	def add_cors_headers(response):
	"""添加CORS头"""
	origin = request.headers.get('Origin', '*')
	allowed_origins = GLOBAL_CONFIG.get("allowed_origins", ["*"])
	if '*' in allowed_origins or origin in allowed_origins:
	response.headers['Access-Control-Allow-Origin'] = origin
	response.headers['Access-Control-Allow-Methods'] = 'GET, POST, PUT, DELETE, OPTIONS, HEAD'
	response.headers['Access-Control-Allow-Headers'] = 'Content-Type, X-API-Key'
	return response

	@app.after_request
	def after_request(response):
	return add_cors_headers(response)

	@app.route('/health', methods=['GET', 'HEAD'])
	def health_check():
	"""健康检查端点 - 支持 GET 和 HEAD 方法"""
	if request.method == 'HEAD':
	# HEAD 请求：极轻量，只返回状态码和必要的响应头
	return Response('', status=200, headers={
	'Content-Type': 'application/json',
	'X-Service-Status': 'healthy',
	'X-Proxies-Count': str(len(PROXY_CONFIGS)),
	'X-Timestamp': datetime.now().isoformat()
	})

	# GET 请求：返回详细信息
	return jsonify({
	"status": "healthy",
	"timestamp": datetime.now().isoformat(),
	"proxies": list(PROXY_CONFIGS.keys()),
	"proxies_detail": {name: config.get("name", name) for name, config in PROXY_CONFIGS.items()}
	})

	@app.route('/api/proxies', methods=['GET', 'HEAD'])
	def list_proxies():
	"""列出所有可用的代理"""
	if request.method == 'HEAD':
	return Response('', status=200, headers={
	'X-Proxies-Count': str(len(PROXY_CONFIGS))
	})

	proxies_info = []
	for name, config in PROXY_CONFIGS.items():
	proxies_info.append({
	"name": name,
	"display_name": config.get("name", name),
	"endpoints": list(config["endpoints"].keys())
	})
	return jsonify({"proxies": proxies_info})

	@app.route('/api/proxy/<proxy_name>/info', methods=['GET', 'HEAD'])
	def proxy_info(proxy_name):
	"""获取代理信息"""
	if proxy_name not in PROXY_CONFIGS:
	return jsonify({"error": "Proxy not found"}), 404

	if request.method == 'HEAD':
	return Response('', status=200)

	proxy_config = PROXY_CONFIGS[proxy_name]
	return jsonify({
	"name": proxy_config.get("name", proxy_name),
	"base_url": proxy_config["base_url"],
	"endpoints": list(proxy_config["endpoints"].keys()),
	"endpoints_detail": proxy_config["endpoints"]
	})

	@app.route('/api/proxy/<proxy_name>', defaults={'sub_path': ''}, methods=['GET', 'POST', 'PUT', 'DELETE', 'HEAD'])
	@app.route('/api/proxy/<proxy_name>/<path:sub_path>', methods=['GET', 'POST', 'PUT', 'DELETE', 'HEAD'])
	@require_auth
	def proxy_endpoint(proxy_name, sub_path):
	"""
	动态代理端点
	使用方式: /api/proxy/hotlist/api/hot/list?type=0
	"""
	# 获取代理配置
	if proxy_name not in PROXY_CONFIGS:
	return jsonify({"error": "Proxy not found", "available": list(PROXY_CONFIGS.keys())}), 404

	proxy_config = PROXY_CONFIGS[proxy_name]

	# 构建完整路径
	if not sub_path:
	sub_path = '/'
	else:
	sub_path = '/' + sub_path

	# 查找匹配的端点配置（仅用于验证）
	matched_config = None
	for endpoint, config in proxy_config["endpoints"].items():
	if sub_path.startswith(endpoint) or sub_path == endpoint:
	matched_config = config
	break

	# HEAD 请求的特殊处理
	if request.method == 'HEAD':
	# 只验证路径是否存在，不实际转发请求
	if matched_config or len(proxy_config["endpoints"]) > 0:
	return Response('', status=200, headers={
	'X-Proxy-Name': proxy_name,
	'X-Path': sub_path
	})
	else:
	return Response('', status=404)

	# 检查方法是否允许
	if matched_config and request.method not in matched_config.get("methods", ["GET"]):
	return jsonify({
	"error": "Method not allowed",
	"allowed_methods": matched_config.get("methods")
	}), 405

	# 构建目标URL
	target_url = f"{proxy_config['base_url']}{sub_path}"

	# 合并参数（请求参数 > 默认参数）
	params = {}
	if matched_config:
	params.update(matched_config.get("params", {}))
	params.update(request.args.to_dict())

	# 合并请求头
	headers = {}
	headers.update(proxy_config.get("headers", {}))
	# 移除客户端可能发送的冲突头
	for key, value in request.headers.items():
	if key.lower() not in ['host', 'content-length', 'content-type']:
	headers[key] = value

	# 处理请求体
	data = request.get_data()
	json_data = None

	if request.is_json:
	json_data = request.get_json()
	data = None

	# 记录请求日志
	if GLOBAL_CONFIG.get("log_requests", True):
	logger.info(f"PROXY: {proxy_name} -> {request.method} {target_url}")

	try:
	# 发送请求
	resp = requests.request(
	method=request.method,
	url=target_url,
	headers=headers,
	params=params,
	data=data,
	json=json_data,
	timeout=GLOBAL_CONFIG.get("timeout", 30),
	verify=True
	)

	# 处理响应
	excluded_headers = ['content-encoding', 'content-length', 'transfer-encoding', 'connection']
	response_headers = [(name, value) for name, value in resp.raw.headers.items()
	if name.lower() not in excluded_headers]

	return Response(resp.content, status=resp.status_code, headers=response_headers)

	except requests.Timeout:
	return jsonify({"error": "Request timeout", "message": f"Target server did not respond within {GLOBAL_CONFIG.get('timeout', 30)} seconds"}), 504
	except requests.RequestException as e:
	logger.error(f"Proxy error for {proxy_name}: {str(e)}")
	return jsonify({"error": "Proxy error", "message": str(e)}), 500

	@app.route('/', defaults={'path': ''}, methods=['GET', 'POST', 'PUT', 'DELETE', 'HEAD'])
	@app.route('/<path:path>', methods=['GET', 'POST', 'PUT', 'DELETE', 'HEAD'])
	def legacy_proxy(path):
	"""兼容模式：如果不使用/api/proxy格式，默认使用第一个代理"""

	# 保活专用：HEAD 请求到根路径
	if request.method == 'HEAD' and path == '':
	return Response('', status=200, headers={
	'Content-Type': 'text/plain',
	'X-Keep-Alive': 'true',
	'X-Service': 'api-proxy',
	'X-Status': 'running',
	'X-Timestamp': datetime.now().isoformat()
	})

	# GET 请求到根路径返回帮助信息
	if request.method == 'GET' and path == '':
	return jsonify({
	"service": "API Proxy Service",
	"version": "1.0.0",
	"endpoints": {
	"health": "/health",
	"proxies": "/api/proxies",
	"proxy_example": "/api/proxy/hotlist/api/hot/list",
	"proxy_info": "/api/proxy/hotlist/info"
	},
	"keep_alive": "Use HEAD request to / for keep-alive"
	})

	if not PROXY_CONFIGS:
	return jsonify({"error": "No proxy configured"}), 500

	# 使用第一个配置的代理
	default_proxy = list(PROXY_CONFIGS.keys())[0]

	# 重新构造请求到新的端点
	if path:
	new_path = f"/api/proxy/{default_proxy}/{path}"
	else:
	new_path = f"/api/proxy/{default_proxy}"

	# 直接调用代理函数（避免重复认证）
	from flask import current_app
	with app.test_request_context(new_path, method=request.method, headers=dict(request.headers), query_string=request.query_string.decode() if request.query_string else None):
	if request.data:
	request._get_data_for_json = lambda: request.data
	return proxy_endpoint(default_proxy, path)

	@app.route('/api/proxy/<proxy_name>/health', methods=['GET', 'HEAD'])
	def proxy_health(proxy_name):
	"""检查特定代理的健康状态"""
	if proxy_name not in PROXY_CONFIGS:
	return jsonify({"error": "Proxy not found"}), 404

	if request.method == 'HEAD':
	return Response('', status=200)

	proxy_config = PROXY_CONFIGS[proxy_name]
	return jsonify({
	"proxy": proxy_name,
	"name": proxy_config.get("name"),
	"base_url": proxy_config["base_url"],
	"status": "configured",
	"endpoints_count": len(proxy_config["endpoints"])
	})

	if __name__ == '__main__':
	app.run(host='0.0.0.0', port=7860, debug=False)