祝各位端午节安康！只要心中无结，每天都是节，开心最重要！
在上一篇文章高考分数查询结果自动推送至微信（卷Ⅰ）-CSDN博客中谈了思路，今天具体实现。文中将敏感信息已做处理，读者根据自己的实际情况替换。
主要逻辑：
- 轮询接口列表；
- 提取IP和端口；
- 替换查询接口的IP和端口；
- 替换验证码接口的IP和端口；
- 检测可达性；
- 成功查询后保存结果并推送消息，跳出循环。
某省去年放出的查询接口主页
主页源码中与查询入口有关的网页源码（部分）
<ul class="cfrk">
<li> <a href="http://xxx.xxx.xxx.xxx:81/n_score/index.jsp" class="gkcj-btn">入口一</a></li>
<li> <a href="http://xxx.xxx.xxx.xxx:82/n_score/index.jsp" class="gkcj-btn">入口二</a></li>
<li><a href="http://xxx.xxx.xxx.xxx:81/n_score/index.jsp" class="gkcj-btn">入口三</a></li>
<li> <a href="http://xxx.xxx.xxx.xxx:83/n_score/index.jsp" class="gkcj-btn">入口四</a></li>
<li><a href="http://xxx.xxx.xxx.xxx:81/n_score/index.jsp" class="gkcj-btn">入口五</a> </li>
中间省略若干行......
<li> <a href="http://xxx.xxx.xxx.xxx:86/n_score/index.jsp" class="gkcj-btn">入口十六</a></li>
</ul>我们从上面接口列表中提取IP和端口，检测可达性，自动查询分数，然后推送微信。
完整程序
# coding=utf-8
import requests
from bs4 import BeautifulSoup
import time
import itertools
import json
import random
import ddddocr
import onnxruntime
import schedule
# 设置日志级别用于去除ddddocr广告
onnxruntime.set_default_logger_severity(3)
# 你的 Server 酱 SCKEY
SERVER_CHAN_SCKEY = 'YOUR_SERVER_CHAN_SCKEY' # 替换成你的 SCKEY
def imgRecognition(img):
"""
识别验证码图片内容
:param img: 图片文件路径
:return: 返回识别结果字符串，如果失败则返回None
"""
try:
ocr = ddddocr.DdddOcr()
with open(img, 'rb') as f:
img_bytes = f.read()
res = ocr.classification(img_bytes)
return res
except Exception as e:
print(f"OCR failed: {e}")
return None
def getVerifyimagepage(session, headers, cookies, captcha_url):
"""
获取验证码图片并保存为 img.jpg
:param session: requests.Session 对象
:param headers: 请求头信息
:param cookies: 请求时带的Cookie
:param captcha_url: 验证码URL
:return: 无
"""
time.sleep(random.random())
t = random.uniform(0.0, 1.0)
params = {
'rnd': t,
}
try:
res = session.get(captcha_url, headers=headers, params=params, cookies=cookies, verify=False)
with open('img.jpg', 'wb') as img:
img.write(res.content)
except Exception as e:
print(f"Failed to get image: {e}")
def get_Kaptcha(session, headers, cookies, captcha_url):
"""
获取验证码图片并进行识别
:param session: requests.Session 对象
:param headers: 请求头信息
:param cookies: 请求时带的Cookie
:param captcha_url: 验证码URL
:return: 返回识别结果字符串，如果失败则返回None
"""
getVerifyimagepage(session, headers, cookies, captcha_url)
return imgRecognition('img.jpg')
def get_query_endpoints(session, base_url):
"""
获取包含查询接口的URL列表
:param session: requests.Session 对象
:param base_url: 主页URL
:return: 返回查询接口URL列表，如果获取失败则返回空列表
"""
try:
response = session.get(base_url)
response.raise_for_status()
html_content = response.text
soup = BeautifulSoup(html_content, 'html.parser')
ul_tag = soup.find('ul', class_='cfrk')
if ul_tag:
return [a['href'] for a in ul_tag.find_all('a')]
else:
print("No query endpoints found.")
return []
except requests.RequestException as e:
print(f"Failed to get query endpoints: {e}")
return []
def extract_ip_port(url):
"""
从URL中提取IP和端口部分
:param url: 完整URL
:return: 返回IP和端口字符串，如果提取失败则返回None
"""
parts = url.split('/')
if len(parts) > 2:
return parts[2]
return None
def ping_website(session, full_url):
"""
检查网站的可达性
:param session: requests.Session 对象
:param full_url: 完整网站URL
:return: 网站可达返回True，否则返回False
"""
try:
response = session.head(full_url)
return response.status_code == 200
except requests.RequestException as e:
print(f"Request failed: {e}")
return False
def send_to_wechat(message):
"""
通过 Server 酱发送消息到微信
:param message: 要发送的消息文本
:return: 返回响应的JSON数据，如果发送失败则返回None
"""
url = f'https://sc.ftqq.com/{SERVER_CHAN_SCKEY}.send'
data = {
'text': '查询结果通知',
'desp': message
}
try:
response = requests.post(url, data=data)
response.raise_for_status()
return response.json()
except requests.RequestException as e:
print(f"Failed to send message: {e}")
return None
def perform_query(session, headers, cookies, query_url, idNumber, ticketNumber, captcha_url):
"""
执行具体的查询任务
:param session: requests.Session 对象
:param headers: 请求头信息
:param cookies: 请求时带的Cookie
:param query_url: 查询接口URL
:param idNumber: 身份证号
:param ticketNumber: 准考证号
:param captcha_url: 验证码URL
:return: 返回查询结果的JSON数据，如果查询失败则返回None
"""
randCode = get_Kaptcha(session, headers, cookies, captcha_url)
if randCode is None:
print("Failed to get CAPTCHA code.")
return None
query_full_url = f'{query_url}?idNumber={idNumber}&ticketNumber={ticketNumber}&randCode={randCode}'
try:
response = session.get(query_full_url, headers=headers, cookies=cookies)
response.raise_for_status()
return response.json()
except requests.RequestException as e:
print(f"Query failed: {e}")
return None
terminate = False # 终止程序标志
def job():
"""
每分钟执行一次的任务，用于进行查询并处理结果
:return: 无，但当查询成功并发送消息后，会返回return schedule.CancelJob取消定时任务
"""
global terminate
for full_url in url_cycle:
ip_port = extract_ip_port(full_url)
if ip_port:
base_query_url = base_query_url_template.format(ip_port)
captcha_url = base_captcha_url_template.format(ip_port)
if ping_website(session, full_url):
print(f"{full_url} is reachable")
# 动态设置 headers 中的 Referer 和 Host
headers["Referer"] = f"http://{ip_port}/n_score/"
headers["Host"] = ip_port
result = perform_query(session, headers, cookies, base_query_url, idNumber, ticketNumber, captcha_url)
if result:
print("Query success:", result)
score_text = json.dumps(result, ensure_ascii=False, indent=4)
json_path = 'result.json'
with open(json_path, 'w') as json_file:
json.dump(result, json_file, ensure_ascii=False, indent=4)
response = send_to_wechat(score_text)
if response and response.get('message') == '':
print("信息发送成功")
terminate = True
return schedule.CancelJob # 成功后返回取消任务
else:
print("发送信息失败！")
else:
print("查询失败")
break # 无论成功还是失败，都跳出循环
else:
print(f"{full_url} 暂时不能访问！")
else:
print(f"IP、端口错误 {full_url}")
# 初始化配置
session = requests.Session()
headers = {
"Accept": "image/avif,image/webp,image/apng,image/svg+xml,image/*,*/*;q=0.8",
"Accept-Encoding": "gzip, deflate",
"Accept-Language": "zh-CN,zh;q=0.9",
"Cache-Control": "no-cache",
"Pragma": "no-cache",
"Proxy-Connection": "keep-alive",
"Referer": "http://xxx.xxx.xxx.xxx:81/n_score/", # 此处应与查询的IP和端口一致
"Host": "xxx.xxx.xxx.xxx:81", # 此处应与查询的IP和端口一致
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/107.0.0.0 Safari/537.36"
}
cookies = {
"JSESSIONID": "FB32CDA83DF30554A9C01A2BE77260B6", # 实际使用时请替换
"Hm_lvt_9b3b578b0b797c6004bfd68445de9e88": "1687654663" # 时间戳可以不要
}
base_url = "https://xxx.xxx.xxx.xxx/cx/" # 查询入口主页
idNumber = "your_id_number" # 替换为实际身份证号
ticketNumber = "your_ticket_number" # 替换为实际准考证号
base_query_url_template = 'http://{}/n_score/rest/api/queryscore/snquery' # 查询接口模板
base_captcha_url_template = 'http://{}/n_score/Kaptcha.do' # 生成验证码接口模板
# 获取查询接口列表
href_list = get_query_endpoints(session, base_url)
url_cycle = itertools.cycle(href_list) # 将查询接口用于循环迭代
# 使用 schedule 模块每分钟执行一次 job
schedule.every(1).minutes.do(job)
while not terminate:
schedule.run_pending()
time.sleep(1)
print("程序结束")几点说明：
1、在Server酱中传送json或text
# 假定得到的结果为字典
text = {"querytime": "查询时间：2022-09-22 12:46:47", "totalScore": " 433"}
# 将结果转为json
json_text = json.dumps(text, ensure_ascii=False)
data = {
'title': '查询结果通知',
'desp': json_text
}
headers = {'Content-Type': 'application/json'} # 在头中设置Content-Type 为 json
response = requests.post("https://sctapi.ftqq.com/你自己的key.send?title={}", data=data, headers=headers)
#############################################################################################
# 如果结果为table, 可将table转为列表，再转成文格传送
# 提取表格数据
table_data = []
for table in soup.find_all('table'):
for row in table.find_all('tr'):
cols = row.find_all(['td'])
cols = [ele.text.strip() for ele in cols]
table_data.append(cols)
# 将列表中的元素都转换为字符串类型
str_list = [str(item) for item in table_data]
# 使用 join() 方法将列表转换为文本
text = ', '.join(str_list)
data = {
'title': '查询结果通知',
'desp': text
}
response = requests.post("https://sctapi.ftqq.com/你自己的key.send?title={}", data=data)2、为了确保在查询分数网页和获取验证码时使用同一个会话，我们需要确保所有与网络交互的请求都通过同一个requests.Session。
3、可以将程序部署到阿里云使用FC功能 ，在高考查分前一天触发，让其自动运行直到查询到结果。
