生活中很多麻烦的事情，比如等人，等饭，等车，等等，很多时候没法不等，虽然习以为常，但总会想着怎么才能更便捷，更高效。今天我们用 Windows 服务、实现一个公交闹钟，让乘车更优雅，少说废话，开干
问题及分析
我明天乘坐公交车上下班，虽然很方便，但经常和班车擦肩而过，眼看着师傅头也不回的离去，那心情真是相当复(狂)杂(躁)，怎么才能不错过班车，而且不会等待太久，我找了很多方法，从撞运气，到估算，在到后来，发现了很多 app、网站可以查询实时公交信息，不过每次打开，选择路线，选择站点，查询，很麻烦，而且还得不断的关注，稍不留神就错过了
既然能从网站上查询到公交实时信息，是否可以用爬虫帮忙呢？应该没问题，然后让程序不断的跑，并且设置一个提醒时间段，比如上班时或者下班时，发现公交离站不远了，提醒自己出发，感觉挺好。
语言选用强大的 Python，为了避免忘记启动，最好做成服务，Linux 最为方便，不过得有台 Linux 主机，因为平时办公用 Windows，所以选用了做成 Windows 服务。另外，虽然也可以用计划任务中执行，但设置提醒时间段不够灵活
确定了方案，就开始行动吧
实践
只有简单的想法，没有简单的项目，将时间过程拆分为 获取到站时间、发送通知、制作服务 和 完善几个部分
获取到站时间
很多城市都有实时公交的查询网站，例如北京的北京公交集团网站 http://www.bjbus.com，可以查询实时公交信息，选择线路，形式方向，上车站点，就可以得到实时公交的信息。
在浏览其上点击 F12，打开网络选项卡，在网上上点击查询，找到查询请求
可以看到时 GET 请求，网址是：http://www.bjbus.com/home/ajax_rtbus_data.php?act=busTime&selBLine=1&selBDir=5276138694316562750&selBStop=2
请求参数含义为：
- act：查询类型，固定值是 busTime
- selBLine：线路，值表示线路名，例如 1 表示 1 路车
- selBDir：行驶方向，值比较复杂，需要通过实际查询获得
- selBStop：上车站点，值为线路在形式方向上的序号，从 1 开始，例如 2 表示第二站
用 httpx， 测试一下
httpx 是基于经典库 requests 实现的，接口更简洁高效，通过
pip install httpx安装
在 python 环境下，执行
>>> import httpx >>> url = "http://www.bjbus.com/home/ajax_rtbus_data.php?act=busTime&selBLine=1&selBDir=5276138694316562750&selBStop=2" >>> r = httpx.get(url) >>> print(r.status_code) 200 >>> print(r.text) '{"html":"<div class=\\"inquiry_header\\"><div class=\\"left fixed\\"> ...
httpx.get 可以发送一个 GET 请求，返回响应对象，status_code 为请求状态编码，text 为响应内容
可以看到，返回的是 JSON 格式数据，通过 httpx 响应对象的 json 方法，可以知道将结果转换为 Python 的词典对象：
>>> ret.json() {'html': '<div class="inquiry_header"><div class="left fixed"><h3 id="lh">1路</h3>< ...
分析返回的结果，发现在开始部分，就有较为详细的公交实时信息，例如：
<p>最近一辆车距离此还有 3 站， <span>589</span> 米，预计到站时间 <span>1</span> 分钟</p>
如果没有车辆信息为：
<p>车辆均已过站</p>
所以只要提取到预计到站时间数值就可以了
利用 BeautifulSoup 对 html 解析，得到到站时间：
import httpx from bs4 import BeautifulSoup as bs4 url = "http://www.bjbus.com/home/ajax_rtbus_data.php?act=busTime&selBLine=1&selBDir=5276138694316562750&selBStop=2" r = httpx.get(url).json() b = bs4(r.get('html'), 'html.parser') info = b.find('article') i = info.find_all('p')[1] ret = re.search(r'\d+(?=\s分钟)', i.text)
BeautifulSoup 可以通过
pip install beautifulsoup4来安装
- 引入 BeautifulSoup 库，起个别名 bs4
- 获取请求响应，并转换为词典对象
- 提取词典中的 html 属性，将其转为 BeautifulSoup 对象
b，使用 Python 自带的html.parser解析器，其他解析器可能需要安装 - 通过分析 html 内容，可知有效信息在 article 标签中，通过 find 来获取只包含 article 标签的 BeautifulSoup 对象
info - 将
info中的 p 标签提取出来，其中第 2 个（列表第一个元素索引为 0 ）元素，就是需要提取的内容，放入i - 从
i中的文本中，利用正则表达式，提取车辆到达分钟数，正则表达式的意思是：匹配有一个或者多个数字组成的后面是空格和字符分钟的数字部分 - 匹配到，返回车辆到达分钟数，返回 None，表示车辆还未发车
上述方法实际上就是一个简单爬虫，反复执行，直到发现合适的时间，发出提醒，就完成了核心任务
发送通知
发送通知有多种方法，例如 Windows 下弹窗或者消息，不过在实践中遇到不少困难，所以选用了邮件通知，不仅实现简单，如果在手机上配置了邮件客户端，收到邮件会给出提醒，更加方便
用 Python 很容易发邮件，直接看代码:
import smtplib from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText msg = MIMEMultipart('alternative') # 实例化email对象 msg['from'] = 'tom@example.com' # 对应发件人邮箱昵称、发件人邮箱账号 msg['to'] = ';'.join(['lily@example.com']) # 对应收件人邮箱昵称、收件人邮箱账号 msg['subject'] = '你好' # 邮件的主题 msg.attach(MIMEText('你好，很高兴认识你...', 'html')) # 附加正文 SMTP_SERVER = 'smtp.example.com' # 邮箱服务器 SSL_PORT = '465' # 端口 USER_NAME = 'username' # 邮箱用户名 USER_PWD = 'password' # 密码 smtp = smtplib.SMTP_SSL(SMTP_SERVER, SSL_PORT) # 邮件服务器地址和端口 smtp.ehlo() # 用户认证 smtp.login(USER_NAME, USER_PWD) # 括号中对应的是发件人邮箱账号、邮箱密码 smtp.sendmail(FROM_MAIL, TO_MAIL, str(msg)) # 收件人邮箱账号、发送邮件 smtp.quit() # 等同 smtp.close() ,关闭连接
例如，我收到的一个邮件通知：
制作服务
万事俱备，只欠东风，接下了，需要将脚本做成可执行程序，注册为 Windows 服务
Windows 服务脚本
用 Python 写 需要借助于 win32api 库
安装 win32api 库
pip install pywin32
这是服务脚本框架:
import win32api import win32event import win32service import win32serviceutil import servicemanager class MyService(win32serviceutil.ServiceFramework): _svc_name_ = "服务名称" _svc_display_name_ = "在服务列表中显示的名称" _svc_description_ = "服务描述" def __init__(self, args): win32serviceutil.ServiceFramework.__init__(self, args) self.stop_event = win32event.CreateEvent(None, 0, 0, None) def SvcDoRun(self): self.ReportServiceStatus(win32service.SERVICE_START_PENDING) # 这里写服务启动后的业务逻辑 win32event.WaitForSingleObject(self.stop_event, win32event.INFINITE) def SvcStop(self): self.ReportServiceStatus(win32service.SERVICE_STOP_PENDING) # 这里写服务即将停止时的业务逻辑 win32event.SetEvent(self.stop_event) self.ReportServiceStatus(win32service.SERVICE_STOPPED) if __name__ == "__main__": if len(sys.argv) == 1: servicemanager.Initialize() servicemanager.PrepareToHostSingle(MyService) servicemanager.StartServiceCtrlDispatcher() else: win32serviceutil.HandleCommandLine(MyService)
- 引入服务相关包
- 定义继承自
win32serviceutil.ServiceFramework的服务类 _svc_name_,_svc_display_name_,_svc_description_为服务声明属性- 服务类初始化
__init__方法中, 定义服务停止事件，实际应用中可以初始化业务相关的属性 SvcDoRun为服务启动时的回调方法，可以写服务执行中的处理逻辑SvcStop为服务结束时的回调方法，可以写服务结束时的处理逻辑ReportServiceStatus为服务状态通知方法，以便服务管理器及时获取到服务状态- 当脚本执行时，如果没有参数，表示服务在启动，如果有参数，将运行服务管理方法，例如 install(安装)、start（启动）等
打包
写好服务代码之后，需要将其打包为 EXE
虽然 上述服务脚本可以不用打包为 EXE 也能注册为服务，但是常常会因为环境、组件引用问题导致注册的服务失败
Pyinstaller 工具可以将 Python 脚本打包成 Windows 的可执行文件
安装:
pip install pyinstaller
然后可以在命令行中直接使用，例如将 service.py 打包为 EXE:
pyinstaller service.py
打包过程较慢，会有大量信息输出，如果没有报错信息，即为打包成功。
打包成后，会在脚步所在目录创建 build 和 dist 目录，dist 目录下会有打包好的 EXE，名称与脚本名一样
注意:打包好的程序，注册服务后，启动时可能会报 win32timezone 找不到的错误，这时需要加一个参数: --hiddenimport win32timezone 打包命令换成：
pyinstaller --hiddenimport win32timezone -F service.py重新打包即可
注册服务
做好了可执行文件，就可以注册为服务了
首先需要用管理员权限运行命令行
注册服务
service.exe install
注册后，可以在计算机管理的服务，或者从任务管理的服务列表中看到，名称为 脚本中 服务类 _svc_display_name_ 所定义的名称
- 启动服务
service.exe start
- 也可以在服务列表中启动
- 停止服务
service.exe stop
- 也可以在服务列表中停止
- 注销服务
service.exe remove
- 注销服务时，需要先停止服务，不然会有个服务尸体在服务列表中
除了通过 install、start 参数管理服务外，还可以使用 Windows 命令 sc 来操作，有兴趣可以了解下
如果启动服务报错，可以在 Windows 的事件管理器中查看错误日志，以便得到详细信息：
完善
从构建公交实时信息爬虫，到启动 Windows 服务，主要的工作已经完成了，整体跑一遍，至少可以确定方案是可行。
如果要实际应用，还有很多细节问题需要处理
设置提醒时间段
作为 Windows 服务运行的话，会长时间处于运行状态，公交提醒功能，只需要在特定事件段有效就行，所以需要判断当前时间是否进入提醒时间窗口， onTime 方法可以做到这一点：
import datetime def onTime(begin, end): d_time = datetime.datetime.strptime( str(datetime.datetime.now().date())+begin, '%Y-%m-%d%H:%M') d_time1 = datetime.datetime.strptime( str(datetime.datetime.now().date())+end, '%Y-%m-%d%H:%M') n_time = datetime.datetime.now() if n_time > d_time and n_time < d_time1: return True else: return False
- 引入 datetime 包
- 方法接受两个参数，开始时间和结束时间，例如"18:00"，"18:30"
- 如果当前时间段在开始时间和结束时间之内，返回
True，否则返回False
在服务的启动方法中，写一个循环，每次循环判断一下当前时间，如果 onTime 返回 True 就进入到提醒业务代码中
支持多条线路
同一路车，但是不同方向需要看成不同的线路，所以在提醒方法 run 中，需要同时对多条线路进行判断：
def run(self): for line in self.config.lines: if self.onTime(line['begin'], line['end']): if line.get('needSentMail', True): bustime = self.getBusTime(line) if bustime is not None: if int(bustime) <= int(line.get('latestLeaveMinute', self.config.latestLeaveMinute)): self.mailClient.send_mail(self.config.alertMail, '班车提醒: '+line['line'], '车辆即将到站，现在出发正当时') line['needSentMail'] = False # 发送通知后，不必再发了 else: line['needSentMail'] = True
其中 needSentMail 表示是否需要发送通知，当在时间窗口中发送了通知，就不必再发了，如果过了时间窗口，需要将其设置为需要发送
线路配置为列表：
lines = [{ "line": "835快", "dir": "5066222788346588777", "stop": "13", "begin": "08:00", "end": "08:30" }, { "line": "835快", "dir": "4997908670784162973", "stop": "3", "begin": "19:00", "end": "20:30" }]
配置
将业务相关信息写死在代码中不是个好主意，所以需要将通知邮件的配置，线路信息等写到配置中，这样如果业务发生变化时，只需修改下配置文件就可以了，我使用 json 格式的配置文件 config.json：
{ "loopWaitSeconds": 60, "spurtWaitSeconds": 10, "latestLeaveMinute": 5, "mailConfig": { "FROM": "tom@example.com", "HOST": "smtp.example.com", "PORT": "465", "USER": "tom", "PASS": "password", "SSL": true }, "alertMail": "lily@example.com", "lines": [{ "line": "835快", "dir": "5066222788346588777", "stop": "13", "begin": "08:00", "end": "08:30" }, { "line": "835快", "dir": "4997908670784162973", "stop": "3", "begin": "19:00", "end": "20:30" }] }
配置字段比较简单，下面这些需要解释下:
- loopWaitSeconds: 空循环时的等待秒数
- spurtWaitSeconds: 进入提醒时间窗口的等待秒数
- latestLeaveMinute: 可以出发的时间，即具车辆到站还有多久时，需要发出通知
Python 可以方便的读取 json 配置文件，读取之后，将其转换为一个类，在代码中使用更方便:
class Config: def __init__(self, config): self.loopWaitSeconds = config.get("loopWaitSeconds", 60) self.spurtWaitSeconds = config.get("spurtWaitSeconds", 10) self.mailConfig = config.get("mailConfig", None) self.latestLeaveMinute = config.get("latestLeaveTime", 5) self.lines = config.get("lines", {}) import json with open(r"C:\config.json", "r", encoding='UTF-8') as config_file: config = Config(json.load(config_file)) # 整体配置
这里需要注意的是，配置文件的位置，当程序以服务的形式运行时，当前路径是个临时目录，因此写绝对路径比较方便
总结
虽然解决了问题，不过想要同很多 app 那样优雅，还需要做很多工作。通过这次实践，可以了解了 Python 打包，Windows 服务，简单爬虫，邮件发送 等功能，为日后做其他应用奠定了基础，比如基于这个框架，可以做一个打卡签到功能，让自己更自由
很多时候舒适让我们懒于行动，而行动带来的惊喜远胜过一时的安逸……
感谢阅读，代码示例中有较为完整的代码，欢迎参考研究
