5.4 实战案例：基于Flak和Floyd-Warshall的航班查询系统
本项目是一个基于Flak框架的在线航班查询系统，通过使用Floyd-Warshall算法来找航班的最短路径。用户可以输入起点、终点和路径优先标准，然后应用会查询数据库，返回最短路径上的航班信息，包括航空公司、航班号和权重。用户可以通过Flak Web界面或命令行界面来使用这个功能，还可以导入CSV数据进行查询。
实例5-8：基于Flak和Floyd-Warshall的航班查询系统（codes/5/connecting_flights）
5.4.1 背景介绍
在现代社会，越来越多的人们选择使用航空出行，而寻找最佳的航班连接以及最短的飞行路线成为了一个重要的需求。然而，随着航班数量的增加和航线网络的复杂性，手动查找最佳连接变得十分繁琐，因此需要一种自动化的方式来帮助人们寻找最佳的航班连接方案。在这样的背景下，本航班查询系统便应运而生。
本项目是一个用Python实现的航班查询应用系统，通过使用Floyd-Warshall算法来寻找最短路径。它提供了一个用户友好的Web界面和命令行界面，让用户可以方便地输入起点、终点和路径优先标准，然后应用会查询数据库，返回最短路径上的航班信息。用户可以选择查询最低价格、最短距离或最短时间的航班连接。总之，本项目为用户提供了一个方便快捷的方式来查询航班信息，帮助他们找到最佳的航班路线，节省时间和精力。
在项目中使用MongoDB数据库来存储航班信息，用户可以通过命令行界面或导入CSV数据来添加新的航班信息。
5.4.2 数据操作
（1）文件Database.py定义了一个名为 Database 的类，实现了与MongoDB数据库的连接和操作相关的功能。
from enum import IntEnum from pymongo import MongoClient, errors class Database(): """方便在数据库中操作数据""" Criterion = IntEnum('Criterion', 'price duration distance') # 用于计算最短路径的权重标准 def __init__(self, ip, port, db_name): """构造函数""" self._client = MongoClient(ip, port) # 获取客户端 self._name = db_name self._db = self._client[db_name] # 检查数据库是否连接 try: self._client.server_info() print("已连接到 {}:{} - {}".format(ip, port, db_name)) except errors.ServerSelectionTimeoutError: print("连接数据库失败") exit() self._flights = self._db.flights self._airports = self._db.airports self._adj = self._db.adj def add_airport(self, airport): """如果机场不存在，则添加机场。机场ID以字符串形式给出""" if self._airports.count_documents({"id": airport}) == 0: self._airports.insert_one({"id": airport}) def add_flight(self, orig, dest, **kwargs): """添加航班到数据库""" self.add_airport(orig) self.add_airport(dest) new_flight = {"orig": orig, "dest": dest} for key, val in kwargs.items(): new_flight[key] = val return self._flights.insert_one(new_flight) def drop_database(self): """删除数据库""" self._client.drop_database(self._name) def all_flights_from(self, orig): """获取所有连接到给定出发机场的航班""" return self._flights.find({"orig": orig}) @property def all_airports(self): """返回所有机场列表""" group = self._airports.aggregate([ {"$group": { "_id": None, "ids": {"$push": "$id"} }} ]) return list(group)[0]["ids"] def add_to_adj(self, criterion, orig, dest, thru, weight, last_flight): """向邻接表中添加条目""" new_adj = { "criterion": criterion.name, "orig": orig, "dest": dest, "thru": thru, "last_flight": last_flight, "weight": float(weight) } self._adj.insert_one(new_adj) def get_adj(self, criterion, orig, dest): """获取邻接表条目""" target = {"criterion": criterion.name, "orig": orig, "dest": dest} return self._adj.find_one(target) def update_adj(self, criterion, orig, dest, thru, weight, last_flight): """更新邻接表条目""" query = { "criterion": criterion.name, "orig": orig, "dest": dest, } new_val = {"$set": { "thru": thru, "weight": float(weight), "last_flight": last_flight } } self._adj.update_one(query, new_val) def get_weight(self, criterion, orig, dest): """获取给定航班的权重""" query = { "orig": orig, "dest": dest } target = self._flights.find_one(query) return float(target[criterion.name]) # 返回权重 def clear_adj(self): """清除邻接表""" self._adj.drop() @property def all_flights(self): """返回所有航班""" return self._flights.find()总的来说，上述代码提供了一个包装器，使得用户可以轻松地连接到MongoDB数据库，并进行机场、航班以及加权邻接表的管理和操作。对上述代码的具体说明如下所示。
- 初始化数据库连接：在构造函数中，通过指定 IP 地址、端口和数据库名称，完成了 MongoDB 数据库连接的初始化。。
- 添加机场和航班信息：方法 add_airport() 和 add_flight()，用于向数据库中添加机场和航班信息。这些方法确保机场和航班的唯一性。
- 获取航班和机场信息：方法 all_flights_from() 用于获取特定起点的所有航班信息，以及 all_airports 属性用于获取所有机场信息。
- 管理加权邻接表：提供了用于管理加权邻接表的方法，包括添加、获取、更新、清空邻接表中的条目。
- 获取航班权重：方法 get_weight() 用于获取给定航班的权重，例如价格、持续时间或距离。
- 清除数据库和邻接表：提供了方法用于清除整个数据库或特定集合（邻接表）的内容。
（2）文件scale_down_csv.py实现了一个CSV文件处理工具，通过读取输入的CSV文件，根据指定条件进行数据筛选和转换，并生成新的CSV文件。能够根据用户需求对数据进行削减和抽样，同时还能根据距离计算价格，并将结果写入输出文件。用户可以通过交互式输入来控制处理过程，包括输入文件路径、读取间隔以及是否紧密连接。
import csv import random COST_FACTOR = 3.75 #CUTDOWN_INDEX = 50000 def convert(in_name, out_name, top_airlines, cutdown, closely): """准备给定的csv以便导入，通过削减和添加价格""" data = {} with open(in_name, 'r') as routedata: routereader = csv.DictReader(routedata) for row in routereader: orig = row["ORIGIN"] dest = row["DEST"] # 如果 closely 为真，并且 orig 或 dest 不在 top_airports 中，则大部分数据来源于给定的顶级机场 if closely and (orig not in top_airports or dest not in top_airports): x = random.randint(0, cutdown) if x != 0: continue airline = row["OP_UNIQUE_CARRIER"] # 如果航空公司不在 top_airlines 中，则跳过 # if airline not in top_airlines: no = row["OP_CARRIER_FL_NUM"] duration = int(float(row["CRS_ELAPSED_TIME"])) distance = int(float(row["DISTANCE"])) price = round(price_cal(distance), 2) # 计算示例价格 flightinfo = [airline, no, orig, dest, duration, distance, price] if (airline, no) not in data: # 忽略重复项 data[(airline, no)] = flightinfo routedata.close() # 写入新的CSV with open(out_name, 'w') as outputcsv: csvwriter = csv.writer(outputcsv) csvwriter.writerow(["OP_UNIQUE_CARRIER", "OP_CARRIER_FL_NUM", "ORIGIN", "DEST", "CRS_ELAPSED_TIME", "DISTANCE", "PRICE"]) flights = list(data.values()) count = 0 # 计数条目 for i in range(0, len(flights), cutdown): csvwriter.writerow(flights[i]) count += 1 outputcsv.close() print("成功将数据转换为 {}. 大小: {}".format(out_name, count)) def price_cal(distance): """从距离计算随机价格""" return distance/COST_FACTOR * random.randint(5, 30)/10 if __name__ == '__main__': top_airports = ["LAX", "ORD", "DFW", "JFK", "SFO", "BOS", "LAS", "SEA", "MIA"] top_airlines = [] in_file = input("输入文件路径: ") cut = input("读取每 __ 条目: ") closely = True if input("紧密连接? (Y/N) ").upper() == "Y" else False out_file = input("输出文件路径: ") convert(in_file, out_file, top_airlines, int(cut), closely)（3）文件db_test.py是用于测试数据库类(Database)的单元测试文件，其中的每个测试方法都针对数据库类中不同的功能进行测试。测试包括添加机场、添加航班、防止重复添加机场、自动添加机场等功能。通过这些单元测试，可以确保数据库类的各项功能能够正常运行，并且在不同情况下能够正确处理数据，保证了代码的稳健性和可靠性。
from .context import Database # 设置IP地址 IP = "localhost" # 设置端口号 PORT = 27017 # 设置数据库名称 DB_NAME = "connecting_flight_test" def setup(): # 初始化数据库对象 db = Database(IP, PORT, DB_NAME) # 删除测试数据库 db.drop_database() # 返回初始化后的数据库对象 return db def test_add_airport(): # 设置测试环境 db = setup() # 定义机场ID id = "test" # 向数据库添加机场 db.add_airport(id) # 在_airports集合中查找机场 result = db._airports.find_one({"id": id}) # 断言机场ID存在于结果中 assert result["id"] == id def test_add_duplicate_airport(): # 设置测试环境 db = setup() # 定义机场ID id = "test" # 向数据库添加重复机场 db.add_airport(id) db.add_airport(id) db.add_airport(id) # 在_airports集合中统计机场ID数量 result = db._airports.count_documents({"id": id}) # 断言只存在一个机场ID assert result == 1 def test_add_flight(): # 设置测试环境 db = setup() # 定义航班信息 airline = "A" no = 123 orig = "SFO" dest = "LAX" price = 200 distance = 5000 # 向数据库添加航班 db.add_flight(orig, dest, airline=airline, no=no, price=price, distance=distance) # 在_flights集合中统计符合条件的文档数量 result = db._flights.count_documents( {"orig": orig, "dest": dest, "no": no, "price": price, "distance": distance }) # 断言只存在一个符合条件的文档 assert result == 1 def test_auto_add_airport(): # 设置测试环境 db = setup() # 定义航班信息 airline = "A" no = 123 orig = "SFO" dest = "LAX" price = 200 distance = 5000 # 向数据库添加航班，自动添加机场 db.add_flight(orig, dest, airline=airline, no=no, price=price, distance=distance) # 统计机场ID数量 result1 = db._airports.count_documents({"id": orig}) result2 = db._airports.count_documents({"id": dest}) # 断言起点和终点机场都存在于数据库中 assert result1 == 1 and result2 == 1
