Python实现命令行监控北京实时公交之一

摘要：开头先放上效果展示在命令行输入，显示基本信息，如下在命令行输入，显示北京路公交车的位置，如下红色的说明在梅园附近有一辆公交车，并且会不断刷新。

在命令行输入 python bus.py -i，显示app基本信息，如下：

在命令行输入 python bus.py 438，显示北京438路公交车的位置，如下：

红色的B说明在梅园附近有一辆438公交车，并且会不断刷新。

GitHub地址 - https://github.com/Harpsichor...

作为一个在北京西二旗郊区上班的苦逼，不敢太早下班，怕领导心里不满，又不敢走太晚，不然赶不上末班公交车了，虽然加班打车能报销，但打不着车！因此实时公交成立必备神器。

目前用的主要两个查公交的途径是车来了微信小程序和北京公交微信公众号，经过用了一段时间发现北京公交的结果是更准的，但是用起来不方便，需要点击多次才能看到结果，如图：

由于想写一个监控公交车的小程序，车快到了能自动提醒。

经过在北京公交官网的一番搜索、分析，发现下面两个可以使用的URL：

http://www.bjbus.com/home/ind...
北京公交官网首页，从这里可以获取所有的公交车编号

http://www.bjbus.com/home/aja...
AJAX接口，获取指定公交车的路线、站名、目前公交车的位置

先看第一个，是官网首页，使用requests去获取，返回的是整个页面的html, 公交车的编号在图中显示的这个dd标签中：

我们可以使用正则表达式结合一些字符串操作，将所有公交车编号获取到一个list中，代码如下：

index_url = r"http://www.bjbus.com/home/index.php"

def get_bus_code_list():
    with open("db/bus.txt", "r", encoding="utf-8") as f:
        db_data = json.loads(f.read())
        if db_data["time"] >= time() - 12*3600:
            print("Getting bus code from db...")
            return db_data["data"]
    resp = requests.get(index_url).content.decode("utf-8")
    print("Getting bus code from web...")
    bus_code_string = re.findall("
([sS]*?)
", resp)
    bus_code_string = bus_code_string[0].strip().replace("", "")
    bus_code_list = bus_code_string.split("")[:-1]
    db_data = {
        "time": time(),
        "data": bus_code_list
    }
    with open("db/bus.txt", "w", encoding="utf-8") as f:
        f.write(json.dumps(db_data, ensure_ascii=False, indent=2))
    return bus_code_list

注意为了避免每次都要联网获取，我将数据用json.dumps保存到了bus.txt里，并设置了一个保存时间，每次请求这个数据的时候，会先从bus.txt里读取，如果数据还在有效期内（12*3600秒），则直接使用。

而如果获取公交车的实时状态，那么需要去GET请求一个这样格式的url：

http://www.bjbus.com/home/ajax_rtbus_data.php?act=busTime&selBLine=17&selBDir=5223175012989176770&selBStop=9

那么可以看到这个url有4个参数，分别是act（固定为busTime)，selBLine（表示公交车编号），selBDir（表示公交车线路的编号），selBStop（表示上车的车站），请求上面的这个url时，返回的结果是json对象，并且有个key是"html"，这个html的部分结构如下图：

首先开头是一段类似提示性的信息，然后一个ul无序列表，每一个li标签中都有一个div，其id是递增的数字或者是数字加一个m，纯数字的div还有对应的车站名，带m的则是空的，我理解带m的表示车站之间的中间区域。注意div中的i标签，有的有class和clstag这两个属性，这代表的就是公交车了，并且clstag的值表示公交车距离我们选择上车的车站selBStop=9还有多远，如果是已经过站的车，这个值则为空或-1。所以直接给出代码如下，代码解释也写在注释里：

main_url = r"http://www.bjbus.com/home/ajax_rtbus_data.php"

# 获取公交车的位置，参数为公交车编号，线路编号，上车站点的编号
def get_bus_status(bus_code, direction, station_no):
    payload = {
        "act": "busTime",
        "selBLine": bus_code,
        "selBDir": direction,
        "selBStop": station_no
    }
    # 带参数的Get方法，将返回对象json化，获取key为"html"的内容
    resp = requests.get(main_url, params=payload).json()["html"]
    print("Getting bus status from web...")
    # 这部分使用正则太复杂，因此使用BeautifulSoup解析html
    soup = BeautifulSoup(resp, "html.parser")
    # html开头的路线，并将bs的string类型转换为普通string
    path = str(soup.find(id="lm").contents[0]) 
    # html开头的提示性信息，获取上车车站的名称，路线的运营时间
    station_name, operation_time, *_ = soup.article.p.string.split("xa0")
    # tip获取html开头的提示性信息的具体文本，例如最近一辆车还有多远
    tip = ""
    for content in soup.article.p.next_sibling.contents:
        if isinstance(content, str):
            tip += content.replace("xa0", "")
        else:
            tip += content.string
    bus_position = []
    # 获取所有有公交车的标签（即有clstag这个属性的）
    for tag in soup.find_all("i", attrs={"clstag": True}):
        temp_dic = dict()
        # 获取车站的id
        station_id = tag.parent["id"]
        # 如果id不带m，说明公交车离车站较近，near_station为True
        temp_dic["near_station"] = False if "m" in station_id else True
        station_id = station_id.replace("m", "")
        temp_dic["station_id"] = station_id
        # 获取公交车离上车车站的距离，如果已过站则为-1
        temp_dic["distance"] = int(tag["clstag"]) if tag["clstag"].isdigit() else -1
        # 此时temp_dic有车站id，距离，及near_station三个属性，将其append到list
        bus_position.append(temp_dic)
    result = {
        "path": path,
        "station_name": station_name,
        "operation_time": operation_time,
        "bus_position": bus_position,   # A list of dict
        "tip": tip
    }
    # 返回的结果包含较多内容，后续按需要选取部分数据使用
    return result

刚刚我们的函数获取公交车的位置，需要公交车编号、路线编号和车站编号三个参数，在一开始我们获取了所有北京公交车的编号，并存储在bus.txt中，那么怎么获取路线的编号的呢？同样用Chrome浏览器分析北京公交官网的访问过程，可以找到这样一个链接：

http://www.bjbus.com/home/ajax_rtbus_data.php?act=getLineDirOption&selBLine=438

其返回的结果是这样的（可以试试直接用浏览器访问）：

很明显option中的value就是公交车路线的代码，因此也很容易写出一个获取公交车路线代码的函数，如下：

main_url = r"http://www.bjbus.com/home/ajax_rtbus_data.php"

def get_bus_direction(bus_code):
    # 先从文本中读取，避免重复联网访问
    with open("db/direction.txt", "r", encoding="utf-8") as f:
        db_data = json.loads(f.read())
        bus_direction = db_data.get(str(bus_code))
        if bus_direction and bus_direction["time"] >= time() - 12*3600:
            print("Getting bus direction from db...")
            return bus_direction["data"]
    payload = {
        "act": "getLineDirOption",
        "selBLine": bus_code
    }
    resp = requests.get(url=main_url, params=payload).content.decode("utf-8")
    print("Getting bus direction from web...")
    # 正则获取编号
    direction_no = re.findall("value="(d+)"", resp)
    if not direction_no:
        print("%s路公交车未找到" % str(bus_code))
        return []
    # 正则获取路线
    direction_path = re.findall(str(bus_code) + "(.*?)<", resp)
    data = []
    for j in range(2):
        direction_path_str = direction_path[j][1:-1]
        data.append([direction_no[j], direction_path_str])
    # 最新数据写入文本
    with open("db/direction.txt", "w+", encoding="utf-8") as f:
        db_data[str(bus_code)] = {
            "time": time(),
            "data": data
        }
        f.write(json.dumps(db_data, ensure_ascii=False, indent=2))
    return data

获取公交车的车站也是类似的，其url是：

http://www.bjbus.com/home/ajax_rtbus_data.php?act=getDirStationOption&selBLine=438&selBDir=5204817486192029180

其返回的结果是：

直接上代码：

def get_bus_stations(bus_code, direction):
    with open("db/station.txt", "r", encoding="utf-8") as f:
        db_data = json.loads(f.read())
        bus_station = db_data.get(str(bus_code) + "#" + str(direction))
        if bus_station and bus_station["time"] >= time() - 12 * 3600:
            print("Getting bus station from db...")
            return bus_station["data"]
    payload = {
        "act": "getDirStationOption",
        "selBLine": bus_code,
        "selBDir": direction
    }
    resp = requests.get(main_url, params=payload).content.decode("utf-8")
    print("Getting bus station from web...")
    stations = re.findall("(.*?)", resp)[1:]
    with open("db/station.txt", "w+", encoding="utf-8") as f:
        db_data[str(bus_code) + "#" + str(direction)] = {
            "time": time(),
            "data": stations
        }
        f.write(json.dumps(db_data, ensure_ascii=False, indent=2))
    return stations

至此，功能函数就都已经写好了，剩余的是实现命令行输出结果的功能，在后续文章说明。