This repository was archived by the owner on Dec 19, 2023. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 9.7k
Expand file tree
/
Copy pathQuery.py
More file actions
194 lines (185 loc) · 11.4 KB
/
Query.py
File metadata and controls
194 lines (185 loc) · 11.4 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
# coding=utf-8
import copy
import random
import wrapcache
from config import urlConf
from config.TicketEnmu import ticket
from myUrllib.httpUtils import HTTPClient
from config.configCommon import seat_conf_2
from config.StatusCode import StatusCode
import TickerConfig
class query:
"""
查询接口
"""
def __init__(self, selectObj, from_station, to_station, from_station_h, to_station_h, _station_seat, station_trains,
ticke_peoples_num, station_dates=None, ):
self.session = selectObj
self.httpClint = HTTPClient(TickerConfig.IS_PROXY)
self.httpClint.set_cookies(self.session.cookies)
self.urls = urlConf.urls
self.from_station = from_station
self.to_station = to_station
self.from_station_h = from_station_h
self.to_station_h = to_station_h
self.station_trains = station_trains
self._station_seat = _station_seat if isinstance(_station_seat, list) else list(_station_seat)
self.station_dates = station_dates if isinstance(station_dates, list) else list(station_dates)
self.ticket_black_list = dict()
self.ticke_peoples_num = ticke_peoples_num
def station_seat(self, index):
"""
获取车票对应坐席
:return:
"""
seat = {'商务座': 32,
'一等座': 31,
'二等座': 30,
'特等座': 25,
'软卧': 23,
'硬卧': 28,
'硬座': 29,
'无座': 26,
'动卧': 33,
}
return seat[index]
def check_is_need_train(self, ticket_info):
"""
判断车次是否为想要的车次，如果ticket_info为空，那么就不校验车次，直接返回True
:param ticket_info:
:return:
"""
if self.station_dates and self.station_trains:
return ticket_info[3] in self.station_trains
else:
return True
def sendQuery(self):
"""
查询
:return:
"""
if TickerConfig.IS_CDN == 1 and self.session.cdn_list:
self.httpClint.cdn = self.session.cdn_list[random.randint(4, len(self.session.cdn_list) - 1)]
for station_date in self.station_dates:
select_url = copy.copy(self.urls["select_url"])
select_url["req_url"] = select_url["req_url"].format(station_date, self.from_station, self.to_station,
self.session.queryUrl)
station_ticket = self.httpClint.send(select_url)
status_code = station_ticket.get("code", StatusCode.OK.value)
if status_code != StatusCode.OK.value:
return station_ticket
value = station_ticket.get("data", "")
if not value:
print(u'{0}-{1} 车次坐席查询为空，查询url: https://kyfw.12306.cn{2}, 可以手动查询是否有票'.format(
self.from_station_h,
self.to_station_h,
select_url["req_url"]))
else:
result = value.get('result', [])
if result:
for i in value['result']:
ticket_info = i.split('|')
if self.session.flag:
print(f"车次：{ticket_info[3]} 出发站：{self.from_station_h} 到达站：{self.to_station_h} 历时：{ticket_info[10]}"
f" 商务/特等座：{ticket_info[32] or '--'}"
f" 一等座：{ticket_info[31] or '--'}"
f" 二等座：{ticket_info[30] or '--'}"
f" 动卧：{ticket_info[33] or '--'}"
f" 硬卧：{ticket_info[28] or '--'}"
f" 软座：{ticket_info[23] or '--'}"
f" 硬座：{ticket_info[29] or '--'}"
f" 无座：{ticket_info[26] or '--'}"
f" {ticket_info[1] or '--'}")
if ticket_info[1] == "预订" and self.check_is_need_train(ticket_info): # 筛选未在开始时间内的车次
for j in self._station_seat:
is_ticket_pass = ticket_info[j]
if ticket_info[11] == "Y":
if is_ticket_pass != '' and is_ticket_pass != '无' and is_ticket_pass != '*': # 过滤有效目标车次
secretStr = ticket_info[0]
train_no = ticket_info[2]
query_from_station_name = ticket_info[6]
query_to_station_name = ticket_info[7]
train_location = ticket_info[15]
stationTrainCode = ticket_info[3]
leftTicket = ticket_info[12]
start_time = ticket_info[8]
arrival_time = ticket_info[9]
distance_time = ticket_info[10]
print(start_time, arrival_time, distance_time)
seat = j
try:
ticket_num = int(ticket_info[j])
except ValueError:
ticket_num = "有"
print(u'车次: {0} 始发车站: {1} 终点站: {2} {3}: {4}'.format(ticket_info[3],
self.from_station_h,
self.to_station_h,
seat_conf_2[j],
ticket_num))
if seat_conf_2[j] == "无座" and ticket_info[3][0] in ["G", "D", "C"]:
seat = 30 # GD开头的无座直接强制改为二等座车次
if wrapcache.get(train_no):
print(ticket.QUERY_IN_BLACK_LIST.format(train_no))
continue
else:
if ticket_num != "有" and self.ticke_peoples_num > ticket_num:
if TickerConfig.IS_MORE_TICKET:
print(
u"余票数小于乘车人数，当前余票数: {}, 删减人车人数到: {}".format(ticket_num, ticket_num))
is_more_ticket_num = ticket_num
else:
print(u"余票数小于乘车人数，当前设置不提交，放弃此次提交机会")
continue
else:
print(u"设置乘车人数为: {}".format(self.ticke_peoples_num))
is_more_ticket_num = self.ticke_peoples_num
print(ticket.QUERY_C)
return {
"secretStr": secretStr,
"train_no": train_no,
"stationTrainCode": stationTrainCode,
"train_date": station_date,
"query_from_station_name": query_from_station_name,
"query_to_station_name": query_to_station_name,
"seat": seat,
"leftTicket": leftTicket,
"train_location": train_location,
"code": StatusCode.OK.value,
"is_more_ticket_num": is_more_ticket_num,
"cdn": self.httpClint.cdn,
"status": True,
}
elif is_ticket_pass == '无' and ticket_info[37] == "1" and TickerConfig.TICKET_TYPE is 2:
"""
is_ticket_pass如果有别的显示，但是可以候补，可以提issues提出来，附上query log，我将添加上
判断车次是否可以候补
目前的候补机制是只要一有候补位置，立马提交候补
"""
# 如果最后一位为1，则是可以候补的，不知道这些正确嘛？
nate = list(ticket_info[38])
if wrapcache.get(f"hb{ticket_info[2]}"):
continue
for set_type in TickerConfig.SET_TYPE:
if TickerConfig.PASSENGER_TICKER_STR[set_type] not in nate:
if ticket_info[3][0] in ["G", "D", "C"] and set_type in ["一等座", "特等座", "二等座", "商务座", "无座"]:
return {
"secretList": ticket_info[0],
"seat": [set_type],
"train_no": ticket_info[2],
"status": True,
"cdn": self.httpClint.cdn,
}
elif ticket_info[3][0] in ["T", "Z", "K"] and set_type in ["硬卧", "硬座", "无座", "软座", "软卧"]:
return {
"secretList": ticket_info[0],
"seat": [set_type],
"train_no": ticket_info[2],
"status": True,
"cdn": self.httpClint.cdn,
}
else:
print(u"车次配置信息有误，或者返回数据异常，请检查 {}".format(station_ticket))
self.session.flag = False
return {"code": StatusCode.UnknownError.value, "status": False, "cdn": self.httpClint.cdn, }
if __name__ == "__main__":
q = query()