python自動監(jiān)控北京醫(yī)院預約掛號平臺是否有號的提醒程序

[重要通告]如您遇疑難雜癥,本站支持知識付費業(yè)務,掃右邊二維碼加博主微信,可節(jié)省您寶貴時間哦!

最近在找醫(yī)院掛號，發(fā)現(xiàn)各種掛號好困難，實數(shù)沒辦法，就找了一些監(jiān)控有號的代碼，然后特此記錄一下；簡單寫了一個 0.1版的自動刷新監(jiān)控固定科室的程序

from time import sleep

from selenium import webdriver

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

my_sender = '寫自己的' # 發(fā)件人郵箱賬號
my_pass = '寫自己的' # 發(fā)件人郵箱密碼(當時申請smtp給的口令)
my_user = ['liang@qq.com', 'laoliang@163.com'] # 收件人郵箱賬號，我這邊發(fā)送給自己

def mail():
ret = True
try:
msg = MIMEText('監(jiān)測到號了，趕緊去掛噢！小可愛', 'plain', 'utf-8')
msg['From'] = formataddr(["你的二狗子", my_sender]) # 括號里的對應發(fā)件人郵箱昵稱、發(fā)件人郵箱賬號
# 括號里的對應收件人郵箱昵稱、收件人郵箱賬號
msg['To'] = ",".join(my_user)
msg['Subject'] = "監(jiān)測到號了，趕緊去掛噢！小可愛" # 郵件的主題，也可以說是標題

server = smtplib.SMTP_SSL("smtp.qq.com", 465) # 發(fā)件人郵箱中的SMTP服務器，端口是465
server.login(my_sender, my_pass) # 括號中對應的是發(fā)件人郵箱賬號、郵箱密碼
# 括號中對應的是發(fā)件人郵箱賬號、收件人郵箱賬號、發(fā)送郵件
server.sendmail(my_sender, my_user, msg.as_string())
server.quit() # 關閉連接
except Exception: # 如果 try 中的語句沒有執(zhí)行，則會執(zhí)行下面的 ret=False
ret = False
return ret

driver = webdriver.Chrome() # 需要 下載 對應瀏覽器 驅(qū)動到 python 安裝目錄
driver.get("https://www.114yygh.com/hospital/142/75fec1a900e3d4c238cf384556de46de/200051666/source") # 刷新網(wǎng)址 填寫你要監(jiān)控的網(wǎng)站

while True:
driver.refresh() # 刷新網(wǎng)頁
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:
"""檢查用戶登錄狀態(tài)
: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: 醫(yī)院代碼?
: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:
"""向事件循環(huán)添加回調(diào)函數(shù)，實現(xiàn)定時任務
:param cors: 協(xié)程
:param args: 協(xié)程參數(shù)
: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號又微信號)右上方掃一掃可加博主微信

所寫所說，是心之所感，思之所悟，行之所得；文當無敷衍，落筆求簡潔。 以所舍，求所獲；有所依，方所成！