好友
阅读权限10
听众
最后登录1970-1-1
|
"""
微信定时自动汇报系统 V2.0
优化要点:错误处理、性能、代码结构、用户体验
"""
import tkinter as tk
from tkinter import ttk, messagebox, scrolledtext
import pyautogui
import pyperclip
import schedule
import threading
import time
from datetime import datetime, timedelta
from typing import Optional, Tuple
import queue
import sys
class WeChatAutoApp:
def __init__(self, root: tk.Tk):
self.root = root
self.root.title("微信定时自动汇报系统 V2.0")
# 线程安全的消息队列
self.message_queue = queue.Queue()
# 初始化状态
self.is_running = False
self.target_datetime: Optional[datetime] = None
self.countdown_thread: Optional[threading.Thread] = None
# 设置窗口
self.setup_window()
# 初始化UI
self.setup_ui()
# 设置默认值
self.set_default_values()
# 启动消息处理器
self.root.after(100, self.process_queue)
# 绑定窗口关闭事件
self.root.protocol("WM_DELETE_WINDOW", self.on_closing)
def setup_window(self):
"""设置窗口大小和位置"""
window_width = 700
window_height = 1000
# 隐藏窗口以计算位置
self.root.withdraw()
self.root.update_idletasks()
screen_width = self.root.winfo_screenwidth()
screen_height = self.root.winfo_screenheight()
center_x = max(0, (screen_width - window_width) // 2)
center_y = max(0, (screen_height - window_height) // 2 - 30)
self.root.geometry(f"{window_width}x{window_height}+{center_x}+{center_y}")
self.root.minsize(700, 1000)
self.root.configure(bg="#f0f2f5")
self.root.deiconify()
def setup_ui(self):
"""设置用户界面"""
# 配置网格权重
self.root.columnconfigure(0, weight=1)
for i in [6]: # 只有日志区域可伸缩
self.root.rowconfigure(i, weight=1)
# 创建样式
self.create_styles()
# 标题
title_label = tk.Label(
self.root,
text="微信定时自动汇报系统 V2.0",
font=('Microsoft YaHei', 20, 'bold'),
bg="#f0f2f5",
fg="#1a1a1a"
)
title_label.grid(row=0, column=0, pady=(20, 10))
# 输入框架
self.setup_input_frame()
# 倒计时显示
self.setup_countdown_frame()
# 按钮区域
self.setup_button_frame()
# 日志区域
self.setup_log_frame()
def create_styles(self):
"""创建样式"""
style = ttk.Style()
style.configure("Red.TLabel", foreground="#e74c3c")
style.configure("Green.TButton", background="#28a745", foreground="white")
style.configure("Red.TButton", background="#dc3545", foreground="white")
def setup_input_frame(self):
"""设置输入区域"""
input_frame = ttk.LabelFrame(self.root, text="任务配置", padding=20)
input_frame.grid(row=1, column=0, padx=40, pady=(0, 10), sticky="ew")
input_frame.columnconfigure(1, weight=1)
# 群聊名称
ttk.Label(input_frame, text="目标群聊名称:",
font=('Microsoft YaHei', 10, 'bold')).grid(row=0, column=0, sticky="w", pady=(0, 5))
self.entry_group = ttk.Entry(input_frame, font=('Microsoft YaHei', 11))
self.entry_group.grid(row=0, column=1, sticky="ew", pady=(0, 10), padx=(10, 0))
# 消息内容
ttk.Label(input_frame, text="发送信息内容:",
font=('Microsoft YaHei', 10, 'bold')).grid(row=1, column=0, sticky="nw", pady=(0, 5))
self.text_msg = tk.Text(input_frame, width=40, height=4,
font=('Microsoft YaHei', 11), wrap="word")
self.text_msg.grid(row=1, column=1, sticky="nsew", pady=(0, 10), padx=(10, 0))
# 发送时间
ttk.Label(input_frame, text="设定发送时间 (HH:MM:SS):",
font=('Microsoft YaHei', 10, 'bold')).grid(row=2, column=0, sticky="w", pady=(0, 5))
self.entry_time = ttk.Entry(input_frame, font=('Consolas', 18, 'bold'),
foreground="#0056b3", justify='center')
self.entry_time.grid(row=2, column=1, sticky="ew", pady=(0, 5), padx=(10, 0))
def setup_countdown_frame(self):
"""设置倒计时显示"""
self.label_countdown = tk.Label(
self.root,
text="等待启动...",
font=('Consolas', 36, 'bold'),
bg="#ffffff",
fg="#333",
height=1,
relief="groove"
)
self.label_countdown.grid(row=2, column=0, pady=(0, 10), padx=40, sticky="ew")
def setup_button_frame(self):
"""设置按钮区域"""
btn_frame = tk.Frame(self.root, bg="#f0f2f5")
btn_frame.grid(row=3, column=0, pady=(0, 10))
self.btn_start = tk.Button(
btn_frame,
text="▶ 开始定时任务",
command=self.start_monitoring,
bg="#28a745",
fg="white",
width=20,
height=2,
font=('Microsoft YaHei', 11, 'bold'),
cursor="hand2",
activebackground="#218838",
activeforeground="white"
)
self.btn_start.pack(side="left", padx=20)
self.btn_stop = tk.Button(
btn_frame,
text="⏹ 停止任务",
command=self.stop_monitoring,
bg="#dc3545",
fg="white",
width=20,
height=2,
font=('Microsoft YaHei', 11, 'bold'),
cursor="hand2",
state="disabled",
activebackground="#c82333",
activeforeground="white"
)
self.btn_stop.pack(side="left", padx=20)
# 测试按钮
self.btn_test = tk.Button(
btn_frame,
text="🔧 测试发送",
command=self.test_send,
bg="#007bff",
fg="white",
width=20,
height=2,
font=('Microsoft YaHei', 11, 'bold'),
cursor="hand2",
activebackground="#0069d9",
activeforeground="white"
)
self.btn_test.pack(side="left", padx=20)
def setup_log_frame(self):
"""设置日志区域"""
# 日志标题
tk.Label(
self.root,
text="系统执行状态日志 (实时更新):",
bg="#f0f2f5",
font=('Microsoft YaHei', 10, 'bold')
).grid(row=4, column=0, pady=(10, 0), padx=40, sticky="w")
# 日志控制按钮
log_btn_frame = tk.Frame(self.root, bg="#f0f2f5")
log_btn_frame.grid(row=5, column=0, pady=(5, 0), padx=40, sticky="w")
tk.Button(
log_btn_frame,
text="📄 清空日志",
command=self.clear_log,
bg="#6c757d",
fg="white",
font=('Microsoft YaHei', 9),
cursor="hand2"
).pack(side="left", padx=(0, 10))
tk.Button(
log_btn_frame,
text="💾 保存日志",
command=self.save_log,
bg="#6c757d",
fg="white",
font=('Microsoft YaHei', 9),
cursor="hand2"
).pack(side="left")
# 日志文本框
self.log_area = scrolledtext.ScrolledText(
self.root,
font=('Consolas', 10),
bg="#2c3e50",
fg="#ecf0f1",
bd=0,
wrap="word"
)
self.log_area.grid(row=6, column=0, pady=(5, 20), padx=40, sticky="nsew")
def set_default_values(self):
"""设置默认值"""
self.entry_group.insert(0, "工作汇报群")
self.text_msg.insert("1.0", "今日工作已完成。")
self.entry_time.insert(0, datetime.now().strftime("%H:%M:%S"))
def log(self, message: str, level: str = "INFO"):
"""线程安全的日志记录"""
self.message_queue.put((message, level))
def process_queue(self):
"""处理消息队列"""
try:
while True:
message, level = self.message_queue.get_nowait()
self._write_log(message, level)
except queue.Empty:
pass
finally:
self.root.after(100, self.process_queue)
def _write_log(self, message: str, level: str):
"""写入日志到文本框"""
self.log_area.config(state='normal')
timestamp = datetime.now().strftime("%H:%M:%S")
# 根据级别设置颜色
colors = {
"INFO": "#ecf0f1",
"SUCCESS": "#2ecc71",
"WARNING": "#f39c12",
"ERROR": "#e74c3c"
}
self.log_area.insert(tk.END, f"[{timestamp}] ", "timestamp")
self.log_area.insert(tk.END, f"[{level}] ", level.lower())
self.log_area.insert(tk.END, f"{message}\n")
# 添加标签配置
self.log_area.tag_configure("timestamp", foreground="#95a5a6")
self.log_area.tag_configure(level.lower(), foreground=colors.get(level, "#ecf0f1"))
self.log_area.see(tk.END)
self.log_area.config(state='disabled')
def update_countdown(self):
"""更新倒计时"""
while self.is_running:
try:
now = datetime.now()
if self.target_datetime:
# 计算时间差
diff = self.target_datetime - now
if diff.total_seconds() <= 0:
# 今天的时间已过,设定为明天
self.target_datetime += timedelta(days=1)
diff = self.target_datetime - now
self.log(f"目标时间已过,已调整到明天 {self.target_datetime.strftime('%H:%M:%S')}", "INFO")
# 更新显示
h, m, s = self._format_timedelta(diff)
countdown_text = f"{h:02d}:{m:02d}:{s:02d}"
# 在主线程中更新UI
self.root.after(0, lambda t=countdown_text:
self.label_countdown.config(text=t, fg="#e74c3c"))
# 最后5分钟变黄色,最后1分钟变红色
if diff.total_seconds() < 60:
self.root.after(0, lambda:
self.label_countdown.config(fg="#ff0000"))
elif diff.total_seconds() < 300:
self.root.after(0, lambda:
self.label_countdown.config(fg="#f39c12"))
# 运行计划任务
schedule.run_pending()
time.sleep(0.1) # 更快的响应速度
except Exception as e:
self.log(f"倒计时更新错误: {str(e)}", "ERROR")
time.sleep(1)
def _format_timedelta(self, td: timedelta) -> Tuple[int, int, int]:
"""格式化时间差"""
total_seconds = int(td.total_seconds())
hours = total_seconds // 3600
minutes = (total_seconds % 3600) // 60
seconds = total_seconds % 60
return hours, minutes, seconds
def send_action(self, group: str, msg: str):
"""执行发送操作"""
try:
self.log(f"开始发送消息到: {group}", "INFO")
# 检查必要输入
if not group or not group.strip():
raise ValueError("群聊名称不能为空")
if not msg or not msg.strip():
raise ValueError("消息内容不能为空")
# 激活微信
pyautogui.hotkey('ctrl', 'alt', 'w')
time.sleep(1)
# 搜索群聊
pyautogui.hotkey('ctrl', 'f')
time.sleep(0.5)
pyperclip.copy(group.strip())
pyautogui.hotkey('ctrl', 'v')
time.sleep(1.2)
pyautogui.press('enter')
time.sleep(0.8)
# 发送消息
pyperclip.copy(msg.strip())
pyautogui.hotkey('ctrl', 'v')
time.sleep(0.5)
pyautogui.press('enter')
time.sleep(0.5)
# 最小化微信
pyautogui.hotkey('ctrl', 'alt', 'w')
self.log(f"✓ 消息已成功发送到: {group.strip()}", "SUCCESS")
except pyautogui.FailSafeException:
self.log("安全保护触发: 鼠标移动到屏幕角落中断了操作", "WARNING")
except Exception as e:
error_msg = f"发送失败: {str(e)}"
self.log(error_msg, "ERROR")
# 显示错误弹窗
self.root.after(0, lambda: messagebox.showerror("发送失败", error_msg))
def start_monitoring(self):
"""开始监控任务"""
try:
# 获取输入
group = self.entry_group.get().strip()
msg = self.text_msg.get("1.0", tk.END).strip()
t_str = self.entry_time.get().strip()
# 验证输入
if not group:
messagebox.showwarning("输入错误", "请输入群聊名称")
return
if not msg:
messagebox.showwarning("输入错误", "请输入消息内容")
return
if not t_str:
messagebox.showwarning("输入错误", "请输入发送时间")
return
# 解析时间
try:
t = datetime.strptime(t_str, "%H:%M:%S")
except ValueError:
messagebox.showerror("格式错误", "时间格式应为 HH:MM:SS")
return
# 计算目标时间
now = datetime.now()
self.target_datetime = now.replace(hour=t.hour, minute=t.minute,
second=t.second, microsecond=0)
if self.target_datetime <= now:
self.target_datetime += timedelta(days=1)
# 清除现有任务
schedule.clear()
# 添加新任务
schedule.every().day.at(t_str).do(
self.send_action,
group=group,
msg=msg
)
# 更新状态
self.is_running = True
self.btn_start.config(state="disabled")
self.btn_stop.config(state="normal")
self.btn_test.config(state="disabled")
# 启动倒计时线程
if self.countdown_thread and self.countdown_thread.is_alive():
self.countdown_thread.join(timeout=1)
self.countdown_thread = threading.Thread(
target=self.update_countdown,
daemon=True
)
self.countdown_thread.start()
self.log(f"✓ 任务已启动,将在 {t_str} 发送消息", "SUCCESS")
self.log(f"目标群聊: {group}", "INFO")
self.log(f"下次发送: {self.target_datetime.strftime('%Y-%m-%d %H:%M:%S')}", "INFO")
except Exception as e:
self.log(f"启动任务失败: {str(e)}", "ERROR")
messagebox.showerror("启动失败", f"启动定时任务失败: {str(e)}")
def stop_monitoring(self):
"""停止监控任务"""
self.is_running = False
schedule.clear()
# 更新UI状态
self.btn_start.config(state="normal")
self.btn_stop.config(state="disabled")
self.btn_test.config(state="normal")
self.label_countdown.config(text="等待启动...", fg="#333")
self.log("✗ 任务已停止", "WARNING")
# 等待线程结束
if self.countdown_thread and self.countdown_thread.is_alive():
self.countdown_thread.join(timeout=2)
def test_send(self):
"""测试发送功能"""
if not self.is_running:
group = self.entry_group.get().strip()
msg = self.text_msg.get("1.0", tk.END).strip()
if not group or not msg:
messagebox.showwarning("测试失败", "请先填写群聊名称和消息内容")
return
response = messagebox.askyesno(
"测试发送",
f"即将发送测试消息到 '{group}',请确保微信已登录并处于可操作状态。\n\n继续吗?"
)
if response:
self.log("开始测试发送...", "INFO")
threading.Thread(target=self.send_action, args=(group, msg), daemon=True).start()
def clear_log(self):
"""清空日志"""
self.log_area.config(state='normal')
self.log_area.delete(1.0, tk.END)
self.log_area.config(state='disabled')
self.log("日志已清空", "INFO")
def save_log(self):
"""保存日志到文件"""
try:
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"wechat_auto_log_{timestamp}.txt"
with open(filename, 'w', encoding='utf-8') as f:
f.write(self.log_area.get(1.0, tk.END))
self.log(f"日志已保存到: {filename}", "SUCCESS")
messagebox.showinfo("保存成功", f"日志已保存到:\n{filename}")
except Exception as e:
self.log(f"保存日志失败: {str(e)}", "ERROR")
messagebox.showerror("保存失败", f"保存日志失败: {str(e)}")
def on_closing(self):
"""关闭窗口时的处理"""
if self.is_running:
response = messagebox.askyesno(
"确认退出",
"定时任务正在运行,退出将停止所有任务。\n\n确定要退出吗?"
)
if not response:
return
self.stop_monitoring()
self.root.destroy()
def main():
"""主函数"""
try:
root = tk.Tk()
app = WeChatAutoApp(root)
root.mainloop()
except Exception as e:
print(f"程序启动失败: {e}")
sys.exit(1)
if __name__ == "__main__":
main() |
|