分享一段B/S自动更新代码[Ai写的]

rhci

原本需求是客户第二屏需要播放视频广告，但是每一次更新工作量巨大，使用一些同步软件啥的，又比较麻烦，索性找豆包AI写了这么一段代码，基本能完美解决我的问题，达到预期的目的，直接上代码
server.py

[Python] 纯文本查看 复制代码

?

001 002 003 004 005 006 007 008 009 010 011 012 013 014 015 016 017 018 019 020 021 022 023 024 025 026 027 028 029 030 031 032 033 034 035 036 037 038 039 040 041 042 043 044 045 046 047 048 049 050 051 052 053 054 055 056 057 058 059 060 061 062 063 064 065 066 067 068 069 070 071 072 073 074 075 076 077 078 079 080 081 082 083 084 085 086 087 088 089 090 091 092 093 094 095 096 097 098 099 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232

import hashlib import os import http.server import socketserver import time import threading import sys import signal from pystray import Icon as icon, Menu as menu, MenuItem as item from PIL import Image import configparser import logging import traceback   # 读取配置文件 config = configparser.ConfigParser() if not config.read('server_config.ini'):     logging.warning("未找到 server_config.ini 文件，将使用默认配置。")     config['Server'] = {         'update_folder': 'root',         'port': 8020,         'log_level': 'INFO'     }     with open('server_config.ini', 'w') as configfile:         config.write(configfile)   # 获取日志级别 log_level_str = config.get('Server', 'log_level', fallback='INFO') log_level = getattr(logging, log_level_str.upper(), logging.INFO)   # 配置日志记录 logging.basicConfig(     level=log_level,     format='%(asctime)s - %(levelname)s - %(message)s',     filename='server.log',     filemode='a' )   def extract_embedded_file(file_name):     if getattr(sys, 'frozen', False):         # 打包后的环境         base_path = sys._MEIPASS         source_path = os.path.join(base_path, file_name)         target_path = os.path.join(os.path.dirname(sys.executable), file_name)         try:             with open(source_path, 'rb') as src, open(target_path, 'wb') as dst:                 dst.write(src.read())             logging.info(f"成功释放文件 {file_name} 到 {target_path}")         except Exception as e:             logging.error(f"释放文件 {file_name} 时出错: {e}")     return os.path.join(os.path.dirname(sys.executable), file_name)   class FileListGenerator:     def __init__(self, update_folder):         self.update_folder = self.get_full_update_folder(update_folder)       def get_full_update_folder(self, update_folder):         if getattr(sys, 'frozen', False):             script_dir = os.path.dirname(sys.executable)         else:             script_dir = os.path.dirname(os.path.abspath(__file__))         full_update_folder = os.path.join(script_dir, update_folder)         if not os.path.exists(full_update_folder):             os.makedirs(full_update_folder)         return full_update_folder       @staticmethod     def calculate_md5(file_path):         hash_md5 = hashlib.md5()         try:             with open(file_path, "rb") as f:                 for chunk in iter(lambda: f.read(4096), b""):                     hash_md5.update(chunk)             return hash_md5.hexdigest()         except Exception as e:             logging.error(f"计算文件 {file_path} 的 MD5 时出错: {e}")             return None       def generate_file_list(self):         file_list = []         try:             for root, dirs, files in os.walk(self.update_folder):                 for file in files:                     file_path = os.path.join(root, file)                     md5 = self.calculate_md5(file_path)                     if md5:                         relative_file = os.path.relpath(file_path, self.update_folder)                         file_list.append(f"{relative_file} || {md5}")               file_list_path = os.path.join(self.update_folder, 'file_list.sha')             file_list_dir = os.path.dirname(file_list_path)             if not os.path.exists(file_list_dir):                 os.makedirs(file_list_dir)               with open(file_list_path, 'w') as f:                 f.write('\n'.join(file_list))             logging.info("文件列表生成成功")         except Exception as e:             logging.error(f"生成文件列表时出错: {e}")   # 自定义请求处理类，将日志输出到日志文件 class CustomHTTPRequestHandler(http.server.SimpleHTTPRequestHandler):     def log_message(self, format, *args):         logging.info("%s - - [%s] %s\n" % (             self.address_string(),             self.log_date_time_string(),             format % args))   class ServerManager:     def __init__(self, update_folder, port):         self.update_folder = update_folder         self.port = port         self.file_list_generator = FileListGenerator(update_folder)         self.httpd = None         self.server_thread = None         self.timer_thread = None         self.tray_icon = None         self.running = True       def start_timer_task(self):         def timer_task():             while self.running:                 try:                     self.file_list_generator.generate_file_list()                 except Exception as e:                     logging.error(f"定时任务出错: {e}\n{traceback.format_exc()}")                 time.sleep(180)           self.timer_thread = threading.Thread(target=timer_task)         self.timer_thread.daemon = True         self.timer_thread.start()       def start_http_server(self):         Handler = CustomHTTPRequestHandler         update_folder = self.file_list_generator.update_folder           def serve():             try:                 # 确保路径是绝对路径                 abs_update_folder = os.path.abspath(update_folder)                 if not os.path.exists(abs_update_folder):                     logging.error(f"更新文件夹 {abs_update_folder} 不存在。")                     return                 os.chdir(abs_update_folder)                 self.httpd = socketserver.TCPServer(("", self.port), Handler)                 logging.info(f"服务端已启动，监听端口 {self.port}，可通过 http://localhost:{self.port} 访问。")                 self.httpd.serve_forever()             except OSError as e:                 if e.errno == 98:  # 端口被占用                     logging.error(f"端口 {self.port} 已被占用，请更换端口。")                 else:                     logging.error(f"启动 HTTP 服务时出错: {e}\n{traceback.format_exc()}")             except Exception as e:                 logging.error(f"启动 HTTP 服务时出错: {e}\n{traceback.format_exc()}")           self.server_thread = threading.Thread(target=serve)         self.server_thread.daemon = True         self.server_thread.start()       def recalculate(self, icon, item):         try:             self.file_list_generator.generate_file_list()         except Exception as e:             logging.error(f"重新计算文件列表时出错: {e}\n{traceback.format_exc()}")       def exit_program(self, icon, item):         self.running = False         try:             if icon:                 icon.stop()         except Exception as e:             logging.error(f"停止系统托盘图标时出错: {e}\n{traceback.format_exc()}")         try:             if self.httpd:                 self.httpd.shutdown()                 self.httpd.server_close()         except Exception as e:             logging.error(f"关闭 HTTP 服务器时出错: {e}\n{traceback.format_exc()}")         try:             if self.server_thread:                 self.server_thread.join(timeout=5)                 if self.server_thread.is_alive():                     logging.warning("HTTP 服务器线程未能在 5 秒内关闭。")         except Exception as e:             logging.error(f"等待 HTTP 服务器线程结束时出错: {e}\n{traceback.format_exc()}")         try:             if self.timer_thread:                 self.timer_thread.join(timeout=5)                 if self.timer_thread.is_alive():                     logging.warning("定时任务线程未能在 5 秒内关闭。")         except Exception as e:             logging.error(f"等待定时任务线程结束时出错: {e}\n{traceback.format_exc()}")         sys.exit(0)       def start_tray_icon(self):         try:             image_path = extract_embedded_file('feather.png')             image = Image.open(image_path)         except FileNotFoundError:             logging.warning("未找到图标文件，使用默认图标。")             image = Image.new('RGB', (16, 16))           self.tray_icon = icon('name', image, '服务端程序', menu=menu(             item('重新计算', self.recalculate),             item('退出程序', self.exit_program)         ))         self.tray_icon.run()       def start(self):         # 生成初始的 file_list.sha         self.file_list_generator.generate_file_list()         # 启动定时任务线程         self.start_timer_task()         # 启动 HTTP 服务器         self.start_http_server()         # 启动系统托盘图标         self.start_tray_icon()   if __name__ == "__main__":     update_folder = config.get('Server', 'update_folder', fallback='root')     port = config.getint('Server', 'port', fallback=8020)       server_manager = ServerManager(update_folder, port)       def signal_handler(sig, frame):         server_manager.exit_program(None, None)       signal.signal(signal.SIGINT, signal_handler)     signal.signal(signal.SIGTERM, signal_handler)       server_manager.start()

原理，新建一个root目录，需要更新下载的文件放置进入，程序运行时会计算一次root目录下所有文件的md5值并保存至file_list.sha文件中，并开启HTTP服务，默认端口为8020；右下角出现菜单，重新计算和退出程序。
重新计算为强制手动刷新root下文件的md5值，这个程序里有时间，3分钟一次自动计算，手动计算是偶尔可能会需要。
已知问题，打包为单文件后，第一次执行时，不会写入日志文件，第二次运行时开始创建日志文件，并写入；
日志文件通过配置文件可指定内容，INFO及DEBUG

clitent.py

[Python] 纯文本查看 复制代码

?

001 002 003 004 005 006 007 008 009 010 011 012 013 014 015 016 017 018 019 020 021 022 023 024 025 026 027 028 029 030 031 032 033 034 035 036 037 038 039 040 041 042 043 044 045 046 047 048 049 050 051 052 053 054 055 056 057 058 059 060 061 062 063 064 065 066 067 068 069 070 071 072 073 074 075 076 077 078 079 080 081 082 083 084 085 086 087 088 089 090 091 092 093 094 095 096 097 098 099 100 101 102 103 104 105 106 107 108 109 110 111

import tkinter as tk from tkinter import ttk import requests import hashlib import os import time import threading import configparser import sys     def calculate_md5(file_path):     hash_md5 = hashlib.md5()     with open(file_path, "rb") as f:         for chunk in iter(lambda: f.read(4096), b""):             hash_md5.update(chunk)     return hash_md5.hexdigest()     def download_file(url, local_filename):     start_time = time.time()     response = requests.get(url, stream=True)     total_size = int(response.headers.get('content-length', 0))     block_size = 1024     wrote = 0     with open(local_filename, 'wb') as f:         for data in response.iter_content(block_size):             elapsed_time = time.time() - start_time             if elapsed_time == 0:                 speed = 0             else:                 speed = (wrote / elapsed_time) / 1024             wrote += len(data)             progress = (wrote / total_size) * 100             progress_bar['value'] = progress             speed_label.config(text=f"下载速度: {speed:.2f} KB/s")             root.update_idletasks()             f.write(data)     def check_and_update(root):     config = configparser.ConfigParser()     # 获取程序所在目录     base_dir = os.path.dirname(sys.executable) if getattr(sys, 'frozen', False) else os.path.dirname(os.path.abspath(__file__))     config_file_path = os.path.join(base_dir, 'config.ini')       if not config.read(config_file_path):         print("未读取到 config.ini 文件，将写入默认配置。")         config['Server'] = {             'url': 'http://192.168.6.2:8020/'         }         with open(config_file_path, 'w') as configfile:             config.write(configfile)         print("已写入默认配置，请根据实际情况修改服务地址。")         root.after(0, root.destroy)         return       base_url = config.get('Server', 'url', fallback='')     if not base_url:         print("config.ini 文件中未正确配置服务地址。")         root.after(0, root.destroy)         return     if not base_url.endswith('/'):         base_url += '/'       # 获取服务端文件列表     try:         response = requests.get(base_url + 'file_list.sha')         lines = response.text.strip().splitlines()         file_list = []         for line in lines:             try:                 file_name, md5 = line.split(' || ')                 file_list.append((file_name, md5))             except ValueError:                 print(f"解析行 '{line}' 失败，跳过该行。")     except Exception as e:         print(f"获取文件列表失败: {e}")         root.after(0, root.destroy)         return       for file_name, remote_md5 in file_list:         if file_name == 'file_list.sha':             continue         local_file = os.path.join(base_dir, file_name)         if os.path.exists(local_file):             local_md5 = calculate_md5(local_file)         else:             local_md5 = None           if local_md5 != remote_md5:             download_file(base_url + file_name, local_file)       root.after(0, root.destroy)     root = tk.Tk() root.title("更新程序")   progress_bar = ttk.Progressbar(root, orient="horizontal", length=300, mode="determinate") progress_bar.pack(pady=20)   speed_label = tk.Label(root, text="下载速度: 0.00 KB/s") speed_label.pack(pady=10)   update_thread = threading.Thread(target=check_and_update, args=(root,)) update_thread.daemon = True  # 设置线程为守护线程，主程序退出时线程也会退出 update_thread.start()   root.mainloop()

这个是客户端代码，程序运行会自动新建一个config.ini文件，其中url为指定服务器IP，理论上也可以是域名

已知客户端问题，第一次运行因为没有配置文件，会创建配置文件，并自动退出。
第二次运行会直接读取配置文件，然后自动下载file_list.sha,并和程序所在目录下的文件MD5值做对比，相同及退出，不同，开始下载，完成后退出。

代码已经进行过打包测试运行,WIN10 ,WIN11 32/64稳定，WIN7没有环境，没测试，理论上可以通吃。

图片就不发了，另外SERVer第一次运行，生成配置文件后，需要手动在右下角退出程序，再次运行即可开始正常更新；
目前固定读取配置文件中URL的root路径，这个逻辑，可以自行修改。

malaxiangguo

终于搞明白了 感谢分享~

rhci

打包后测试，win7无法运行，可以安装python3.110运行源码。

kong1988

详细说明一下使用场景呗！

rhci

kong1988 发表于 2025-4-13 16:41
详细说明一下使用场景呗！

主要用于局域网部署，我的场景是大批量传输视频文件到客户机第二屏进行本地播放。所以需要把视频文件分发到客户机，之前我们是远程一台一台的替换，很麻烦，现在有这个脚本，只需要把文件放置在服务端，客户端设定个定时运行，就可以自动更新了。按逻辑代码，大约更新时间为5分钟即可。