高考分数查询结果自动推送至微信（卷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' # 替换成你的 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功能 ，在高考查分前一天触发，让其自动运行直到查询到结果。

此程序是基于我去年成功运行的版本进行修改的。目前，由于今年的查分接口尚未发布，我只能使用去年的接口列表进行开发。因此，当下的接口无法Ping通。基于此限制，我在理论上完成了整个程序，但尚未进行实际测试。

希望以后高考分数，家长和考生第一时间知晓，自动推送。家长无需费心，考生无需烦恼，无需查询，消息早已送达。此时我无需多言，只愿学子一举夺魁，金榜题名。

原文链接：https://blog.csdn.net/chinagaobo/article/details/139569873?ops_request_misc=%257B%2522request%255Fid%2522%253A%2522171910925216800178568569%2522%252C%2522scm%2522%253A%252220140713.130102334.pc%255Fblog.%2522%257D&request_id=171910925216800178568569&biz_id=0&utm_medium=distribute.pc_search_result.none-task-blog-2~blog~first_rank_ecpm_v1~times_rank-3-139569873-null-null.nonecase&utm_term=2024%E9%AB%98%E8%80%83%E6%9F%A5%E5%88%86