feat: 新增定时任务模块

cron_tasks.py

@@ -0,0 +1,40 @@
+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)