Python 获取车票信息

时间：2021-09-13 18:29:57

提示:该代码仅供学习使用，切勿滥用！！！

先来一个git地址:https://gitee.com/wang_li/li_wang

效果图:

Python 获取车票信息

逻辑:

1.获取Json文件的内容
2.根据信息生成URL
3.获取URL的内容，根据时间进行分析，如果有票则发送邮件，如果没有，则继续监听

1.定义Json文件

内容如下:

{
  "_Nodes" : "定义起始站",
  "from_address" : "成都东",

  "_Nodes" : "定义终点站",
  "to_address" : "遂宁",

  "_Nodes" : "定义车次",
  "departure_date" : "2018-12-30",

  "_Nodes" : "定义时间检查区间",
  "start_time" : "09:00",

  "_Nodes" : "定义时间结束区间",
  "stop_time" : "12:00",

  "_Nodes" : "定义列车类型 [ D:动车 Z:直达 T:特快 K:快速 A:全部 ]",
  "type" : "A",

  "_Nodes" : "定义是否发送邮件 true为发送 false为不发送",
  "send_mail" : "false",

  "_Nodes" : "如果上述定义为true,则下面则定义邮箱的信息包括，SMTP 和 认证信息",
  "mail_smtp" : "smtp.qq.com",
  "user_name" : "to_user@qq.com",
  "user_password" : "password",
  "mail_subject" : "仅仅是为了好玩",
  "_Nodes" : "将信息发送给谁",
  "mail_to_address" : "user1@qq.com;user2@163.com",

  "_Nodes" : "有票时，提示的最大次数，当有间隔时，则重新计数 , 这里最大值为100",
  "send_mail_max" : 3,

  "_Nodes" : "刷新间隔时间,单位:秒",
  "interval_time" : 30
}

2.Python代码处理Json文件并且返回结果

代码如下:

#!/usr/bin/env python3

import requests
import json
import time
import sys
import email.mime.text
import smtplib

#解析json文件

def Dealwith_jsonfile(jsonfile) :

    return_dirct = {}
    #定义json标题
    json_title = ['from_address' , 'to_address' , 'departure_date' , 'start_time' , 'stop_time' , 'type' , 'send_mail' , 'mail_smtp' , 'user_name' , 'user_password' , 'mail_subject', 'mail_to_address' , 'send_mail_max','interval_time']

    #读取文件
    open_json_file = open(jsonfile , 'rb')

    #初始化send_mail参数
    send_mail_on = 0

    #开启try except 捕获json错误的信息
    try:
        dict_json_file = json.load(open_json_file)

        #查询是否开启发送邮件
        if 'true' == dict_json_file['send_mail'] :
            send_mail_on = 1

        # 0 为不开启，则不需要记录smtp/username/password/send_mail的信息
        if 0 == send_mail_on :
            for title_configure in json_title :
                if 'mail_smtp' == title_configure or 'user_name' == title_configure or 'user_password' == title_configure or 'send_mail_max' == title_configure or 'mail_subject' == title_configure or 'mail_to_address' == title_configure:
                    continue
                else :
                    return_dirct[title_configure] = dict_json_file[title_configure]

        else :
            for title_configure in json_title :
                return_dirct[title_configure] = dict_json_file[title_configure]

        #关闭文件
        open_json_file.close()
        print(return_dirct)
        return return_dirct

    except Exception as e:
        print("处理Json文件失败，详情如下:")
        print(e)

if __name__ == "__main__" :
     Dealwith_jsonfile("go_gome.json")

执行的结果如下:

{'from_address': '成都东', 'to_address': '遂宁', 'departure_date': '2019-01-04', 'start_time': '09:30', 'stop_time': '12:00', 'type': 'D', 'send_mail': 'true', 'mail_smtp': 'smtp.qq.com', 'user_name': '2081795560@QQ.COM', 'user_password': 'idfqdnnrfjttbjbe', 'mail_subject': '仅仅是为了好玩', 'mail_to_address': '2859413527@QQ.COM', 'send_mail_max': 2, 'interval_time': 10}

3.获取站别ID

网址:https://kyfw.12306.cn/otn/resources/js/framework/station_name.js?station_version=1.9085

具体内容如下:

Python 获取车票信息

通过分析得到，名词的后一位就是站别ID

Python获取站别ID如下:

#!/usr/bin/env python3

import requests

def Get_Address_resolution(from_address,to_address , Address_type) :

    return_station_name = []

    # 站点信息可以在12306网站上找到
    station_name_url = 'https://kyfw.12306.cn/otn/resources/js/framework/station_name.js?station_version=1.9085'

    #获取网页信息
    station_information = requests.get(station_name_url).text

    #以 | 分割
    list_station_information = station_information.split("|")

    # 获取from_address 和 to_address 地址的ID
    if 1 == Address_type :
        address_com = 1
    elif 2 == Address_type :
        address_com = -1
    else :
        return  return_station_name

    for address in from_address , to_address  :
        try:
            if list_station_information.index(address):
                return_station_name.append(
                    list_station_information[list_station_information.index(address) + address_com])
        except Exception as e:
            print(e)

    return return_station_name

if __name__ == "__main__" :
    print(Get_Address_resolution('成都东','遂宁',1))
    print(Get_Address_resolution('ICW', 'NIW', 2))

执行结果如下:

['ICW', 'NIW']
['成都东', '遂宁']

3.获取实际URL

通过分析，我们可以得知，每次请求网页，实际上实在请求一个数据包，例如:

https://kyfw.12306.cn/otn/leftTicket/queryZ?leftTicketDTO.train_date=2019-01-05&leftTicketDTO.from_station=BJP&leftTicketDTO.to_station=SHH&purpose_codes=ADULT

我们只需要将数据拼接上就可以了

Python 获取车票信息

代码如下:

#!/usr/bin/env python3

def Generate_Url () :
    try:
        station_code = ['ICW', 'NIW']
        #获取最新的地址解析时间(此信息也是固定的)
        return_url = "https://kyfw.12306.cn/otn/leftTicket/queryA?leftTicketDTO.train_date=%s&leftTicketDTO.from_station=%s&leftTicketDTO.to_station=%s&purpose_codes=ADULT" \
                     %('2019-01-03' ,station_code[0] , station_code[1] )
        return return_url
    except Exception as e:
        print("生成URL失败")
        print(e)

if __name__ == "__main__" :
    print(Generate_Url())
https://kyfw.12306.cn/otn/leftTicket/queryA?leftTicketDTO.train_date=2019-01-03&leftTicketDTO.from_station=ICW&leftTicketDTO.to_station=NIW&purpose_codes=ADULT

综合代码

#!/usr/bin/env python3

import requests
import json
import time
import sys
import email.mime.text
import smtplib

#解析json文件

def Dealwith_jsonfile(jsonfile) :

    return_dirct = {}
    #定义json标题
    json_title = ['from_address' , 'to_address' , 'departure_date' , 'start_time' , 'stop_time' , 'type' , 'send_mail' , 'mail_smtp' , 'user_name' , 'user_password' , 'mail_subject', 'mail_to_address' , 'send_mail_max','interval_time']

    #读取文件
    open_json_file = open(jsonfile , 'rb')

    #初始化send_mail参数
    send_mail_on = 0

    #开启try except 捕获json错误的信息
    try:
        dict_json_file = json.load(open_json_file)

        #查询是否开启发送邮件
        if 'true' == dict_json_file['send_mail'] :
            send_mail_on = 1

        # 0 为不开启，则不需要记录smtp/username/password/send_mail的信息
        if 0 == send_mail_on :
            for title_configure in json_title :
                if 'mail_smtp' == title_configure or 'user_name' == title_configure or 'user_password' == title_configure or 'send_mail_max' == title_configure or 'mail_subject' == title_configure or 'mail_to_address' == title_configure:
                    continue
                else :
                    return_dirct[title_configure] = dict_json_file[title_configure]

        else :
            for title_configure in json_title :
                return_dirct[title_configure] = dict_json_file[title_configure]

        #关闭文件
        open_json_file.close()
        return return_dirct

    except Exception as e:
        print("处理Json文件失败，详情如下:")
        print(e)

#抓取城市名称对应的ID
def Get_Address_resolution(from_address,to_address , Address_type) :

    #定义返回列表
    return_station_name = []

    # 站点信息可以在12306网站上找到
    station_name_url = 'https://kyfw.12306.cn/otn/resources/js/framework/station_name.js?station_version=1.9085'

    #获取网页信息
    station_information = requests.get(station_name_url).text

    #以 | 分割
    list_station_information = station_information.split("|")

    # 获取from_address 和 to_address 地址的ID
    if 1 == Address_type :
        address_com = 1
    elif 2 == Address_type :
        address_com = -1
    else :
        return  return_station_name

    for address in from_address , to_address  :
        try:
            if list_station_information.index(address):
                return_station_name.append(
                    list_station_information[list_station_information.index(address) + address_com])
        except Exception as e:
            print(e)

    return return_station_name

    #     try:
    #         if list_station_information.index(from_address) :
    #             return_station_name.append(list_station_information[list_station_information.index(from_address) + 1])
    #     except Exception as e:
    #         print(e)
    #
    #     try:
    #         if list_station_information.index(to_address):
    #             return_station_name.append(list_station_information[list_station_information.index(to_address) + 1])
    #     except Exception as e:
    #         print(e)
    #
    #     #将ID返回回去
    #     return return_station_name
    #
    # #代码和上述一致，可以合并为到一块
    # elif 2 == Address_type :
    #     try:
    #         if list_station_information.index(from_address) :
    #             return_station_name.append(list_station_information[list_station_information.index(from_address) - 1])
    #     except Exception as e:
    #         print(e)
    #
    #     try:
    #         if list_station_information.index(to_address):
    #             return_station_name.append(list_station_information[list_station_information.index(to_address) - 1])
    #     except Exception as e:
    #         print(e)
    #
    #     #将ID返回回去
    #     return return_station_name

#生成URL连接
def Generate_Url (operation_parameters) :

    try:
        station_code = Get_Address_resolution(operation_parameters['from_address'],operation_parameters['to_address'],1)
        #获取最新的地址解析时间(此信息也是固定的)
        return_url = "https://kyfw.12306.cn/otn/leftTicket/queryA?leftTicketDTO.train_date=%s&leftTicketDTO.from_station=%s&leftTicketDTO.to_station=%s&purpose_codes=ADULT" \
                     %(operation_parameters['departure_date'] ,station_code[0] , station_code[1] )
        return return_url
    except Exception as e:
        print("生成URL失败")
        print(e)

#爬取是否有票
def Crawling_Informations (url , operation_parameters) :

    #初始化检查失败参数
    check_failed = 0

    #初始化票send-mail次数
    send_mail_number = {}

    #定义无限循环
    while True:
        if check_failed > 10 :
            print("连续10次抓取失败，程序退出")
            break

        try:
            #获取内容并且转换为json
            crawling_text = requests.get(url).text
            crawling_json = json.loads(crawling_text)

            #定义空的列表用于存放票务信息
            checkmessageall = []

            #便利json文件的['data']['result']信息
            for crawling_list in crawling_json['data']['result'] :
                checkmessage = []

                #将票务信息按 | 分割
                for split_list_json in crawling_list.split('|'):
                    #排除长度问20字符的行
                    if len(split_list_json) > 20:
                        pass
                    else:
                        #将其他值压进checkmessage
                        checkmessage.append(split_list_json)

                #将整个checkmessage 押进checkmessageall
                checkmessageall.append(checkmessage)

            #开始处理我们爬过来的票务信息
            for dealwithforlist in checkmessageall:

                # 计算列车类型
                train_type = operation_parameters['type']

                # 判断 [ D:动车 Z:直达 T:特快 K:快速 A:全部 ] 如果都不是的话，默认设置为A 全部
                if train_type != 'D' and train_type != 'Z' and train_type != 'T' and train_type != 'K' and train_type != 'A' :
                    train_type = 'A'

                # 开始匹配列车类型
                if train_type in dealwithforlist[dealwithforlist.index('预订') + 2] or train_type == 'A' :

                    #获取设置的开始时间和结束时间
                    start_time = operation_parameters['start_time']
                    stop_time = operation_parameters['stop_time']

                    # 判断开始时间是否大于结束时间
                    if stop_time <= start_time :
                        print("开始时间不能超过结束时间")
                        sys.exit(-1)

                    # 开始匹配我们的 开始时间 和 结束时间
                    if (start_time <= dealwithforlist[dealwithforlist.index('预订') + 7] ) and (stop_time >= dealwithforlist[dealwithforlist.index('预订') + 7] ) :

                        # 判断是否有票
                        if 'Y' == (dealwithforlist[dealwithforlist.index('预订') + 10]):
                            print("有票")
                            print(dealwithforlist)

                            # 判断是否发送邮件
                            if 'true' == operation_parameters['send_mail']:
                                #获取邮件发送最大次数
                                send_mail_max = operation_parameters['send_mail_max']

                                # 如果没有邮件发送计数，则重新生成
                                if (dealwithforlist[dealwithforlist.index('预订') + 2]) not in send_mail_number:
                                    send_mail_number[dealwithforlist[dealwithforlist.index('预订') + 2]] = 0

                                # 判断发送邮件
                                if (send_mail_number[dealwithforlist[dealwithforlist.index('预订') + 2]]) <= send_mail_max :
                                    Send_Mail_Only_You_Yan(dealwithforlist , operation_parameters)
                                    pass

                                if send_mail_number[dealwithforlist[dealwithforlist.index('预订') + 2]] <= 1200 :
                                    send_mail_number[dealwithforlist[dealwithforlist.index('预订') + 2]] = send_mail_number[dealwithforlist[dealwithforlist.index('预订') + 2]] + 1

                        else:
                            print("没有票")
                            print(dealwithforlist)
                            #将sned_mail_number至0
                            send_mail_number[dealwithforlist[dealwithforlist.index('预订') + 2]] = 0

            # 获取间隔时间
            interval_time = operation_parameters['interval_time']
            #定义睡眠时间
            time.sleep(interval_time)

            check_failed = 0
        except Exception as e:
            print(e)
            check_failed = check_failed + 1

def Send_Mail_Only_You_Yan (dealwithforlist , operation_parameters):

    #获取票的信息
    network_list_info_station_address = Get_Address_resolution(dealwithforlist[dealwithforlist.index('预订') + 3],dealwithforlist[dealwithforlist.index('预订') + 4],2)
    network_list_info_train_number = dealwithforlist[dealwithforlist.index('预订') + 2]
    network_list_info_start_time = dealwithforlist[dealwithforlist.index('预订') + 7]
    network_list_info_stop_time = dealwithforlist[dealwithforlist.index('预订') + 8]
    network_list_info_run_time = dealwithforlist[dealwithforlist.index('预订') + 9]
    network_list_info_date = dealwithforlist[dealwithforlist.index('预订') + 11]
    local_list_info_start_address = operation_parameters['from_address']
    local_list_info_to_address = operation_parameters['to_address']

    #获取邮件信息
    local_list_info_mail_smtp = operation_parameters['mail_smtp']
    local_list_info_mail_user = operation_parameters['user_name']
    local_list_info_mail_password = operation_parameters['user_password']
    local_list_info_mail_subject = operation_parameters['mail_subject']
    local_list_info_mail_to_address = operation_parameters['mail_to_address']

    HOST = local_list_info_mail_smtp
    SUBJECT = local_list_info_mail_subject
    TO = local_list_info_mail_to_address
    FROM = local_list_info_mail_user

    msg = email.mime.text.MIMEText(
    """
        <html>
                <head>
                    <title>仅仅是为了好玩</title>
                        <style>
                            body {
                                width: 35em;
                                margin: 0 auto;
                                font-family: Tahoma, Verdana, Arial, sans-serif;
                            }
                        </style>
                </head>

                <body>
                        <table style="padding: 1px;background-color: #2300f9;">
                            <tr>
                                <th style="background-color: white">车次</th>
                                <th style="background-color: white">需求出发站</th>
                                <th style="background-color: white">需求到达站</th>
                                <th style="background-color: white">车票出发站</th>
                                <th style="background-color: white">车票到达站</th>
                                <th style="background-color: white">时间</th>
                                <th style="background-color: white">发车时间</th>
                                <th style="background-color: white">到达时间</th>
                                <th style="background-color: white">历时</th>
                            </tr>
                            <tr>
                                <td style="background-color: white">%s</td>
                                <td style="background-color: white">%s</td>
                                <td style="background-color: white">%s</td>
                                <td style="background-color: white">%s</td>
                                <td style="background-color: white">%s</td>
                                <td style="background-color: white">%s</td>
                                <td style="background-color: white">%s</td>
                                <td style="background-color: white">%s</td>
                                <td style="background-color: white">%s</td>
                            </tr>
                        </table>
                </body>
        </html>
    """ %(network_list_info_train_number ,local_list_info_start_address ,local_list_info_to_address , network_list_info_station_address[0] , network_list_info_station_address[1] ,network_list_info_date , network_list_info_start_time , network_list_info_stop_time , network_list_info_run_time ), "html","utf-8")

    msg['Subject'] = SUBJECT
    msg['From'] = FROM
    msg['TO'] = TO

    server = smtplib.SMTP_SSL(HOST,')
    server.login(local_list_info_mail_user,local_list_info_mail_password)
    server.sendmail(FROM,TO.split(';'),msg.as_string())
    server.quit

if __name__ == "__main__" :

    # #解析json文件
    # operation_parameters = Dealwith_jsonfile("go_gome.json")
    #
    # #生成URL连接
    # url = Generate_Url(operation_parameters)
    #
    # #开始查询票票
    # Crawling_Informations(url , operation_parameters)

    Crawling_Informations(Generate_Url(Dealwith_jsonfile("go_gome.json")) , Dealwith_jsonfile("go_gome.json"))

相关文章