# -*- coding: utf-8 -*- """ 获取双色球结果, 并匹配自己购买的号码, 配合定时执行使用 """ import sys import os sys.path.append(os.path.join(os.path.abspath(__file__).split('auto')[0] + 'auto')) from utils.utils_send_gotify import * from utils.utils_send_serverchan import * from utils.utils_send_email import * class CheckDlt: def __init__(self): self.url = 'https://webapi.sporttery.cn/gateway/lottery/getHistoryPageListV1.qry?gameNo=85&provinceId=0&pageSize=1&isVerify=1&pageNo=1' self.headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)", "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Accept-Language": "en-US,en;q=0.5", "Accept-Encoding": "gzip, deflate, br", "Connection": "keep-alive", "Content-Type": "application/json" } self.my_dlt = [ ['10', '11', '16', '17', '18', '11', '12'], ['02', '03', '11', '12', '23', '05', '06'], ['07', '09', '15', '17', '22', '09', '11'], ['05', '06', '07', '34', '35', '02', '09'], ['09', '10', '11', '21', '22', '04', '05'] ] def req(self): resp = httpx.get(self.url, headers=self.headers) if resp.status_code != 200: print('state code: {}'.format(resp.status_code)) log_detail = '访问失败, 状态码:{},url:{}'.format(resp.status_code, self.url) print(log_detail) exit(0) resp_json = resp.json() value = resp_json.setdefault('value') data_list = value.setdefault('list') if not data_list: print('获取数据为空') exit(0) return data_list def data_handle(self, data_list): result_data = [] for d in data_list: numbers = d.setdefault('lotteryUnsortDrawresult') try: if len(numbers.split(' ')) < 7: continue except Exception as e: print('numbers: {}, err: {}'.format(numbers, e)) continue red_list = numbers.split(' ')[:5] blue_list = numbers.split(' ')[5:] red_list.sort() blue_list.sort() try: # 切开红球,蓝球数组 red1 = red_list[0] red2 = red_list[1] red3 = red_list[2] red4 = red_list[3] red5 = red_list[4] blue1 = blue_list[0] blue2 = blue_list[1] except Exception as e: print('红球或蓝球数据丢失') continue result_data.append({ 'serial': d.setdefault('lotteryDrawNum'), 'red1': red1 or '', 'red2': red2 or '', 'red3': red3 or '', 'red4': red4 or '', 'red5': red5 or '', 'blue1': blue1 or '', 'blue2': blue2 or '', 'drawPdfUrl': d.setdefault('drawPdfUrl'), 'date': d.setdefault('lotteryDrawTime'), 'pool': d.setdefault('poolBalanceAfterdraw') }) if result_data: return result_data else: print('返回的数据为空, 获取数据失败') exit(0) def data_compare(self, all_data): text = '' data = all_data[0] red_list = [data['red1'], data['red2'], data['red3'], data['red4'], data['red5']] blue_list = [data['blue1'], data['blue2']] # 期号 subject = '{}'.format(data['serial']) # 组成每期数据的text serial_text = 'serial: {}\t\tlottery draw date: {}\nbonus pool: {} RMB\n{}\nlottery draw num: {} + {}\n\n'.format( data['serial'], data['date'], data['pool'], '*' * 90, red_list, blue_list) for my_num in self.my_dlt: my_red_list = my_num[:5] my_blue_list = my_num[5:] # 使用列表推导式找出两个列表中都存在的元素 red_common_elements = [element for element in red_list if element in my_red_list] blue_common_elements = [element for element in blue_list if element in my_blue_list] # 计算相等元素的数量 red_equal_count = len(red_common_elements) blue_equal_count = len(blue_common_elements) serial_text += 'my nums: {} + {}\nred hit: {}\nblue hit: {}\n\n'.format(my_red_list, my_blue_list, red_equal_count, blue_equal_count) text += serial_text text += '{}\n\n\n\n'.format('*' * 90) return text, subject def send_message(self, text, subject): if not text: exit(0) title = f'大乐透 {subject} 期' # 推送到 message GotifyNotifier(title, text).send_message() # 推送到 serverchan ServerChanNotifier(title, text.replace('\n', '\n\n')).send_message() # 发送到邮件 SendEmail(title, title, text).send() def main(self): data_list = self.req() result_data = self.data_handle(data_list) text, subject = self.data_compare(result_data) self.send_message(text, subject) if __name__ == '__main__': CheckDlt().main()