开发环境准备与依赖安装
在开始接入全国流量API之前，需要确保本地开发环境已经配置好Python 3.8及以上版本。本文将以Python作为开发语言，结合Flask框架搭建一个简易的流量充值服务。请在终端中执行以下命令安装必要的依赖库：
pip install requests flask
如果尚未安装Python，请前往Python官网下载对应操作系统的安装包并完成安装。安装完成后，创建一个项目文件夹，例如national_flow_project，并在其中新建以下文件结构：
- config.py：存放配置信息，如API地址、账号密钥等。
- utils.py：存放工具函数，主要是签名生成算法。
- app.py：主程序入口，处理Web请求和API调用。
配置文件编写
打开config.py文件，填入从流量供应商处获取的接口信息。为了确保安全性，建议不要将真实的密钥硬编码在代码中，但在本示例中，我们将直接在此文件中进行定义，方便读者直接复制运行。请将以下内容完整复制到config.py中：
```python
config.py
流量供应商API接口地址，请替换为实际供应商提供的地址
API_BASE_URL = "http://api.example.com/v2"
供应商分配的商户账号
MERCHANT_ACCOUNT = "your_account_here"
供应商分配的商户密钥
MERCHANT_KEY = "your_secret_key_here"
接口超时时间（秒）
TIMEOUT = 30
本地服务端口
LOCAL_PORT = 5000
```
注意：API_BASE_URL、MERCHANT_ACCOUNT和MERCHANT_KEY需要替换为你实际持有的凭证。大多数流量接口采用MD5签名方式，通常涉及对参数进行字典排序后拼接密钥的操作。
签名工具函数实现
流量接口的核心安全机制在于签名验证。如果签名计算错误，所有请求都将被拒绝。打开utils.py，我们需要编写一个通用的MD5签名生成函数。该函数将接收一个字典参数，去除空值，按Key的字母顺序排序，然后拼接成字符串进行MD5加密。以下是完整的代码实现：
```python
utils.py
import hashlib
import urllib.parse
def generate_sign(params, key):
"""
生成API接口签名
:param params: 请求参数字典
:param key: 商户密钥
:return: 32位小写MD5签名值
"""
1. 过滤掉sign参数和值为空的参数
filtered_params = {k: v for k, v in params.items() if v is not None and v != "" and k != "sign"}
2. 对参数Key进行字典序升序排列
sorted_keys = sorted(filtered_params.keys())
3. 拼接字符串：key1=value1&key2=value2...key
sign_str = ""
for k in sorted_keys:
对值进行URL编码，防止特殊字符导致签名不一致
value = str(filtered_params[k])
sign_str += f"{k}={value}&"
4. 在末尾追加商户密钥
sign_str += key
5. 进行MD5加密并转为小写
m = hashlib.md5()
m.update(sign_str.encode('utf-8'))
return m.hexdigest().lower()
def format_response(code, msg, data=None):
"""
统一返回格式
"""
result = {
"code": code,
"msg": msg,
"data": data
}
return result
```
核心业务逻辑：发起流量充值
接下来编写app.py的核心部分。我们需要创建一个函数，专门负责向供应商接口发起HTTP POST请求。该函数需要组装业务参数（如手机号、流量包大小），调用generate_sign生成签名，并处理返回结果。以下是app.py的前半部分代码：
```python
app.py
import requests
import config
from flask import Flask, request, jsonify
from utils import generate_sign, format_response
app = Flask(__name__)
def recharge_flow(mobile, size, order_id):
"""
调用第三方流量充值接口
:param mobile: 手机号
:param size: 流量大小（单位通常为M，具体看供应商文档，如100代表100M）
:param order_id: 本地生成的唯一订单号
:return: 接口返回结果字典
"""
构造业务参数
params = {
"account": config.MERCHANT_ACCOUNT,
"mobile": mobile,
"size": size,
"out_order_id": order_id,
"timestamp": "20231027120000" 实际场景应使用当前时间戳
}
生成签名
sign = generate_sign(params, config.MERCHANT_KEY)
params["sign"] = sign
完整的请求URL
url = f"{config.API_BASE_URL}/flow/recharge"
try:
发起POST请求
response = requests.post(url, data=params, timeout=config.TIMEOUT)
response.raise_for_status() 检查HTTP错误
解析JSON返回
res_data = response.json()
根据供应商返回的code判断逻辑成功与否
假设供应商返回 code=0 表示成功提交
if res_data.get("code") == 0:
return True, "提交成功", res_data
else:
return False, res_data.get("message", "未知错误"), res_data
except requests.exceptions.RequestException as e:
return False, f"网络请求异常: {str(e)}", None
except Exception as e:
return False, f"系统异常: {str(e)}", None
```
搭建Web服务与路由接口
为了让本地系统能够接收前端的充值请求，我们需要在app.py中添加Flask路由。这里我们将创建一个/recharge接口，接收JSON格式的数据。请将以下代码追加到app.py中：
```python
app.py (接续上文)
import time
import random
@app.route('/recharge', methods=['POST'])
def handle_recharge():
"""
处理流量充值请求
"""
try:
获取前端传递的JSON参数
req_data = request.get_json()
if not req_data:
return jsonify(format_response(400, "请求参数不能为空"))
mobile = req_data.get("mobile")
size = req_data.get("size")
参数校验
if not mobile or not size:
return jsonify(format_response(400, "手机号和流量大小不能为空"))
if len(mobile) != 11 or not mobile.isdigit():
return jsonify(format_response(400, "手机号格式不正确"))
生成本地唯一订单号：时间戳 + 随机数
timestamp_str = str(int(time.time()))
random_str = str(random.randint(1000, 9999))
local_order_id = f"ORD{timestamp_str}{random_str}"
调用核心充值逻辑
is_success, message, remote_data = recharge_flow(mobile, size, local_order_id)
if is_success:
记录日志：此处可添加数据库操作，保存订单状态为“处理中”
print(f"订单 {local_order_id} 提交成功。")
return jsonify(format_response(200, "充值请求已提交", {"order_id": local_order_id}))
else:
记录日志：保存订单状态为“失败”
print(f"订单 {local_order_id} 提交失败：{message}")
return jsonify(format_response(500, message))
except Exception as e:
return jsonify(format_response(500, f"服务器内部错误: {str(e)}"))
if __name__ == '__main__':
启动Flask服务
print(f"流量服务启动中，监听端口 {config.LOCAL_PORT} ...")
app.run(host='0.0.0.0', port=config.LOCAL_PORT, debug=True)
```
处理异步回调通知
流量充值通常是异步过程：提交请求后立即返回“提交成功”，但实际充值结果可能几分钟后才返回。供应商会通过回调接口通知你的服务器。我们需要在app.py中添加一个回调路由。将以下代码添加到app.py的if __name__ == '__main__':之前：
```python
app.py (回调处理部分)
@app.route('/callback/flow', methods=['POST'])
def handle_flow_callback():
"""
处理流量供应商的异步回调通知
"""
获取供应商POST过来的表单数据
callback_data = request.form.to_dict()
1. 验证签名，防止伪造请求
注意：部分供应商回调签名算法与请求签名算法略有不同，请根据具体文档调整
这里假设算法一致
received_sign = callback_data.get("sign")
calculated_sign = generate_sign(callback_data, config.MERCHANT_KEY)
if received_sign != calculated_sign:
print("回调签名验证失败")
return "sign error", 400
2. 获取业务参数
out_order_id = callback_data.get("out_order_id") 本地订单号
status = callback_data.get("status") 充值状态：1成功，2失败
3. 处理业务逻辑
实际项目中，这里需要连接数据库，根据 out_order_id 查找订单并更新状态
print(f"收到回调：订单 {out_order_id} 状态更新为 {status}")
if status == "1":
更新数据库订单状态为“充值成功”
pass
else:
更新数据库订单状态为“充值失败”，并记录错误原因
pass
4. 必须返回特定字符串给供应商，告知已接收
大多数供应商要求返回 "success" 或 "ok"
return "success"
```
实操测试与验证
代码编写完毕后，即可进行测试。在终端运行启动命令：
python app.py
看到提示“流量服务启动中”后，使用Postman或curl工具向本地接口发送测试请求。以下是使用curl发送POST请求的示例命令：
```bash
curl -X POST http://127.0.0.1:5000/recharge \
-H "Content-Type: application/json" \
-d '{"mobile": "13800138000", "size": "100"}'
```
如果一切配置正确，你将看到类似以下的JSON返回：
```json
{
"code": 200,
"msg": "充值请求已提交",
"data": {
"order_id": "ORD16984000001234"
}
}
``>
此时检查控制台日志，应该能看到“订单...提交成功”的打印信息。对于回调接口，由于是外网访问本地，建议使用内网穿透工具（如ngrok）将本地5000端口映射到公网，并在供应商后台配置回调地址为http://your-ngrok-url/callback/flow进行联调。
