273 lines
9.6 KiB
Python
273 lines
9.6 KiB
Python
import sys,os
|
|
import platform
|
|
import requests
|
|
from PyQt5.QtWidgets import QApplication, QMainWindow, QSystemTrayIcon, QMenu, QAction,QActionGroup
|
|
from PyQt5.QtGui import QIcon
|
|
from PyQt5.QtCore import 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.traffic_thread = TrafficThread(self)
|
|
self.traffic_thread.data_ready.connect(self.update_speed_data)
|
|
self.traffic_thread.memory_data.connect(self.update_memory_data)
|
|
self.traffic_thread.start()
|
|
|
|
def init_ui(self):
|
|
self.setWindowTitle('Clash Tray')
|
|
|
|
# 创建系统托盘图标
|
|
script_dir = os.path.dirname(os.path.realpath('__file__'))
|
|
print(script_dir)
|
|
# 构建图标文件的路径
|
|
icon_path = os.path.join(script_dir, 'img', 'icon.png')
|
|
tray_icon = QSystemTrayIcon(QIcon(icon_path), 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.inuse_memory=QAction('内存占用',self)
|
|
self.inuse_memory.setCheckable(False)
|
|
self.inuse_memory.setEnabled(False)
|
|
|
|
# 更多 子菜单项设置
|
|
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.addSeparator()
|
|
context_menu.addAction(self.inuse_memory)
|
|
context_menu.addActions([self.upload_speed,self.download_speed])
|
|
context_menu.addSeparator()
|
|
context_menu.addAction(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)
|
|
|
|
@staticmethod
|
|
def convert_speed(speed):
|
|
units = ['B/s', 'KB/s', 'MB/s', 'GB/s']
|
|
index = 0
|
|
while speed >= 1024 and index < len(units) - 1:
|
|
speed /= 1024.0
|
|
index += 1
|
|
return f'{speed:.2f} {units[index]}'
|
|
|
|
|
|
# 设置网速
|
|
def update_speed_data(self,data):
|
|
print(data)
|
|
upload=data['up']
|
|
down=data['down']
|
|
converted_upload_speed = self.convert_speed(upload)
|
|
converted_download_speed = self.convert_speed(down)
|
|
self.upload_speed.setText(f'↑ {converted_upload_speed}')
|
|
self.download_speed.setText(f'↓ {converted_download_speed}')
|
|
|
|
@staticmethod
|
|
def convert_memory(speed):
|
|
units = ['B', 'KB', 'MB', 'GB']
|
|
index = 0
|
|
while speed >= 1024 and index < len(units) - 1:
|
|
speed /= 1024.0
|
|
index += 1
|
|
return f'{speed:.2f} {units[index]}'
|
|
# 设置内存
|
|
def update_memory_data(self,data):
|
|
print(data)
|
|
memory=data['inuse']
|
|
convert_memory=self.convert_memory(memory)
|
|
self.inuse_memory.setText(f'内存占用：{convert_memory}')
|
|
|
|
def exit_application(self):
|
|
# 退出前关闭线程
|
|
if self.traffic_thread.isRunning():
|
|
# self.traffic_thread.terminate()
|
|
self.traffic_thread.stop()
|
|
self.traffic_thread.wait()
|
|
QApplication.quit()
|
|
|
|
|
|
class TrafficThread(QThread):
|
|
data_ready = pyqtSignal(dict)
|
|
memory_data= pyqtSignal(dict)
|
|
def run(self):
|
|
loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(loop)
|
|
# 多个协程
|
|
loop.run_until_complete(asyncio.gather(self.fetch_json_data("http://127.0.0.1:9090/traffic",self.data_ready),
|
|
self.fetch_json_data("http://127.0.0.1:9090/memory",self.memory_data)))
|
|
|
|
async def fetch_json_data(self,url,signal):
|
|
self.should_stop = False
|
|
# url = "http://127.0.0.1:9090/traffic"
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(url, timeout=None) as response:
|
|
while not self.should_stop:
|
|
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)
|
|
signal.emit(partial_data_json)
|
|
except json.JSONDecodeError as e:
|
|
print(f"解析 JSON 时出错：{e}")
|
|
await asyncio.sleep(1)
|
|
def stop(self):
|
|
self.should_stop = True
|
|
|
|
if __name__ == '__main__':
|
|
app = QApplication(sys.argv)
|
|
window = TrayIconExample()
|
|
sys.exit(app.exec_())
|