好友
阅读权限10
听众
最后登录1970-1-1
|
我这边打开使用后发现悬浮窗口不显示 + 没有 config.json无法
个人简易修复，请大佬看下：
修复点：
程序启动自动生成默认 config.json，打开文件夹就能看到；
修复 DPI 适配、屏幕外窗口问题，强制窗口居中兜底；
增加启动日志，cmd 运行能看到报错；
优化透明窗口渲染逻辑，避免看不见；
托盘图标异常兼容处理；
网络请求超时隔离，不阻塞 UI。
import tkinter as tk
from tkinter import ttk, messagebox
import requests
import json
import ctypes
import random
import os
import time
import threading
import queue
import sys
import smtplib
from email.mime.text import MIMEText
from collections import deque
from PIL import Image, ImageDraw
import pystray
from pystray import MenuItem as item
# DPI 适配修复
try:
ctypes.windll.shcore.SetProcessDpiAwareness(2)
except Exception:
try:
ctypes.windll.user32.SetProcessDPIAware()
except:
pass
DEFAULT_CONFIG = {
"refresh_interval": 2,
"opacity": 0.55,
"au_upper_enabled": False,
"au_upper_value": 0.0,
"au_lower_enabled": True,
"au_lower_value": 1130.0,
"intl_upper_enabled": False,
"intl_upper_value": 0.0,
"intl_lower_enabled": False,
"intl_lower_value": 0.0,
"extreme_window_min": 4,
"extreme_threshold": 5.0,
"intl_extreme_window_min": 4,
"intl_extreme_threshold": 30,
"extreme_enabled": True,
"extreme_flash_times": 6,
"extreme_flash_interval_ms": 150,
"extreme_cooldown_sec": 60,
"alert_cooldown_sec": 10,
"email_enabled": False,
"email_sender": "",
"email_password": "",
"email_receiver": "",
"window_x": None,
"window_y": None
}
def _config_path():
base_dir = os.path.dirname(os.path.abspath(sys.argv[0]))
return os.path.join(base_dir, "config.json")
# 新增：启动自动生成默认配置文件
def init_default_config():
cfg_path = _config_path()
if not os.path.exists(cfg_path):
with open(cfg_path, "w", encoding="utf-8") as f:
json.dump(DEFAULT_CONFIG, f, ensure_ascii=False, indent=2)
print(f"已生成默认配置文件：{cfg_path}")
class SettingsWindow(tk.Toplevel):
def __init__(self, parent, app):
super().__init__(parent)
self.app = app
self.title("设置")
self.geometry("380x520")
self.resizable(False, False)
self.attributes("-topmost", True)
self.protocol("WM_DELETE_WINDOW", self._on_close)
self.notebook = ttk.Notebook(self)
self.notebook.pack(padx=12, pady=12, expand=True, fill="both")
self.tab_general = ttk.Frame(self.notebook)
self.tab_alerts = ttk.Frame(self.notebook)
self.tab_extreme = ttk.Frame(self.notebook)
self.tab_email = ttk.Frame(self.notebook)
self.notebook.add(self.tab_general, text="通用")
self.notebook.add(self.tab_alerts, text="提醒")
self.notebook.add(self.tab_extreme, text="异动")
self.notebook.add(self.tab_email, text="邮件")
self._build_general()
self._build_alerts()
self._build_extreme()
self._build_email()
footer = ttk.Frame(self)
footer.pack(fill="x", padx=12, pady=(0, 12))
ttk.Button(footer, text="保存并应用", command=self._save).pack(side="right")
ttk.Button(footer, text="取消", command=self._on_close).pack(side="right", padx=(0, 8))
self.after(50, self.focus_force)
def _build_general(self):
frame = ttk.Frame(self.tab_general)
frame.pack(fill="both", expand=True, padx=12, pady=12)
ttk.Label(frame, text="刷新间隔（秒）").grid(row=0, column=0, sticky="w")
self.var_interval = tk.StringVar(value=str(self.app.interval))
ttk.Entry(frame, textvariable=self.var_interval, width=12).grid(row=0, column=1, sticky="w", padx=(10, 0))
ttk.Label(frame, text="透明度（0.2~1.0）").grid(row=1, column=0, sticky="w", pady=(12, 0))
self.var_opacity = tk.DoubleVar(value=float(self.app.root.attributes("-alpha")))
ttk.Scale(frame, variable=self.var_opacity, from_=0.2, to=1.0, orient="horizontal",
command=self._on_opacity_change).grid(row=1, column=1, sticky="we", padx=(10, 0), pady=(12, 0))
self.opacity_value_label = ttk.Label(frame, text=f"{self.var_opacity.get():.2f}")
self.opacity_value_label.grid(row=1, column=2, sticky="e", padx=(8, 0), pady=(12, 0))
frame.columnconfigure(1, weight=1)
def _build_alerts(self):
frame = ttk.Frame(self.tab_alerts)
frame.pack(fill="both", expand=True, padx=12, pady=12)
self.var_au_upper_enabled = tk.BooleanVar(value=self.app.au_upper_target is not None)
self.var_au_upper_value = tk.StringVar(value=str(self.app.au_upper_target or 0.0))
self.var_au_lower_enabled = tk.BooleanVar(value=self.app.au_lower_target is not None)
self.var_au_lower_value = tk.StringVar(value=str(self.app.au_lower_target or 0.0))
self.var_intl_upper_enabled = tk.BooleanVar(value=self.app.intl_upper_target is not None)
self.var_intl_upper_value = tk.StringVar(value=str(self.app.intl_upper_target or 0.0))
self.var_intl_lower_enabled = tk.BooleanVar(value=self.app.intl_lower_target is not None)
self.var_intl_lower_value = tk.StringVar(value=str(self.app.intl_lower_target or 0.0))
ttk.Label(frame, text="沪金 上破提醒").grid(row=0, column=0, sticky="w")
ttk.Checkbutton(frame, variable=self.var_au_upper_enabled).grid(row=0, column=1, sticky="w")
ttk.Entry(frame, textvariable=self.var_au_upper_value, width=12).grid(row=0, column=2, sticky="e")
ttk.Label(frame, text="沪金 下破提醒").grid(row=1, column=0, sticky="w", pady=(10, 0))
ttk.Checkbutton(frame, variable=self.var_au_lower_enabled).grid(row=1, column=1, sticky="w", pady=(10, 0))
ttk.Entry(frame, textvariable=self.var_au_lower_value, width=12).grid(row=1, column=2, sticky="e", pady=(10, 0))
ttk.Separator(frame).grid(row=2, column=0, columnspan=3, sticky="we", pady=16)
ttk.Label(frame, text="国际 上破提醒").grid(row=3, column=0, sticky="w")
ttk.Checkbutton(frame, variable=self.var_intl_upper_enabled).grid(row=3, column=1, sticky="w")
ttk.Entry(frame, textvariable=self.var_intl_upper_value, width=12).grid(row=3, column=2, sticky="e")
ttk.Label(frame, text="国际 下破提醒").grid(row=4, column=0, sticky="w", pady=(10, 0))
ttk.Checkbutton(frame, variable=self.var_intl_lower_enabled).grid(row=4, column=1, sticky="w", pady=(10, 0))
ttk.Entry(frame, textvariable=self.var_intl_lower_value, width=12).grid(row=4, column=2, sticky="e", pady=(10, 0))
frame.columnconfigure(1, weight=1)
def _build_extreme(self):
frame = ttk.Frame(self.tab_extreme)
frame.pack(fill="both", expand=True, padx=12, pady=12)
self.var_extreme_enabled = tk.BooleanVar(value=self.app.extreme_enabled)
ttk.Checkbutton(frame, text="启用异动提醒", variable=self.var_extreme_enabled).grid(
row=0, column=0, columnspan=2, sticky="w")
ttk.Label(frame, text="沪金 统计窗口（分钟）").grid(row=1, column=0, sticky="w", pady=(10, 0))
self.var_window_min = tk.StringVar(value=str(int(self.app.extreme_window_sec / 60)))
ttk.Entry(frame, textvariable=self.var_window_min, width=12).grid(row=1, column=1, sticky="w", padx=(10, 0), pady=(10, 0))
ttk.Label(frame, text="沪金 异动阈值").grid(row=2, column=0, sticky="w", pady=(10, 0))
self.var_extreme_threshold = tk.StringVar(value=str(self.app.extreme_threshold))
ttk.Entry(frame, textvariable=self.var_extreme_threshold, width=12).grid(row=2, column=1, sticky="w", padx=(10, 0), pady=(10, 0))
ttk.Label(frame, text="国际 统计窗口（分钟）").grid(row=3, column=0, sticky="w", pady=(10, 0))
self.var_intl_window_min = tk.StringVar(value=str(int(self.app.intl_extreme_window_sec / 60)))
ttk.Entry(frame, textvariable=self.var_intl_window_min, width=12).grid(row=3, column=1, sticky="w", padx=(10, 0), pady=(10, 0))
ttk.Label(frame, text="国际 异动阈值").grid(row=4, column=0, sticky="w", pady=(10, 0))
self.var_intl_extreme_threshold = tk.StringVar(value=str(self.app.intl_extreme_threshold))
ttk.Entry(frame, textvariable=self.var_intl_extreme_threshold, width=12).grid(row=4, column=1, sticky="w", padx=(10, 0), pady=(10, 0))
ttk.Separator(frame).grid(row=5, column=0, columnspan=2, sticky="we", pady=16)
ttk.Label(frame, text="闪烁次数").grid(row=6, column=0, sticky="w", pady=(10, 0))
self.var_flash_times = tk.StringVar(value=str(self.app.extreme_flash_times))
ttk.Entry(frame, textvariable=self.var_flash_times, width=12).grid(row=6, column=1, sticky="w", padx=(10, 0), pady=(10, 0))
ttk.Label(frame, text="闪烁间隔（毫秒）").grid(row=7, column=0, sticky="w", pady=(10, 0))
self.var_flash_interval = tk.StringVar(value=str(self.app.extreme_flash_interval_ms))
ttk.Entry(frame, textvariable=self.var_flash_interval, width=12).grid(row=7, column=1, sticky="w", padx=(10, 0), pady=(10, 0))
ttk.Label(frame, text="冷却时间（秒）").grid(row=8, column=0, sticky="w", pady=(10, 0))
self.var_cooldown = tk.StringVar(value=str(self.app.extreme_cooldown_sec))
ttk.Entry(frame, textvariable=self.var_cooldown, width=12).grid(row=8, column=1, sticky="w", padx=(10, 0), pady=(10, 0))
def _build_email(self):
frame = ttk.Frame(self.tab_email)
frame.pack(fill="both", expand=True, padx=12, pady=12)
self.var_email_enabled = tk.BooleanVar(value=self.app.email_enabled)
ttk.Checkbutton(
frame,
text="启用邮件提醒",
variable=self.var_email_enabled,
).grid(row=0, column=0, columnspan=2, sticky="w")
ttk.Label(frame, text="发件人地址").grid(row=1, column=0, sticky="w", pady=(12, 0))
self.var_email_sender = tk.StringVar(value=self.app.email_sender)
ttk.Entry(frame, textvariable=self.var_email_sender, width=26).grid(
row=1, column=1, sticky="we", padx=(10, 0), pady=(12, 0))
ttk.Label(frame, text="授权码").grid(row=2, column=0, sticky="w", pady=(10, 0))
self.var_email_password = tk.StringVar(value=self.app.email_password)
ttk.Entry(frame, textvariable=self.var_email_password, width=26, show="*").grid(
row=2, column=1, sticky="we", padx=(10, 0), pady=(10, 0))
ttk.Label(frame, text="收件人地址").grid(row=3, column=0, sticky="w", pady=(10, 0))
self.var_email_receiver = tk.StringVar(value=self.app.email_receiver)
ttk.Entry(frame, textvariable=self.var_email_receiver, width=26).grid(
row=3, column=1, sticky="we", padx=(10, 0), pady=(10, 0))
ttk.Separator(frame).grid(row=4, column=0, columnspan=2, sticky="we", pady=14)
ttk.Label(frame, text="SMTP 服务器").grid(row=5, column=0, sticky="w")
ttk.Label(frame, text="smtp.163.com 端口 465 SSL", foreground="#9AA0A6").grid(
row=5, column=1, sticky="w", padx=(10, 0))
ttk.Button(frame, text="发送测试邮件", command=self._test_email).grid(
row=6, column=0, columnspan=2, sticky="w", pady=(14, 0))
frame.columnconfigure(1, weight=1)
def _test_email(self):
sender = self.var_email_sender.get().strip()
password = self.var_email_password.get().strip()
receiver = self.var_email_receiver.get().strip()
if not all([sender, password, receiver]):
messagebox.showerror("错误", "请先填写发件人、授权码和收件人", parent=self)
return
def do_send():
ok, err = self.app.send_email(
sender, password, receiver,
subject="黄金监控 — 测试邮件",
body="邮件提醒功能配置成功，此为测试邮件。"
)
if ok:
self.after(0, lambda: messagebox.showinfo("成功", "测试邮件发送成功", parent=self))
else:
self.after(0, lambda: messagebox.showerror("发送失败", f"错误信息：{err}", parent=self))
threading.Thread(target=do_send, daemon=True).start()
def _on_opacity_change(self, value):
try:
self.opacity_value_label.config(text=f"{float(value):.2f}")
except Exception:
return
def _save(self):
try:
interval = float(self.var_interval.get().strip())
if interval <= 0:
raise ValueError()
except Exception:
messagebox.showerror("错误", "刷新间隔必须是有效的正数", parent=self)
return
opacity = float(self.var_opacity.get())
if not (0.2 <= opacity <= 1.0):
messagebox.showerror("错误", "透明度必须在 0.2 到 1.0 之间", parent=self)
return
def parse_enabled_value(enabled_var, value_var):
if not enabled_var.get():
return None
try:
return float(value_var.get().strip())
except Exception:
return None
au_upper = parse_enabled_value(self.var_au_upper_enabled, self.var_au_upper_value)
au_lower = parse_enabled_value(self.var_au_lower_enabled, self.var_au_lower_value)
intl_upper = parse_enabled_value(self.var_intl_upper_enabled, self.var_intl_upper_value)
intl_lower = parse_enabled_value(self.var_intl_lower_enabled, self.var_intl_lower_value)
try:
window_min = int(float(self.var_window_min.get().strip()))
if window_min <= 0:
raise ValueError()
except Exception:
messagebox.showerror("错误", "沪金统计窗口必须是有效的正整数（分钟）", parent=self)
return
try:
intl_window_min = int(float(self.var_intl_window_min.get().strip()))
if intl_window_min <= 0:
raise ValueError()
except Exception:
messagebox.showerror("错误", "国际金统计窗口必须是有效的正整数（分钟）", parent=self)
return
try:
extreme_threshold = float(self.var_extreme_threshold.get().strip())
if extreme_threshold <= 0:
raise ValueError()
except Exception:
messagebox.showerror("错误", "沪金异动阈值必须是有效的正数", parent=self)
return
try:
intl_extreme_threshold = float(self.var_intl_extreme_threshold.get().strip())
if intl_extreme_threshold <= 0:
raise ValueError()
except Exception:
messagebox.showerror("错误", "国际金异动阈值必须是有效的正数", parent=self)
return
try:
flash_times = int(float(self.var_flash_times.get().strip()))
if flash_times < 1:
raise ValueError()
except Exception:
messagebox.showerror("错误", "闪烁次数必须是正整数", parent=self)
return
try:
flash_interval = int(float(self.var_flash_interval.get().strip()))
if flash_interval < 50:
raise ValueError()
except Exception:
messagebox.showerror("错误", "闪烁间隔必须是不小于 50 的整数（毫秒）", parent=self)
return
try:
cooldown = int(float(self.var_cooldown.get().strip()))
if cooldown < 1:
raise ValueError()
except Exception:
messagebox.showerror("错误", "冷却时间必须是有效的正整数（秒）", parent=self)
return
new_config = {
"refresh_interval": interval,
"opacity": opacity,
"au_upper_enabled": au_upper is not None,
"au_upper_value": float(au_upper or 0.0),
"au_lower_enabled": au_lower is not None,
"au_lower_value": float(au_lower or 0.0),
"intl_upper_enabled": intl_upper is not None,
"intl_upper_value": float(intl_upper or 0.0),
"intl_lower_enabled": intl_lower is not None,
"intl_lower_value": float(intl_lower or 0.0),
"extreme_enabled": self.var_extreme_enabled.get(),
"extreme_window_min": window_min,
"extreme_threshold": extreme_threshold,
"intl_extreme_window_min": intl_window_min,
"intl_extreme_threshold": intl_extreme_threshold,
"extreme_flash_times": flash_times,
"extreme_flash_interval_ms": flash_interval,
"extreme_cooldown_sec": cooldown,
"alert_cooldown_sec": 10,
"email_enabled": self.var_email_enabled.get(),
"email_sender": self.var_email_sender.get().strip(),
"email_password": self.var_email_password.get().strip(),
"email_receiver": self.var_email_receiver.get().strip(),
"window_x": self.app.root.winfo_x(),
"window_y": self.app.root.winfo_y()
}
self.app.apply_config(new_config)
self.app.save_config()
messagebox.showinfo("成功", "设置已保存并应用", parent=self)
def _on_close(self):
self.app.settings_window = None
self.destroy()
class GoldTaskbarDoubleLine:
def __init__(self):
# 启动自动生成配置文件
init_default_config()
self.config = self.load_config()
self.root = tk.Tk()
self.root.title("黄金监控")
self.root.overrideredirect(True)
self.root.attributes("-topmost", True)
self.bg_color = "#000000"
self.card_color = "#101218"
self.card_border = "#2A2D36"
self.card_highlight_color = "#1B1F2A"
self.card_border_highlight = "#3A4150"
self.text_color = "#F5F5F7"
self.muted_color = "#9AA0A6"
self.up_color = "#FF3B30"
self.down_color = "#00C853"
self.font_family = "Segoe UI"
self.text_font_size = 10
self.flash_text_font_size = 12
self.arrow_font_size = 15
self.card_width = 135
self.card_height = 60
self.corner_radius = 15
self.root.configure(bg=self.bg_color)
self.root.attributes("-transparentcolor", self.bg_color)
self.root.attributes("-alpha", float(self.config.get("opacity", 0.55)))
user32 = ctypes.windll.user32
self.sw = user32.GetSystemMetrics(0)
self.sh = user32.GetSystemMetrics(1)
position = self._load_position()
# 修复：窗口超出屏幕兜底居中
if position:
x, y = position
# 坐标越界重置到右下角
if x < 0 or x > self.sw or y < 0 or y > self.sh:
x = self.sw - self.card_width - 20
y = self.sh - self.card_height - 80
else:
x = self.sw - self.card_width - 20
y = self.sh - self.card_height - 80
self.root.geometry(f"{self.card_width}x{self.card_height}+{x}+{y}")
self.canvas = tk.Canvas(
self.root, width=self.card_width, height=self.card_height,
bg=self.bg_color, highlightthickness=0, bd=0
)
self.canvas.pack(fill="both", expand=True)
self.card_fill_items = []
self.card_border_items = []
self._draw_card()
self.au_icon = self.canvas.create_text(
18, 20, text="Ⓐ", fill=self.text_color,
font=(self.font_family, self.text_font_size, "bold"))
self.au_value_text = self.canvas.create_text(
36, 20, text="--", fill=self.text_color,
font=(self.font_family, self.text_font_size, "bold"), anchor="w")
self.au_arrow_text = self.canvas.create_text(
120, 20, text="•", fill=self.muted_color,
font=(self.font_family, self.arrow_font_size, "bold"), anchor="e")
self.intl_icon = self.canvas.create_text(
18, 44, text="Ⓖ", fill=self.text_color,
font=(self.font_family, self.text_font_size, "bold"))
self.intl_value_text = self.canvas.create_text(
36, 44, text="--", fill=self.text_color,
font=(self.font_family, self.text_font_size, "bold"), anchor="w")
self.intl_arrow_text = self.canvas.create_text(
120, 44, text="•", fill=self.muted_color,
font=(self.font_family, self.arrow_font_size, "bold"), anchor="e")
self._bind_drag()
self.session = requests.Session()
self.session.headers.update({
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
"Accept": "application/json, text/plain, */*",
"Accept-Language": "zh-CN,zh;q=0.9",
"Connection": "keep-alive",
})
self.au_upper_target = None
self.au_lower_target = 1130.00
self.intl_upper_target = None
self.intl_lower_target = None
self.au_upper_triggered = False
self.au_lower_triggered = False
self.intl_upper_triggered = False
self.intl_lower_triggered = False
self.au_upper_last_ts = 0.0
self.au_lower_last_ts = 0.0
self.intl_upper_last_ts = 0.0
self.intl_lower_last_ts = 0.0
self.alert_cooldown_sec = 10
self.email_enabled = False
self.email_sender = ""
self.email_password = ""
self.email_receiver = ""
self.au = None
self.prev_au = None
self.intl = None
self.prev_intl = None
self.interval = 2
self.extreme_window_sec = 300
self.intl_extreme_window_sec = 300
self.extreme_threshold = 5.0
self.intl_extreme_threshold = 5.0
self.extreme_enabled = True
self.extreme_flash_times = 6
self.extreme_flash_interval_ms = 150
self.extreme_cooldown_sec = 60
self.au_history = deque()
self.intl_history = deque()
self.extreme_last_ts = 0.0
self.intl_extreme_last_ts = 0.0
self.flash_active = False
self._price_lock = threading.Lock()
self._action_queue = queue.Queue()
self.apply_config(self.config)
self.settings_window = None
self.tray = None
self.hidden = False
self._setup_tray()
self.root.protocol("WM_DELETE_WINDOW", self.quit_app)
self.data_thread = threading.Thread(target=self._data_fetch_loop, daemon=True)
self.data_thread.start()
self._update_ui_cycle()
print("黄金监控已启动，右下角托盘可操作")
def load_config(self):
try:
path = _config_path()
if not os.path.exists(path):
return DEFAULT_CONFIG.copy()
with open(path, "r", encoding="utf-8") as f:
saved = json.load(f)
return {**DEFAULT_CONFIG, **saved}
except Exception as e:
print(f"读取配置异常：{e}")
return DEFAULT_CONFIG.copy()
def save_config(self):
try:
with open(_config_path(), "w", encoding="utf-8") as f:
json.dump(self.config, f, ensure_ascii=False, indent=2)
except Exception as e:
print(f"保存配置失败：{e}")
def apply_config(self, cfg):
self.config = {**self.config, **cfg}
self.interval = float(self.config.get("refresh_interval", self.interval))
self.root.attributes("-alpha", float(self.config.get("opacity", self.root.attributes("-alpha"))))
def target(enabled_key, value_key):
if not self.config.get(enabled_key, False):
return None
try:
return float(self.config.get(value_key, 0.0))
except Exception:
return None
self.au_upper_target = target("au_upper_enabled", "au_upper_value")
self.au_lower_target = target("au_lower_enabled", "au_lower_value")
self.intl_upper_target = target("intl_upper_enabled", "intl_upper_value")
self.intl_lower_target = target("intl_lower_enabled", "intl_lower_value")
self.extreme_window_sec = int(self.config.get("extreme_window_min", int(self.extreme_window_sec / 60))) * 60
self.extreme_threshold = float(self.config.get("extreme_threshold", self.extreme_threshold))
self.intl_extreme_window_sec = int(self.config.get("intl_extreme_window_min", int(self.intl_extreme_window_sec / 60))) * 60
self.intl_extreme_threshold = float(self.config.get("intl_extreme_threshold",self.intl_extreme_threshold))
self.extreme_enabled = bool(self.config.get("extreme_enabled", self.extreme_enabled))
self.extreme_flash_times = int(self.config.get("extreme_flash_times", self.extreme_flash_times))
self.extreme_flash_interval_ms = int(self.config.get("extreme_flash_interval_ms", self.extreme_flash_interval_ms))
self.extreme_cooldown_sec = int(self.config.get("extreme_cooldown_sec", self.extreme_cooldown_sec))
self.alert_cooldown_sec = int(self.config.get("alert_cooldown_sec", self.alert_cooldown_sec))
self.email_enabled = bool(self.config.get("email_enabled", self.email_enabled))
self.email_sender = str(self.config.get("email_sender", self.email_sender))
self.email_password = str(self.config.get("email_password", self.email_password))
self.email_receiver = str(self.config.get("email_receiver", self.email_receiver))
self.au_history.clear()
self.intl_history.clear()
def _setup_tray(self):
try:
img = Image.new("RGB", (64, 64), color=(18, 18, 18))
d = ImageDraw.Draw(img)
d.ellipse([8, 8, 56, 56], fill=(255, 215, 0))
d.text((22, 22), "Au", fill=(0, 0, 0))
except Exception:
img = None
menu = (
item("显示/隐藏", self._tray_toggle),
item("设置", self._tray_settings),
item("退出", self._tray_quit),
)
self.tray = pystray.Icon("GoldMonitor", img, "Gold Monitor", menu)
threading.Thread(target=self.tray.run, daemon=True).start()
def _tray_toggle(self, icon=None, menu_item=None): self.root.after(0, self.toggle_visible)
def _tray_settings(self, icon=None, menu_item=None): self.root.after(0, self.open_settings)
def _tray_quit(self, icon=None, menu_item=None): self.root.after(0, self.quit_app)
def toggle_visible(self):
if self.hidden:
self.root.deiconify()
self.hidden = False
else:
self.root.withdraw()
self.hidden = True
def open_settings(self):
if self.settings_window is not None and self.settings_window.winfo_exists():
self.settings_window.deiconify()
self.settings_window.lift()
self.settings_window.focus_force()
return
self.settings_window = SettingsWindow(self.root, self)
def quit_app(self):
try:
if self.tray is not None:
self.tray.stop()
except Exception:
pass
try:
self.root.quit()
self.root.destroy()
except Exception:
pass
def _draw_card(self):
for it in self.card_fill_items: self.canvas.delete(it)
for it in self.card_border_items: self.canvas.delete(it)
self.card_fill_items = []
self.card_border_items = []
r, (x1, y1, x2, y2) = self.corner_radius, (0, 0, self.card_width, self.card_height)
cc, cb = self.card_color, self.card_border
self.card_fill_items += [
self.canvas.create_rectangle(x1+r, y1, x2-r, y2, fill=cc, outline=cc),
self.canvas.create_rectangle(x1, y1+r, x2, y2-r, fill=cc, outline=cc),
self.canvas.create_oval(x1, y1, x1+2*r, y1+2*r, fill=cc, outline=cc),
self.canvas.create_oval(x2-2*r, y1, x2, y1+2*r, fill=cc, outline=cc),
self.canvas.create_oval(x1, y2-2*r, x1+2*r, y2, fill=cc, outline=cc),
self.canvas.create_oval(x2-2*r, y2-2*r, x2, y2, fill=cc, outline=cc),
]
self.card_border_items += [
self.canvas.create_rectangle(x1+r, y1+1, x2-r, y1+2, fill=cb, outline=cb),
self.canvas.create_rectangle(x1+r, y2-2, x2-r, y2-1, fill=cb, outline=cb),
self.canvas.create_rectangle(x1+1, y1+r, x1+2, y2-r, fill=cb, outline=cb),
self.canvas.create_rectangle(x2-2, y1+r, x2-1, y2-r, fill=cb, outline=cb),
]
def _apply_card_style(self, fill_color, border_color):
for it in self.card_fill_items: self.canvas.itemconfig(it, fill=fill_color, outline=fill_color)
for it in self.card_border_items: self.canvas.itemconfig(it, fill=border_color, outline=border_color)
def _bind_drag(self):
for w in (self.canvas, self.root):
w.bind("<ButtonPress-1>", self._start_drag)
w.bind("<B1-Motion>", self._on_drag)
w.bind("<ButtonRelease-1>",self._end_drag)
def _start_drag(self, event):
self.drag_offset_x = event.x_root - self.root.winfo_x()
self.drag_offset_y = event.y_root - self.root.winfo_y()
def _on_drag(self, event):
self.root.geometry(f"{self.card_width}x{self.card_height}"
f"+{event.x_root - self.drag_offset_x}"
f"+{event.y_root - self.drag_offset_y}")
def _end_drag(self, event):
self._save_position()
def _save_position(self):
try:
self.config["window_x"] = self.root.winfo_x()
self.config["window_y"] = self.root.winfo_y()
self.save_config()
except Exception:
pass
def _load_position(self):
try:
x = self.config.get("window_x")
y = self.config.get("window_y")
if x is None or y is None:
return None
return int(x), int(y)
except Exception:
return None
def _fetch_au(self):
url = "https://ms.jr.jd.com/gw2/generic/CreatorSer/pc/m/pcQueryGoldProduct"
params = {"reqData": '{"goldType":"2"}'}
headers = {"Origin": "https://jdjr.jd.com", "Referer": "https://jdjr.jd.com/"}
try:
res = self.session.get(url, params=params, headers=headers, timeout=2)
res.raise_for_status()
return float(res.json()["resultData"]["data"]["priceValue"])
except Exception as e:
print(f"沪金接口请求失败：{e}")
return None
def _fetch_intl(self):
url = "https://ms.jr.jd.com/gw2/generic/CaiFuPC/pc/m/getQuoteExtendUseUniqueCodeWithCache"
payload = {"ticket": "jd-jr-pc", "uniqueCode": "WG-XAUUSD"}
try:
res = self.session.post(url, json=payload, timeout=2)
res.raise_for_status()
return float(json.loads(res.json()["resultData"]["data"])["lastPrice"])
except Exception as e:
print(f"国际金接口请求失败：{e}")
return None
def _notify(self, msg):
if self.tray is not None:
self.tray.notify(msg, "波动提醒")
else:
print(f"通知: {msg}")
def send_email(self, sender, password, receiver, subject, body):
try:
msg = MIMEText(body, "plain", "utf-8")
msg["Subject"] = subject
msg["From"] = sender
msg["To"] = receiver
with smtplib.SMTP_SSL("smtp.163.com", 465, timeout=10) as smtp:
smtp.login(sender, password)
smtp.sendmail(sender, [receiver], msg.as_string())
return True, ""
except Exception as e:
return False, str(e)
def _send_alert_email(self, subject, body):
if not self.email_enabled:
return
if not all([self.email_sender, self.email_password, self.email_receiver]):
return
threading.Thread(
target=self.send_email,
args=(self.email_sender, self.email_password,
self.email_receiver, subject, body),
daemon=True
).start()
def _check_alert(self, price, symbol):
if price is None:
return
now = time.time()
if symbol == "AU":
if self.au_upper_target is not None:
if price >= self.au_upper_target and not self.au_upper_triggered:
if now - self.au_upper_last_ts >= self.alert_cooldown_sec:
msg = f"Au {self.au_upper_target}，当前价格：{price:.2f}"
self._notify(msg)
self._send_alert_email("沪金提醒 — 上破", msg)
self.au_upper_last_ts = now
self.au_upper_triggered = True
if price < self.au_upper_target:
self.au_upper_triggered = False
if self.au_lower_target is not None:
if price <= self.au_lower_target and not self.au_lower_triggered:
if now - self.au_lower_last_ts >= self.alert_cooldown_sec:
msg = f"Au {self.au_lower_target}，当前价格：{price:.2f}"
self._notify(msg)
self._send_alert_email("沪金提醒 — 下破", msg)
self.au_lower_last_ts = now
self.au_lower_triggered = True
if price > self.au_lower_target:
self.au_lower_triggered = False
if symbol == "INTL":
if self.intl_upper_target is not None:
if price >= self.intl_upper_target and not self.intl_upper_triggered:
if now - self.intl_upper_last_ts >= self.alert_cooldown_sec:
msg = f"Gold {self.intl_upper_target}，当前价格：{price:.2f}"
self._notify(msg)
self._send_alert_email("国际提醒 — 上破", msg)
self.intl_upper_last_ts = now
self.intl_upper_triggered = True
if price < self.intl_upper_target:
self.intl_upper_triggered = False
if self.intl_lower_target is not None:
if price <= self.intl_lower_target and not self.intl_lower_triggered:
if now - self.intl_lower_last_ts >= self.alert_cooldown_sec:
msg = f"Gold {self.intl_lower_target}，当前价格：{price:.2f}"
self._notify(msg)
self._send_alert_email("国际提醒 — 下破", msg)
self.intl_lower_last_ts = now
self.intl_lower_triggered = True
if price > self.intl_lower_target:
self.intl_lower_triggered = False
def _track_au_extreme(self, price):
if not self.extreme_enabled:
return
now = time.time()
self.au_history.append((now, price))
cutoff = now - self.extreme_window_sec
while self.au_history and self.au_history[0][0] < cutoff:
self.au_history.popleft()
if len(self.au_history) < 2:
return
delta = price - self.au_history[0][1]
if abs(delta) < self.extreme_threshold:
return
if now - self.extreme_last_ts < self.extreme_cooldown_sec:
return
self.extreme_last_ts = now
self._action_queue.put(("flash", self.up_color if delta > 0 else self.down_color, self.extreme_flash_times))
def _track_intl_extreme(self, price):
if not self.extreme_enabled:
return
now = time.time()
self.intl_history.append((now, price))
cutoff = now - self.intl_extreme_window_sec
while self.intl_history and self.intl_history[0][0] < cutoff:
self.intl_history.popleft()
if len(self.intl_history) < 2:
return
delta = price - self.intl_history[0][1]
if abs(delta) < self.intl_extreme_threshold:
return
if now - self.intl_extreme_last_ts < self.extreme_cooldown_sec:
return
self.intl_extreme_last_ts = now
self._action_queue.put(("flash", self.up_color if delta > 0 else self.down_color, self.extreme_flash_times))
def _flash_text(self, color, times):
if times <= 0 or self.flash_active:
return
self.flash_active = True
items = [self.au_icon, self.au_value_text, self.intl_icon, self.intl_value_text]
total = times * 2
def toggle(remaining):
if remaining <= 0:
for it in items:
self.canvas.itemconfig(it, fill=self.text_color,
font=(self.font_family, self.text_font_size, "bold"))
self._apply_card_style(self.card_color, self.card_border)
self.flash_active = False
return
use_color = remaining % 2 == 0
fill = color if use_color else self.text_color
size = self.flash_text_font_size if use_color else self.text_font_size
for it in items:
self.canvas.itemconfig(it, fill=fill, font=(self.font_family, size, "bold"))
if use_color:
self._apply_card_style(self.card_highlight_color, self.card_border_highlight)
else:
self._apply_card_style(self.card_color, self.card_border)
self.root.after(self.extreme_flash_interval_ms, lambda: toggle(remaining - 1))
toggle(total)
def _update_ui_cycle(self):
try:
while True:
action = self._action_queue.get_nowait()
if action[0] == "flash":
self._flash_text(action[1], action[2])
except queue.Empty:
pass
with self._price_lock:
au = self.au
prev_au = self.prev_au
intl = self.intl
prev_intl = self.prev_intl
if au is not None:
self.canvas.itemconfig(self.au_value_text, text=f"{au:.2f}")
if prev_au is not None:
arrow = "↑" if au > prev_au else ("↓" if au < prev_au else "•")
clr = self.up_color if au > prev_au else (self.down_color if au < prev_au else self.muted_color)
self.canvas.itemconfig(self.au_arrow_text, text=arrow, fill=clr)
if intl is not None:
self.canvas.itemconfig(self.intl_value_text, text=f"{intl:.2f}")
if prev_intl is not None:
arrow = "↑" if intl > prev_intl else ("↓" if intl < prev_intl else "•")
clr = self.up_color if intl > prev_intl else (self.down_color if intl < prev_intl else self.muted_color)
self.canvas.itemconfig(self.intl_arrow_text, text=arrow, fill=clr)
self.root.after(200, self._update_ui_cycle)
def _data_fetch_loop(self):
while True:
try:
new_au = self._fetch_au()
new_intl = self._fetch_intl()
if new_au is not None:
with self._price_lock:
self.prev_au = self.au
self.au = new_au
self._check_alert(new_au, "AU")
self._track_au_extreme(new_au)
if new_intl is not None:
with self._price_lock:
self.prev_intl = self.intl
self.intl = new_intl
self._check_alert(new_intl, "INTL")
self._track_intl_extreme(new_intl)
except Exception as e:
print(f"数据轮询异常: {e}")
time.sleep(self.interval + random.uniform(0.08, 0.7))
def run(self):
try:
self.root.mainloop()
except KeyboardInterrupt:
pass
finally:
try:
self.root.destroy()
except Exception:
pass
if __name__ == "__main__":
# 运行前检查依赖
try:
import requests, PIL, pystray
except ImportError:
print("缺少依赖库，请执行：pip install requests pillow pystray")
os.system("pause")
sys.exit(1)
app = GoldTaskbarDoubleLine()
app.run() |
免费评分
-
查看全部评分
|
[复制链接]
发表于 2026-3-18 22:31
|
发表于 2026-3-19 15:10
