简单热更新系统（rmxp）

939034448 · 发表于 5 天前

加入我们，或者，欢迎回来。

您需要登录才可以下载或查看，没有帐号？注册会员

x

本帖最后由 939034448 于 2025-4-14 12:04 编辑

一个简单的热更新系统，没有做文件校验，实现原理是在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*([a-zA-Z0-9_\-\.]+(?:,[a-zA-Z0-9_\-\.]+)*)\s+文件大小：\s*(\d+(?:,\d+)*)/i
  # UTF8与宽字符转换
  def self.utf8_to_wide(str)
    str.unpack("U*").pack("S*")+ "\0\0"
  end
  # 获取网页内容（核心方法）
  def self.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(',')[0]}"=>n.to_s.split(',')[0],"#{f.to_s.split(',')[1]}"=>n.to_s.split(',')[1]}   
       fcontent={"version"=>v.to_s.strip,"flist"=>f.to_s.split(',')}   
       j=0
       for i in fcontent["flist"]
         fcontent[i]=n.to_s.split(',')[j]
         j+=1
       end  
       return fcontent
      else
        show_error("内容格式不匹配", url)
        nil
      end
    rescue => e
      show_error("获取失败: #{e.message}", url)
      nil
    end
  end
 
  # 错误提示
  def self.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
    units = ["B", "KB", "MB", "GB"]
    exp = (Math.log(bytes) / Math.log(1024)).to_i
    exp = [exp, 0].max  # 替换 clamp(0,3)
    exp = [exp, 3].min  # 手动实现范围限制
    "%.1f %s" % [bytes.to_f / 1024 ** exp, units[exp]]
  end
 
  def dispose
    @window.dispose
    @bg.dispose
    @viewport.dispose
  end
end
#==============================
# 增强版文件大小获取方法（修正字节序）
#==============================
def get_content_length(fname)#通过文件名获取大小
  msg=$msg_update[fname]#返回云端包含的信息
  #version=版本号，flist=一个存储所有需要下载文件的名字数组，“文件名”=该文件大小
  return msg.to_i#返回此次下载文件的大小
end
#==============================
#对比版本号
#==============================
def check_verison
  $msg_update=get_update_msg#获取云端信息
  #version=版本号，flist=一个存储所有需要下载文件的名字数组，“文件名”=该文件大小
  nv=$msg_update["version"]#云端版本号
  cv=CURRENT
  nv=nv.to_s.split('.')
  cv=cv.to_s.split('.')
  for i in 0...nv.size
    cv[i]=0 if cv[i]==nil#补丁，防止云端版本需要增加版本号后缀
    if nv[i].to_i-cv[i].to_i>0#版本号迭代，需要更新
      return true
    end  
  end  
  return false
end  
#==============================
# 下载核心
#==============================
def xp_download(url,save_path,fn)#文件直链，保存路径，文件名
  # 清理残留文件
  File.delete(save_path) rescue nil
  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
          loop do
            bytes_read = [0].pack('L')
            Win32API.new('wininet', 'InternetReadFile', 'lplp', 'l').call(
              h_url, buffer, CHUNK_SIZE, bytes_read)
            read_size = bytes_read.unpack('L').first
 
            break if read_size == 0
 
            f.write(buffer[0, read_size])
            downloaded += read_size
 
            # 优化刷新频率
            if Time.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)  # 指数退避
          retry if retry_count < MAX_RETRIES
        ensure
          Win32API.new('wininet', 'InternetCloseHandle', 'l', 'l').call(h_url) rescue nil
        end
      end
    end
 
  rescue => e
   # p "错误发生在：#{e.backtrace.first}"
    show_message("最终错误: #{e.message}", "系统错误")
  ensure
    scene.dispose if scene
    Win32API.new('wininet', 'InternetCloseHandle', 'l', 'l').call(h_internet) rescue nil
    # 重命名临时文件
  #  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：显示确认对话框
     return unless 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

#==============================


#  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*([a-zA-Z0-9_\-\.]+(?:,[a-zA-Z0-9_\-\.]+)*)\s+文件大小：\s*(\d+(?:,\d+)*)/i


  # UTF8与宽字符转换


  def self.utf8_to_wide(str)


    str.unpack("U*").pack("S*")+ "\0\0"


  end


  # 获取网页内容（核心方法）


  def self.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(',')[0]}"=>n.to_s.split(',')[0],"#{f.to_s.split(',')[1]}"=>n.to_s.split(',')[1]}   


       fcontent={"version"=>v.to_s.strip,"flist"=>f.to_s.split(',')}   


       j=0


       for i in fcontent["flist"]


         fcontent[i]=n.to_s.split(',')[j]


         j+=1


       end  


       return fcontent


      else


        show_error("内容格式不匹配", url)


        nil


      end


    rescue => e


      show_error("获取失败: #{e.message}", url)


      nil


    end


  end


 


  # 错误提示


  def self.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


    units = ["B", "KB", "MB", "GB"]


    exp = (Math.log(bytes) / Math.log(1024)).to_i


    exp = [exp, 0].max  # 替换 clamp(0,3)


    exp = [exp, 3].min  # 手动实现范围限制


    "%.1f %s" % [bytes.to_f / 1024 ** exp, units[exp]]


  end


 


  def dispose


    @window.dispose


    @bg.dispose


    @viewport.dispose


  end


end


#==============================


# 增强版文件大小获取方法（修正字节序）


#==============================


def get_content_length(fname)#通过文件名获取大小


  msg=$msg_update[fname]#返回云端包含的信息


  #version=版本号，flist=一个存储所有需要下载文件的名字数组，“文件名”=该文件大小


  return msg.to_i#返回此次下载文件的大小


end


#==============================


#对比版本号


#==============================


def check_verison


  $msg_update=get_update_msg#获取云端信息


  #version=版本号，flist=一个存储所有需要下载文件的名字数组，“文件名”=该文件大小


  nv=$msg_update["version"]#云端版本号


  cv=CURRENT


  nv=nv.to_s.split('.')


  cv=cv.to_s.split('.')


  for i in 0...nv.size


    cv[i]=0 if cv[i]==nil#补丁，防止云端版本需要增加版本号后缀


    if nv[i].to_i-cv[i].to_i>0#版本号迭代，需要更新


      return true


    end  


  end  


  return false


end  


#==============================


# 下载核心


#==============================


def xp_download(url,save_path,fn)#文件直链，保存路径，文件名


  # 清理残留文件


  File.delete(save_path) rescue nil


  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


          loop do


            bytes_read = [0].pack('L')


            Win32API.new('wininet', 'InternetReadFile', 'lplp', 'l').call(


              h_url, buffer, CHUNK_SIZE, bytes_read)


            read_size = bytes_read.unpack('L').first


 


            break if read_size == 0


 


            f.write(buffer[0, read_size])


            downloaded += read_size


 


            # 优化刷新频率


            if Time.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)  # 指数退避


          retry if retry_count < MAX_RETRIES


        ensure


          Win32API.new('wininet', 'InternetCloseHandle', 'l', 'l').call(h_url) rescue nil


        end


      end


    end


 


  rescue => e


   # p "错误发生在：#{e.backtrace.first}"


    show_message("最终错误: #{e.message}", "系统错误")


  ensure


    scene.dispose if scene


    Win32API.new('wininet', 'InternetCloseHandle', 'l', 'l').call(h_internet) rescue nil


    # 重命名临时文件


  #  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：显示确认对话框


     return unless 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

imsy · 发表于 4 天前

讲真非常厉害的工作由衷支持技术力楼主
顺便提供一下个人收藏的另外几个帖子，也许会对您未来的开发有所助力？
https://rpg.blue/thread-403455-1-1.html
https://rpg.blue/thread-376885-1-1.html
https://rpg.blue/thread-412154-1-1.html

帐号		自动登录	找回密码
密码			注册会员

[原创发布] 简单热更新系统（rmxp）

加入我们，或者，欢迎回来。

点评

点评

浏览过的版块