python自動監控北京醫院預約掛號平臺是否有號的提醒程序
[重要通告]如您遇疑難雜癥,本站支持知識付費業務,掃右邊二維碼加博主微信,可節省您寶貴時間哦!
最近在找醫院掛號,發現各種掛號好困難,實數沒辦法,就找了一些監控有號的代碼,然后特此記錄一下;簡單寫了一個 0.1版的自動刷新監控固定科室的程序
from time import sleep
from selenium import webdriver
import smtplib
from email.mime.text import MIMEText
from email.utils import formataddrmy_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 retdriver = 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 MozillaCookieJarheaders = {
'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 textasync 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 Falseasync 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 clientasync 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號又微信號)右上方掃一掃可加博主微信
所寫所說,是心之所感,思之所悟,行之所得;文當無敷衍,落筆求簡潔。 以所舍,求所獲;有所依,方所成!