"""
pytcper TCP调试助手 - 工具函数与常量
====================================
纯函数与全局常量:编码转换、HEX 解析、时长格式化、路径与超时配置。
"""
__version__ = "2.0.001"
import os
import socket
ENCODINGS = ("UTF-8", "UTF-8(BOM)", "UTF-16", "UTF-16LE", "UTF-16BE",
"UTF-32", "UTF-32LE", "UTF-32BE",
"GBK", "GB2312", "BIG5", "Latin-1", "ASCII")
_ENCODING_ALIASES = {"UTF-8(BOM)": "utf_8_sig"}
def _codec_name(label):
"""把界面上的编码标签转换为 Python 可用的编码名。"""
return _ENCODING_ALIASES.get(label, label)
FONT = ("Consolas", 10)
CONFIG_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".tcp_tool_config.json")
HISTORY_MAX = 50
SOCK_TIMEOUT = 30.0
SEND_TIMEOUT = 5.0
LOG_MAX_LINES = 5000
def encode_text(text, encoding, hex_mode):
"""将待发送文本按编码或 HEX 字符串转换为字节(HEX 非法时抛 ValueError)。"""
if hex_mode:
return bytes.fromhex("".join(text.split()))
return text.encode(_codec_name(encoding), errors="replace")
def decode_bytes(data, encoding, hex_mode):
"""将接收到的字节按编码或 HEX 格式转换为可显示文本。"""
if hex_mode:
return " ".join("%02X" % b for b in data)
return data.decode(_codec_name(encoding), errors="replace")
_AUTO_ENCODINGS = ("UTF-8", "GBK", "GB2312", "BIG5", "UTF-16LE", "Latin-1")
def guess_decodable(data):
"""尝试用常见编码解码字节,返回 (文本, 替换字符数)。
服务端无法获知客户端发送字节用的编码;当按当前编码解码出大量替换
字符(乱码)时,用本函数在候选编码中挑一个替换字符最少的方案,
从而跨编码正确显示。替换字符数相同则返回 None 表示无更优解。
"""
picked = None
best_repl = None
for enc in _AUTO_ENCODINGS:
try:
text = data.decode(_codec_name(enc), errors="replace")
except (LookupError, UnicodeDecodeError):
continue
repl = text.count("\ufffd")
if best_repl is None or repl < best_repl:
best_repl = repl
picked = (enc, text)
if repl == 0:
break
if best_repl is not None:
return picked[1], best_repl
return None, None
def fmt_duration(secs):
"""把秒数格式化为 HH:MM:SS。"""
secs = int(secs)
h, rem = divmod(secs, 3600)
m, s = divmod(rem, 60)
return "%02d:%02d:%02d" % (h, m, s)
def close_socket(sock):
"""安全关闭 socket:先 shutdown 唤醒阻塞中的 recv,再 close。
Linux 上若直接 close 一个正被其它线程 recv 的 socket,关闭不会真正
生效(对端收不到 FIN,阻塞的 recv 线程也无法感知断开),必须先
shutdown 才能正确断开连接并唤醒线程。
"""
try:
sock.shutdown(socket.SHUT_RDWR)
except OSError:
pass
try:
sock.close()
except OSError:
pass