267 lines
9.6 KiB
Python
267 lines
9.6 KiB
Python
import sys
|
||
import platform
|
||
import requests
|
||
from PyQt5.QtWidgets import QApplication, QMainWindow, QSystemTrayIcon, QMenu, QAction,QActionGroup
|
||
from PyQt5.QtGui import QIcon
|
||
from PyQt5.QtCore import QTimer,pyqtSignal,QThread
|
||
import json
|
||
import aiohttp
|
||
import asyncio
|
||
|
||
class TrayIconExample(QMainWindow):
|
||
|
||
speed_data = pyqtSignal(dict)
|
||
def __init__(self):
|
||
super(TrayIconExample, self).__init__()
|
||
|
||
self.init_ui()
|
||
# 启动异步任务
|
||
|
||
self.worker_thread = WorkerThread(self)
|
||
self.worker_thread.data_ready.connect(self.update_speed_data)
|
||
self.worker_thread.start()
|
||
|
||
def init_ui(self):
|
||
self.setWindowTitle('Clash Tray')
|
||
|
||
# 创建系统托盘图标
|
||
tray_icon = QSystemTrayIcon(QIcon('img/icon.png'), self)
|
||
tray_icon.setToolTip('Clash Tray')
|
||
|
||
# 创建右键菜单
|
||
context_menu = QMenu(self)
|
||
proxy_mode = QMenu('代理模式', self)
|
||
exit_action = QAction('退出', self)
|
||
self.tun_mode = QAction('TUN模式', self)
|
||
self.tun_mode.setCheckable(True)
|
||
# 获取环境变量
|
||
self.copy_env=QAction('复制环境变量',self)
|
||
|
||
self.more_menu=QMenu('更多',self)
|
||
self.upload_speed=QAction('⬆️上传速度',self)
|
||
self.download_speed=QAction('⬇️下载速度',self)
|
||
self.upload_speed.setCheckable(False)
|
||
self.upload_speed.setEnabled(False)
|
||
self.download_speed.setCheckable(False)
|
||
self.download_speed.setEnabled(False)
|
||
|
||
# self.timer = QTimer(self)
|
||
# self.timer.timeout.connect(self.update_data)
|
||
# self.timer.start(1000) # 定时器间隔为1000毫秒（1秒）
|
||
|
||
# 更多 子菜单项设置
|
||
self.version=self.get_version()
|
||
self.show_version = QAction('内核版本：'+self.version, self)
|
||
self.show_version.setCheckable(False)
|
||
self.show_version.setEnabled(False)
|
||
self.restart_core=QAction('重启Clash',self)
|
||
|
||
self.more_menu.addAction(self.restart_core)
|
||
self.more_menu.addAction(self.show_version)
|
||
|
||
|
||
self.restart_core.triggered.connect(self.restart_clash)
|
||
|
||
|
||
# 创建代理模式 QActionGroup
|
||
proxy_mode_group = QActionGroup(self)
|
||
proxy_mode_group.setExclusive(True) # 设置为单选模式
|
||
# 设置代理模式菜单
|
||
self.direct_mode=QAction('直连模式',self)
|
||
self.rule_mode=QAction('规则模式',self)
|
||
self.global_mode=QAction('全局模式',self)
|
||
self.direct_mode.setCheckable(True)
|
||
self.rule_mode.setCheckable(True)
|
||
self.global_mode.setCheckable(True)
|
||
|
||
proxy_mode_group.addAction(self.direct_mode)
|
||
proxy_mode_group.addAction(self.rule_mode)
|
||
proxy_mode_group.addAction(self.global_mode)
|
||
|
||
self.tun_mode.triggered.connect(self.change_tun_mode)
|
||
exit_action.triggered.connect(self.exit_application)
|
||
# 将菜单添加到托盘图标
|
||
context_menu.addAction(self.tun_mode)
|
||
context_menu.addMenu(proxy_mode)
|
||
context_menu.addAction(self.copy_env)
|
||
context_menu.addMenu(self.more_menu)
|
||
context_menu.addActions([self.upload_speed,self.download_speed,exit_action])
|
||
proxy_mode.addActions([self.direct_mode,self.rule_mode,self.global_mode])
|
||
|
||
self.copy_env.triggered.connect(self.copy_env_var)
|
||
|
||
|
||
|
||
self.direct_mode.triggered.connect(lambda:self.set_proxy_mode(0))
|
||
self.rule_mode.triggered.connect(lambda:self.set_proxy_mode(1))
|
||
self.global_mode.triggered.connect(lambda:self.set_proxy_mode(2))
|
||
|
||
|
||
# 将右键菜单设置给托盘图标
|
||
tray_icon.setContextMenu(context_menu)
|
||
|
||
# 显示托盘图标
|
||
tray_icon.show()
|
||
self.get_config()
|
||
|
||
self.setGeometry(200, 200, 1200, 800)
|
||
|
||
def restart_clash(self):
|
||
url,port=self.get_connect()
|
||
data={'path': '','payload': ''}
|
||
response=requests.post(url=url+':'+port+'/restart').json()
|
||
if response['status']=='ok':
|
||
print('重启成功')
|
||
|
||
def get_connect(self,url=None,port=None):
|
||
url='http://127.0.0.1'
|
||
port=str(9090)
|
||
return url,port
|
||
|
||
def get_config(self):
|
||
url,port=self.get_connect(self)
|
||
response=requests.get(url=url+':'+port+'/configs')
|
||
# print(response.json['mode'])
|
||
if response.status_code==200:
|
||
res=response.json()
|
||
mode=res['mode']
|
||
if mode=='direct':
|
||
self.direct_mode.setChecked(True)
|
||
elif mode=='rule':
|
||
self.rule_mode.setChecked(True)
|
||
else:
|
||
self.global_mode.setChecked(True)
|
||
tun=res['tun']['enable']
|
||
print('tun:',tun)
|
||
if tun:
|
||
self.tun_mode.setChecked(True)
|
||
else:
|
||
self.tun_mode.setChecked(False)
|
||
return response
|
||
|
||
def get_version(self):
|
||
url,port=self.get_connect()
|
||
response=requests.get(url=url+':'+port+'/version')
|
||
if response.status_code==200:
|
||
res=response.json()
|
||
print('version',res)
|
||
return res['version']
|
||
def set_proxy_mode(self,type):
|
||
mode=None
|
||
if type==0:
|
||
mode='direct'
|
||
elif type==1:
|
||
mode='rule'
|
||
elif type==2:
|
||
mode='global'
|
||
url,port=self.get_connect()
|
||
data = {"mode": mode}
|
||
response=requests.patch(url=url+':'+port+'/configs?force=true',json=data)
|
||
if response.status_code==204:
|
||
self.get_config()
|
||
def change_tun_mode(self):
|
||
url,port=self.get_connect()
|
||
print(self.tun_mode.isChecked())
|
||
data = {"tun": {"enable": False if self.tun_mode.isChecked() else True}}
|
||
print(data)
|
||
response=requests.patch(url=url+':'+port+'/configs?force=true',json=data)
|
||
if response.status_code==204:
|
||
self.get_config()
|
||
|
||
def copy_env_var(self):
|
||
sys_info=platform.uname()
|
||
clipboard = QApplication.clipboard()
|
||
env_var=''
|
||
# print(sys_info.system)
|
||
if sys_info.system == ('Linux' or 'Darwin'):
|
||
env_var='export https_proxy=http://127.0.0.1:7890 http_proxy=http://127.0.0.1:7890 all_proxy=socks5://127.0.0.1:7890'
|
||
elif sys_info.system == 'Windows':
|
||
env_var='setx https_proxy=http://127.0.0.1:7890\nsetx http_proxy=http://127.0.0.1:7890\nsetx all_proxy=socks5://127.0.0.1:7890'
|
||
|
||
# print(env_var)
|
||
clipboard.setText(env_var)
|
||
|
||
def update_speed_data(self,data):
|
||
print(data)
|
||
upload=data['up']
|
||
down=data['down']
|
||
|
||
|
||
if upload /1024<=1024 or down /1024<=1024:
|
||
upload=round(upload/1024,2)
|
||
down=round(down/1024,2)
|
||
self.upload_speed.setText('上传速度：'+str(upload)+'KB/s')
|
||
self.download_speed.setText('下载速度：'+str(down)+'KB/s')
|
||
else:
|
||
upload=round(upload/1024/1024,2)
|
||
down=round(down/1024/1024,2)
|
||
self.upload_speed.setText('上传速度：'+str(upload)+'MB/s')
|
||
self.download_speed.setText('下载速度：'+str(down)+'MB/s')
|
||
|
||
pass
|
||
|
||
def exit_application(self):
|
||
QApplication.quit()
|
||
|
||
|
||
# async def fetch_and_output_data(self):
|
||
# url = "http://127.0.0.1:9090/traffic" # 替换为实际的 API 地址
|
||
# async with aiohttp.ClientSession() as session:
|
||
# async with session.get(url, timeout=None) as response: # timeout=None 表示不设置超时
|
||
# while True:
|
||
# partial_data = await response.content.read(100) # 每次读取100字节的数据示例
|
||
|
||
# if not partial_data:
|
||
# break # 如果没有更多数据，退出循环
|
||
|
||
# # 在此处可以进行异步加载部分数据的操作，例如使用 Qt 的异步加载机制
|
||
# # print("Partial data:", partial_data)
|
||
|
||
# # 将字节数据转换为字符串
|
||
# partial_data_str = partial_data.decode('utf-8')
|
||
|
||
# print(partial_data_str)
|
||
|
||
# # 将字符串解析为 JSON 对象
|
||
# try:
|
||
# partial_data_json = json.loads(partial_data_str)
|
||
# print("Partial data:", partial_data_json)
|
||
# self.speed_data.emit(partial_data_json)
|
||
# except json.JSONDecodeError as e:
|
||
# print(f"Error decoding JSON: {e}")
|
||
|
||
# await asyncio.sleep(1) # 休眠1秒，模拟每秒输出一部分数据
|
||
|
||
class WorkerThread(QThread):
|
||
data_ready = pyqtSignal(dict)
|
||
|
||
def run(self):
|
||
loop = asyncio.new_event_loop()
|
||
asyncio.set_event_loop(loop)
|
||
loop.run_until_complete(self.fetch_and_output_data())
|
||
|
||
async def fetch_and_output_data(self):
|
||
url = "http://127.0.0.1:9090/traffic"
|
||
async with aiohttp.ClientSession() as session:
|
||
async with session.get(url, timeout=None) as response:
|
||
while True:
|
||
partial_data = await response.content.read(100)
|
||
|
||
if not partial_data:
|
||
break
|
||
|
||
partial_data_str = partial_data.decode('utf-8')
|
||
|
||
try:
|
||
partial_data_json = json.loads(partial_data_str)
|
||
self.data_ready.emit(partial_data_json)
|
||
except json.JSONDecodeError as e:
|
||
print(f"解析 JSON 时出错：{e}")
|
||
|
||
await asyncio.sleep(1)
|
||
|
||
|
||
if __name__ == '__main__':
|
||
app = QApplication(sys.argv)
|
||
window = TrayIconExample()
|
||
sys.exit(app.exec_())
|