mirror of
https://gitee.com/xmz7891/ykxmz.git
synced 2026-10-18 16:00:43 +00:00
xmz7891
b62ac865ea
被控端 - 修复 MTU 问题：夹 MSS + 路径 MTU 探测（公网可用率 48% -> 100%） - 复用 HTTP 连接（免掉每次新建 TLS 的 30% 握手失败） - 每分钟回头找局域网，自动切回快 400 倍的路径 - 画面心跳帧：静止时省流量，客户端仍知道流是活的 - 锁屏时给出可操作的提示 手表端 - 内置 GTS Root R4 / GlobalSign Root CA，修 SSLHandshakeException - 连接错误翻译成人话（证书/时间/解析/超时分别提示） - 经中转连上后用服务器报的地址刷新已配对记录 - 新增局域网连接页面、服务器延迟检测 服务器端 - 支持 chunked 请求体（修每 16 秒一次的连接错乱与大面积 502） - 网络调优脚本 net-tune.sh（开机自动应用） - 管理页：批量删除、网络体检、下载区 - 隧道改 HTTP/2 + 双连接 文档 - 许可协议 / 安装前必读 / 安装后须知（作者：小麦子）
655 lines
26 KiB
Python
655 lines
26 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""再加 25 个功能，重点在娱乐和极客（总数到 105）。
|
||
|
||
娱乐（课堂课间用）：
|
||
弹幕、滚动字幕、抽奖、掷骰子、抛硬币、3-2-1 倒计时、
|
||
影院模式、随机放歌、放视频、计分板、加油动画
|
||
|
||
极客：
|
||
数字雨、星空、实时监控全屏、二维码、哈希、Base64、
|
||
UUID、时间戳、JSON 格式化、屏幕取色、网络测速、端口检查
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
FEATURES = Path(r"C:\Users\shu\Desktop\deepseekagent\LanRemote\windows"
|
||
r"\lanremote\features.py")
|
||
|
||
DEFS_ANCHOR = ' add("gpu_info", "显卡信息", "action", "媒体")'
|
||
|
||
DEFS_ADD = ''' add("gpu_info", "显卡信息", "action", "媒体")
|
||
|
||
# ---- 娱乐 ----
|
||
add("danmaku", "全屏弹幕", "input", "娱乐", hint="弹幕内容", key="text")
|
||
add("marquee", "滚动字幕", "input", "娱乐", hint="要滚动的字", key="text")
|
||
add("lottery", "抽奖", "action", "娱乐")
|
||
add("dice", "掷骰子", "action", "娱乐")
|
||
add("coin_flip", "抛硬币", "action", "娱乐")
|
||
add("countdown_321", "3-2-1 倒计时", "action", "娱乐")
|
||
add("cinema_mode", "影院模式（黑屏）", "action", "娱乐")
|
||
add("random_song", "随机放一首歌", "action", "娱乐")
|
||
add("play_video", "播放视频文件", "input", "娱乐",
|
||
hint="视频路径", key="path")
|
||
add("score_board", "显示计分板", "action", "娱乐")
|
||
add("score_change", "改分数", "input", "娱乐",
|
||
hint="A 或 B", key="team", hint2="加几分（负数减）", key2="delta")
|
||
add("cheer", "加油动画", "action", "娱乐")
|
||
|
||
# ---- 极客 ----
|
||
add("matrix_rain", "数字雨（黑客帝国）", "action", "极客")
|
||
add("starfield", "星空穿梭", "action", "极客")
|
||
add("dashboard", "全屏实时监控", "action", "极客")
|
||
add("qr_code", "生成二维码", "input", "极客",
|
||
hint="要变成二维码的内容", key="text")
|
||
add("hash_text", "算哈希（MD5/SHA）", "input", "极客",
|
||
hint="要计算的文字", key="text")
|
||
add("base64_encode", "Base64 编码", "input", "极客",
|
||
hint="原文", key="text")
|
||
add("base64_decode", "Base64 解码", "input", "极客",
|
||
hint="Base64 内容", key="text")
|
||
add("uuid_gen", "生成 UUID", "action", "极客")
|
||
add("timestamp_now", "当前时间戳", "action", "极客")
|
||
add("json_format", "JSON 格式化", "input", "极客",
|
||
hint="JSON 文本", key="text")
|
||
add("screen_color", "鼠标位置取色", "action", "极客")
|
||
add("speed_test", "网络测速", "action", "极客")
|
||
add("port_check", "检查常用端口", "input", "极客",
|
||
hint="目标 IP 或主机名", key="host")'''
|
||
|
||
IMPL_ANCHOR = " # ---------------- 持久化 ----------------"
|
||
|
||
IMPL_ADD = ''' # ==================================================================
|
||
# 娱乐
|
||
# ==================================================================
|
||
|
||
def _do_danmaku(self, params) -> str:
|
||
text = str(params.get("text") or "").strip() or "大家好"
|
||
speed = 6
|
||
|
||
def builder(win, close):
|
||
import tkinter as tk
|
||
win.configure(bg="#000000")
|
||
width = win.winfo_screenwidth()
|
||
height = win.winfo_screenheight()
|
||
canvas = tk.Canvas(win, bg="#000000", highlightthickness=0,
|
||
width=width, height=height)
|
||
canvas.pack(fill="both", expand=True")
|
||
canvas.configure(highlightthickness=0)
|
||
items = []
|
||
colors = ["#FFFFFF", "#FFD33D", "#5AC8FA", "#FF6B6B", "#8EE36A"]
|
||
for i in range(14):
|
||
y = int(height * (i + 1) / 15)
|
||
item = canvas.create_text(width + 200, y, text=text,
|
||
fill=colors[i % len(colors)],
|
||
font=("Microsoft YaHei UI", 26, "bold"),
|
||
anchor="w")
|
||
items.append([item, speed + (i % 4)])
|
||
|
||
def tick():
|
||
alive = False
|
||
for entry in items:
|
||
canvas.move(entry[0], -entry[1], 0)
|
||
x, _y = canvas.coords(entry[0])
|
||
if x > -600:
|
||
alive = True
|
||
if alive:
|
||
win.after(30, tick)
|
||
else:
|
||
close()
|
||
|
||
tick()
|
||
|
||
self._fullscreen_window(builder)
|
||
return f"弹幕已开始：{text[:20]}"
|
||
|
||
def _do_marquee(self, params) -> str:
|
||
text = str(params.get("text") or "").strip() or "欢迎"
|
||
|
||
def builder(win, close):
|
||
import tkinter as tk
|
||
win.configure(bg="#0B1020")
|
||
width = win.winfo_screenwidth()
|
||
height = win.winfo_screenheight()
|
||
canvas = tk.Canvas(win, bg="#0B1020", highlightthickness=0)
|
||
canvas.pack(fill="both", expand=True)
|
||
item = canvas.create_text(width, height // 2, text=text,
|
||
fill="#FFD33D",
|
||
font=("Microsoft YaHei UI", 60, "bold"),
|
||
anchor="w")
|
||
|
||
def tick():
|
||
canvas.move(item, -4, 0)
|
||
x, _y = canvas.coords(item)
|
||
if x < -len(text) * 70:
|
||
close()
|
||
return
|
||
win.after(25, tick)
|
||
|
||
tick()
|
||
|
||
self._fullscreen_window(builder)
|
||
return f"字幕滚动中：{text[:20]}"
|
||
|
||
def _do_lottery(self, params) -> str:
|
||
import random
|
||
names = self.store.get("roll_call_names") or []
|
||
pool = names or [str(i) for i in range(1, 51)]
|
||
|
||
def builder(win, close):
|
||
import tkinter as tk
|
||
win.configure(bg="#0B1020")
|
||
label = tk.Label(win, text="", bg="#0B1020", fg="#FFD33D",
|
||
font=("Microsoft YaHei UI", 120, "bold"))
|
||
label.pack(expand=True, fill="both")
|
||
state = {"n": 0, "winner": random.choice(pool)}
|
||
|
||
def tick():
|
||
state["n"] += 1
|
||
if state["n"] <= 22:
|
||
label.configure(text=random.choice(pool), fg="#FFD33D")
|
||
win.after(60 + state["n"] * 12, tick)
|
||
else:
|
||
label.configure(text=state["winner"], fg="#5AC8FA")
|
||
try:
|
||
win.bell()
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
|
||
tick()
|
||
|
||
self._fullscreen_window(builder)
|
||
return "抽奖中…"
|
||
|
||
def _do_dice(self, params) -> str:
|
||
import random
|
||
faces = ["⚀", "⚁", "⚂", "⚃", "⚄", "⚅"]
|
||
result = random.randint(1, 6)
|
||
|
||
def builder(win, close):
|
||
import tkinter as tk
|
||
win.configure(bg="#0B1020")
|
||
label = tk.Label(win, text="", bg="#0B1020", fg="#FFFFFF",
|
||
font=("Segoe UI Symbol", 200, "bold"))
|
||
label.pack(expand=True, fill="both")
|
||
state = {"n": 0, "final": faces[result - 1]}
|
||
|
||
def tick():
|
||
state["n"] += 1
|
||
if state["n"] <= 14:
|
||
label.configure(text=random.choice(faces))
|
||
win.after(70, tick)
|
||
else:
|
||
label.configure(text=state["final"], fg="#5AC8FA")
|
||
|
||
tick()
|
||
|
||
self._fullscreen_window(builder)
|
||
return f"掷出 {result} 点"
|
||
|
||
def _do_coin_flip(self, params) -> str:
|
||
import random
|
||
result = random.choice(["正", "反"])
|
||
|
||
def builder(win, close):
|
||
import tkinter as tk
|
||
win.configure(bg="#0B1020")
|
||
label = tk.Label(win, text="", bg="#0B1020", fg="#FFFFFF",
|
||
font=("Microsoft YaHei UI", 200, "bold"))
|
||
label.pack(expand=True, fill="both")
|
||
state = {"n": 0}
|
||
|
||
def tick():
|
||
state["n"] += 1
|
||
if state["n"] <= 16:
|
||
label.configure(text=random.choice(["正", "反"]))
|
||
win.after(80, tick)
|
||
else:
|
||
label.configure(text=result, fg="#FFD33D")
|
||
|
||
tick()
|
||
|
||
self._fullscreen_window(builder)
|
||
return f"硬币：{result}"
|
||
|
||
def _do_countdown_321(self, params) -> str:
|
||
def builder(win, close):
|
||
import tkinter as tk
|
||
win.configure(bg="#0B1020")
|
||
label = tk.Label(win, text="", bg="#0B1020", fg="#FFFFFF",
|
||
font=("Consolas", 260, "bold"))
|
||
label.pack(expand=True, fill="both")
|
||
state = {"n": 3}
|
||
|
||
def tick():
|
||
if state["n"] <= 0:
|
||
label.configure(text="开始！", fg="#8EE36A",
|
||
font=("Microsoft YaHei UI", 120, "bold"))
|
||
win.after(900, close)
|
||
return
|
||
label.configure(text=str(state["n"]),
|
||
fg="#FF6B6B" if state["n"] == 1 else "#FFFFFF")
|
||
try:
|
||
win.bell()
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
state["n"] -= 1
|
||
win.after(1000, tick)
|
||
|
||
tick()
|
||
|
||
self._fullscreen_window(builder)
|
||
return "倒计时开始"
|
||
|
||
def _do_cinema_mode(self, params) -> str:
|
||
def builder(win, close):
|
||
import tkinter as tk
|
||
win.configure(bg="#000000")
|
||
tk.Label(win, text="影院模式", bg="#000000", fg="#1A1A1A",
|
||
font=("Microsoft YaHei UI", 40, "bold")).pack(
|
||
expand=True, fill="both")
|
||
|
||
self._fullscreen_window(builder)
|
||
return "影院模式（Esc 退出）"
|
||
|
||
def _do_random_song(self, params) -> str:
|
||
import random
|
||
files = self.actions.list_local_audio()
|
||
if not files:
|
||
raise ValueError("音乐目录里没有音频文件")
|
||
picked = random.choice(files)
|
||
self.actions.play_audio(picked["path"])
|
||
return f"正在播放：{picked['name']}"
|
||
|
||
def _do_play_video(self, params) -> str:
|
||
target = self.resolve_in_roots(str(params.get("path") or ""))
|
||
if not target.is_file():
|
||
raise ValueError("视频文件不存在")
|
||
subprocess.Popen(["cmd", "/c", "start", "", str(target)],
|
||
creationflags=_NO_WINDOW)
|
||
return f"已用播放器打开 {target.name}"
|
||
|
||
def _scores(self) -> dict:
|
||
value = self.store.get("score_board")
|
||
if not isinstance(value, dict):
|
||
value = {"A": 0, "B": 0}
|
||
value.setdefault("A", 0)
|
||
value.setdefault("B", 0)
|
||
return value
|
||
|
||
def _show_score_board(self) -> None:
|
||
scores = self._scores()
|
||
|
||
def builder(win, close):
|
||
import tkinter as tk
|
||
win.configure(bg="#0B1020")
|
||
row = tk.Frame(win, bg="#0B1020")
|
||
row.pack(expand=True, fill="both")
|
||
for team, color in (("A", "#5AC8FA"), ("B", "#FFD33D")):
|
||
box = tk.Frame(row, bg="#0B1020")
|
||
box.pack(side="left", expand=True, fill="both")
|
||
tk.Label(box, text=f"{team} 队", bg="#0B1020", fg=color,
|
||
font=("Microsoft YaHei UI", 40, "bold")).pack(pady=(60, 0))
|
||
tk.Label(box, text=str(scores.get(team, 0)), bg="#0B1020",
|
||
fg="#FFFFFF", font=("Consolas", 200, "bold")).pack(
|
||
expand=True)
|
||
|
||
self._fullscreen_window(builder)
|
||
|
||
def _do_score_board(self, params) -> str:
|
||
self._show_score_board()
|
||
scores = self._scores()
|
||
return f"A 队 {scores['A']} : {scores['B']} B 队"
|
||
|
||
def _do_score_change(self, params) -> str:
|
||
team = str(params.get("team") or "A").strip().upper()
|
||
if team not in ("A", "B"):
|
||
raise ValueError("队伍只能填 A 或 B")
|
||
try:
|
||
delta = int(params.get("delta") or 0)
|
||
except (TypeError, ValueError):
|
||
raise ValueError("分数要填数字")
|
||
scores = self._scores()
|
||
scores[team] = scores[team] + delta
|
||
self.store.set("score_board", scores)
|
||
self._show_score_board()
|
||
return f"A 队 {scores['A']} : {scores['B']} B 队"
|
||
|
||
def _do_cheer(self, params) -> str:
|
||
def builder(win, close):
|
||
import tkinter as tk
|
||
import random
|
||
win.configure(bg="#0B1020")
|
||
width = win.winfo_screenwidth()
|
||
height = win.winfo_screenheight()
|
||
canvas = tk.Canvas(win, bg="#0B1020", highlightthickness=0)
|
||
canvas.pack(fill="both", expand=True)
|
||
canvas.create_text(width // 2, height // 2, text="加 油 !",
|
||
fill="#FFD33D",
|
||
font=("Microsoft YaHei UI", 120, "bold"))
|
||
colors = ["#FF6B6B", "#FFD33D", "#5AC8FA", "#8EE36A", "#FFFFFF"]
|
||
flakes = []
|
||
for _ in range(90):
|
||
x = random.randint(0, width)
|
||
y = random.randint(-height, 0)
|
||
item = canvas.create_rectangle(x, y, x + 10, y + 16,
|
||
fill=random.choice(colors),
|
||
outline="")
|
||
flakes.append(item)
|
||
|
||
def tick():
|
||
for item in flakes:
|
||
canvas.move(item, 0, 7)
|
||
_x, y = canvas.coords(item)[:2]
|
||
if y > height:
|
||
canvas.coords(item, random.randint(0, width), -20,
|
||
random.randint(0, width) + 10, 0)
|
||
win.after(40, tick)
|
||
|
||
tick()
|
||
|
||
self._fullscreen_window(builder)
|
||
return "加油！"
|
||
|
||
# ==================================================================
|
||
# 极客
|
||
# ==================================================================
|
||
|
||
def _do_matrix_rain(self, params) -> str:
|
||
def builder(win, close):
|
||
import tkinter as tk
|
||
import random
|
||
win.configure(bg="#000000")
|
||
width = win.winfo_screenwidth()
|
||
height = win.winfo_screenheight()
|
||
canvas = tk.Canvas(win, bg="#000000", highlightthickness=0)
|
||
canvas.pack(fill="both", expand=True)
|
||
font_size = 18
|
||
columns = max(1, width // font_size)
|
||
drops = [random.randint(-40, 0) for _ in range(columns)]
|
||
chars = "01アイウエオカキクケコサシスセソタチツテト"
|
||
|
||
def tick():
|
||
canvas.delete("fade")
|
||
for i, y in enumerate(drops):
|
||
char = random.choice(chars)
|
||
canvas.create_text(i * font_size + font_size // 2,
|
||
y * font_size, text=char,
|
||
fill="#00FF66",
|
||
font=("Consolas", font_size, "bold"))
|
||
drops[i] = y + 1 if y * font_size < height else random.randint(-20, 0)
|
||
win.after(60, tick)
|
||
|
||
tick()
|
||
|
||
self._fullscreen_window(builder)
|
||
return "数字雨已开始（Esc 退出）"
|
||
|
||
def _do_starfield(self, params) -> str:
|
||
def builder(win, close):
|
||
import tkinter as tk
|
||
import random
|
||
win.configure(bg="#000010")
|
||
width = win.winfo_screenwidth()
|
||
height = win.winfo_screenheight()
|
||
canvas = tk.Canvas(win, bg="#000010", highlightthickness=0)
|
||
canvas.pack(fill="both", expand=True)
|
||
stars = []
|
||
for _ in range(240):
|
||
x = random.uniform(-width, width)
|
||
y = random.uniform(-height, height)
|
||
z = random.uniform(1, width)
|
||
item = canvas.create_oval(0, 0, 0, 0, fill="#FFFFFF", outline="")
|
||
stars.append([x, y, z, item])
|
||
|
||
def tick():
|
||
for star in stars:
|
||
star[2] -= 16
|
||
if star[2] <= 1:
|
||
star[0] = random.uniform(-width, width)
|
||
star[1] = random.uniform(-height, height)
|
||
star[2] = width
|
||
k = 220 / star[2]
|
||
px = star[0] * k + width / 2
|
||
py = star[1] * k + height / 2
|
||
size = max(1, int((1 - star[2] / width) * 4))
|
||
canvas.coords(star[3], px, py, px + size, py + size)
|
||
win.after(33, tick)
|
||
|
||
tick()
|
||
|
||
self._fullscreen_window(builder)
|
||
return "星空穿梭中（Esc 退出）"
|
||
|
||
def _do_dashboard(self, params) -> str:
|
||
def builder(win, close):
|
||
import tkinter as tk
|
||
win.configure(bg="#0B1020")
|
||
frame = tk.Frame(win, bg="#0B1020")
|
||
frame.pack(expand=True, fill="both", padx=60, pady=60)
|
||
titles = {}
|
||
for key, title, color in (("cpu", "CPU", "#5AC8FA"),
|
||
("mem", "内存", "#FFD33D"),
|
||
("net", "网络", "#8EE36A")):
|
||
box = tk.Frame(frame, bg="#0B1020")
|
||
box.pack(fill="x", pady=16)
|
||
tk.Label(box, text=title, bg="#0B1020", fg=color,
|
||
font=("Microsoft YaHei UI", 28, "bold"),
|
||
width=6, anchor="w").pack(side="left")
|
||
value = tk.Label(box, text="--", bg="#0B1020", fg="#FFFFFF",
|
||
font=("Consolas", 60, "bold"), anchor="w")
|
||
value.pack(side="left", fill="x", expand=True)
|
||
titles[key] = value
|
||
|
||
state = {"last": None, "last_bytes": None}
|
||
|
||
def tick():
|
||
info = self.actions.perf()
|
||
titles["cpu"].configure(text=f"{info.get('cpu_percent', 0)} %")
|
||
titles["mem"].configure(
|
||
text=f"{info.get('mem_percent', 0)} % "
|
||
f"({info.get('mem_used_gb')}/{info.get('mem_total_gb')} GB)")
|
||
_code, out = _run(["powershell", "-NoProfile", "-Command",
|
||
"(Get-NetAdapterStatistics | Measure-Object "
|
||
"-Property ReceivedBytes -Sum).Sum"])
|
||
digits = re.findall(r"\\d+", out or "")
|
||
now_bytes = int(digits[0]) if digits else 0
|
||
if state["last_bytes"] is not None and state["last"] is not None:
|
||
delta = now_bytes - state["last_bytes"]
|
||
elapsed = max(0.1, time.time() - state["last"])
|
||
mbps = delta / elapsed / 1048576
|
||
titles["net"].configure(text=f"{mbps:.2f} MB/s")
|
||
state["last_bytes"] = now_bytes
|
||
state["last"] = time.time()
|
||
win.after(1500, tick)
|
||
|
||
tick()
|
||
|
||
self._fullscreen_window(builder)
|
||
return "实时监控已打开（Esc 退出）"
|
||
|
||
def _do_qr_code(self, params) -> str:
|
||
text = str(params.get("text") or "").strip()
|
||
if not text:
|
||
raise ValueError("内容为空")
|
||
import tempfile
|
||
import urllib.parse
|
||
import urllib.request
|
||
url = ("https://api.qrserver.com/v1/create-qr-code/?size=520x520&data="
|
||
+ urllib.parse.quote(text))
|
||
target = Path(tempfile.gettempdir()) / "_lanremote_qr.png"
|
||
try:
|
||
with urllib.request.urlopen(url, timeout=10) as resp:
|
||
target.write_bytes(resp.read())
|
||
except Exception as exc: # noqa: BLE001
|
||
raise ValueError(f"生成二维码失败（要联网）: {exc}") from exc
|
||
self.actions.show_image(str(target), f"二维码 - {text[:20]}")
|
||
return f"二维码已显示（内容 {len(text)} 字）"
|
||
|
||
def _do_hash_text(self, params) -> str:
|
||
import hashlib
|
||
text = str(params.get("text") or "")
|
||
if not text:
|
||
raise ValueError("内容为空")
|
||
data = text.encode("utf-8")
|
||
return ("MD5: " + hashlib.md5(data).hexdigest()
|
||
+ "\\nSHA1: " + hashlib.sha1(data).hexdigest()
|
||
+ "\\nSHA256: " + hashlib.sha256(data).hexdigest())
|
||
|
||
def _do_base64_encode(self, params) -> str:
|
||
import base64
|
||
text = str(params.get("text") or "")
|
||
if not text:
|
||
raise ValueError("内容为空")
|
||
result = base64.b64encode(text.encode("utf-8")).decode("ascii")
|
||
self.actions.clip_set(result)
|
||
return result
|
||
|
||
def _do_base64_decode(self, params) -> str:
|
||
import base64
|
||
text = str(params.get("text") or "").strip()
|
||
if not text:
|
||
raise ValueError("内容为空")
|
||
try:
|
||
result = base64.b64decode(text).decode("utf-8", errors="replace")
|
||
except Exception as exc: # noqa: BLE001
|
||
raise ValueError(f"不是合法的 Base64: {exc}") from exc
|
||
self.actions.clip_set(result)
|
||
return result[:500]
|
||
|
||
def _do_uuid_gen(self, params) -> str:
|
||
import uuid
|
||
value = str(uuid.uuid4())
|
||
self.actions.clip_set(value)
|
||
return f"已生成并复制：\\n{value}"
|
||
|
||
def _do_timestamp_now(self, params) -> str:
|
||
now = time.time()
|
||
return (f"秒：{int(now)}\\n毫秒：{int(now * 1000)}\\n"
|
||
f"本地：{time.strftime('%Y-%m-%d %H:%M:%S')}\\n"
|
||
f"UTC：{time.strftime('%Y-%m-%d %H:%M:%S', time.gmtime(now))}")
|
||
|
||
def _do_json_format(self, params) -> str:
|
||
import json
|
||
text = str(params.get("text") or "").strip()
|
||
if not text:
|
||
raise ValueError("内容为空")
|
||
try:
|
||
obj = json.loads(text)
|
||
except ValueError as exc:
|
||
raise ValueError(f"不是合法 JSON: {exc}") from exc
|
||
pretty = json.dumps(obj, ensure_ascii=False, indent=2)
|
||
self.actions.clip_set(pretty)
|
||
return pretty[:800]
|
||
|
||
def _do_screen_color(self, params) -> str:
|
||
if not IS_WINDOWS:
|
||
raise ValueError("仅支持 Windows")
|
||
|
||
class Point(ctypes.Structure):
|
||
_fields_ = [("x", ctypes.c_long), ("y", ctypes.c_long)]
|
||
|
||
point = Point()
|
||
ctypes.windll.user32.GetCursorPos(ctypes.byref(point))
|
||
hdc = ctypes.windll.user32.GetDC(0)
|
||
color = ctypes.windll.gdi32.GetPixel(hdc, point.x, point.y)
|
||
ctypes.windll.user32.ReleaseDC(0, hdc)
|
||
if color < 0:
|
||
raise ValueError("取色失败")
|
||
r = color & 0xFF
|
||
g = (color >> 8) & 0xFF
|
||
b = (color >> 16) & 0xFF
|
||
hex_color = f"#{r:02X}{g:02X}{b:02X}"
|
||
self.actions.clip_set(hex_color)
|
||
return (f"鼠标位置 ({point.x}, {point.y})\\n"
|
||
f"颜色 {hex_color}\\nRGB({r}, {g}, {b})")
|
||
|
||
def _do_speed_test(self, params) -> str:
|
||
import urllib.request
|
||
url = "https://speed.cloudflare.com/__down?bytes=5000000"
|
||
try:
|
||
start = time.time()
|
||
with urllib.request.urlopen(url, timeout=20) as resp:
|
||
total = 0
|
||
while True:
|
||
chunk = resp.read(65536)
|
||
if not chunk:
|
||
break
|
||
total += len(chunk)
|
||
if time.time() - start > 12:
|
||
break
|
||
elapsed = max(0.01, time.time() - start)
|
||
except Exception as exc: # noqa: BLE001
|
||
raise ValueError(f"测速失败（要联网）: {exc}") from exc
|
||
mbps = total * 8 / elapsed / 1_000_000
|
||
return (f"下载 {total / 1048576:.1f} MB，用时 {elapsed:.1f} 秒\\n"
|
||
f"约 {mbps:.1f} Mbps（{total / elapsed / 1048576:.2f} MB/s）")
|
||
|
||
def _do_port_check(self, params) -> str:
|
||
import socket
|
||
host = str(params.get("host") or "").strip()
|
||
if not host:
|
||
raise ValueError("地址为空")
|
||
if not re.match(r"^[A-Za-z0-9_.:\\-]+$", host):
|
||
raise ValueError("地址含有不合法字符")
|
||
common = [21, 22, 23, 25, 53, 80, 110, 135, 139, 443, 445, 1433,
|
||
3306, 3389, 5432, 5900, 6379, 8080, 8765]
|
||
open_ports = []
|
||
for port in common:
|
||
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||
sock.settimeout(0.45)
|
||
try:
|
||
if sock.connect_ex((host, port)) == 0:
|
||
open_ports.append(port)
|
||
except OSError:
|
||
pass
|
||
finally:
|
||
sock.close()
|
||
if not open_ports:
|
||
return f"{host}：常用端口都没开"
|
||
return f"{host} 开放端口：{', '.join(str(p) for p in open_ports)}"
|
||
|
||
''' + IMPL_ANCHOR
|
||
|
||
|
||
def main() -> int:
|
||
text = FEATURES.read_text(encoding="utf-8")
|
||
changed = 0
|
||
|
||
if '"danmaku"' in text:
|
||
print(" defs 已存在，跳过")
|
||
elif DEFS_ANCHOR in text:
|
||
text = text.replace(DEFS_ANCHOR, DEFS_ADD, 1)
|
||
changed += 1
|
||
print(" defs: 已插入娱乐/极客")
|
||
else:
|
||
print(" [失败] defs 锚点未找到")
|
||
|
||
if "_do_matrix_rain" in text:
|
||
print(" 实现已存在，跳过")
|
||
elif IMPL_ANCHOR in text:
|
||
text = text.replace(IMPL_ANCHOR, IMPL_ADD, 1)
|
||
changed += 1
|
||
print(" 实现: 已插入 25 个方法")
|
||
else:
|
||
print(" [失败] 实现锚点未找到")
|
||
|
||
if changed:
|
||
FEATURES.write_text(text, encoding="utf-8")
|
||
print(f" 已写入（{changed} 处）")
|
||
|
||
print("\n语法检查：")
|
||
import ast
|
||
try:
|
||
ast.parse(FEATURES.read_text(encoding="utf-8"))
|
||
print(" features.py: OK")
|
||
except SyntaxError as exc:
|
||
print(f" features.py: 语法错误 {exc}")
|
||
return 1
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|