hejialiangya 发表于 2023-10-26 21:05:17

[Python] 我想实现局域网自动更新文件功能

新手,写了个程序,编译后的Main程序在我本机里
想加个自动更新的功能,点登录后就会自动下载保持程序一致性
但我从来没接触过目前毫无头绪
请问是用requests库吗?具体应该怎么实现?
有人做过相关功能可以指点指点吗

isdkz 发表于 2023-10-26 21:05:29

是的,你可以使用Python的requests库来实现局域网自动更新文件的功能。具体步骤如下:

1. 首先,你需要确定文件的下载地址。可以是一个HTTP或FTP链接,或者是一个局域网内部的共享文件夹路径。

2. 使用requests库发送HTTP请求,下载文件。你可以使用 requests.get() 方法来发送GET请求,并将文件保存到本地。

下面是一个简单的示例代码:

import requests

def download_file(url, save_path):
    response = requests.get(url)
    with open(save_path, 'wb') as file:
      file.write(response.content)

# 调用示例
url = 'http://example.com/file.txt'# 文件的下载地址
save_path = 'path/to/save/file.txt'# 文件保存的路径
download_file(url, save_path)


在这个示例中, download_file() 函数接受文件的下载地址和保存路径作为参数,使用 requests.get() 方法下载文件,并使用 open() 函数将文件保存到本地。

你可以根据自己的需求,将上述代码集成到你的程序中。当用户登录后,调用 download_file() 函数来下载文件,以保持程序的一致性。

请注意,这只是一个简单的示例,你可能还需要处理一些异常情况,例如网络连接错误或文件不存在等。你可以根据具体需求进行相应的错误处理。

hejialiangya 发表于 2023-10-26 21:09:57

isdkz 发表于 2023-10-26 21:05
是的,你可以使用Python的requests库来实现局域网自动更新文件的功能。具体步骤如下:

1. 首先,你需要 ...

这是AI的,下午我已经尝试过AI,解释的不是非常通透。

歌者文明清理员 发表于 2023-10-26 21:23:51

可以把项目开源到 GitHub,然后每次更新都调用 git pull

hejialiangya 发表于 2023-10-26 21:43:23

歌者文明清理员 发表于 2023-10-26 21:23
可以把项目开源到 GitHub,然后每次更新都调用 git pull

我们是纯内网,这个方法可能行不通吧

歌者文明清理员 发表于 2023-10-27 17:03:05

hejialiangya 发表于 2023-10-26 21:43
我们是纯内网,这个方法可能行不通吧

确实有点麻烦了

hejialiangya 发表于 2023-10-31 21:08:42

歌者文明清理员 发表于 2023-10-27 17:03
确实有点麻烦了

学了2天学了点皮毛,打算用通过ftp协议来实现。
瞎整了一段代码,但是运行起来有点问题,请大佬帮我看看


from ftplib import FTP
import os
import sys
import time
import socket

class MyFTP():
    # 初始化FTP客户端
    def __init__(self, host, port=21):
      self.host = host
      self.port = port
      self.ftp = FTP()
      # 重新设置下编码方式
      self.ftp.encoding = 'gbk'
      self.log_file = open("log.txt", "a")
      self.file_list = []

    # FTP客户端登录
    def login(self, username, password):
      try:
            timeout = 60
            socket.setdefaulttimeout(timeout)
            # 0主动模式 1 #被动模式
            self.ftp.set_pasv(True)
            # 打开调试级别2,显示详细信息
            # self.ftp.set_debuglevel(2)

            self.debug_print('开始尝试连接到 %s' % self.host)
            self.ftp.connect(self.host, self.port)
            self.debug_print('成功连接到 %s' % self.host)

            self.debug_print('开始尝试登录到 %s' % self.host)
            self.ftp.login(username, password)
            self.debug_print('成功登录到 %s' % self.host)

            self.debug_print(self.ftp.welcome)
      except Exception as error:
            self.deal_error("FTP 连接或登录失败 ,错误描述为:%s" % error)
            pass

    #用于下载前判断远程文件和本地文件大小是否一致
    def is_same_size(self, local_file, remote_file):
      try:
            remote_file_size = self.ftp.size(remote_file)
      except Exception as error:
            # self.debug_print("is_same_size() 错误描述为:%s" % error)
            remote_file_size = -1

      try:
            local_file_size = os.path.getsize(local_file)
      except Exception as error:
            # self.debug_print("is_same_size() 错误描述为:%s" % error)
            local_file_size = -1

      self.debug_print('local_file_size:%d, remote_file_size:%d' % (local_file_size, remote_file_size))
      if remote_file_size == local_file_size:
            return 1
      else:
            return 0

    #从ftp下载单个文件
    def download_file(self, local_file, remote_file):
      self.debug_print("download_file()---> local_path = %s ,remote_path = %s" % (local_file, remote_file))

      if self.is_same_size(local_file, remote_file):
            self.debug_print('%s 文件大小相同,无需下载' % local_file)
            return
      else:
            try:
                self.debug_print('>>>>>>>>>>>>下载文件 %s ... ...' % local_file)
                buf_size = 1024
                file_handler = open(local_file, 'wb')
                self.ftp.retrbinary('RETR %s' % remote_file, file_handler.write, buf_size)
                file_handler.close()
            except Exception as error:
                self.debug_print('下载文件出错,出现异常:%s ' % error)
                return

    #从远程目录下载多个文件到本地目录
    def download_file_tree(self, local_path, remote_path):
      print("download_file_tree()--->local_path = %s ,remote_path = %s" % (local_path, remote_path))
      try:
            self.ftp.cwd(remote_path)
      except Exception as error:
            self.debug_print('远程目录%s不存在,继续...' % remote_path + " ,具体错误描述为:%s" % error)
            return

      if not os.path.isdir(local_path):
            self.debug_print('本地目录%s不存在,先创建本地目录' % local_path)
            os.makedirs(local_path)

      self.debug_print('切换至目录: %s' % self.ftp.pwd())

      self.file_list = []
      # 方法回调
      self.ftp.dir(self.get_file_list)

      remote_names = self.file_list
      self.debug_print('远程目录 列表: %s' % remote_names)
      for item in remote_names:
            file_type = item
            file_name = item
            local = os.path.join(local_path, file_name)
            if file_type == 'd':
                print("download_file_tree()---> 下载目录: %s" % file_name)
                self.download_file_tree(local, file_name)
            elif file_type == '-':
                print("download_file()---> 下载文件: %s" % file_name)
                self.download_file(local, file_name)
            self.ftp.cwd("..")
            self.debug_print('返回上层目录 %s' % self.ftp.pwd())
      return True

    # 从本地上传文件到ftp
    def upload_file(self, local_file, remote_file):
      if not os.path.isfile(local_file):
            self.debug_print('%s 不存在' % local_file)
            return

      if self.is_same_size(local_file, remote_file):
            self.debug_print('跳过相等的文件: %s' % local_file)
            return

      buf_size = 1024
      file_handler = open(local_file, 'rb')
      self.ftp.storbinary('STOR %s' % remote_file, file_handler, buf_size)
      file_handler.close()
      self.debug_print('上传: %s' % local_file + "成功!")

    # 从本地上传目录下多个文件到ftp
    def upload_file_tree(self, local_path, remote_path):
      if not os.path.isdir(local_path):
            self.debug_print('本地目录 %s 不存在' % local_path)
            return

      self.ftp.cwd(remote_path)
      self.debug_print('切换至远程目录: %s' % self.ftp.pwd())

      local_name_list = os.listdir(local_path)
      for local_name in local_name_list:
            src = os.path.join(local_path, local_name)
            if os.path.isdir(src):
                try:
                  self.ftp.mkd(local_name)
                except Exception as error:
                  self.debug_print("目录已存在 %s ,具体错误描述为:%s" % (local_name, error))
                self.debug_print("upload_file_tree()---> 上传目录: %s" % local_name)
                self.upload_file_tree(src, local_name)
            else:
                self.debug_print("upload_file_tree()---> 上传文件: %s" % local_name)
                self.upload_file(src, local_name)
      self.ftp.cwd("..")

    def close(self):
      self.debug_print("close()---> FTP退出")
      self.ftp.quit()
      self.log_file.close()

    # 打印日志
    def debug_print(self, s):
      self.write_log(s)

    # 处理错误异常
    def deal_error(self, error):
      log_str = '发生错误: %s' % error
      self.write_log(log_str)
      sys.exit()

    # 记录日志
    def write_log(self, log_str):
      time_now = time.localtime()
      date_now = time.strftime('%Y-%m-%d', time_now)
      format_log_str = "%s ---> %s \n " % (date_now, log_str)
      print(format_log_str)
      self.log_file.write(format_log_str)

    # 获取文件列表
    def get_file_list(self, line):
      file_arr = self.get_file_name(line)
      # 去除. 和..
      if file_arr not in ['.', '..']:
            self.file_list.append(file_arr)

    # 获取文件名
    def get_file_name(self, line):
      pos = line.rfind(':')
      while (line != ' '):
            pos += 1
      while (line == ' '):
            pos += 1
      file_arr = , line]
      return file_arr


if __name__ == "__main__":
    my_ftp = MyFTP("130.1.11.70")
    my_ftp.login("xxk", "xxk@2021")
    # 下载单个文件
    # my_ftp.download_file("D:/download", "./update/")
    # 下载目录
    my_ftp.download_file_tree(r"D:\download", "/update/")
    # 上传单个文件
    # my_ftp.upload_file("G:/f/", "/App/")
    # 上传目录
    # my_ftp.upload_file_tree("G:/f/", "/App/")
    my_ftp.close()



ftp服务器上有4个文件
https://t1.wodetu.cn/2023/10/31/f52828bae8be94430b5bc60220a55a8f.jpg
下载到本地只有1个文件
https://t4.wodetu.cn/2023/10/31/c9706217bd3a1306c03d1295ccac67e8.jpg
下载文件夹的时候会报错
https://t3.wodetu.cn/2023/10/31/adedc65f091ce8815a77ed7dd11b2767.jpg
以下是报错日志:
https://t4.wodetu.cn/2023/10/31/ff0c957f45052806a2e1a7da8c5a8797.jpg

歌者文明清理员 发表于 2023-10-31 21:56:17

hejialiangya 发表于 2023-10-31 21:08
学了2天学了点皮毛,打算用通过ftp协议来实现。
瞎整了一段代码,但是运行起来有点问题,请大佬帮我看看 ...

好家伙,给我整不会了{:10_257:}

参考:
https://www.cnblogs.com/kaituorensheng/p/4480512.html
https://www.jianshu.com/p/8e645e6de8bf

Mike_python小 发表于 2023-10-31 22:03:51

可以每次连接时服务端发送版本号,一致则进入,不一致则传输最新版本文件,我之前写的聊天室里面有点,就在论坛里面,你觉得麻烦看看这个思路也行{:10_245:}{:10_245:}
页: [1]
查看完整版本: [Python] 我想实现局域网自动更新文件功能