简单热更新系统（rmxp）

じ☆ve冰风 发表于 2025-4-14 00:29:18

一个简单的热更新系统，没有做文件校验，实现原理是在Gitee创建一个仓库，通过读取仓库简介中的信息获得版本号和需要更新的文件清单，然后把需要被下载的文件挂到仓库里。
简介信息示例：
版本：1.0.1 文件列表：update_part1.zip,update_part2.zip 文件大小：473858,3139291
文件大小需要手动输入，每个文件大小顺序需要和前面的文件名数据一致，文件大小可以属性里看，但是要注意填到网页上的时候数字中间不要有逗号，不然会被分割导致文件大小读取异常
下面代码：
RUBY 代码
#==============================
#Auto Update System
#Version: XP-Final-Plus
#TIME:20250413
#by：金牛
#发布论坛：rpg.blue
#部分声明在56行
#网页信息格式示例：版本：1.0.1 文件列表：update_part1.zip,update_part2.zip 文件大小：473858,3139291
#使用Gitee仓库挂载更新文件和更新信息，信息写在仓库简介里。
#==============================
module WebContentFetcher#网页模块
# Win32API 声明
URLDownloadToCacheFile = Win32API.new('Urlmon', 'URLDownloadToCacheFile', 'ippiii', 'i')
User32_msgbox       = Win32API.new('user32', 'MessageBoxW', 'LppL', 'L')
# 正则表达式配置
CONTENT_REGEX =/版本：\s*(\d+\.\d+\.\d+)\s+文件列表：\s*(+(?:,+)*)\s+文件大小：\s*(\d+(?:,\d+)*)/i
# UTF8与宽字符转换
defself.utf8_to_wide(str)
str.unpack("U*").pack("S*")+ "\0\0"
end
# 获取网页内容（核心方法）
defself.fetch(url)
begin
   URLDownloadToCacheFile.call(0,url,buf = "\0" * 1024,1024,0,0)
   content = open(buf.sub(/\0+$/){}, 'rb'){ |f| f.read}=~ CONTENT_REGEX
   v= $1
   f= $2
   n= $3
   if content !=nil
   # fcontent={"version"=>v.to_s.strip,"flist"=>f.to_s.split(','),"#{f.to_s.split(',')}"=>n.to_s.split(','),"#{f.to_s.split(',')}"=>n.to_s.split(',')}
   fcontent={"version"=>v.to_s.strip,"flist"=>f.to_s.split(',')}
   j=0
   for i in fcontent["flist"]
      fcontent=n.to_s.split(',')
      j+=1
   end
   return fcontent
   else
   show_error("内容格式不匹配", url)
   nil
   end
rescue => e
   show_error("获取失败: #{e.message}", url)
   nil
end
end

# 错误提示
defself.show_error(msg, url)
title = utf8_to_wide(msg)
text = utf8_to_wide("请检查网页内容格式或访问:\n#{url}")
User32_msgbox.call(0, text, title, 16)
end
end
def get_update_msg
content = WebContentFetcher.fetch(REPO_URL)
return content#返回云端获取到的信息
#version=版本号，flist=一个存储所有需要下载文件的名字数组，“文件名”=该文件大小
end
#==============================
#Auto Update System
#Version: XP-Final-Plus
#TIME:20250413
#==============================
$msg_update#存储云端信息
REPO_URL = 'https://gitee.com/你的用户名/你的仓库名'# 仓库主页地址
UPBASE_URL= 'https://gitee.com/你的用户名//你的仓库名/releases/download/你发布的版本名/'
#TMP_ZIP = './update.temp' # 临时文件名不用管
#FINAL_ZIP = './update.zip' # 最终文件名不用管
GAME_DIR = './'          # 游戏根目录
CHUNK_SIZE= 4096#4096          # 增大分块提升稳定性
MAX_RETRIES = 3             # 最大重试次数
CURRENT="1.0.0"             #设定当前版本号，当云端版本号出现迭代时进入更新流程
User32_msgbox = Win32API.new('user32' , 'MessageBoxW' , 'LppL', 'L')

#==============================
# 下载进度窗口
#==============================
class Scene_Download
def initialize
# 创建半透明背景
@viewport = Viewport.new(0, 0, 640, 480)
@bg = Sprite.new(@viewport)
@bg.bitmap = Bitmap.new(640, 480)
@bg.bitmap.fill_rect(0, 0, 640, 480, Color.new(0,0,0,180))

# 进度窗口
@window = Window_Base.new(160, 180, 320, 140)
@window.contents = Bitmap.new(@window.width-32, @window.height-32)
@start_time = Time.now
refresh(0, 0, 0,"")
end

def refresh(downloaded, total, retry_count,fn)
@window.contents.clear
# 动态处理未知大小
text = "下载#{fn}: "
if total > 0
   percent = (downloaded.to_f / total * 100).round
   text += "#{percent}% (#{filesize_format(downloaded)}/#{filesize_format(total)})"
   bar_width = (downloaded.to_f / total * 276).to_i
else
   phase = (Time.now - @start_time) * 2
   bar_width = (Math.sin(phase) * 50 + 50).to_i
   text += "正在连接服务器#{'.' * (3 - (Time.now.to_i % 3))}"
end

# 重试提示
text += "\n重试次数：#{retry_count}"if retry_count > 0

# 绘制内容
@window.contents.font.color = Color.new(255,255,255)
@window.contents.draw_text(4, 0, 292, 48, text)
@window.contents.fill_rect(20, 60, 276, 16, Color.new(100,100,100))
@window.contents.fill_rect(20, 60, bar_width, 16, Color.new(0,200,0))

Graphics.update
Input.update
end

#==============================
# 文件大小格式化（兼容Ruby 1.8）
#==============================
def filesize_format(bytes)
return"0 B"if bytes 0#版本号迭代，需要更新
   returntrue
end
end
returnfalse
end
#==============================
# 下载核心
#==============================
def xp_download(url,save_path,fn)#文件直链，保存路径，文件名
# 清理残留文件
File.delete(save_path)rescuenil
scene = Scene_Download.new
success = false
retry_count = 0

begin
internet_open = Win32API.new('wininet', 'InternetOpenA', 'plppl', 'l')
h_internet = internet_open.call("RGSS Player", 0, 0, 0, 0)
raise"网络初始化失败"if h_internet == 0

File.open(save_path, 'wb')do |f|
   until success || retry_count >= MAX_RETRIES
   begin
      scene.refresh(0, 0, retry_count,fn)

      # 建立连接
      h_url = Win32API.new('wininet', 'InternetOpenUrlA', 'lpplll', 'l').call(h_internet, url, 0, 0, 0x80000000 | 0x00800000 | 0x00001000, 0)
      raise"连接服务器失败"if h_url == 0
      # 获取文件大小
      total_size = get_content_length(fn)
      dynamic_mode = total_size == 0
      # 下载循环
      buffer = "\0" * CHUNK_SIZE
      downloaded = 0
      last_update = Time.now
      loopdo
         bytes_read = .pack('L')
         Win32API.new('wininet', 'InternetReadFile', 'lplp', 'l').call(
         h_url, buffer, CHUNK_SIZE, bytes_read)
         read_size = bytes_read.unpack('L').first

         breakif read_size == 0

         f.write(buffer)
         downloaded += read_size

         # 优化刷新频率
         ifTime.now - last_update > 0.3
         scene.refresh(downloaded, total_size, retry_count,fn)

         last_update = Time.now
         end
      end

      success = true
   rescue => e
      retry_count += 1
      scene.refresh(downloaded, total_size, retry_count,fn)
      sleep(2**retry_count)# 指数退避
      retryif retry_count < MAX_RETRIES
   ensure
      Win32API.new('wininet', 'InternetCloseHandle', 'l', 'l').call(h_url)rescuenil
   end
   end
end

rescue => e
# p "错误发生在：#{e.backtrace.first}"
show_message("最终错误: #{e.message}", "系统错误")
ensure
scene.disposeif scene
Win32API.new('wininet', 'InternetCloseHandle', 'l', 'l').call(h_internet)rescuenil
# 重命名临时文件
#File.rename(TMP_ZIP, fn) if success && File.exist?(TMP_ZIP)
end

success
end

#==============================
# 安全解压方法
#==============================
def unzip_file(zip_path, dest_dir)

# success = system("powershell -Command \"Expand-Archive -Path '#{zip_path}' -DestinationPath '#{dest_dir}'\"")
success = system("powershell -WindowStyle Hidden -Command \"Expand-Archive -Path '#{zip_path}' -DestinationPath '#{dest_dir}'\"")
success
end

#==============================
# 增强版更新流程
#==============================
def perform_update

# 步骤1：同步云端版本信息，判断是否需要更新
if check_verison
   # 步骤2：显示确认对话框
returnunless confirm_update?#不需要确认的话注释掉就行
# 步骤3：获取更新列表并下载更新
for fn in$msg_update["flist"]#遍历更新列表
   url=UPBASE_URL+fn
   if xp_download(url, GAME_DIR+fn,fn)
   else
      show_message("#{fn}下载失败，请检查网络连接", "网络错误")
   end
end
for fn in$msg_update["flist"]
   Graphics.update
   # 步骤3：解压文件
   begin
   if unzip_file(GAME_DIR+fn, GAME_DIR)
         File.delete(fn)
         # show_message("#{fn}更新成功！", "更新完成")
   else
         show_message("#{fn}解压失败，请手动解压文件", "解压错误")
   end
   rescue => e
      show_message("解压过程发生错误：#{e.message}", "严重错误")
   end
end
#步骤4：更新完成，重载数据库
show_message("更新成功！部分更新游戏重启后生效！", "更新完成")
#重新加载数据库的方法
#需根据自身需要实现重载方法。
end
end

#==============================
# 辅助方法
#==============================
def utf8_to_wide(str)
str.unpack("U*").pack("S*")+ "\0\0"
end

def show_message(text, title="提示")
User32_msgbox.call(0, utf8_to_wide(text), utf8_to_wide(title), 0x40)
end
def confirm_update?
response = User32_msgbox.call(0,
utf8_to_wide("检测到新版本，是否立即更新？"),
utf8_to_wide("版本更新"),
0x34)# 带取消按钮的警告图标
response == 6# IDYES
end

         本帖来自P1论坛作者939034448，因Project1站服务器在国外有时候访问缓慢不方便作者交流学习，经联系P1站长fux2同意署名转载一起分享游戏制作经验，共同为国内独立游戏作者共同创造良好交流环境，原文地址：https://rpg.blue/forum.php?mod=viewthread&tid=497224若有侵权，发帖作者可联系底部站长QQ在线咨询功能删除，谢谢。

页: [1]

同能RPG制作大师's Archiver

简单热更新系统（rmxp）