python脚本
文章摘要 时光AI 加载中...|
钉钉机器人推送Umami网站统计信息脚本
python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import json
import requests
import time
import hmac
import hashlib
import base64
import urllib.parse
from datetime import datetime, timedelta
import sys
import os
# ========== 替换为你的信息 ==========
UMAMI_DOMAIN = "" # Umami部署域名
UMAMI_USERNAME = "" # Umami登录账号
UMAMI_PASSWORD = "" # Umami登录密码
WEBSITE_ID = "" # Umami站点ID
DINGTALK_WEBHOOK = "" # 钉钉机器人access_token
DINGTALK_SECRET = "" # 钉钉机器人密钥(SECRET)
# ==================================
class UmamiDingtalkReporter:
def __init__(self):
self.session = requests.Session()
self.session.headers.update({
"Content-Type": "application/json",
"User-Agent": "Mozilla/5.0"
})
def get_current_time(self):
return datetime.now().strftime('%Y-%m-%d %H:%M:%S')
def print_with_time(self, message):
print(f"{self.get_current_time()} {message}")
def login_umami(self):
"""登录Umami获取token"""
self.print_with_time("正在登录Umami获取token...")
login_url = f"{UMAMI_DOMAIN}/api/auth/login"
login_data = {
"username": UMAMI_USERNAME,
"password": UMAMI_PASSWORD
}
try:
response = self.session.post(login_url, json=login_data, timeout=10)
response.raise_for_status()
result = response.json()
token = result.get('token')
if not token:
self.print_with_time(f"Umami登录失败: {result}")
return None
self.print_with_time("成功获取token")
return token
except Exception as e:
self.print_with_time(f"Umami登录失败: {str(e)}")
return None
def get_website_info(self, token):
"""获取网站基本信息"""
self.print_with_time("正在获取网站信息...")
url = f"{UMAMI_DOMAIN}/api/websites/{WEBSITE_ID}"
headers = {"Authorization": f"Bearer {token}"}
try:
response = self.session.get(url, headers=headers, timeout=10)
response.raise_for_status()
result = response.json()
website_name = result.get('name', '未命名网站')
website_domain = result.get('domain', '')
return website_name, website_domain
except Exception as e:
self.print_with_time(f"获取网站信息失败: {str(e)}")
return "未命名网站", ""
def get_stats_data(self, token, start_at, end_at):
"""获取统计数据"""
url = f"{UMAMI_DOMAIN}/api/websites/{WEBSITE_ID}/stats"
headers = {"Authorization": f"Bearer {token}"}
params = {
"startAt": start_at,
"endAt": end_at
}
try:
response = self.session.get(url, headers=headers, params=params, timeout=10)
response.raise_for_status()
return response.json()
except Exception as e:
self.print_with_time(f"获取统计数据失败: {str(e)}")
return {}
def calculate_timestamp(self, time_str):
"""计算时间戳（毫秒）"""
try:
if time_str == "now":
dt = datetime.now()
elif "ago" in time_str:
days = int(time_str.split()[0])
dt = datetime.now() - timedelta(days=days)
elif time_str.startswith("today"):
dt = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
if ":" in time_str:
time_part = time_str.split()[1]
hour, minute, second = map(int, time_part.split(":"))
dt = dt.replace(hour=hour, minute=minute, second=second)
elif time_str.startswith("yesterday"):
dt = datetime.now() - timedelta(days=1)
dt = dt.replace(hour=0, minute=0, second=0, microsecond=0)
if ":" in time_str:
time_part = time_str.split()[1]
hour, minute, second = map(int, time_part.split(":"))
dt = dt.replace(hour=hour, minute=minute, second=second)
else:
dt = datetime.now()
return int(dt.timestamp() * 1000)
except Exception:
return int(time.time() * 1000)
def calculate_growth(self, current, previous):
"""计算增长率"""
if previous == 0:
return "N/A"
try:
growth = ((current - previous) / previous) * 100
return round(growth, 2)
except:
return 0
def generate_ding_signature(self):
"""生成钉钉签名"""
# 修正：使用毫秒级时间戳字符串
timestamp = str(round(time.time() * 1000))
secret_enc = DINGTALK_SECRET.encode('utf-8')
string_to_sign = f"{timestamp}\n{DINGTALK_SECRET}"
string_to_sign_enc = string_to_sign.encode('utf-8')
hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
return timestamp, sign
def send_dingtalk_message(self, markdown_content, title="Umami网站统计报告"):
"""发送钉钉消息"""
if not DINGTALK_WEBHOOK or not DINGTALK_SECRET:
return "未配置钉钉WEBHOOK或secret"
self.print_with_time("正在推送数据到钉钉...")
try:
# 生成签名
timestamp, signature = self.generate_ding_signature()
self.print_with_time(f"钉钉签名参数 - timestamp: {timestamp}, signature: {signature}")
# 构建钉钉Webhook URL
webhook_url = f"{DINGTALK_WEBHOOK}×tamp={timestamp}&sign={signature}"
self.print_with_time(f"钉钉Webhook URL: {webhook_url}")
# 构建消息体
message = {
"msgtype": "markdown",
"markdown": {
"title": title,
"text": markdown_content
},
"at": {
"isAtAll": False
}
}
# 发送请求
response = self.session.post(webhook_url, json=message, timeout=10)
result = response.json()
if result.get('errcode') == 0 and result.get('errmsg') == 'ok':
return "成功"
else:
errcode = result.get('errcode', 'unknown')
errmsg = result.get('errmsg', 'unknown')
self.print_with_time(f"钉钉推送失败详情 - 错误码: {errcode}, 错误信息: {errmsg}")
return "失败"
except Exception as e:
self.print_with_time(f"钉钉推送异常: {str(e)}")
return "失败"
def run(self):
"""主函数"""
self.print_with_time("开始执行Umami数据推送")
# 1. 登录Umami获取token
token = self.login_umami()
if not token:
return False
# 2. 获取网站信息
website_name, website_domain = self.get_website_info(token)
# 3. 计算时间范围
today_start = self.calculate_timestamp("today 00:00:00")
today_end = self.calculate_timestamp("now")
yesterday_start = self.calculate_timestamp("yesterday 00:00:00")
yesterday_end = today_start
last_month_start = self.calculate_timestamp("30 days ago")
last_year_start = self.calculate_timestamp("365 days ago")
today_date = datetime.now().strftime('%Y-%m-%d')
# 4. 获取统计数据
self.print_with_time("正在抓取Umami统计数据...")
today_data = self.get_stats_data(token, today_start, today_end)
yesterday_data = self.get_stats_data(token, yesterday_start, yesterday_end)
last_month_data = self.get_stats_data(token, last_month_start, today_end)
last_year_data = self.get_stats_data(token, last_year_start, today_end)
# 5. 解析数据
today_uv = today_data.get('visitors', 0) or 0
today_pv = today_data.get('pageviews', 0) or 0
today_bounce = today_data.get('bounces', 0) or 0
today_visits = today_data.get('visits', 0) or 0
today_totaltime = today_data.get('totaltime', 0) or 0
yesterday_uv = yesterday_data.get('visitors', 0) or 0
yesterday_pv = yesterday_data.get('pageviews', 0) or 0
last_month_pv = last_month_data.get('pageviews', 0) or 0
last_year_pv = last_year_data.get('pageviews', 0) or 0
# 6. 计算平均访问时长和跳出率
if today_visits > 0:
avg_duration = round(today_totaltime / today_visits, 2)
bounce_rate = round((today_bounce / today_visits) * 100, 2)
else:
avg_duration = 0
bounce_rate = 0
# 7. 计算环比增长率
uv_growth = self.calculate_growth(today_uv, yesterday_uv)
pv_growth = self.calculate_growth(today_pv, yesterday_pv)
# 趋势符号
def get_trend_symbol(growth):
if growth == "N/A":
return "➖"
elif growth >= 0:
return "📈"
else:
return "📉"
uv_trend = get_trend_symbol(uv_growth)
pv_trend = get_trend_symbol(pv_growth)
# 格式化增长率
def format_growth(growth):
if growth == "N/A":
return "N/A"
else:
return f"{growth:.2f}%"
uv_growth_formatted = format_growth(uv_growth)
pv_growth_formatted = format_growth(pv_growth)
# 8. 输出数据到终端
data_log = f"""
{self.get_current_time()} 数据统计：
网站名称: {website_name}
网站域名: {website_domain}
今日访客数(UV): {today_uv} 人
今日访问量(PV): {today_pv} 次
今日访问次数: {today_visits} 次
平均访问时长: {avg_duration} 秒
跳出率: {bounce_rate}%
昨日访客数: {yesterday_uv} 人
昨日访问量: {yesterday_pv} 次
最近30天访问量: {last_month_pv} 次
最近365天访问量: {last_year_pv} 次
访客环比: {uv_growth_formatted} {uv_trend}
访问量环比: {pv_growth_formatted} {pv_trend}
"""
print(data_log)
# 9. 构造钉钉机器人消息
markdown_content = f"""# 📊 Umami网站统计报告
**网站名称：** {website_name}
**统计时间：** {today_date} {datetime.now().strftime('%H:%M:%S')}
**网站域名：** {website_domain}
---
## 📈 今日核心数据
| 指标 | 数值 | 环比昨日 |
|------|------|----------|
| 👥 独立访客(UV) | {today_uv} 人 | {uv_growth_formatted} {uv_trend} |
| 🔄 页面浏览量(PV) | {today_pv} 次 | {pv_growth_formatted} {pv_trend} |
| 🚶♂️ 访问次数 | {today_visits} 次 | - |
| ⏱️ 平均访问时长 | {avg_duration} 秒 | - |
| 🚪 跳出率 | {bounce_rate}% | - |
---
## 📊 历史数据对比
| 时间段 | 独立访客(UV) | 页面浏览量(PV) |
|--------|--------------|----------------|
| 昨日({(datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')}) | {yesterday_uv} 人 | {yesterday_pv} 次 |
| 最近30天 | - | {last_month_pv} 次 |
| 最近365天 | - | {last_year_pv} 次 |
---
## 📋 数据说明
- **UV (Unique Visitors)**: 独立访客数，统计去重的访问用户
- **PV (Page Views)**: 页面浏览量，统计所有页面访问次数
- **跳出率**: 只访问一个页面就离开的会话占比
- **环比**: 与昨日同时段数据对比
**报告生成时间：** {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
"""
# 10. 发送钉钉消息
ding_status = self.send_dingtalk_message(markdown_content)
# 11. 输出最终结果
self.print_with_time(f"数据推送{ding_status}")
return ding_status == "成功"
def main():
reporter = UmamiDingtalkReporter()
success = reporter.run()
sys.exit(0 if success else 1)
if __name__ == "__main__":
main()bash
#!/bin/bash
# ========== 配置信息 ==========
# 请替换为你的实际信息
UMAMI_DOMAIN="" # Umami部署域名
UMAMI_USERNAME="" # Umami登录账号
UMAMI_PASSWORD="" # Umami登录密码
WEBSITE_ID="" # Umami站点ID
DINGTALK_WEBHOOK="" # 钉钉机器人access_token
DINGTALK_SECRET="" # 钉钉机器人密钥(SECRET)
# ========== 全局变量 ==========
UMAMI_TOKEN=""
WEBSITE_NAME="未命名网站"
WEBSITE_DOMAIN=""
CURRENT_TIME=$(date '+%Y-%m-%d %H:%M:%S')
TODAY_DATE=$(date '+%Y-%m-%d')
YESTERDAY_DATE=$(date -d "1 day ago" '+%Y-%m-%d')
# ========== 颜色定义 ==========
RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
BLUE='\033[0;34m'
NC='\033[0m' # No Color
# ========== 工具函数 ==========
# 带时间戳的输出
print_with_time() {
echo -e "$(date '+%Y-%m-%d %H:%M:%S') $1"
}
# 检查命令是否存在
check_command() {
if ! command -v "$1" &> /dev/null; then
echo -e "${RED}错误: 缺少命令 '$1'${NC}"
echo "请安装:"
if [[ "$1" == "jq" ]]; then
echo " Ubuntu/Debian: sudo apt-get install jq"
echo " CentOS/RHEL: sudo yum install jq"
elif [[ "$1" == "bc" ]]; then
echo " Ubuntu/Debian: sudo apt-get install bc"
echo " CentOS/RHEL: sudo yum install bc"
else
echo " sudo apt-get install $1 # Ubuntu/Debian"
echo " sudo yum install $1 # CentOS/RHEL"
fi
exit 1
fi
}
# 计算时间戳（毫秒）
calculate_timestamp() {
local time_str="$1"
case "$time_str" in
"now")
date +%s%3N
;;
"today 00:00:00")
date -d "today 00:00:00" +%s%3N 2>/dev/null || date +%s000
;;
"yesterday 00:00:00")
date -d "yesterday 00:00:00" +%s%3N 2>/dev/null || echo $(($(date +%s) - 86400))000
;;
"30 days ago")
date -d "30 days ago" +%s%3N 2>/dev/null || echo $(($(date +%s) - 2592000))000
;;
"365 days ago")
date -d "365 days ago" +%s%3N 2>/dev/null || echo $(($(date +%s) - 31536000))000
;;
*)
date +%s%3N
;;
esac
}
# 登录Umami获取token
login_umami() {
print_with_time "正在登录Umami获取token..."
local login_url="${UMAMI_DOMAIN}/api/auth/login"
local login_data="{\"username\":\"${UMAMI_USERNAME}\",\"password\":\"${UMAMI_PASSWORD}\"}"
local response
response=$(curl -s -X POST "$login_url" \
-H "Content-Type: application/json" \
-d "$login_data" \
--max-time 10 2>/dev/null)
if [ $? -ne 0 ]; then
print_with_time "Umami登录失败: 网络请求错误"
return 1
fi
local token
token=$(echo "$response" | jq -r '.token // empty')
if [ -z "$token" ] || [ "$token" == "null" ]; then
print_with_time "Umami登录失败: 无法获取token"
echo "响应: $response"
return 1
fi
UMAMI_TOKEN="$token"
print_with_time "成功获取token"
return 0
}
# 获取网站信息
get_website_info() {
print_with_time "正在获取网站信息..."
local url="${UMAMI_DOMAIN}/api/websites/${WEBSITE_ID}"
local response
response=$(curl -s -X GET "$url" \
-H "Authorization: Bearer ${UMAMI_TOKEN}" \
--max-time 10 2>/dev/null)
if [ $? -ne 0 ]; then
print_with_time "获取网站信息失败: 网络请求错误"
return 1
fi
WEBSITE_NAME=$(echo "$response" | jq -r '.name // "未命名网站"')
WEBSITE_DOMAIN=$(echo "$response" | jq -r '.domain // ""')
print_with_time "网站名称: ${WEBSITE_NAME}"
print_with_time "网站域名: ${WEBSITE_DOMAIN}"
return 0
}
# 获取统计数据
get_stats_data() {
local start_at="$1"
local end_at="$2"
local url="${UMAMI_DOMAIN}/api/websites/${WEBSITE_ID}/stats"
local params="startAt=${start_at}&endAt=${end_at}"
local response
response=$(curl -s -X GET "${url}?${params}" \
-H "Authorization: Bearer ${UMAMI_TOKEN}" \
--max-time 10 2>/dev/null)
if [ $? -ne 0 ]; then
echo "{}"
return 1
fi
echo "$response"
}
# 计算增长率
calculate_growth() {
local current="$1"
local previous="$2"
if [ "$previous" -eq 0 ]; then
echo "N/A"
return
fi
local growth
growth=$(echo "scale=2; ($current - $previous) / $previous * 100" | bc 2>/dev/null)
if [ $? -eq 0 ]; then
echo "$growth"
else
echo "0"
fi
}
# 生成钉钉签名
generate_ding_signature() {
local timestamp
timestamp=$(date +%s%3N)
local string_to_sign="${timestamp}\n${DINGTALK_SECRET}"
local sign
# 使用openssl生成HMAC-SHA256签名
sign=$(echo -en "$string_to_sign" | openssl dgst -hmac "$DINGTALK_SECRET" -sha256 -binary | base64)
# URL编码
sign=$(echo -n "$sign" | sed 's/+/%2B/g;s/\//%2F/g;s/=/%3D/g')
echo "$timestamp $sign"
}
# 发送钉钉消息
send_dingtalk_message() {
local markdown_content="$1"
local title="${2:-Umami网站统计报告}"
if [ -z "$DINGTALK_WEBHOOK" ] || [ -z "$DINGTALK_SECRET" ]; then
print_with_time "未配置钉钉WEBHOOK或secret"
return 1
fi
print_with_time "正在推送数据到钉钉..."
# 生成签名
local signature_info
signature_info=$(generate_ding_signature)
local timestamp=$(echo "$signature_info" | cut -d' ' -f1)
local signature=$(echo "$signature_info" | cut -d' ' -f2)
print_with_time "钉钉签名参数 - timestamp: ${timestamp}, signature: ${signature}"
# 构建钉钉Webhook URL
local webhook_url="${DINGTALK_WEBHOOK}×tamp=${timestamp}&sign=${signature}"
# 构建消息体
local message="{
\"msgtype\": \"markdown\",
\"markdown\": {
\"title\": \"${title}\",
\"text\": \"${markdown_content//\"/\\\"}\"
},
\"at\": {
\"isAtAll\": false
}
}"
# 发送请求
local response
response=$(curl -s -X POST "$webhook_url" \
-H "Content-Type: application/json" \
-d "$message" \
--max-time 10 2>/dev/null)
if [ $? -ne 0 ]; then
print_with_time "钉钉推送失败: 网络请求错误"
return 1
fi
local errcode
local errmsg
errcode=$(echo "$response" | jq -r '.errcode // "unknown"')
errmsg=$(echo "$response" | jq -r '.errmsg // "unknown"')
if [ "$errcode" = "0" ] && [ "$errmsg" = "ok" ]; then
print_with_time "钉钉推送成功"
return 0
else
print_with_time "钉钉推送失败 - 错误码: ${errcode}, 错误信息: ${errmsg}"
return 1
fi
}
# 转义Markdown内容中的特殊字符
escape_markdown() {
echo "$1" | sed 's/_/\\_/g;s/*/\\*/g;s/\[/\\[/g;s/\]/\\]/g;s/(/\\(/g;s/)/\\)/g;s/`/\\`/g'
}
# ========== 主函数 ==========
main() {
print_with_time "开始执行Umami数据推送"
# 检查必需的命令
check_command curl
check_command jq
check_command bc
check_command openssl
check_command base64
# 1. 登录Umami获取token
if ! login_umami; then
echo -e "${RED}Umami登录失败，请检查配置${NC}"
exit 1
fi
# 2. 获取网站信息
if ! get_website_info; then
echo -e "${YELLOW}警告: 获取网站信息失败，使用默认值${NC}"
fi
# 3. 计算时间范围
print_with_time "正在计算时间范围..."
TODAY_START=$(calculate_timestamp "today 00:00:00")
TODAY_END=$(calculate_timestamp "now")
YESTERDAY_START=$(calculate_timestamp "yesterday 00:00:00")
YESTERDAY_END=$TODAY_START
LAST_MONTH_START=$(calculate_timestamp "30 days ago")
LAST_YEAR_START=$(calculate_timestamp "365 days ago")
# 4. 获取统计数据
print_with_time "正在抓取Umami统计数据..."
TODAY_DATA=$(get_stats_data "$TODAY_START" "$TODAY_END")
YESTERDAY_DATA=$(get_stats_data "$YESTERDAY_START" "$YESTERDAY_END")
LAST_MONTH_DATA=$(get_stats_data "$LAST_MONTH_START" "$TODAY_END")
LAST_YEAR_DATA=$(get_stats_data "$LAST_YEAR_START" "$TODAY_END")
# 5. 解析数据
TODAY_UV=$(echo "$TODAY_DATA" | jq -r '.visitors // 0')
TODAY_PV=$(echo "$TODAY_DATA" | jq -r '.pageviews // 0')
TODAY_BOUNCE=$(echo "$TODAY_DATA" | jq -r '.bounces // 0')
TODAY_VISITS=$(echo "$TODAY_DATA" | jq -r '.visits // 0')
TODAY_TOTALTIME=$(echo "$TODAY_DATA" | jq -r '.totaltime // 0')
YESTERDAY_UV=$(echo "$YESTERDAY_DATA" | jq -r '.visitors // 0')
YESTERDAY_PV=$(echo "$YESTERDAY_DATA" | jq -r '.pageviews // 0')
LAST_MONTH_PV=$(echo "$LAST_MONTH_DATA" | jq -r '.pageviews // 0')
LAST_YEAR_PV=$(echo "$LAST_YEAR_DATA" | jq -r '.pageviews // 0')
# 设置默认值
TODAY_UV=${TODAY_UV:-0}
TODAY_PV=${TODAY_PV:-0}
TODAY_BOUNCE=${TODAY_BOUNCE:-0}
TODAY_VISITS=${TODAY_VISITS:-0}
TODAY_TOTALTIME=${TODAY_TOTALTIME:-0}
YESTERDAY_UV=${YESTERDAY_UV:-0}
YESTERDAY_PV=${YESTERDAY_PV:-0}
LAST_MONTH_PV=${LAST_MONTH_PV:-0}
LAST_YEAR_PV=${LAST_YEAR_PV:-0}
# 6. 计算平均访问时长和跳出率
local AVG_DURATION=0
local BOUNCE_RATE=0
if [ "$TODAY_VISITS" -gt 0 ]; then
AVG_DURATION=$(echo "scale=2; $TODAY_TOTALTIME / $TODAY_VISITS" | bc 2>/dev/null || echo "0")
BOUNCE_RATE=$(echo "scale=2; $TODAY_BOUNCE / $TODAY_VISITS * 100" | bc 2>/dev/null || echo "0")
fi
# 7. 计算环比增长率
UV_GROWTH=$(calculate_growth "$TODAY_UV" "$YESTERDAY_UV")
PV_GROWTH=$(calculate_growth "$TODAY_PV" "$YESTERDAY_PV")
# 趋势符号
local UV_TREND="➖"
local PV_TREND="➖"
if [ "$UV_GROWTH" != "N/A" ]; then
if (( $(echo "$UV_GROWTH >= 0" | bc -l 2>/dev/null || echo "1") )); then
UV_TREND="📈"
else
UV_TREND="📉"
fi
fi
if [ "$PV_GROWTH" != "N/A" ]; then
if (( $(echo "$PV_GROWTH >= 0" | bc -l 2>/dev/null || echo "1") )); then
PV_TREND="📈"
else
PV_TREND="📉"
fi
fi
# 格式化增长率
local UV_GROWTH_FORMATTED="N/A"
local PV_GROWTH_FORMATTED="N/A"
if [ "$UV_GROWTH" != "N/A" ]; then
UV_GROWTH_FORMATTED=$(printf "%.2f%%" "$UV_GROWTH")
fi
if [ "$PV_GROWTH" != "N/A" ]; then
PV_GROWTH_FORMATTED=$(printf "%.2f%%" "$PV_GROWTH")
fi
# 8. 输出数据到终端
echo ""
print_with_time "数据统计："
echo " 网站名称: ${WEBSITE_NAME}"
echo " 网站域名: ${WEBSITE_DOMAIN}"
echo " 今日访客数(UV): ${TODAY_UV} 人"
echo " 今日访问量(PV): ${TODAY_PV} 次"
echo " 今日访问次数: ${TODAY_VISITS} 次"
echo " 平均访问时长: ${AVG_DURATION} 秒"
echo " 跳出率: ${BOUNCE_RATE}%"
echo " 昨日访客数: ${YESTERDAY_UV} 人"
echo " 昨日访问量: ${YESTERDAY_PV} 次"
echo " 最近30天访问量: ${LAST_MONTH_PV} 次"
echo " 最近365天访问量: ${LAST_YEAR_PV} 次"
echo " 访客环比: ${UV_GROWTH_FORMATTED} ${UV_TREND}"
echo " 访问量环比: ${PV_GROWTH_FORMATTED} ${PV_TREND}"
echo ""
# 9. 构造钉钉机器人消息
# 转义特殊字符
local ESCAPED_WEBSITE_NAME=$(escape_markdown "$WEBSITE_NAME")
local ESCAPED_WEBSITE_DOMAIN=$(escape_markdown "$WEBSITE_DOMAIN")
# 获取当前时间
local CURRENT_DATETIME=$(date '+%Y-%m-%d %H:%M:%S')
# 构建Markdown内容
local MARKDOWN_CONTENT="# 📊 Umami网站统计报告
**网站名称：** ${ESCAPED_WEBSITE_NAME}
**统计时间：** ${TODAY_DATE} $(date '+%H:%M:%S')
**网站域名：** ${ESCAPED_WEBSITE_DOMAIN}
---
## 📈 今日核心数据
| 指标 | 数值 | 环比昨日 |
|------|------|----------|
| 👥 独立访客(UV) | ${TODAY_UV} 人 | ${UV_GROWTH_FORMATTED} ${UV_TREND} |
| 🔄 页面浏览量(PV) | ${TODAY_PV} 次 | ${PV_GROWTH_FORMATTED} ${PV_TREND} |
| 🚶♂️ 访问次数 | ${TODAY_VISITS} 次 | - |
| ⏱️ 平均访问时长 | ${AVG_DURATION} 秒 | - |
| 🚪 跳出率 | ${BOUNCE_RATE}% | - |
---
## 📊 历史数据对比
| 时间段 | 独立访客(UV) | 页面浏览量(PV) |
|--------|--------------|----------------|
| 昨日(${YESTERDAY_DATE}) | ${YESTERDAY_UV} 人 | ${YESTERDAY_PV} 次 |
| 最近30天 | - | ${LAST_MONTH_PV} 次 |
| 最近365天 | - | ${LAST_YEAR_PV} 次 |
---
## 📋 数据说明
- **UV (Unique Visitors)**: 独立访客数，统计去重的访问用户
- **PV (Page Views)**: 页面浏览量，统计所有页面访问次数
- **跳出率**: 只访问一个页面就离开的会话占比
- **环比**: 与昨日同时段数据对比
**报告生成时间：** ${CURRENT_DATETIME}"
# 10. 发送钉钉消息
if send_dingtalk_message "$MARKDOWN_CONTENT"; then
echo -e "${GREEN}✓ 数据推送成功${NC}"
exit 0
else
echo -e "${RED}✗ 数据推送失败${NC}"
exit 1
fi
}
# ========== 脚本入口 ==========
# 处理命令行参数
case "${1:-}" in
-h|--help)
echo "Umami网站统计推送脚本"
echo "用法: $0 [选项]"
echo ""
echo "选项:"
echo " -h, --help 显示此帮助信息"
echo " -t, --test 测试模式，只输出数据不发送到钉钉"
echo ""
echo "环境变量配置:"
echo " 编辑脚本开头的配置部分，设置您的Umami和钉钉信息"
exit 0
;;
-t|--test)
echo -e "${YELLOW}测试模式: 将只输出数据，不发送到钉钉${NC}"
# 这里可以添加测试逻辑
;;
esac
# 运行主函数
main "$@"钉钉机器人推送示例（原理）
python
import time
import hmac
import hashlib
import base64
import urllib.parse
import requests
import json
class DingTalkRobot:
def __init__(self):
# 替换成你自己的钉钉机器人webhook地址
self.URL = ""
# 替换成你自己的加签密钥
self.DINGDING_SECRET = ""
def get_sign(self):
"""
生成钉钉机器人的加签参数（timestamp + sign）
"""
# 获取当前时间戳（毫秒级）
timestamp = str(round(time.time() * 1000))
# 拼接待签名字符串
string_to_sign = f"{timestamp}\n{self.DINGDING_SECRET}"
# 使用HmacSHA256算法计算签名
hmac_code = hmac.new(
self.DINGDING_SECRET.encode('utf-8'),
string_to_sign.encode('utf-8'),
digestmod=hashlib.sha256
).digest()
# base64编码并URL编码
sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
# 返回拼接后的参数
return f"×tamp={timestamp}&sign={sign}"
def build_dingding_message(self, phone, message):
"""
构建钉钉消息的JSON结构
:param phone: 要@的手机号（None则不@任何人）
:param message: 要发送的文本消息内容
:return: 消息的字典结构
"""
# 基础文本消息结构
msg = {
"msgtype": "text",
"text": {
"content": message
}
}
# 如果手机号不为空，添加@配置
if phone and phone.strip():
msg["at"] = {
"atMobiles": [phone.strip()]
}
return msg
def do_send_message(self, phone=None, message=None):
"""
发送钉钉消息
:param phone: 要@的手机号（可选）
:param message: 要发送的消息内容（必填）
:raises ValueError: 消息内容为空时抛出异常
"""
# 检查消息内容是否为空
if not message or not message.strip():
raise ValueError("请输入钉钉服务机器人要输出的信息")
# 构建消息体
msg_data = self.build_dingding_message(phone, message)
# 拼接完整的请求URL（包含加签参数）
url = self.URL + self.get_sign()
try:
# 发送POST请求
response = requests.post(
url=url,
headers={"Content-Type": "application/json"},
data=json.dumps(msg_data),
timeout=(60, 300) # connect_timeout=60s, read_timeout=300s
)
# 解析响应结果
response_json = response.json()
if response_json.get("errmsg") != "ok":
print(f"钉钉消息发送失败: {response_json.get('errmsg')}")
else:
print("钉钉消息发送成功")
except Exception as e:
print(f"发送钉钉消息时发生异常: {str(e)}")
# 测试使用示例
if __name__ == "__main__":
# 创建机器人实例
robot = DingTalkRobot()
# 1. 发送普通消息（不@任何人）
robot.do_send_message(message="这是一条测试消息")
# 2. 发送@特定人的消息
# robot.do_send_message(phone="13800138000", message="这是一条@特定人的测试消息")