RoyirinRosmontis/Private-Proxy
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 21 Wiki Activity

上传文件至「lite」

Browse Source
This commit is contained in:
RoyirinRosmontis
2026-08-23 02:29:27 +08:00
parent 3a521daefd
commit 6c9a04b754
1 changed files with 175 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lite/proxy_lite.py
+175 -46
View File
@@ -6,7 +6,7 @@ from pathlib import Path
from PyQt5.QtWidgets import (
QApplication, QWidget, QLabel, QPushButton,
QVBoxLayout, QHBoxLayout, QFrame, QMessageBox, QTextEdit,
QCheckBox,
QCheckBox, QMenu,
)
from PyQt5.QtCore import Qt, QTimer, pyqtSignal
from PyQt5.QtGui import QFont, QIcon, QPixmap, QImage, QColor
@@ -93,10 +93,38 @@ def _load_nodes():
log_red(f"nodes.json 加载失败: {e}")
return _NODES
_SELECTED_NODE = 0 # 当前选中节点下标（默认第一条）
def _current_node():
"""当前使用的节点：默认取列表第一个；无节点时返回 None。"""
nodes = _load_nodes()
return nodes[0] if nodes else None
if not nodes:
return None
i = _SELECTED_NODE
return nodes[i] if 0 <= i < len(nodes) else nodes[0]
def _selected_index():
"""当前选中节点下标。"""
return _SELECTED_NODE
def _switch_node(index):
"""切换当前节点下标；相同或越界返回 False。"""
global _SELECTED_NODE
nodes = _load_nodes()
if not nodes or not (0 <= index < len(nodes)):
log_red("节点切换失败: 列表为空或下标越界")
return False
if index == _SELECTED_NODE:
log("已在当前节点")
return False
_SELECTED_NODE = index
return True
def _reload_nodes():
"""强制重新读取 nodes.json（菜单打开时调用，外部改动即时生效）。"""
global _NODES
_NODES = None
return _load_nodes()
def _node_val(key, default=""):
node = _current_node()
@@ -243,7 +271,7 @@ def log_red(msg: str):
def _run(cmd, warn=True):
"""无窗口执行子进程，失败时记录警告日志。"""
r = subprocess.run(cmd, capture_output=True, text=True,
creationflags=subprocess.CREATE_NO_WINDOW)
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0))
if warn and r.returncode != 0:
err = r.stderr.strip() if r.stderr else ""
log(f"[警告] {' '.join(str(c) for c in cmd)} -> {r.returncode} {err}")
@@ -819,6 +847,35 @@ def load_icon():
img.fill(QColor(C_TITLE))
return QIcon(QPixmap.fromImage(img))
# ── 可点击节点标签 ─────────────────────────────────────
_NL_BASE = f"color: {C_TITLE}; font-size: 12px;"
_NL_HOVER = _NL_BASE + " text-decoration: underline;"
class NodeLabel(QLabel):
"""节点标签：hover 显示下划线+手型光标，点击发出 clicked 信号。"""
clicked = pyqtSignal()
def __init__(self, text=""):
super().__init__(text)
self.setStyleSheet(_NL_BASE)
self.setToolTip("点击切换节点")
def enterEvent(self, e):
self.setStyleSheet(_NL_HOVER)
self.setCursor(Qt.PointingHandCursor)
super().enterEvent(e)
def leaveEvent(self, e):
self.setStyleSheet(_NL_BASE)
self.unsetCursor()
super().leaveEvent(e)
def mouseReleaseEvent(self, e):
if e.button() == Qt.LeftButton:
self.clicked.emit()
super().mouseReleaseEvent(e)
# ── 状态指示灯（小圆点） ───────────────────────────────
class StatusDot(QLabel):
@@ -914,11 +971,17 @@ class MainWindow(QWidget):
title.setAlignment(Qt.AlignCenter)
layout.addWidget(title)
# 节点副标题
subtitle = QLabel(f"节点: {_endpoint_domain()} ({_endpoint_country()})")
subtitle.setStyleSheet(f"color: {C_TITLE}; font-size: 12px;")
subtitle.setAlignment(Qt.AlignCenter)
layout.addWidget(subtitle)
# 节点副标题（域名可点击切换：hover 下划线+手型提示，点击弹节点列表）
node_row = QHBoxLayout()
node_row.addStretch()
prefix = QLabel("节点: ")
prefix.setStyleSheet(f"color: {C_TITLE}; font-size: 12px;")
node_row.addWidget(prefix)
self.subtitle = NodeLabel(f"{_endpoint_domain()} ({_endpoint_country()})")
self.subtitle.clicked.connect(self._show_node_menu)
node_row.addWidget(self.subtitle)
node_row.addStretch()
layout.addLayout(node_row)
# 连接方式选择区
pac_layout = QHBoxLayout()
@@ -929,7 +992,7 @@ class MainWindow(QWidget):
self.cb_bypass_lan.setStyleSheet(f"color: {C_SUBTEXT}; font-size: 12px;")
pac_layout.addWidget(self.cb_bypass_lan)
self.cb_pac = QCheckBox("PAC 规则模式")
self.cb_pac.setChecked(False) # 默认关闭 PAC，由用户按需开启
self.cb_pac.setChecked(True) # 默认开启智能分流（域名表/中国IP判定已足够完善）
self.cb_pac.setStyleSheet(f"color: {C_SUBTEXT}; font-size: 12px;")
self.cb_pac.setToolTip("只代理海外服务，使用域名和解析 IP 匹配。")
pac_layout.addWidget(self.cb_pac)
@@ -1086,48 +1149,52 @@ class MainWindow(QWidget):
threading.Thread(target=self._start_thread, args=(use_pac, use_bypass_lan), daemon=True).start()
def _start_thread(self, use_pac, use_bypass_lan):
try:
_traffic_reset()
log("TLS 直连模式")
ep = _endpoint_domain()
log(f" 目标域名: {ep}")
# PAC 模式：全部流量进转发器，按域名表/解析 IP 智能分流
if use_pac:
_load_cn_ranges()
_load_force_proxy_suffixes()
_tls_start(ep, _endpoint_port(), use_tls=True, bypass_lan=use_bypass_lan, smart=use_pac)
if use_pac:
# 本地生成全量转发 PAC（不再从服务器拉取），全部流量交给转发器做智能分流
pac_content = _build_pac(f"{PROXY_HOST}:{PROXY_PORT}")
if use_bypass_lan:
pac_content = _apply_lan_bypass(pac_content)
# 在内存中启动本地 PAC HTTP 服务，不往 EXE 目录写任何文件
pac_port = _start_pac_server(pac_content)
pac_url = f"http://127.0.0.1:{pac_port}/proxy.pac"
log(f"使用 PAC: {pac_url}")
proxy_on(pac_url=pac_url)
else:
proxy_on(bypass_lan=use_bypass_lan)
if use_bypass_lan:
log("不代理内网 IP：内网地址将直连")
self._proxy_set_by_us = True # 标记系统代理由本程序设置，退出时据此清理
log("验证代理连通性...")
import urllib.request
# baidu 验证国内直连路径，google 验证境外隧道路径
test_urls = [("baidu.com", "http://www.baidu.com"), ("google.com", "http://www.google.com")]
for name, url in test_urls:
try:
req = urllib.request.Request(url)
req.set_proxy(f"{PROXY_HOST}:{PROXY_PORT}", "http")
urllib.request.urlopen(req, timeout=10)
log(f" {name} 可达")
except Exception as e:
log(f" {name} 不可达: {e}")
log("开启成功")
self._start_proxy(use_pac, use_bypass_lan)
self._op_done.emit("start", True)
except Exception as e:
log(f"[错误] 开启失败: {e}")
self._op_done.emit("start", False)
def _start_proxy(self, use_pac, use_bypass_lan):
"""启动本地转发器并设置系统代理（PAC/固定两种模式），可在后台线程调用。"""
_traffic_reset()
log("TLS 直连模式")
ep = _endpoint_domain()
log(f" 目标域名: {ep}")
# PAC 模式：全部流量进转发器，按域名表/解析 IP 智能分流
if use_pac:
_load_cn_ranges()
_load_force_proxy_suffixes()
_tls_start(ep, _endpoint_port(), use_tls=True, bypass_lan=use_bypass_lan, smart=use_pac)
if use_pac:
# 本地生成全量转发 PAC（不再从服务器拉取），全部流量交给转发器做智能分流
pac_content = _build_pac(f"{PROXY_HOST}:{PROXY_PORT}")
if use_bypass_lan:
pac_content = _apply_lan_bypass(pac_content)
# 在内存中启动本地 PAC HTTP 服务，不往 EXE 目录写任何文件
pac_port = _start_pac_server(pac_content)
pac_url = f"http://127.0.0.1:{pac_port}/proxy.pac"
log(f"使用 PAC: {pac_url}")
proxy_on(pac_url=pac_url)
else:
proxy_on(bypass_lan=use_bypass_lan)
if use_bypass_lan:
log("不代理内网 IP：内网地址将直连")
self._proxy_set_by_us = True # 标记系统代理由本程序设置，退出时据此清理
log("验证代理连通性...")
import urllib.request
# baidu 验证国内直连路径，google 验证境外隧道路径
test_urls = [("baidu.com", "http://www.baidu.com"), ("google.com", "http://www.google.com")]
for name, url in test_urls:
try:
req = urllib.request.Request(url)
req.set_proxy(f"{PROXY_HOST}:{PROXY_PORT}", "http")
urllib.request.urlopen(req, timeout=10)
log(f" {name} 可达")
except Exception as e:
log(f" {name} 不可达: {e}")
log("开启成功")
def on_stop(self):
"""禁用按钮并在后台线程执行关闭流程。"""
self.btn_toggle.setEnabled(False)
@@ -1146,6 +1213,67 @@ class MainWindow(QWidget):
self._op_done.emit("stop", False)
# ── 节点切换 ──
def _show_node_menu(self):
"""弹出节点切换菜单（每次打开重新读取 nodes.json，外部改动即时生效）。"""
if self._proxy_set_by_us:
log("代理运行中，请先关闭代理再切换节点")
return
nodes = _reload_nodes()
if not nodes:
QMessageBox.information(self, "提示", "nodes.json 中没有可用节点")
return
menu = QMenu(self)
menu.setStyleSheet(f"""
QMenu {{
background-color: {C_CARD};
border: 1px solid {C_BORDER};
border-radius: 8px;
padding: 4px;
}}
QMenu::item {{
padding: 6px 18px;
border-radius: 6px;
color: {C_TEXT};
font-size: 12px;
}}
QMenu::item:selected {{
background-color: #e3e8ee;
}}
""")
cur = _selected_index()
for i, n in enumerate(nodes):
domain = n.get("domain") or "???"
country = n.get("country") or ""
remark = n.get("remark") or ""
text = f"{domain} ({country})"
if remark and remark != domain:
text += f"　{remark}"
if i == cur:
text = "✓ " + text
menu.addAction(text).triggered.connect(
lambda _=False, idx=i: self._on_node_selected(idx))
menu.exec_(self.subtitle.mapToGlobal(self.subtitle.rect().bottomLeft()))
def _on_node_selected(self, idx):
"""切换节点：更新标题。"""
if not _switch_node(idx):
return
node = _current_node()
self._refresh_subtitle()
log(f"已切换节点: {node.get('domain')} ({node.get('country')})")
def _refresh_subtitle(self):
"""刷新节点副标题文字。"""
self.subtitle.setText(f"{_endpoint_domain()} ({_endpoint_country()})")
def _update_node_label(self):
"""代理运行中时提示禁止切换节点。"""
if self._proxy_set_by_us:
self.subtitle.setToolTip("代理运行中，请先关闭代理再切换节点")
else:
self.subtitle.setToolTip("点击切换节点")
def _after_op(self, op, ok):
"""开启/关闭操作完成后刷新界面状态并恢复按钮/控件可用性。"""
self.refresh_status()
@@ -1155,6 +1283,7 @@ class MainWindow(QWidget):
is_off = (op == "stop") or (op == "start" and not ok)
self.cb_pac.setEnabled(is_off)
self.cb_bypass_lan.setEnabled(is_off)
self._update_node_label()
# ── 日志 ──
_early_logs = [] # 窗口初始化期间（log_text 创建前）到达的日志缓存