From ad83fd261a0174b7d02812feaf25efbc5c97eea2 Mon Sep 17 00:00:00 2001 From: Jalin Date: Thu, 16 May 2019 13:17:40 +0800 Subject: [PATCH] Add verify of query result --- py12306/app/query.py | 117 ++++++++++++++++++++++++++++++++++++++++-- py12306/lib/func.py | 9 ++++ py12306/lib/helper.py | 10 ++++ tests/test_query.py | 48 ++++++++++++++--- 4 files changed, 174 insertions(+), 10 deletions(-) diff --git a/py12306/app/query.py b/py12306/app/query.py index 55e8d84..7b2010c 100644 --- a/py12306/app/query.py +++ b/py12306/app/query.py @@ -4,15 +4,32 @@ from typing import List from py12306.app.app import Logger from py12306.lib.api import API_QUERY_INIT_PAGE, API_LEFT_TICKETS from py12306.lib.exceptions import RetryException -from py12306.lib.func import retry -from py12306.lib.helper import DataHelper +from py12306.lib.func import retry, number_of_time_period +from py12306.lib.helper import DataHelper, TrainSeat from py12306.lib.request import Request +class TicketSeatData(DataHelper): + name: str + num: str + raw: str + + class QueryTicketData(DataHelper): left_date: str left_station: str arrive_station: str + left_periods: List[tuple] = [] + allow_train_numbers: List[str] = [] + execpt_train_numbers: List[str] = [] + allow_seats: List[str] = [] + available_seat: TicketSeatData + members: list + member_num: int + less_member: bool = False + + def _after(self): + self.member_num = len(self.members) class TicketData(DataHelper): @@ -76,11 +93,105 @@ class QueryTicket: if not result: Logger.error('车票查询失败, %s' % resp.reason) tickets = QueryParser.parse_ticket(result) + for ticket in tickets: + self.is_ticket_valid(ticket, data) + if not data: + continue + # 验证完成，准备下单 + Logger.info('[ 查询到座位可用 出发时间 {left_date} 车次 {train_number} 座位类型 {seat_type} 余票数量 {rest_num} ]'.format( + left_date=data.left_date, train_number=ticket.train_number, seat_type=data.available_seat.name, + rest_num=data.available_seat.raw)) + + def is_ticket_valid(self, ticket: TicketData, query: QueryTicketData) -> bool: + """ + 验证 Ticket 信息是否可用 + ) 出发日期验证 + ) 车票数量验证 + ) 时间点验证(00:00 - 24:00) + ) 车次验证 + ) 座位验证 + ) 乘车人数验证 + :param ticket: 车票信息 + :param query: 查询条件 + :return: + """ + if not self.verify_ticket_num(ticket): + return False + + if not self.verify_period(ticket.left_time, query.left_periods): + return False + + if query.allow_train_numbers and ticket.train_no.upper() not in map(str.upper, query.allow_train_numbers): + return False + + if query.execpt_train_numbers and ticket.train_no.upper() in map(str.upper, query.execpt_train_numbers): + return False + + if not self.verify_seat(ticket, query): + return False + if not self.verify_member_count(query): + return False + + return True + + @staticmethod + def verify_period(period: str, available_periods: List[tuple]): + if not available_periods: + return True + period = number_of_time_period(period) + for available_period in available_periods: + if period < number_of_time_period(available_period[0]) or period > number_of_time_period( + available_period[1]): + return False + return True + + @staticmethod + def verify_ticket_num(ticket: TicketData): + return ticket.ticket_num == 'Y' and ticket.order_text == '预订' + + def verify_seat(self, ticket: TicketData, query: QueryTicketData) -> bool: + """ + 检查座位是否可用 + TODO 小黑屋判断 通过 车次 + 座位 + :param ticket: + :param query: + :return: + """ + allow_seats = query.allow_seats + for seat in allow_seats: + seat_num = TrainSeat.types[seat] + raw = ticket.get_origin()[seat_num] + if self.verify_seat_text(raw): + query.available_seat = TicketSeatData({ + 'name': seat, + 'num': seat_num, + 'raw': raw + }) + return True + return False + + @staticmethod + def verify_seat_text(seat: str) -> bool: + return seat != '' and seat != '无' and seat != '*' + + @staticmethod + def verify_member_count(query: QueryTicketData) -> bool: + seat = query.available_seat + if not (seat.raw == '有' or query.member_num <= int(seat.raw)): + rest_num = int(seat.raw) + if query.less_member: + query.member_num = rest_num + Logger.info( + '余票数小于乘车人数，当前余票数: %d, 实际人数 %d, 删减人车人数到: %d' % (rest_num, query.member_num, query.member_num)) + else: + Logger.info('余票数 %d 小于乘车人数 %d，放弃此次提交机会' % (rest_num, query.member_num)) + return False + return True class QueryParser: @classmethod - def parse_ticket(cls, items: dict) -> List[TicketData]: + def parse_ticket(cls, items: List[dict]) -> List[TicketData]: res = [] for item in items: info = item.split('|') diff --git a/py12306/lib/func.py b/py12306/lib/func.py index 02a2b69..8924577 100644 --- a/py12306/lib/func.py +++ b/py12306/lib/func.py @@ -72,3 +72,12 @@ def retry(num: int = 3): return wrapper return decorator + + +def number_of_time_period(period: str) -> int: + """ + Example: 23:00 -> 2300 + :param period: + :return: + """ + return int(period.replace(':', '')) diff --git a/py12306/lib/helper.py b/py12306/lib/helper.py index ff7b08b..bcc616a 100644 --- a/py12306/lib/helper.py +++ b/py12306/lib/helper.py @@ -41,6 +41,8 @@ class DataHelper: self.__dict__[self.__mappers[str(key)]] = val elif key in self.__annotations__: self.__dict__[key] = val + if getattr(self, '_after', None): + self._after() def __generate_mappers(self): for key, val in self.__annotations__.items(): @@ -58,3 +60,11 @@ class DataHelper: except (KeyError, AttributeError): pass + + def get_origin(self) -> dict: + return self.__origin + + +class TrainSeat: + types = {'特等座': 25, '商务座': 32, '一等座': 31, '二等座': 30, '软卧': 23, '硬卧': 28, '硬座': 29, '无座': 26, } + order_types = {'特等座': 'P', '商务座': 9, '一等座': 'M', '二等座': 'O', '软卧': 4, '硬卧': 3, '硬座': 1, '无座': 1} diff --git a/tests/test_query.py b/tests/test_query.py index 9b7bd60..844bc55 100644 --- a/tests/test_query.py +++ b/tests/test_query.py @@ -1,25 +1,59 @@ from unittest import TestCase from tests.helper import BaseTest -from py12306.app.query import QueryTicket, QueryParser +from py12306.app.query import * class TestQueryTicket(BaseTest): task = { 'name': 'admin', } + query_dict = { + 'left_date': '2019-05-18', + 'left_station': 'BJP', + 'arrive_station': 'LZJ', + 'allow_seats': ['二等座'], + 'members': ['test'] + } + query: QueryTicketData + ticket_str = 'iV6uPpzX3CcwqhHe4yzrJHp9hFVCouaXtS01wlUB8f%2BuA%2BKD%2FTV5KLu37w1aKHO2zlAlwMDa%2FDYY%0A2xykUxU964zvkfI3qZZ6uGEKWi0tXCT8fhkQTVvnRI43%2FAinVozab2W1Cq%2FMzJtGBv3D1Q3CscAj%0ANA1XmfNzd6Carglhzvyyy63MkbLIRvxrngx9F01W9jhKXnQupQNTOM3Pw4UIxbesBWkmQfYNj%2Fj%2F%0A3mU33kluoI5vbGVsm115Ec%2BS29KPeaM%2B4%2F2h2UZsiCb%2F5hKfew8Hijodr2VuFftbkge1meSTRRvz%0A|预订|240000G42704|G427|BXP|LAJ|BXP|LAJ|06:21|13:44|07:23|Y|nhsXhn1BbGmb4MI%2BEto43zoslKFQlIY8c356nXAHAEk9Zb2G|20190518|3|P3|01|05|0|0|||||||||||有|无|5||O090M0|O9M|0|0|null' + ticket: TicketData + + def setUp(self) -> None: + super().setUp() + self.ticket = QueryParser.parse_ticket([self.ticket_str])[0] + self.query = QueryTicketData(self.query_dict) def test_get_query_api_type(self): res = QueryTicket().get_query_api_type() self.assertEqual('leftTicket/query', res) def test_get_ticket(self): - data = { - 'left_date': '2019-05-18', - 'left_station': 'BJP', - 'arrive_station': 'LZJ' - } - res = QueryTicket().get_ticket(data) + res = QueryTicket().get_ticket(self.query_dict) + + def test_is_ticket_valid(self): + res = QueryTicket().is_ticket_valid(self.ticket, self.query) + self.assertEqual(res, True) + + def test_verify_period(self): + self.query.left_periods = [('08:00', '16:00')] + res = QueryTicket.verify_period('12:00', self.query.left_periods) + self.assertEqual(res, True) + res = QueryTicket.verify_period('16:00', self.query.left_periods) + self.assertEqual(res, True) + res = QueryTicket.verify_period('16:01', self.query.left_periods) + self.assertEqual(res, False) + + def test_verify_ticket_num(self): + self.ticket.ticket_num = 'Y' + res = QueryTicket.verify_ticket_num(self.ticket) + self.assertEqual(res, True) + + def test_verify_seat(self): + self.query.allow_seats = ['硬座', '二等座'] + res = QueryTicket().verify_seat(self.ticket, self.query) + self.assertEqual(res, True) + self.assertEqual(self.query.available_seat.num, 30) class TestQueryParser(TestCase):