下载地址：https://www.pan38.com/dow/share.php?code=JCnzE 提取密码：3899
这个系统包含三个主要模块：物流信息生成器、API服务和单元测试。物流信息生成器可以生成虚拟的快递单号、寄收件人信息和完整的物流轨迹。API服务提供了查询和生成接口，可以部署为Web服务。单元测试确保代码质量。
import randomimport timeimport jsonfrom datetime import datetime, timedeltafrom faker import Faker
class VirtualLogisticsGenerator: def init(self): self.fake = Faker('zh_CN') self.couriers = [ '顺丰速运', '中通快递', '圆通速递', '韵达快递', '申通快递', 'EMS', '京东物流', '极兔速递', '德邦快递' ] self.status_sequence = [ '已揽收', '运输中', '到达分拣中心', '运输中', '到达目的地城市', '派送中', '已签收' ]
def generate_tracking_number(self, courier=None):
"""生成虚拟快递单号"""
if courier is None:
courier = random.choice(self.couriers)
prefix_map = {
'顺丰速运': 'SF',
'中通快递': 'ZTO',
'圆通速递': 'YT',
'韵达快递': 'YD',
'申通快递': 'STO',
'EMS': 'EMS',
'京东物流': 'JD',
'极兔速递': 'JT',
'德邦快递': 'DB'
}
prefix = prefix_map.get(courier, 'XX')
numbers = ''.join([str(random.randint(0, 9)) for _ in range(12)])
return f"{prefix}{numbers}"
def generate_sender_info(self):
"""生成寄件人信息"""
return {
'name': self.fake.name(),
'phone': self.fake.phone_number(),
'address': self.fake.address()
}
def generate_receiver_info(self):
"""生成收件人信息"""
return {
'name': self.fake.name(),
'phone': self.fake.phone_number(),
'address': self.fake.address()
}
def generate_logistics_details(self, tracking_number, days=3):
"""生成物流详细信息"""
details = []
start_time = datetime.now() - timedelta(days=days)
for i, status in enumerate(self.status_sequence):
time_delta = timedelta(
hours=random.randint(1, 4),
minutes=random.randint(0, 59)
)
start_time += time_delta
detail = {
'time': start_time.strftime('%Y-%m-%d %H:%M:%S'),
'status': status,
'location': self.fake.city()
}
if status == '已揽收':
detail['operator'] = self.fake.name()
detail['remark'] = '快递员已取件'
elif status == '已签收':
detail['operator'] = '客户本人'
detail['remark'] = '已签收，感谢使用'
else:
detail['operator'] = '系统自动'
detail['remark'] = '运输中'
details.append(detail)
return details
def generate_full_logistics(self, count=1):
"""生成完整的物流信息"""
results = []
for _ in range(count):
courier = random.choice(self.couriers)
tracking_number = self.generate_tracking_number(courier)
logistics = {
'courier': courier,
'tracking_number': tracking_number,
'sender': self.generate_sender_info(),
'receiver': self.generate_receiver_info(),
'details': self.generate_logistics_details(tracking_number),
'create_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
}
results.append(logistics)
return results
def save_to_json(self, data, filename='logistics_data.json'):
"""保存为JSON文件"""
with open(filename, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=4)
return filename
if name == 'main': generator = VirtualLogisticsGenerator()
# 生成10条物流信息
logistics_data = generator.generate_full_logistics(10)
# 保存到文件
saved_file = generator.save_to_json(logistics_data)
print(f"已生成10条虚拟物流信息，并保存到 {saved_file}")
# 打印第一条信息示例
print("\n示例物流信息：")
print(json.dumps(logistics_data[0], ensure_ascii=False, indent=4))
flask import Flask, jsonify, requestfrom virtual_logistics import VirtualLogisticsGeneratorimport json
app = Flask(name)generator = VirtualLogisticsGenerator()
预先生成1000条物流数据
with open('logistics_database.json', 'w', encoding='utf-8') as f: json.dump(generator.generate_full_logistics(1000), f, ensure_ascii=False, indent=4)
@app.route('/api/logistics', methods=['GET'])def get_logistics(): """查询物流信息API""" tracking_number = request.args.get('tracking_number')
if not tracking_number:
return jsonify({'error': '缺少快递单号参数'}), 400
with open('logistics_database.json', 'r', encoding='utf-8') as f:
all_data = json.load(f)
for item in all_data:
if item['tracking_number'] == tracking_number:
return jsonify(item)
return jsonify({'error': '未找到该快递单号的信息'}), 404
@app.route('/api/generate', methods=['POST'])def generate_new(): """生成新的物流信息API""" count = request.json.get('count', 1) if count > 100: return jsonify({'error': '单次最多生成100条'}), 400
new_data = generator.generate_full_logistics(count)
# 追加到数据库
with open('logistics_database.json', 'r+', encoding='utf-8') as f:
data = json.load(f)
data.extend(new_data)
f.seek(0)
json.dump(data, f, ensure_ascii=False, indent=4)
return jsonify({'success': True, 'count': count, 'data': new_data})
if name == 'main': app.run(host='0.0.0.0', port=5000, debug=True)