40 lines
1.3 KiB
Python
40 lines
1.3 KiB
Python
import time
|
|
|
|
from schedule import every, repeat, run_pending
|
|
from requests_html import HTML, HTMLSession
|
|
|
|
from lottery import Lottery
|
|
|
|
|
|
session = HTMLSession()
|
|
|
|
def get_data_job(lottery_type='pls'):
|
|
print("pls: I'm working...")
|
|
url = f"https://kaijiang.500.com/{lottery_type}.shtml"
|
|
r = session.get(url)
|
|
if lottery_type == 'pls':
|
|
issue_elem = r.html.search('排列3 第 {}期')
|
|
else:
|
|
issue_elem = r.html.search('排列5 第 {}期')
|
|
draw_date = r.html.xpath('//td[@class="td_title01"]/span[@class="span_right"]/text()', first=True)
|
|
result = r.html.xpath('//li[@class="ball_orange"]/text()')
|
|
if issue_elem is not None and draw_date is not None and result is not None and len(result) != 0 :
|
|
draw_issue = HTML(html=issue_elem[0]).text
|
|
draw_code = "".join(result)
|
|
print(f"result:{draw_code}")
|
|
lottery = Lottery(lottery_type=lottery_type)
|
|
last_id = lottery.insert(draw_issue, draw_date, draw_code)
|
|
if last_id:
|
|
print(f"issue:{draw_issue}数据写入完成。。。")
|
|
|
|
@repeat(every().day.at("00:01"))
|
|
def pls_job():
|
|
get_data_job(lottery_type='pls')
|
|
|
|
@repeat(every().day.at("00:02"))
|
|
def plw_job():
|
|
get_data_job(lottery_type='plw')
|
|
|
|
while True:
|
|
run_pending()
|
|
time.sleep(1) |