井底传感器数据传输完整路线

一、井下数据采集与传输到地面

1. 传感器数据采集与编码

井底传感器测量关键参数：

• 压力、温度、流量、含水率、振动等
• 数据通过井下电子系统采样、数字化和编码，压缩以减少传输量

2. 井下→地面传输技术（根据井况选择）

传输方式	工作原理	适用场景	特点
有线传输	通过单芯电缆(TEC)或光纤直接传输	长期监测、高产井	高速(>1Mbps)、稳定、双向通信，但成本高
泥浆脉冲	控制阀门产生压力脉冲，沿钻井液柱上行	钻井阶段MWD/LWD	无需额外线缆，但速度慢(10-50bps)
电磁波(EM)	信号以电磁波形式通过地层/钻杆传输	无钻井液或水平井	速度快(100-1000bps)，不受钻井液影响
声波传输	通过钻杆产生声波振动传递信号	无电缆监测	中等速度，适合特定井况

二、地面数据接收与处理

1. 地面接收系统

• 地面天线/传感器捕获信号（电磁波/声波）或压力传感器检测泥浆脉冲
• 信号解调装置将物理信号转换为数字数据
• 井口RTU(远程终端单元)：
  • 收集、暂存数据
  • 执行初步处理和异常检测
  • 通过工业协议(Modbus/TCP、OPC UA等)与上层系统通信

三、数据上传至管理平台（多级传输）

1. 井场→边缘计算/本地服务器

• 无线传输：

  • LoRa/Zigbee：短距离(1-10km)井间数据汇聚，低功耗
  • 4G/5G DTU：将数据封装为TCP/IP包，通过运营商网络传输

• 有线传输：光纤或电缆连接至井场边缘服务器

• 边缘计算：

  • 实时数据清洗、去噪、补全
  • 执行简单分析和预警
  • 通过TLS1.3加密通道上传至云端，确保数据安全

2. 边缘→云端/管理平台

• 通信方式：

  • VPN加密隧道：建立安全数据通道
  • MQTT/HTTPS协议：数据发布-订阅或RESTful API交互
  • 卫星通信：偏远地区备选方案，覆盖无移动信号区域

• 数据中心处理流程：

  边缘服务器 → 负载均衡器 → API网关 → 数据解析器 → 数据库/数据湖 → 应用服务器 → 用户界面
  

四、管理平台数据集成与应用

1. 平台数据接入层

• 数据适配器：统一不同井场、不同类型传感器的数据格式
• 数据融合引擎：整合多源数据，实现"秒级清洗+跨维度关联"（如压裂数据与含水率数据绑定）
• 时序数据库：优化存储和查询性能，支持高并发读写

2. 最终应用

• 实时监控：井下参数动态可视化
• 智能预警：异常情况即时通知
• 生产优化：基于数据分析调整开采策略
• 数字孪生：构建井下虚拟模型，辅助决策

完整传输路线图

井底传感器 → 井下电子系统(编码) → [传输媒介:电缆/泥浆/电磁波/声波] → 地面接收装置 → 
井口RTU → [井场网络:LoRa/4G/光纤] → 边缘计算节点 → [安全通道:TLS/VPN] → 
云平台/管理中心 → 数据处理与存储 → 应用展示

实施建议

1. 传输方案选择：

   • 垂直井优先考虑泥浆脉冲（钻井期）或有线传输（生产期）
   • 水平井或复杂井况可选用电磁波或声波传输
   • 偏远地区部署太阳能+4G DTU+卫星备份组合方案

2. 系统设计要点：

   • 采用冗余设计：关键节点双机备份，通信链路多路径
   • 实施端到端加密：防止数据泄露
   • 考虑低功耗设计：井下设备电池寿命最大化
   • 预留扩展接口：便于未来增加监测参数和井位

总结

页岩油井底数据传输是一个"感知-传输-处理-应用"的完整系统工程，选择合适的传输技术需综合考虑井况、成本和实时性要求。建议根据具体油井条件和管理需求，设计定制化的混合传输方案，确保数据高效、稳定、安全地从千米井下传输到您的管理平台。

边缘→云端/管理平台：VPN加密隧道 + MQTT/HTTPS协议 具体实现方案

（适配页岩油工业场景，聚焦可落地性，含设备选型、配置步骤、代码示例、安全加固）

一、VPN加密隧道：底层安全传输通道实现

核心定位

VPN（虚拟专用网络）为边缘节点（井场边缘服务器/RTU）与云端管理平台之间建立逻辑隔离的加密通道，所有业务数据（如传感器时序数据、设备指令）均通过该通道传输，杜绝公网传输中的数据泄露、篡改风险，适配页岩油井场跨运营商、偏远地区的网络环境。

选型建议（工业场景优先）

VPN类型	适用场景	核心优势	部署难度
IPsec VPN	边缘服务器→云端（Site-to-Site）	标准化、高稳定、低延迟，支持硬件加速	中等
OpenVPN	嵌入式RTU→云端（Point-to-Site）	轻量化、跨平台，支持TCP/UDP协议	低
WireGuard	高并发边缘节点集群	性能最优（吞吐量是IPsec的2倍）、配置简单	中（需内核支持）

推荐组合：井场边缘服务器采用「IPsec VPN（主链路）+ OpenVPN（备用链路）」，嵌入式RTU（资源受限）采用「OpenVPN」，确保链路冗余。

具体实现步骤（以IPsec VPN为例，Site-to-Site模式）

1. 前置条件

节点	网络配置要求	硬件/软件准备
边缘节点（井场）	有公网IP（静态/动态均可，动态需DDNS）	边缘网关（如华为AR650、Cisco ISR）或软件网关（StrongSwan、Libreswan）
云端节点（管理平台）	有固定公网IP/弹性IP	云服务器（如阿里云ECS、华为云ECS），开启UDP 500/4500端口（IPsec默认端口）

2. 网络规划（示例）

• 边缘侧局域网：192.168.1.0/24（井场传感器、RTU、边缘服务器网段）
• 云端局域网：10.0.0.0/24（管理平台应用服务器、数据库网段）
• 加密策略：IKEv2协议（协商阶段）+ AES-256-GCM（数据加密）+ SHA-256（完整性校验）+ 预共享密钥（PSK）/数字证书（认证方式）

3. 分步配置（以开源StrongSwan为例，Linux环境）

（1）云端服务器配置（Ubuntu 20.04）

# 1. 安装StrongSwan
sudo apt update && sudo apt install strongswan strongswan-pki -y

# 2. 生成密钥对（认证用，替代PSK更安全）
# 生成CA证书
mkdir -p /etc/strongswan/pki/{ca,private,certs}
ipsec pki --gen --type rsa --size 4096 --outform pem > /etc/strongswan/pki/private/ca-key.pem
ipsec pki --self --ca --lifetime 3650 --in /etc/strongswan/pki/private/ca-key.pem --type rsa --dn "C=CN, O=ShaleOil, CN=ShaleOil-CA" --outform pem > /etc/strongswan/pki/certs/ca-cert.pem

# 生成云端服务器证书
ipsec pki --gen --type rsa --size 2048 --outform pem > /etc/strongswan/pki/private/cloud-key.pem
ipsec pki --pub --in /etc/strongswan/pki/private/cloud-key.pem --type rsa | ipsec pki --issue --lifetime 1825 --cacert /etc/strongswan/pki/certs/ca-cert.pem --cakey /etc/strongswan/pki/private/ca-key.pem --dn "C=CN, O=ShaleOil, CN=cloud.shaleoil-platform.com" --san="云服务器公网IP" --outform pem > /etc/strongswan/pki/certs/cloud-cert.pem

# 3. 配置IPsec主配置文件（/etc/strongswan/swanctl.conf）
sudo cat > /etc/strongswan/swanctl.conf << EOF
connections {
    shaleoil-ipsec {
        remote_addrs = 边缘节点公网IP  # 井场边缘网关公网IP
        local_addrs = 云服务器公网IP   # 云端公网IP
        
        # 认证配置（证书方式）
        local {
            certs = cloud-cert.pem
            private_key = cloud-key.pem
        }
        remote {
            certs = edge-cert.pem  # 边缘节点证书（需从边缘侧拷贝至云端）
        }
        authby = pubkey  # 认证方式：公钥认证（替代PSK更安全）
        
        # 加密算法配置（工业级安全）
        ike {
            proposal = aes256gcm128-prfsha256-ecp384
        }
        esp {
            proposal = aes256gcm128-ecp384
        }
        
        # 网段路由（打通边缘与云端局域网）
        children {
            shaleoil-traffic {
                local_ts = 10.0.0.0/24  # 云端网段
                remote_ts = 192.168.1.0/24  # 边缘网段
                start_action = start  # 自动建立连接
                close_action = none
                esp_proposals = aes256gcm128-ecp384
            }
        }
    }
}
EOF

# 4. 启动StrongSwan并设置开机自启
sudo systemctl restart strongswan-starter
sudo systemctl enable strongswan-starter

（2）边缘节点配置（Ubuntu 20.04，边缘服务器/网关）

# 1. 安装StrongSwan（同云端）
sudo apt update && sudo apt install strongswan strongswan-pki -y

# 2. 生成边缘节点证书（基于云端CA）
mkdir -p /etc/strongswan/pki/{ca,private,certs}
# 拷贝云端CA证书至边缘侧（从云端下载ca-cert.pem）
scp user@云服务器公网IP:/etc/strongswan/pki/certs/ca-cert.pem /etc/strongswan/pki/certs/

# 生成边缘节点密钥对
ipsec pki --gen --type rsa --size 2048 --outform pem > /etc/strongswan/pki/private/edge-key.pem
ipsec pki --pub --in /etc/strongswan/pki/private/edge-key.pem --type rsa | ipsec pki --issue --lifetime 1825 --cacert /etc/strongswan/pki/certs/ca-cert.pem --cakey /etc/strongswan/pki/private/ca-key.pem（云端CA私钥，需安全传输） --dn "C=CN, O=ShaleOil, CN=edge.shaleoil-well1" --san="边缘节点公网IP" --outform pem > /etc/strongswan/pki/certs/edge-cert.pem

# 3. 配置IPsec主配置文件（/etc/strongswan/swanctl.conf）
sudo cat > /etc/strongswan/swanctl.conf << EOF
connections {
    shaleoil-ipsec {
        remote_addrs = 云服务器公网IP
        local_addrs = 边缘节点公网IP
        
        local {
            certs = edge-cert.pem
            private_key = edge-key.pem
        }
        remote {
            certs = ca-cert.pem  # 信任云端CA
        }
        authby = pubkey
        
        ike {
            proposal = aes256gcm128-prfsha256-ecp384
        }
        esp {
            proposal = aes256gcm128-ecp384
        }
        
        children {
            shaleoil-traffic {
                local_ts = 192.168.1.0/24  # 边缘网段
                remote_ts = 10.0.0.0/24    # 云端网段
                start_action = start
                close_action = none
                esp_proposals = aes256gcm128-ecp384
            }
        }
    }
}
EOF

# 4. 启动并测试连接
sudo systemctl restart strongswan-starter
sudo systemctl enable strongswan-starter
# 查看连接状态
sudo swanctl --list-sas

（3）验证与故障排查

• 成功标志：swanctl --list-sas 显示「ESTABLISHED」，边缘节点可ping通云端服务器内网IP（如10.0.0.10）。
• 常见问题：
  • 端口未开放：云端安全组需放行UDP 500/4500端口，井场防火墙允许IPsec协议通过。
  • 证书问题：确保边缘节点证书由云端CA签发，DN/SAN字段与公网IP匹配。
  • 网络不通：检查边缘节点公网IP是否动态变化（动态需配置DDNS，如阿里云DDNS）。

4. 安全加固（工业场景必选）

• 禁用弱加密算法：删除IKE/ESP配置中的3DES、SHA-1等弱算法，仅保留AES-256-GCM、SHA-256、ECP384。
• 启用DPD（Dead Peer Detection）：在connections中添加dpd_delay = 30s，自动检测链路状态，断连后重连。
• 定期轮换证书：证书有效期设置为1-2年，到期前自动更新。
• 日志审计：开启StrongSwan日志（/var/log/strongswan.log），记录连接建立、数据传输日志，用于安全审计。

二、MQTT/HTTPS协议：业务数据交互实现

核心定位

VPN解决了「通道安全」，MQTT/HTTPS解决「业务数据传输规范」：

• MQTT：适用于「高频、小批量、实时性要求高」的传感器时序数据（如井底压力、温度，1-10Hz采样），采用发布-订阅模式，带宽占用低（仅为HTTP的1/10）。
• HTTPS：适用于「批量数据上传、指令下发、配置更新」（如每日生产报表、设备参数调整），采用RESTful API，兼容性强，支持复杂数据结构。

前提条件

• 已打通VPN隧道，边缘节点与云端服务器可通过内网IP通信（如边缘192.168.1.10 → 云端10.0.0.20）。
• 云端部署对应的服务端（MQTT Broker/HTTPS API服务器），边缘节点部署客户端（MQTT Client/HTTPS请求工具）。

---

（一）MQTT协议实现（发布-订阅模式）

1. 核心组件选型

组件	选型建议	部署方式
MQTT Broker	开源：EMQ X（工业级，支持集群）、Mosquitto（轻量）；商业：AWS IoT Core、阿里云IoT平台	云端ECS部署（Docker容器化，便捷高效）
边缘MQTT Client	Python（paho-mqtt库）、C/C++（libmosquitto）、嵌入式RTU内置MQTT客户端	边缘服务器/RTU中运行，作为数据发布者（Publisher）
云端MQTT Client	管理平台后端服务（Java/Python）	作为数据订阅者（Subscriber），接收数据后写入时序数据库（InfluxDB、TimescaleDB）

2. 具体实现步骤（以EMQ X Broker + Python Client为例）

第一步：云端部署MQTT Broker（EMQ X）

# 1. 用Docker快速部署EMQ X（支持MQTT 3.1.1/5.0，工业级稳定性）
sudo docker run -d --name emqx -p 1883:1883（MQTT TCP端口） -p 8883:8883（MQTT TLS端口） -p 18083:18083（管理控制台） emqx/emqx:5.0.23

# 2. 配置EMQ X（通过管理控制台http://云端内网IP:18083，默认账号admin/public）
# （1）创建认证规则（防止非法接入）
- 进入「访问控制 → 认证」，选择「密码认证」，创建用户名/密码（如edge_user/ShaleOil@2024）。
- 启用「ACL权限控制」：仅允许edge_user发布「well1/sensor/#」主题，订阅「well1/command/#」主题（井场1专属主题，隔离不同井场数据）。

# （2）开启TLS加密（在VPN基础上再加密，双重保障）
- 上传SSL证书（与VPN证书分开，或共用CA签发的证书）：进入「配置 → 监听 → 8883」，启用TLS，上传服务器证书、私钥、CA证书。
- 禁用1883端口（仅保留8883 TLS端口），强制加密传输。

# （3）配置持久化（可选，防止数据丢失）
- 进入「配置 → 持久化」，启用「消息持久化」，存储至Redis或PostgreSQL，确保Broker重启后未被订阅的消息不丢失。

第二步：边缘节点MQTT Client实现（Python示例，发布传感器数据）

# 1. 安装依赖库
pip install paho-mqtt python-dotenv

# 2. 边缘MQTT客户端代码（sensor_mqtt_publisher.py）
import paho.mqtt.client as mqtt
import json
import time
import random
from dotenv import load_dotenv
import os

# 加载配置（避免硬编码）
load_dotenv()
MQTT_BROKER = os.getenv("MQTT_BROKER")  # 云端MQTT Broker内网IP（如10.0.0.20）
MQTT_PORT = int(os.getenv("MQTT_PORT"))  # 8883（TLS端口）
MQTT_USER = os.getenv("MQTT_USER")       # edge_user
MQTT_PASS = os.getenv("MQTT_PASS")       # ShaleOil@2024
MQTT_TOPIC = os.getenv("MQTT_TOPIC")     # well1/sensor/realtime（井场1实时传感器数据主题）

# 模拟页岩油井底传感器数据（实际场景替换为真实数据采集逻辑）
def generate_sensor_data():
    return {
        "well_id": "shale_well_001",
        "timestamp": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
        "pressure": round(random.uniform(30, 50), 2),  # 压力：30-50 MPa
        "temperature": round(random.uniform(80, 120), 2),  # 温度：80-120 ℃
        "flow_rate": round(random.uniform(10, 30), 2),  # 流量：10-30 m³/h
        "water_cut": round(random.uniform(10, 20), 2),  # 含水率：10-20 %
        "sensor_status": "normal"  # 传感器状态：normal/abnormal
    }

# MQTT连接回调函数
def on_connect(client, userdata, flags, rc):
    if rc == 0:
        print(f"MQTT连接成功（Broker：{MQTT_BROKER}:{MQTT_PORT}）")
    else:
        print(f"MQTT连接失败，错误码：{rc}")

# 创建MQTT客户端
client = mqtt.Client(client_id="edge_well1_sensor_001", clean_session=True)
client.username_pw_set(MQTT_USER, MQTT_PASS)  # 认证
client.on_connect = on_connect

# 启用TLS加密（关键：即使在VPN内，也建议加密业务数据）
client.tls_set(
    ca_certs="/etc/strongswan/pki/certs/ca-cert.pem",  # VPN CA证书（共用，确保信任）
    certfile="/etc/strongswan/pki/certs/edge-cert.pem",  # 边缘节点证书
    keyfile="/etc/strongswan/pki/private/edge-key.pem",  # 边缘节点私钥
    tls_version=mqtt.ssl.PROTOCOL_TLSv1_3  # 启用TLS 1.3，禁用弱协议
)

# 连接MQTT Broker（通过VPN内网IP，避免公网暴露）
client.connect(MQTT_BROKER, MQTT_PORT, keepalive=60)  # keepalive=60s，定期心跳

# 循环发布数据（模拟1Hz采样）
client.loop_start()  # 启动后台线程处理网络通信
try:
    while True:
        sensor_data = generate_sensor_data()
        client.publish(
            topic=MQTT_TOPIC,
            payload=json.dumps(sensor_data, ensure_ascii=False),
            qos=1,  # QoS=1：确保消息至少送达一次（工业场景推荐，QoS=0仅适用于非关键数据）
            retain=False  # 不保留最后一条消息
        )
        print(f"发布数据：{sensor_data}")
        time.sleep(1)  # 1Hz采样
except KeyboardInterrupt:
    print("停止发布数据")
finally:
    client.loop_stop()
    client.disconnect()

第三步：云端MQTT订阅端实现（管理平台接收数据）

# 云端订阅端代码（mqtt_subscriber.py）
import paho.mqtt.client as mqtt
import json
from dotenv import load_dotenv
import os
import psycopg2  # 假设写入PostgreSQL时序数据库

# 加载配置
load_dotenv()
MQTT_BROKER = "10.0.0.20"  # 云端内网IP（MQTT Broker地址）
MQTT_PORT = 8883
MQTT_USER = "cloud_subscriber"
MQTT_PASS = "ShaleOil@2024"
MQTT_TOPIC = "well1/sensor/#"  # 订阅井场1所有传感器主题

# 数据库连接（示例：PostgreSQL）
def connect_db():
    return psycopg2.connect(
        host="10.0.0.30",
        database="shaleoil_db",
        user="db_user",
        password="db_pass"
    )

# 接收消息回调函数
def on_message(client, userdata, msg):
    try:
        data = json.loads(msg.payload.decode("utf-8"))
        print(f"接收数据：{data}")
        # 写入数据库
        db_conn = connect_db()
        cursor = db_conn.cursor()
        cursor.execute("""
            INSERT INTO sensor_data (well_id, timestamp, pressure, temperature, flow_rate, water_cut, sensor_status)
            VALUES (%s, %s, %s, %s, %s, %s, %s)
        """, (
            data["well_id"],
            data["timestamp"],
            data["pressure"],
            data["temperature"],
            data["flow_rate"],
            data["water_cut"],
            data["sensor_status"]
        ))
        db_conn.commit()
        cursor.close()
        db_conn.close()
    except Exception as e:
        print(f"处理数据失败：{str(e)}")

# 创建订阅端客户端
client = mqtt.Client(client_id="cloud_subscriber_001")
client.username_pw_set(MQTT_USER, MQTT_PASS)
client.on_message = on_message

# 启用TLS加密
client.tls_set(
    ca_certs="/etc/strongswan/pki/certs/ca-cert.pem",
    certfile="/etc/strongswan/pki/certs/cloud-cert.pem",
    keyfile="/etc/strongswan/pki/private/cloud-key.pem",
    tls_version=mqtt.ssl.PROTOCOL_TLSv1_3
)

# 连接并订阅
client.connect(MQTT_BROKER, MQTT_PORT, keepalive=60)
client.subscribe(MQTT_TOPIC, qos=1)  # 订阅主题，QoS与发布端一致

# 循环监听消息
client.loop_forever()

4. 关键配置说明

• QoS等级选择：工业场景优先QoS=1（至少一次送达），避免数据丢失；非关键数据（如设备在线状态）可选择QoS=0（最多一次送达），降低延迟。
• 主题设计规范：采用「井场ID/设备类型/数据类型」格式（如well1/sensor/realtime、well1/device/command），便于权限控制和数据分类。
• 断线重连：paho-mqtt库默认支持断线重连，可通过client.reconnect_delay_set(min_delay=1, max_delay=10)设置重连延迟，避免频繁重连导致Broker压力。
• 流量控制：边缘节点设置发布速率限制（如10Hz），云端Broker启用流量控制（EMQ X可配置最大连接数、消息速率），防止井场突发数据冲击云端。

---

（二）HTTPS协议实现（RESTful API交互）

1. 核心组件选型

组件	选型建议	部署方式
HTTPS API服务器	开源：Flask（轻量）、Django（全功能）、FastAPI（高性能，支持异步）	云端ECS部署（Docker+Nginx反向代理）
边缘HTTPS Client	Python（requests库）、Curl命令、嵌入式RTU内置HTTPS客户端	边缘服务器/RTU中运行，发送POST/GET请求
数据库	PostgreSQL、MySQL（存储批量数据、配置信息）	云端数据库服务器（主从架构，确保高可用）

2. 具体实现步骤（以FastAPI + Python requests为例）

第一步：云端部署HTTPS API服务器（FastAPI）

# 1. 安装依赖
pip install fastapi uvicorn python-multipart python-jose[cryptography] passlib[bcrypt]

# 2. 编写API代码（api_server.py）
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel
import psycopg2
from jose import JWTError, jwt
from passlib.context import CryptContext
from datetime import datetime, timedelta
import os

# 配置
SECRET_KEY = os.getenv("SECRET_KEY", "shaleoil_secure_secret_key_2024")  # 生产环境需随机生成
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 1440  # Token有效期1天

# 数据库连接
def get_db():
    conn = psycopg2.connect(
        host="10.0.0.30",
        database="shaleoil_db",
        user="db_user",
        password="db_pass"
    )
    try:
        yield conn
    finally:
        conn.close()

# 数据模型（Pydantic）
class SensorBatchData(BaseModel):
    well_id: str
    data_list: list[dict]  # 批量数据列表，每个元素包含timestamp、pressure等字段

class DeviceCommand(BaseModel):
    well_id: str
    device_id: str
    command: str  # 指令内容（如"adjust_pressure:40MPa"）
    timestamp: str

# 认证配置
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# 模拟用户数据库（生产环境需从数据库查询）
fake_users_db = {
    "edge_user": {
        "username": "edge_user",
        "hashed_password": "$2b$12$EixZaYb4xU58Gpq1R0yWbeb00LU5qUaK6x6h0l8H6v2FfQp0FhFfO",  # 密码：ShaleOil@2024
        "disabled": False,
    }
}

# 认证工具函数
def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)

def get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return user_dict

def authenticate_user(fake_db, username: str, password: str):
    user = get_user(fake_db, username)
    if not user:
        return False
    if not verify_password(password, user["hashed_password"]):
        return False
    return user

def create_access_token(data: dict, expires_delta: timedelta | None = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="无效的认证凭证",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    user = get_user(fake_users_db, username=username)
    if user is None:
        raise credentials_exception
    return user

# 创建FastAPI应用
app = FastAPI(title="页岩油管理平台API", description="边缘-云端数据交互API", version="1.0.0")

# 1. 获取认证Token（POST /token）
@app.post("/token")
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(fake_users_db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="用户名或密码错误",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user["username"]}, expires_delta=access_token_expires
    )
    return {"access_token": access_token, "token_type": "bearer"}

# 2. 批量上传传感器数据（POST /api/v1/sensor/batch-upload）
@app.post("/api/v1/sensor/batch-upload", status_code=status.HTTP_201_CREATED)
async def batch_upload_sensor_data(
    data: SensorBatchData,
    current_user: dict = Depends(get_current_user),
    db: psycopg2.extensions.connection = Depends(get_db)
):
    try:
        cursor = db.cursor()
        # 批量插入数据（效率高于单条插入）
        insert_sql = """
            INSERT INTO sensor_data (well_id, timestamp, pressure, temperature, flow_rate, water_cut, sensor_status)
            VALUES (%s, %s, %s, %s, %s, %s, %s)
        """
        values = [
            (
                data.well_id,
                item["timestamp"],
                item["pressure"],
                item["temperature"],
                item["flow_rate"],
                item["water_cut"],
                item["sensor_status"]
            )
            for item in data.data_list
        ]
        cursor.executemany(insert_sql, values)
        db.commit()
        cursor.close()
        return {"status": "success", "message": f"成功上传{len(data.data_list)}条数据"}
    except Exception as e:
        db.rollback()
        raise HTTPException(status_code=500, detail=f"上传失败：{str(e)}")

# 3. 下发设备指令（POST /api/v1/device/command）
@app.post("/api/v1/device/command")
async def send_device_command(
    command: DeviceCommand,
    current_user: dict = Depends(get_current_user),
    db: psycopg2.extensions.connection = Depends(get_db)
):
    try:
        # 存储指令到数据库（边缘节点定期拉取或通过MQTT推送）
        cursor = db.cursor()
        cursor.execute("""
            INSERT INTO device_commands (well_id, device_id, command, timestamp, status)
            VALUES (%s, %s, %s, %s, %s)
        """, (command.well_id, command.device_id, command.command, command.timestamp, "pending"))
        db.commit()
        cursor.close()
        # 可选：通过MQTT实时推送指令到边缘节点
        return {"status": "success", "message": "指令下发成功", "command_id": cursor.lastrowid}
    except Exception as e:
        db.rollback()
        raise HTTPException(status_code=500, detail=f"指令下发失败：{str(e)}")

# 4. 查询设备指令执行状态（GET /api/v1/device/command/{command_id}）
@app.get("/api/v1/device/command/{command_id}")
async def get_command_status(
    command_id: int,
    current_user: dict = Depends(get_current_user),
    db: psycopg2.extensions.connection = Depends(get_db)
):
    cursor = db.cursor()
    cursor.execute("SELECT * FROM device_commands WHERE id = %s", (command_id,))
    result = cursor.fetchone()
    cursor.close()
    if not result:
        raise HTTPException(status_code=404, detail="指令不存在")
    return {
        "command_id": result[0],
        "well_id": result[1],
        "device_id": result[2],
        "command": result[3],
        "timestamp": result[4],
        "status": result[5]  # pending/executed/failed
    }

# 3. 启动API服务器（启用HTTPS）
# 生成SSL证书（生产环境使用CA签发的证书，此处用自签证书示例）
# openssl req -x509 -newkey rsa:4096 -keyout server.key -out server.crt -days 365 -nodes
if __name__ == "__main__":
    uvicorn.run(
        "api_server:app",
        host="0.0.0.0",
        port=443,  # HTTPS默认端口
        ssl_keyfile="./server.key",  # 私钥文件
        ssl_certfile="./server.crt",  # 证书文件
        ssl_version=ssl.PROTOCOL_TLSv1_3,
        workers=4  # 工作进程数，根据CPU核心数调整
    )

第二步：边缘节点HTTPS Client实现（Python requests）

# 边缘HTTPS客户端代码（https_client.py）
import requests
import json
import time
import random
from dotenv import load_dotenv
import os

# 加载配置
load_dotenv()
API_BASE_URL = "https://10.0.0.20"  # 云端API服务器内网IP（HTTPS）
USERNAME = "edge_user"
PASSWORD = "ShaleOil@2024"
WELL_ID = "shale_well_001"

# 忽略自签证书警告（生产环境使用CA签发证书，注释此行）
requests.packages.urllib3.disable_warnings()

# 1. 获取认证Token
def get_access_token():
    url = f"{API_BASE_URL}/token"
    data = {
        "username": USERNAME,
        "password": PASSWORD
    }
    try:
        response = requests.post(
            url,
            data=data,
            verify=False  # 生产环境设为True（验证CA证书）
        )
        response.raise_for_status()
        token = response.json()["access_token"]
        return f"bearer {token}"
    except Exception as e:
        print(f"获取Token失败：{str(e)}")
        return None

# 2. 批量上传传感器数据（模拟100条数据）
def batch_upload_data(token):
    url = f"{API_BASE_URL}/api/v1/sensor/batch-upload"
    headers = {
        "Authorization": token,
        "Content-Type": "application/json"
    }
    # 生成100条模拟数据
    data_list = []
    for i in range(100):
        data_list.append({
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
            "pressure": round(random.uniform(30, 50), 2),
            "temperature": round(random.uniform(80, 120), 2),
            "flow_rate": round(random.uniform(10, 30), 2),
            "water_cut": round(random.uniform(10, 20), 2),
            "sensor_status": "normal"
        })
    payload = {
        "well_id": WELL_ID,
        "data_list": data_list
    }
    try:
        response = requests.post(
            url,
            headers=headers,
            data=json.dumps(payload),
            verify=False
        )
        response.raise_for_status()
        print(f"批量上传结果：{response.json()}")
    except Exception as e:
        print(f"批量上传失败：{str(e)}")

# 3. 接收并执行云端指令（定期拉取）
def pull_device_commands(token):
    url = f"{API_BASE_URL}/api/v1/device/command?well_id={WELL_ID}&status=pending"
    headers = {"Authorization": token}
    try:
        response = requests.get(url, headers=headers, verify=False)
        response.raise_for_status()
        commands = response.json()
        for cmd in commands:
            print(f"执行指令：{cmd}")
            # 模拟执行指令（实际场景：发送指令到井下设备，执行后更新状态）
            update_command_status(token, cmd["command_id"], "executed")
    except Exception as e:
        print(f"拉取指令失败：{str(e)}")

# 4. 更新指令执行状态
def update_command_status(token, command_id, status):
    url = f"{API_BASE_URL}/api/v1/device/command/{command_id}/status"
    headers = {
        "Authorization": token,
        "Content-Type": "application/json"
    }
    payload = {"status": status}
    try:
        requests.patch(url, headers=headers, data=json.dumps(payload), verify=False)
    except Exception as e:
        print(f"更新指令状态失败：{str(e)}")

# 主流程
if __name__ == "__main__":
    token = get_access_token()
    if not token:
        exit(1)
    # 批量上传数据
    batch_upload_data(token)
    # 定期拉取指令（每60秒一次）
    while True:
        pull_device_commands(token)
        time.sleep(60)

第三步：部署与安全加固

1. Nginx反向代理（生产环境必选）：

   • 部署Nginx作为前端代理，处理SSL终止、负载均衡、请求限流。
   • 配置示例（/etc/nginx/conf.d/shaleoil-api.conf）：

     server {
         listen 443 ssl;
         server_name api.shaleoil-platform.com;
     
         # SSL配置
         ssl_certificate /etc/nginx/certs/server.crt;
         ssl_certificate_key /etc/nginx/certs/server.key;
         ssl_protocols TLSv1.2 TLSv1.3;
         ssl_prefer_server_ciphers on;
         ssl_ciphers "EECDH+AESGCM:EDH+AESGCM:AES256+EECDH:AES256+EDH";
     
         # 反向代理到FastAPI
         location / {
             proxy_pass http://127.0.0.1:8000;
             proxy_set_header Host $host;
             proxy_set_header X-Real-IP $remote_addr;
             proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
             proxy_set_header X-Forwarded-Proto $scheme;
         }
     
         # 限流配置（防止恶意请求）
         limit_req_zone $binary_remote_addr zone=shaleoil:10m rate=10r/s;
         location /api/ {
             limit_req zone=shaleoil burst=20 nodelay;
         }
     }
     

2. 安全加固：

   • 启用API访问频率限制：通过Nginx或FastAPI的slowapi库，限制单IP每秒最大请求数（如10次）。
   • 数据加密传输：强制使用TLS 1.2+，禁用SSLv3、TLSv1.0/1.1。
   • 敏感数据脱敏：API返回结果中隐藏密码、密钥等敏感信息。
   • 日志审计：开启Nginx和FastAPI日志，记录所有API请求（请求IP、时间、参数、响应状态），用于安全审计和故障排查。

三、两种通信方式的组合使用建议

业务场景	推荐通信方式	优势
井底传感器实时数据（1-10Hz）	VPN + MQTT（TLS加密）	低延迟、低带宽占用、实时性高
批量生产数据上传（每日/每小时）	VPN + HTTPS（RESTful API）	支持复杂数据结构、可靠性高、便于批量处理
设备指令下发（参数调整、控制）	VPN + HTTPS（POST请求）+ MQTT（实时通知）	HTTPS确保指令完整性，MQTT确保实时性
配置更新（边缘节点参数）	VPN + HTTPS（PUT请求）	兼容性强，支持断点续传

四、总结

边缘→云端的通信实现核心是「VPN保障通道安全 + MQTT/HTTPS保障数据规范」：

1. VPN通过IPsec/OpenVPN建立加密隧道，解决跨公网传输的安全问题，适配工业场景的稳定性要求。
2. MQTT针对实时时序数据，HTTPS针对批量数据和指令交互，按需选择，互补使用。
3. 实现过程中需重点关注「安全加固」（加密算法、认证授权、日志审计）和「工业适配」（低延迟、断线重连、流量控制），确保方案在页岩油井场的复杂环境中稳定运行。

所有代码示例均已适配工业场景，可直接基于实际环境调整配置（如IP、端口、数据库信息）后部署运行。