197 lines
6.1 KiB
Python
197 lines
6.1 KiB
Python
import pandas as pd
|
||
import requests
|
||
import json
|
||
from datetime import datetime
|
||
import sys
|
||
|
||
|
||
def read_today_work(excel_path):
|
||
"""
|
||
从Excel文件读取今天的工作记录
|
||
"""
|
||
try:
|
||
# 使用openpyxl引擎读取Excel，并指定读取第一个sheet
|
||
df = pd.read_excel(excel_path, sheet_name=0, engine='openpyxl')
|
||
|
||
# 获取今天的日期
|
||
today = datetime.now().strftime('%Y-%m-%d')
|
||
|
||
# 将日期列转换为字符串格式以便比较
|
||
# 先确保日期列是datetime类型
|
||
if not pd.api.types.is_datetime64_any_dtype(df['日期']):
|
||
df['日期'] = pd.to_datetime(df['日期'], errors='coerce')
|
||
|
||
# 提取日期部分（去掉时间）
|
||
df['日期_仅日期'] = df['日期'].dt.strftime('%Y-%m-%d')
|
||
|
||
# 筛选今天的工作记录
|
||
today_work = df[df['日期_仅日期'] == today]
|
||
|
||
if today_work.empty:
|
||
print(f"未找到今天({today})的工作记录")
|
||
# 显示最近几天的记录
|
||
print("\n最近几天的记录:")
|
||
recent = df.tail(3)
|
||
for idx, row in recent.iterrows():
|
||
date_str = row['日期'].strftime('%Y-%m-%d') if isinstance(row['日期'], pd.Timestamp) else row['日期']
|
||
work_summary = str(row['工作'])[:30] + "..." if len(str(row['工作'])) > 30 else str(row['工作'])
|
||
print(f" {date_str}: {work_summary}")
|
||
return None
|
||
|
||
# 提取工作记录（取第一条，因为通常每天只有一条）
|
||
work_record = today_work.iloc[0]
|
||
|
||
print(f"✅ 找到今天({today})的工作记录")
|
||
return work_record
|
||
|
||
except Exception as e:
|
||
print(f"读取Excel文件时出错: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return None
|
||
|
||
|
||
def format_work_message(work_record):
|
||
"""
|
||
格式化工作记录为企业微信消息
|
||
"""
|
||
try:
|
||
# 获取日期
|
||
date_str = work_record['日期_仅日期'] if '日期_仅日期' in work_record else work_record['日期']
|
||
|
||
# 获取工作内容和详情，处理NaN值
|
||
work_content = str(work_record['工作']) if not pd.isna(work_record['工作']) else "无"
|
||
work_detail = str(work_record['详情']) if not pd.isna(work_record['详情']) else "无"
|
||
|
||
# 清理文本中的换行和多余空格
|
||
work_content = work_content.strip()
|
||
work_detail = work_detail.strip()
|
||
|
||
# 处理工作内容中的多行（保持原格式，去掉-符号）
|
||
# 将换行符替换为两个空格和换行，这样在markdown中会显示为换行
|
||
work_content_formatted = work_content.replace('\n', ' \n')
|
||
|
||
# 处理详情中的多行
|
||
work_detail_formatted = work_detail.replace('\n', ' \n')
|
||
|
||
# 构建Markdown格式的消息
|
||
message = f"""## 📅 今日工作日报 ({date_str})
|
||
|
||
### 📋 主要工作
|
||
{work_content_formatted}
|
||
|
||
### 📝 工作详情
|
||
{work_detail_formatted}
|
||
|
||
---
|
||
⏰ 发送时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"""
|
||
|
||
return message
|
||
|
||
except Exception as e:
|
||
print(f"格式化消息时出错: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return None
|
||
|
||
|
||
def send_to_wechat_work(webhook_url, message):
|
||
"""
|
||
发送消息到企业微信机器人
|
||
"""
|
||
try:
|
||
# 构建消息数据
|
||
data = {
|
||
"msgtype": "markdown",
|
||
"markdown": {
|
||
"content": message
|
||
}
|
||
}
|
||
|
||
print("准备发送的消息内容:")
|
||
print(message)
|
||
print("-" * 50)
|
||
|
||
# 发送请求
|
||
headers = {'Content-Type': 'application/json'}
|
||
response = requests.post(
|
||
webhook_url,
|
||
data=json.dumps(data, ensure_ascii=False).encode('utf-8'),
|
||
headers=headers,
|
||
timeout=10
|
||
)
|
||
|
||
if response.status_code == 200:
|
||
result = response.json()
|
||
if result.get('errcode') == 0:
|
||
print("✅ 消息发送成功！")
|
||
return True
|
||
else:
|
||
print(f"❌ 消息发送失败: {result.get('errmsg')}")
|
||
return False
|
||
else:
|
||
print(f"❌ 请求失败，状态码: {response.status_code}")
|
||
return False
|
||
|
||
except Exception as e:
|
||
print(f"❌ 发送消息时出错: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
|
||
def main():
|
||
# 配置信息
|
||
excel_path = r"C:\Users\29924\Desktop\工作\工作记录.xlsx"
|
||
webhook_url = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=1c29b826-3489-4f48-9456-0eb9c73e0201"
|
||
|
||
print("=" * 50)
|
||
print("开始读取今日工作记录...")
|
||
print(f"Excel文件路径: {excel_path}")
|
||
print(f"当前时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||
print("=" * 50)
|
||
|
||
# 读取今日工作记录
|
||
work_record = read_today_work(excel_path)
|
||
|
||
if work_record is not None:
|
||
print("\n✅ 找到今日工作记录，正在格式化消息...\n")
|
||
|
||
# 格式化消息
|
||
message = format_work_message(work_record)
|
||
|
||
if message:
|
||
print("🚀 正在发送消息到企业微信...\n")
|
||
|
||
# 发送消息
|
||
success = send_to_wechat_work(webhook_url, message)
|
||
|
||
if success:
|
||
print("\n🎉 今日工作日报已成功发送到企业微信！")
|
||
else:
|
||
print("\n❌ 消息发送失败，请检查网络连接或企业微信机器人配置")
|
||
else:
|
||
print("\n❌ 消息格式化失败")
|
||
else:
|
||
print("\n❌ 没有找到今日的工作记录")
|
||
print("提示：请检查Excel文件中是否有今天日期的工作记录")
|
||
|
||
print("\n" + "=" * 50)
|
||
print("任务完成")
|
||
print("=" * 50)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
# 先检查必要的依赖
|
||
try:
|
||
import pandas as pd
|
||
import requests
|
||
|
||
print("✅ 依赖检查通过")
|
||
except ImportError as e:
|
||
print(f"❌ 缺少必要的依赖包: {e}")
|
||
print("请运行以下命令安装依赖:")
|
||
print("pip install pandas openpyxl requests")
|
||
sys.exit(1)
|
||
|
||
main() |