高考分数查询结果自动推送至微信（卷II）

祝各位端午节安康！只要心中无结，每天都是节，开心最重要！

在上一篇文章高考分数查询结果自动推送至微信（卷Ⅰ）-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'  # 替换成你的 SCKEYdef 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 resexcept Exception as e:print(f"OCR failed: {e}")return Nonedef 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.textsoup = 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 Nonedef 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 == 200except requests.RequestException as e:print(f"Request failed: {e}")return Falsedef 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 Nonedef 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 Nonequery_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 Noneterminate = False  # 终止程序标志def job():"""每分钟执行一次的任务，用于进行查询并处理结果:return: 无，但当查询成功并发送消息后，会返回return schedule.CancelJob取消定时任务"""global terminatefor 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 和 Hostheaders["Referer"] = f"http://{ip_port}/n_score/"headers["Host"] = ip_portresult = 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 = Truereturn 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)