拼车行程存证程序,行程,费用,路线上链,发生纠纷可追溯,防止绕路,临时加价。
解决一个出行领域的老大难问题——网约车的“信任博弈”。
作为一名经常出差、对“幽灵车”和“天价账单”深恶痛绝的开发者,我设计了这套拼车行程存证系统。它不仅能防绕路,还能在发生纠纷时充当“数字黑匣子”。
下面是完整的 Python 实现方案。
1. 实际应用场景描述
场景: 城际拼车或网约车服务(类似滴滴顺风车或哈啰出行)。
人物: 乘客 Alice、Bob 和司机 David。
过程: Alice 和 Bob 预约了从机场到市区的拼车。行程开始后,David 为了赚更多高速费,故意绕路走了拥堵的外环;到达后,又以“夜间服务费”为由临时加价。Alice 拿出手机想投诉,却发现行程记录模糊,平台客服互相推诿。
解决方案: 引入“链上行车黑匣子”。行程开始时,起点、终点、预估价、路线规划即被打包上链。行驶中,GPS 坐标点定时上链(或存证)。行程结束,实际费用根据链上记录自动结算,任何偏离路线的行为都有据可查。
2. 引入痛点
1. 中心化平台黑箱: 行程数据存储在平台服务器,平台可以为了自身利益(如冲单量)修改数据,用户无法自证。
2. 临时加价乱象: 司机利用信息不对称,借口“堵车”“夜间”“天气不好”随意加价。
3. 绕路举证难: 即使有导航截图,平台也可以辩称“当时路况就是这样”,缺乏不可篡改的第三方证据。
4. 拼车纠纷: 拼车中,先上车的乘客和后上车的乘客路线不一致,容易产生口角,责任界定不清。
3. 核心逻辑讲解 (创新点)
本项目采用“预言机 (Oracle) + 智能合约自动执行”的架构:
* 行程 NFT (Trip NFT): 每次拼车订单生成一个唯一的 Trip ID,作为链上凭证,锚定本次行程的所有数据。
* GPS 坐标锚定: 利用 Chainlink Oracle(概念上)将现实世界的 GPS 数据流定期写入区块链。我们将简化为定时提交
"Location Ping"。
* 智能合约计费: 费用不是由司机说了算,而是由智能合约根据链上记录的里程和时间结合预设费率自动计算得出。
* 争议仲裁: 一旦发生纠纷,任何一方都可以发起仲裁,系统将比对链上记录的路线与标准地图路线,判定责任方。
4. 代码模块化实现
项目结构如下:
carpool_chain/
├── app.py # FastAPI Web 应用入口
├── blockchain.py # 区块链核心逻辑
├── trip_logic.py # 行程计算与计费逻辑
├── models.py # Pydantic 数据模型
└── README.md
4.1 models.py (数据结构)
from pydantic import BaseModel
from typing import List, Tuple
from datetime import datetime
class LocationPoint(BaseModel):
"""GPS 坐标点"""
lat: float # 纬度
lng: float # 经度
timestamp: datetime
class TripRequest(BaseModel):
"""拼车请求"""
trip_id: str
driver_address: str
passenger_addresses: List[str]
start_point: LocationPoint
end_point: LocationPoint
estimated_fare: float
class TripRecord(BaseModel):
"""行程存证记录"""
trip_id: str
route_hash: str # 路线数据的哈希
actual_distance: float # 实际里程 (km)
actual_duration: int # 实际时长 (min)
final_fare: float # 最终结算费用
location_log: List[LocationPoint] # GPS轨迹日志
4.2 trip_logic.py (业务逻辑)
# 简化版行程计算器
class TripCalculator:
"""
根据链上GPS数据计算费用和路线偏差
"""
BASE_RATE = 3.0 # 起步价
PER_KM = 2.0 # 每公里价格
PER_MINUTE = 0.5 # 每分钟价格
@classmethod
def calculate_fare(cls, distance_km: float, duration_min: int) -> float:
"""根据预设规则计算费用"""
fare = cls.BASE_RATE + (distance_km * cls.PER_KM) + (duration_min * cls.PER_MINUTE)
return round(fare, 2)
@classmethod
def check_route_deviation(cls, planned_route: list, actual_route: list) -> float:
"""
检查路线偏离度 (简化版:比较总距离)
返回偏离百分比
"""
# 实际项目中应使用地图API计算相似度
planned_dist = cls._calc_total_distance(planned_route)
actual_dist = cls._calc_total_distance(actual_route)
if planned_dist == 0:
return 0
deviation = (actual_dist - planned_dist) / planned_dist
return deviation * 100
@staticmethod
def _calc_total_distance(route: list) -> float:
# 简化的距离计算(欧几里得距离)
total = 0
for i in range(1, len(route)):
total += ((route[i]['lat'] - route[i-1]['lat'])**2 +
(route[i]['lng'] - route[i-1]['lng'])**2)**0.5
return total
4.3 blockchain.py (核心逻辑)
import hashlib
import json
from .models import TripRecord, LocationPoint
from .trip_logic import TripCalculator
class CarpoolBlockchain:
"""
拼车行程存证链
"""
def __init__(self):
self.chain = []
self.trip_records = {} # {trip_id: TripRecord}
self.create_genesis_block()
def create_genesis_block(self):
self.chain.append({
"index": 0,
"data": "Genesis Block for Carpool System",
"hash": "genesis"
})
def _get_last_block(self) -> dict:
return self.chain[-1]
def submit_location_ping(self, trip_id: str, location: LocationPoint):
"""
提交GPS点位 (模拟Oracle喂价)
"""
if trip_id not in self.trip_records:
# 初始化行程记录
self.trip_records[trip_id] = TripRecord(
trip_id=trip_id,
route_hash="",
actual_distance=0,
actual_duration=0,
final_fare=0,
location_log=[]
)
record = self.trip_records[trip_id]
record.location_log.append(location)
# 更新路线哈希 (将所有点打包哈希)
route_data = json.dumps([loc.dict() for loc in record.location_log])
record.route_hash = hashlib.sha256(route_data.encode()).hexdigest()
print(f"Trip {trip_id}: Location ping received at {location.timestamp}")
def finalize_trip(self, trip_id: str) -> TripRecord:
"""
行程结束,自动结算
"""
record = self.trip_records[trip_id]
log = record.location_log
if len(log) < 2:
raise ValueError("Invalid trip log")
# 计算实际里程和时长
start_time = log[0].timestamp
end_time = log[-1].timestamp
record.actual_duration = int((end_time - start_time).total_seconds() / 60)
record.actual_distance = TripCalculator._calc_total_distance([loc.dict() for loc in log])
# 智能合约自动计费
record.final_fare = TripCalculator.calculate_fare(
record.actual_distance,
record.actual_duration
)
# 写入区块
last_block = self._get_last_block()
new_block = {
"index": len(self.chain),
"data": record.json(),
"previous_hash": last_block["hash"],
"hash": hashlib.sha256(record.json().encode()).hexdigest()
}
self.chain.append(new_block)
return record
4.4 app.py (API 接口)
from fastapi import FastAPI
from .blockchain import CarpoolBlockchain
from .models import LocationPoint
from datetime import datetime
app = FastAPI()
blockchain = CarpoolBlockchain()
@app.post("/trip/{trip_id}/ping")
def submit_location(trip_id: str, lat: float, lng: float):
"""司机端提交GPS位置"""
location = LocationPoint(lat=lat, lng=lng, timestamp=datetime.now())
blockchain.submit_location_ping(trip_id, location)
return {"message": "Location recorded on chain"}
@app.post("/trip/{trip_id}/finish")
def finish_trip(trip_id: str):
"""结束行程,触发自动结算"""
final_record = blockchain.finalize_trip(trip_id)
return {
"message": "行程已结束,费用已自动结算",
"final_fare": final_record.final_fare,
"actual_distance": f"{final_record.actual_distance:.2f} km"
}
5. README 文件与使用说明
CarpoolChain (拼车行程存证系统)
基于区块链的网约车/拼车行程存证与自动结算系统。
核心特性
* GPS 轨迹上链: 实时记录行车轨迹,防止绕路。
* 智能合约计费: 费用根据链上里程和时间自动计算,杜绝临时加价。
* 不可篡改证据: 所有行程数据永久上链,可作为法律纠纷的电子证据。
使用流程
1. 启动服务
uvicorn app:app --reload
2. 模拟行程开始司机接单后,系统分配
"trip_id"。
3. 定时上报位置司机端 App 每隔 10 秒调用一次接口:
curl -X POST http://localhost:8000/trip/TRIP-001/ping?lat=31.23&lng=121.47
4. 结束行程
curl -X POST http://localhost:8000/trip/TRIP-001/finish
系统将返回最终的结算金额。
6. 核心知识点卡片
知识点 解释
Oracle (预言机) 将现实世界的数据(如 GPS、温度、股价)可信地传输到区块链上的中间件。
Smart Contract (智能合约) 部署在链上的自动化脚本,满足条件(如行程结束)即自动执行(如转账付款)。
Geofencing (地理围栏) 结合 GIS 技术,判断车辆是否偏离预定区域,可用于高级风控。
Data Provenance (数据溯源) 追踪数据的来源和历史变更,确保数据从采集到存储的全过程可信。
Immutable Ledger (不可变账本) 区块链的基础特性,保证司机无法抵赖其行驶轨迹。
7. 总结
作为一名全栈工程师,我认为这个项目最能体现区块链在物联网 (IoT) 领域的巨大潜力。
* 技术深度: 我们模拟了 Oracle 的工作方式,解决了“如何让现实世界的数据进入封闭区块链”这一核心难题。
* 产品思维: 传统的解决方案是“出了问题找客服”,而本方案是“从源头上消灭问题”。通过将规则代码化(Code is Law),我们建立了一个无需信任中介的出行环境。
* 商业价值: 这种模式特别适合城际拼车和货运物流。司机不用担心被恶意投诉,乘客不用担心被宰客。所有的信任成本都被技术手段抹平了。
利用AI解决实际问题,如果你觉得这个工具好用,欢迎关注长安牧笛!
