231 lines
9.3 KiB
Python
231 lines
9.3 KiB
Python
import datetime
|
||
import time
|
||
import hmac
|
||
import hashlib
|
||
import base64
|
||
import requests
|
||
import json
|
||
from lxml import etree
|
||
|
||
|
||
# ====================== 1. 钉钉机器人类（保持不变，修复中文编码） ======================
|
||
class DingDingRobot:
|
||
"""钉钉机器人消息发送类"""
|
||
|
||
def __init__(self, token, secret):
|
||
self.token = token
|
||
self.secret = secret
|
||
self.base_url = "https://oapi.dingtalk.com/robot/send"
|
||
|
||
def _sign(self):
|
||
timestamp = str(round(time.time() * 1000))
|
||
string_to_sign = f"{timestamp}\n{self.secret}"
|
||
hmac_code = hmac.new(
|
||
self.secret.encode('utf-8'),
|
||
string_to_sign.encode('utf-8'),
|
||
digestmod=hashlib.sha256
|
||
).digest()
|
||
sign = base64.b64encode(hmac_code).decode('utf-8')
|
||
return timestamp, sign
|
||
|
||
def _build_request_url(self):
|
||
timestamp, sign = self._sign()
|
||
url = f"{self.base_url}?access_token={self.token}×tamp={timestamp}&sign={sign}"
|
||
return url
|
||
|
||
def build_markdown_data(self, title, text):
|
||
return {
|
||
"msgtype": "markdown",
|
||
"markdown": {"title": title, "text": text}
|
||
}
|
||
|
||
def send_message(self, data):
|
||
try:
|
||
url = self._build_request_url()
|
||
headers = {'Content-Type': 'application/json;charset=utf-8'} # 明确指定utf-8
|
||
response = requests.post(
|
||
url=url,
|
||
data=json.dumps(data, ensure_ascii=False), # 关键：保留中文
|
||
headers=headers
|
||
)
|
||
resp_json = response.json()
|
||
if response.status_code == 200 and resp_json.get("errcode") == 0:
|
||
return True, "消息发送成功"
|
||
else:
|
||
return False, f"消息发送失败：{resp_json.get('errmsg', '未知错误')}"
|
||
except Exception as e:
|
||
return False, f"发送异常：{str(e)}"
|
||
|
||
|
||
# ====================== 2. 赛果转换工具函数 ======================
|
||
def saiguo(saiguo1):
|
||
"""将数字赛果转为中文（3→胜，1→平，0→负）"""
|
||
saiguo_map = {'3': '胜', '1': '平', '0': '负'}
|
||
return saiguo_map.get(saiguo1.strip(), '-')
|
||
|
||
|
||
# ====================== 3. 安全提取XPath文本（核心修复：避免索引报错） ======================
|
||
def safe_xpath(elem, xpath_expr):
|
||
"""安全提取XPath文本，空值返回'-'"""
|
||
res = elem.xpath(xpath_expr)
|
||
return res[0].strip() if res and res[0].strip() else '-'
|
||
|
||
|
||
# ====================== 4. 获取html数据（修复日期格式） ======================
|
||
def req(yesterday_str, today_str):
|
||
"""爬取竞彩开奖数据，修复日期格式为YYYYMMDD"""
|
||
url = 'https://www.okooo.com/jingcai/kaijiang/'
|
||
headers = {
|
||
'user-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/143.0.0.0 Safari/537.36',
|
||
'Referer': 'https://www.okooo.com/', # 增加Referer避免反爬
|
||
}
|
||
# 关键修复：StartDate/EndDate要求YYYYMMDD格式（无横线）
|
||
params = {
|
||
'LotteryType': 'SportteryWDL',
|
||
'StartDate': yesterday_str.replace('-', ''),
|
||
'EndDate': today_str.replace('-', ''),
|
||
}
|
||
try:
|
||
response = requests.get(url, headers=headers, params=params, timeout=10)
|
||
response.encoding = response.apparent_encoding # 自动识别编码
|
||
return response.text
|
||
except Exception as e:
|
||
print(f"爬取页面失败：{e}")
|
||
return ""
|
||
|
||
|
||
# ====================== 5. 结果处理成列表（核心修复：XPath安全提取+半全场分割） ======================
|
||
def jieguo_(html,today):
|
||
if not html:
|
||
return []
|
||
html_etree = etree.HTML(html)
|
||
# 修复：直接获取所有数据行，避免按results长度偏移
|
||
tr_list = html_etree.xpath('//table[@class="tableborder tableborder02"]//tr')
|
||
jieguo = []
|
||
|
||
# 遍历行（跳过前2行表头）
|
||
for tr_idx, tr in enumerate(tr_list):
|
||
if tr_idx < 2: # 跳过表头行（根据实际页面调整）
|
||
continue
|
||
|
||
# 提取半全场数据（修复分割逻辑）
|
||
saiguo_ = safe_xpath(tr, './td[17]/b/text()')
|
||
saiguo1 = '-'
|
||
if saiguo_ and '-' in saiguo_:
|
||
parts = saiguo_.split('-', 1) # 只分割一次
|
||
if len(parts) == 2:
|
||
a = saiguo(parts[0])
|
||
b = saiguo(parts[1])
|
||
saiguo1 = f"{a}-{b}"
|
||
|
||
# 安全提取所有字段（基于当前tr节点，避免全局XPath）
|
||
try:
|
||
data = {
|
||
'场次': safe_xpath(tr, './td[1]/text()'),
|
||
'联赛': safe_xpath(tr, './td[2]/span/text()'),
|
||
'时间': safe_xpath(tr, './td[3]/text()'),
|
||
'主队': safe_xpath(tr, './td[4]/a/text()'),
|
||
'客队': safe_xpath(tr, './td[5]/a/text()'),
|
||
'半场': safe_xpath(tr, './td[6]/text()'),
|
||
'全场': safe_xpath(tr, './td[7]/text()'),
|
||
'胜平负': saiguo(safe_xpath(tr, './td[8]/b/text()')),
|
||
'胜平负赔率': safe_xpath(tr, './td[9]/text()'),
|
||
'让球': safe_xpath(tr, './td[10]/span/text()'),
|
||
'让胜平负': saiguo(safe_xpath(tr, './td[11]/b/text()')),
|
||
'让胜平负赔率': safe_xpath(tr, './td[12]/text()'),
|
||
'比分': safe_xpath(tr, './td[13]/b/text()'),
|
||
'比分赔率': safe_xpath(tr, './td[14]/text()'),
|
||
'总进球': safe_xpath(tr, './td[15]/b/text()'),
|
||
'总进球赔率': safe_xpath(tr, './td[16]/text()'),
|
||
'半全场': saiguo1,
|
||
'半全场赔率': safe_xpath(tr, './td[18]/text()')
|
||
}
|
||
# 过滤空场次数据
|
||
weekday_cn = ["周一", "周二", "周三", "周四", "周五", "周六", "周日"][today.weekday() - 1]
|
||
if data['场次'] != '-' and weekday_cn in data['场次']:
|
||
jieguo.append(data)
|
||
except Exception as e:
|
||
print(f"提取第{tr_idx}行数据失败：{e}")
|
||
continue
|
||
return jieguo
|
||
|
||
|
||
# ====================== 6. 生成Markdown表格（核心修复：去掉缩进，规范语法） ======================
|
||
def generate_markdown_table(data_list, table_headers=None):
|
||
"""生成钉钉兼容的Markdown表格（无缩进、语法规范）"""
|
||
if not data_list:
|
||
return "### 竞猜比赛结果\n暂无有效数据"
|
||
|
||
# 自定义表头（只保留你需要的字段）
|
||
if table_headers is None:
|
||
table_headers = {
|
||
'场次':' 场次 ',
|
||
'联赛':' 联赛 ',
|
||
"主队": " 主队 ",
|
||
'客队':' 客队 ',
|
||
"胜平负": " 胜平负 ",
|
||
"让胜平负": " 让胜平负 ",
|
||
"比分": " 比分 ",
|
||
"总进球": " 总进球 ",
|
||
"半全场": " 半全场 "
|
||
}
|
||
|
||
header_keys = list(table_headers.keys())
|
||
# 第一步：构建表头（无缩进！）
|
||
header_str = "| " + " | ".join([table_headers[key] for key in header_keys]) + " |"
|
||
# 第二步：构建分隔行（必须和表头列数一致）
|
||
separator_str = "| " + " | ".join(["---" for _ in header_keys]) + " |"
|
||
# 第三步：构建数据行
|
||
data_rows = []
|
||
for data in data_list:
|
||
row_values = [data.get(key, "-") for key in header_keys]
|
||
row_str = "| " + " | ".join(row_values) + " |"
|
||
data_rows.append(row_str)
|
||
data_rows_str = "\n".join(data_rows)
|
||
# 关键：拼接时去掉所有多余缩进/空格，行尾不要有多余空格
|
||
markdown_table = f"""### 竞猜比赛结果
|
||
{header_str}
|
||
{separator_str}
|
||
{data_rows_str}"""
|
||
return markdown_table
|
||
|
||
|
||
# ====================== 7. 主函数（修复日期格式+分批发送避免字数超限） ======================
|
||
def main():
|
||
# 1. 处理日期（统一为YYYY-MM-DD格式，再传给req函数处理）
|
||
today = datetime.date.today()
|
||
today_str = today.strftime('%Y-%m-%d')
|
||
yesterday = today - datetime.timedelta(days=1)
|
||
yesterday_str = yesterday.strftime("%Y-%m-%d")
|
||
|
||
# 2. 爬取数据
|
||
html = req(yesterday_str, today_str)
|
||
data = jieguo_(html,today)
|
||
if not data:
|
||
print("未爬取到有效数据")
|
||
return
|
||
|
||
# 3. 钉钉机器人配置（替换为你的实际token/secret）
|
||
DING_TOKEN = '5c2777357a4aa5960799d01a02588c19fa90c3f498783376075dafda7d5d706a'
|
||
DING_SECRET = 'SEC058dab30c973edc0dd70b2a7e97eddfd33f20b45a5117b7c5d55c8f2d0dec474'
|
||
ding_robot = DingDingRobot(token=DING_TOKEN, secret=DING_SECRET)
|
||
|
||
# 4. 分批生成表格（避免钉钉字数限制，每批最多15行）
|
||
batch_size = 15
|
||
batches = [data[i:i + batch_size] for i in range(0, len(data), batch_size)]
|
||
|
||
for batch_idx, batch_data in enumerate(batches):
|
||
# 生成单批表格
|
||
mar_data = generate_markdown_table(batch_data)
|
||
# 分批标题（避免重复）
|
||
title = f"竞猜比赛结果（第{batch_idx + 1}批）" if len(batches) > 1 else "竞猜比赛结果"
|
||
# 构建消息体
|
||
msg_data = ding_robot.build_markdown_data(title=title, text=mar_data)
|
||
# 发送消息
|
||
success, message = ding_robot.send_message(msg_data)
|
||
print(f"第{batch_idx + 1}批推送结果：{'成功' if success else '失败'}，提示：{message}")
|
||
# 间隔1秒，避免触发钉钉频率限制
|
||
time.sleep(1)
|
||
|
||
if __name__ == '__main__':
|
||
main() |