当前位置：首页 > news >正文

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

news 来源：原创 2025/5/6 15:26:49

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

在上一篇文章高考分数查询结果自动推送至微信（卷Ⅰ）-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)