ClashTray/main/main.py
2024-01-07 16:52:31 +08:00

235 lines
8 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.traffic_thread = TrafficThread(self)
self.traffic_thread.data_ready.connect(self.update_speed_data)
self.traffic_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.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)
@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}')
def exit_application(self):
QApplication.quit()
class TrafficThread(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_())