python自动监控北京医院预约挂号平台是否有号的提醒程序

[重要通告]如您遇疑难杂症,本站支持知识付费业务,扫右边二维码加博主微信,可节省您宝贵时间哦!

最近在找医院挂号，发现各种挂号好困难，实数没办法，就找了一些监控有号的代码，然后特此记录一下；简单写了一个 0.1版的自动刷新监控固定科室的程序

from time import sleep

from selenium import webdriver

import smtplib
from email.mime.text import MIMEText
from email.utils import formataddr

my_sender = '写自己的' # 发件人邮箱账号
my_pass = '写自己的' # 发件人邮箱密码(当时申请smtp给的口令)
my_user = ['liang@qq.com', 'laoliang@163.com'] # 收件人邮箱账号，我这边发送给自己

def mail():
ret = True
try:
msg = MIMEText('监测到号了，赶紧去挂噢！小可爱', 'plain', 'utf-8')
msg['From'] = formataddr(["你的二狗子", my_sender]) # 括号里的对应发件人邮箱昵称、发件人邮箱账号
# 括号里的对应收件人邮箱昵称、收件人邮箱账号
msg['To'] = ",".join(my_user)
msg['Subject'] = "监测到号了，赶紧去挂噢！小可爱" # 邮件的主题，也可以说是标题

server = smtplib.SMTP_SSL("smtp.qq.com", 465) # 发件人邮箱中的SMTP服务器，端口是465
server.login(my_sender, my_pass) # 括号中对应的是发件人邮箱账号、邮箱密码
# 括号中对应的是发件人邮箱账号、收件人邮箱账号、发送邮件
server.sendmail(my_sender, my_user, msg.as_string())
server.quit() # 关闭连接
except Exception: # 如果 try 中的语句没有执行，则会执行下面的 ret=False
ret = False
return ret

driver = webdriver.Chrome() # 需要 下载 对应浏览器 驱动到 python 安装目录
driver.get("https://www.114yygh.com/hospital/142/75fec1a900e3d4c238cf384556de46de/200051666/source") # 刷新网址 填写你要监控的网站

while True:
driver.refresh() # 刷新网页
sleep(30) # 五秒一次
el = driver.find_elements_by_class_name("calendar-list-wrapper")
if(len(el) > 0):
if("有号" in el[0].text):
mail()
# divlist = driver.find_elements_by_class_name("calendar-item")
# for div in divlist: # 第一个实例
# div.click()
# try:
# button = driver.find_elements_by_class_name("v-button")
# if(len(button) > 0):
# for bt in button: # 第一个实例
# if("剩余" in bt.text):
# bt.click()
# gh = driver.find_elements_by_class_name("v-button")
# gh.click()

# except IOError:
# print("1")
# else:
# print("1")

手机验证如下：

# 2020/9/27 8:48 下午
"""
requirement:
httpx==0.15.4
pycrypto==2.6.1
"""
import os
import time
import base64
import asyncio
from datetime import datetime
from Crypto.Cipher import AES
from httpx import AsyncClient
from typing import Coroutine, Optional
from http.cookiejar import MozillaCookieJar

headers = {
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_14_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/85.0.4183.121 Safari/537.36',
'Referer': 'https://www.114yygh.com/',
'Request-Source': 'PC',
'Origin': 'https://www.114yygh.com',
'Host': 'www.114yygh.com',
'Content-Type': 'application/json;charset=UTF-8'
}

def encrypt(text: str) -> str:
"""AES加密"""
key = 'hyde2019hyde2019'
BS = AES.block_size
pad = lambda s: s + (BS - len(s) % BS) * chr(BS - len(s) % BS)
encryptor = AES.new(key, AES.MODE_ECB)
text = encryptor.encrypt(pad(text))
text = base64.b64encode(text).decode('utf8')
text = text.replace('+', '-').replace('/', '_')
return text

async def check_status(client: AsyncClient) -> bool:
"""检查用户登录状态
:param client:
:return:
"""
url = f'https://www.114yygh.com/web/user/info?_time={int(time.time() * 1000)}'
r = await client.get(url, headers=headers)
r = r.json()
if r.get('resCode') == 0 and r.get('data'):
return True
return False

async def login(username: str) -> AsyncClient:
"""手机号登录114"""
cookies = MozillaCookieJar(f'{username}.ck')
if os.path.exists(f'{username}.ck'):
cookies.load()
client = AsyncClient(cookies=cookies)
if await check_status(client):
return client
await client.get('https://www.114yygh.com/')
url = 'https://www.114yygh.com/web/common/verify-code/get?_time' \
f'={int(time.time() * 1000)}&mobile={username}&smsKey=LOGIN'
await client.get(url, headers=headers)
code = input(f'{username}手机验证码: ')
url = f'https://www.114yygh.com/web/login?_time={int(time.time() * 1000)}'
payload = {
'mobile': encrypt(username),
'code': encrypt(code)
}
res = await client.post(url, data=str(payload), headers=headers)
res = res.json()
if res.get('resCode') != 0:
raise RuntimeError(res.get('msg', ''))
cookies.save()
return client

async def check_order(client: AsyncClient, firstDeptCode: str,
hosCode: str, secondDeptCode: str) -> None:
"""检查是否有号
:param client:
:param firstDeptCode: 科室一级代码?
:param hosCode: 医院代码?
:param secondDeptCode: 科室二级代码?
:return:
"""
url = f'https://www.114yygh.com/web/product/list?_time={int(time.time() * 1000)}'
payload = {
'firstDeptCode': firstDeptCode,
'hosCode': hosCode,
'secondDeptCode': secondDeptCode,
'week': 1
}
r = await client.post(url, data=str(payload), headers=headers)
for data in r.json().get('data', {}).get('calendars', []):
if data.get('status') == 'AVAILABLE':
# TODO call mail()
pass
break
print(f'{datetime.now().strftime("%X")} success call check_order')

def listen_event(cors: Coroutine, args: tuple = (),
interval: Optional[float] = None) -> None:
"""向事件循环添加回调函数，实现定时任务
:param cors: 协程
:param args: 协程参数
:param interval: 非None时为定时事件
:return:
"""
loop = asyncio.get_event_loop()
loop.create_task(cors(*args))
if interval:
loop.call_later(interval, listen_event, cors, args, interval)

async def main():
# for m in ['手机号1', '手机号2', ]:
# client = await login(m)
# listen_event(
# check_order,
# (client, '75fec1a900e3d4c238cf384556de46de', '142', '200051666'),
# interval=30
# )
client = await login('155xxxxx609')
listen_event(
check_order,
(client, '75fec1a900e3d4c238cf384556de46de', '142', '200051666'),
interval=30
)

if __name__ == '__main__':
loop = asyncio.get_event_loop()
loop.create_task(main())
loop.run_forever()

问题未解决？付费解决问题加Q或微信 2589053300 (即Q号又微信号)右上方扫一扫可加博主微信

所写所说，是心之所感，思之所悟，行之所得；文当无敷衍，落笔求简洁。 以所舍，求所获；有所依，方所成！