一、shutil模块
1、介绍
shutil模块是对os中文件操作的补充。--移动 复制 打包 压缩 解压
2、基本使用
1. shutil.copyfileobj(文件1, 文件2, 长度) 将文件1的数据覆盖copy给文件2，可以copy指定大小的内容 文件1和2都是文件对象，都需要打开后才能进行复制操作 import shutil f1 = open('testdir/1.txt', 'r', encoding='utf8') f2 = open('testdir/2.txt', 'w', encoding='utf8') shutil.copyfileobj(f1, f2, length=1024) 2. shutil.copyfile(文件1，文件2) 不用打开文件，直接复制，实际上copyfile调用了copyfileobj import shutil # 1.txt必须要有，3.txt没有就创建，有就覆盖内容 shutil.copyfile('testdir/1.txt', 'testdir/3.txt') 3. shutil.copymode(文件1，文件2) 仅copy权限，不更改文件内容，组和用户 # 先看两个文件的权限 $ ls -l -rw-r--r-- 1 Administrator 197121 5 1月 5 10:22 1.txt -rwxr-xr-x 1 Administrator 197121 8 1月 5 10:24 2.txt # 运行命令 import shutil shutil.copymode('testdir/1.txt', 'testdir/2.txt') #查看结果 $ ls -l -rw-r--r-- 1 Administrator 197121 5 1月 5 10:22 1.txt -rw-r--r-- 1 Administrator 197121 8 1月 5 10:24 2.txt 4. shutil.copystat(src,dst) 复制所有的状态信息，包括权限，组，用户，时间等 import shutil shutil.copystat('testdir/1.txt', 'testdir/2.txt') #查看结果(时间都变了) $ ls -l -rw-r--r-- 1 Administrator 197121 5 1月 5 10:22 1.txt -rw-r--r-- 1 Administrator 197121 8 1月 5 10:22 2.txt 5. shutil.copy(src,dst) 复制文件的内容以及权限，先copyfile后copymode import shutil shutil.copy('testdir/1.txt', 'testdir/2.txt') 6. shutil.copy2(src,dst) 复制文件的内容以及文件的所有状态信息。先copyfile后copystat import shutil shutil.copy2('testdir/1.txt', 'testdir/2.txt') 7. shutil.copytree(src, dst) 递归的复制文件内容及状态信息(就是拷贝整个目录) import shutil shutil.copytree('testdir', 'testdir2') 结果 /testdir $ ls -l total 2 -rw-r--r-- 1 Administrator 197121 5 1月 5 10:22 1.txt -rw-r--r-- 1 Administrator 197121 8 1月 5 10:22 2.txt testdir2 $ ls -l total 2 -rw-r--r-- 1 Administrator 197121 5 1月 5 10:22 1.txt -rw-r--r-- 1 Administrator 197121 8 1月 5 10:22 2.txt 8. shutil.rmtree(path) 递归地删除文件 import shutil shutil.rmtree('testdir2') 结果 testdir2目录还在，但是里面的文件全删了，也就是testdir2现在是一个空文件夹 9. shutil.move(src, dst) 递归的移动文件 import shutil shutil.move('3.txt', 'testdir') # 把3.txt移动到testdir这个文件夹下 10. make_archive(base_name, format, root_dir=None) 压缩打包 base_name：压缩打包后的文件名或者路径名 format： 压缩或者打包格式 "zip", "tar", "bztar"or "gztar" root_dir : 将哪个目录或者文件打包（也就是源文件） import shutil shutil.make_archive('E:/MyProjects/test', 'zip', root_dir='testdir') 把testdir这个目录(文件也行)压缩到E:/MyProjects目录下，压缩后的名字为test，格式为zip， 也就是说压缩完成后，在E:/MyProjects下会生成一个test.zip包 11. 解压 shutil 对压缩包的处理是调用 ZipFile模块来进行的 11-1、zipfile 压缩解压 # 压缩 # 把testdir文件夹下的1.txt和2.txt压缩到E:/MyProjects/test1.zip z = zipfile.ZipFile('E:/MyProjects/test1.zip', 'w') # 可以用 a 模块，那么zip包里面可以追加内容 z.write('testdir/1.txt') z.write('testdir/2.txt') z.close() # 解压 import zipfile z = zipfile.ZipFile('E:/MyProjects/test1.zip', 'r') z.extractall('E:/MyProjects/test111')#将E:/MyProjects/test1.zip解压到E:/MyProjects/test111目录下 # 高级应用 zipfile.is_zipfile(filename) 判断一个文件是不是压缩文件 ZipFile.namelist() 返回文件列表 ZipFile.open(name[, mode[, password]]) 打开压缩文档中的某个文件 11-2、使用shutil的内在方法解压 shutil._unpack_zipfile(file, path) import shutil shutil._unpack_zipfile('E:/MyProjects/test1.zip', 'E:/MyProjects/test1') # 把E:/MyProjects/test1.zip这个zip包解压到E:/MyProjects/test1文件夹下
二、上传Demo
1、socket上传文件
import json import struct import socketserver import operate_handler class MyFTP(socketserver.BaseRequestHandler): def handle(self): conn = self.request length = conn.recv(4) length = struct.unpack('i',length)[0] operate = (conn.recv(length)).decode('utf-8') operate_dic = json.loads(operate) opt = operate_dic['operate'] usr = operate_dic['user'] print(opt,usr) getattr(operate_handler,opt)(conn,usr) socketserver.TCPServer.allow_reuse_address = True server = socketserver.ThreadingTCPServer(('127.0.0.1',9000),MyFTP) server.serve_forever()
import os import json import struct base_path = r'E:\PythonProject\ftp\server\root' def upload(conn,usr): fileinfo_len = conn.recv(4) fileinfo_len = struct.unpack('i',fileinfo_len)[0] fileinfo = (conn.recv(fileinfo_len)).decode('utf-8') fileinfo = json.loads(fileinfo) file_path = os.path.join(base_path,usr,fileinfo['filename']) file_path = os.path.abspath(file_path) with open(file_path,'wb') as f: while fileinfo['filesize']: content = conn.recv(20480) fileinfo['filesize'] -= len(content) f.write(content) print('接收完毕')
import os import json import struct import socket # 发送信息 def my_send(sk,operate_info): b_optinfo = (json.dumps(operate_info)).encode('utf-8') num = struct.pack('i',len(b_optinfo)) sk.send(num) sk.send(b_optinfo) sk = socket.socket() sk.connect(('127.0.0.1',9000)) # [登录,注册,退出] # 要进行的操作 operate_info = {'operate':'upload','user':'xiaoming'} my_send(sk,operate_info) # 选择一个文发送到server端 file_path = r'F:\电影\电影\荒野生存.mp4' # 发送文件信息 file_name = os.path.basename(file_path) file_size = os.path.getsize(file_path) file_info = {'filename':file_name,'filesize':file_size} my_send(sk,file_info) # server端接收写入 with open(file_path,'rb') as f: while file_size: content = f.read(20480) file_size -= len(content) sk.send(content) print('上传完毕') sk.close()
2、Django中使用form表单上传文件
from django.shortcuts import render, HttpResponse, redirect from django import views import os, time from django.conf import settings class UploadView(views.View): def get(self, request): return render(request, 'upload.html') def post(self, request): # 文本类型的数据用request.POST获取，键对应的值是字符串类型 print(request.POST) # 文件类型的数据用request.FILES获取，键对应的值是类的对象 print(request.FILES) # 获取文件对象(是一大串bytes类型的字节码) file_obj = request.FILES.get('file') # 文件对象有个内置的属性name，用于获取接收到的文件名 filename = file_obj.name # 判断本地是否有同名的文件存在 if os.path.exists(os.path.join(settings.BASE_DIR, filename)): # 重命名文件 prefix_name = filename.split('.')[0] # 前缀 suffix_name = filename.split('.')[1] # 后缀 filename = prefix_name + str(time.time()) + suffix_name # 建一个同名的文件接收上传的数据 # with open(filename, 'wb') as f: # for i in file_obj: # f.write(i) # 跟上面的方法是一样的，只不过chunks是一个内置的方法，可以设置每次接收的数据大小 with open(filename, 'wb') as f: for chunk in file_obj.chunks(chunk_size=1024): f.write(chunk) return HttpResponse('收到啦！')
<!DOCTYPE html> <html lang="zh-CN"> <head> <meta http-equiv="content-type" charset="utf-8"> <meta http-equiv="X-UA-Compatible" content="IE=edge"> <meta name="viewport" content="width=device-width, initial-scale=1"> <title>Upload</title> </head> <body> <form action="" method="post" enctype="multipart/form-data"> {% csrf_token %} <p><input type="text" name="username"></p> <p><input type="file" name="file"></p> <p><input type="submit"></p> </form> </body> </html>
3、ajax文件上传
<!DOCTYPE html> <html lang="en"> <head> <meta http-equiv="content-type" charset="utf-8"> <meta http-equiv="X-UA-Compatible" content="IE=edge"> <meta name="viewport" content="width=device-width, initial-scale=1"> <title>Upload</title> </head> <body> {% csrf_token %} <h1>上传文件</h1> <div> <input type="file" id="f1"> <button id="b1"> 上传</button> </div> <span style="color: red" id="s1"></span> <script src="https://code.jquery.com/jquery-3.3.1.min.js"></script> <script> $('#b1').click(function () { // 取到要上传的文件数据，存到一个对象中 var fdObj = new FormData(); fdObj.append('xx', $('#f1')[0].files[0]); // 在请求的数据中添加csrftokrn var csrfToken = $('[name="csrfmiddlewaretoken"]').val(); fdObj.append('csrfmiddlewaretoken', csrfToken); // 发ajax请求 $.ajax({ url: '/upload/', type: 'post', data: fdObj, processData: false, // 告诉jQuery不要处理我上传的数据 contentType: false, // 告诉jQuery不要设置请求的文件类型,这两个参数相当于enctype="multipart/form-data" success: function (res) { $("#s1").text(res); } }) }) </script> </body> </html>
class UploadView(views.View): def get(self, request): return render(request, 'upload.html') def post(self, request): file_obj = request.FILES.get('xx') file_name = file_obj.name with open(file_name, 'wb') as f: for c in file_obj.chunks(): f.write(c) return HttpResponse('上传成功！')
4、Flask的文件上传并统计代码行数
""" app.config.root_path: 项目的根路径 os.walk: 遍历你给的路径下的所有文件(会递归遍历) 每次循环的根文件夹的路径，文件夹的名字组成的列表，和文件组成的列表 dirpath, dirnames, filenames zipfile: 压缩解压文件的模块 shutil: 也是压缩解压文件的模块，还能移动啥的 """ from flask import Blueprint, request, render_template from flask import current_app as app import shutil from uploadCode.models import CodeRecord from uploadCode import db import os import time uploadBlue = Blueprint('uploadBlue', __name__) # zip包上传 @uploadBlue.route('/upload', methods=['GET', 'POST']) def upload(): if request.method == "GET": return render_template("upload.html", error="") # 先获取前端传过来的文件 file = request.files.get("zip_file") # 判断是否是zip包 zip_file_type = file.filename.rsplit(".", 1) if zip_file_type[-1] != "zip": return render_template("upload.html", error="文件必须是zip包") # 解压路径 upload_path = os.path.join(app.config.root_path, "files", zip_file_type[0]+str(time.time())) print(upload_path) # 解压前端传过来的文件file到upload_path这个路径 shutil._unpack_zipfile(file, upload_path) # 遍历保存的文件夹得到所有.py文件 file_list = [] for (dirpath, dirnames, filenames) in os.walk(upload_path): for filename in filenames: file_type = filename.rsplit(".", 1) if file_type[-1] != "py": continue file_path = os.path.join(dirpath, filename) file_list.append(file_path) # 打开每个文件读取行数 sum_num = 0 for path in file_list: with open(path, mode="rb") as f: for line in f: if line.strip().startswith(b"#"): continue sum_num += 1 # 得到总行数去保存数据库 return str(sum_num)
<!DOCTYPE html> <html lang="en"> <head> <meta charset="UTF-8"> <title>Title</title> </head> <body> <form action="" method="post" enctype="multipart/form-data"> 请上传你的代码： <input type="file" name="zip_file"> <button type="submit">提交</button> {{error}} </form> </body> </html>