场景描述

在工作中经常需要把一些文件上传到服务器，以方便使用和管理，如：将文件上传到腾讯云的对象存储(COS)。那么该如何使用Python实现文件上传呢？

方案一

import requests


class Upload:
    def __init__(self):
        """
        基础配置
        """
        # 请求头
        self.headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/96.0.4664.45 Safari/537.36'
        }
        # 上传链接
        self.upload_url = 'https://api.test.cn/upload'  # POST方法

    def upload(self):
        """
        上传文件
        """
        data = {
            'bucketName': 'test',
            'objectName': '测试附件.xlsx',
            'fileDownloadName': '测试附件.xlsx',
        }
        files = {
            'file': open('测试文件.xls', 'rb')
        }
        res = requests.post(url=self.upload_url, headers=self.headers, data=data, files=files)
        print(res.json())


if __name__ == '__main__':
    upload = Upload()
    upload.upload()

方案二

import os
import sys

import pymysql
import requests
import you_get


class Upload:
    def __init__(self):
        """
        基础配置
        """
        # 请求头
        self.headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/99.0.4844.51 Safari/537.36'
        }
        # 上传链接
        self.upload_url = 'https://api.test.cn/upload'  # POST方法

    def update_data(self):
        """
        更新数据，补全数据库中视频对应的oss链接
        """
        connection = pymysql.connect(
            host='host',
            user='user',
            password='password',
            database='db',
            charset='utf8mb4',
            cursorclass=pymysql.cursors.DictCursor
        )
        with connection:
            with connection.cursor() as cursor:
                # 获取所有数据
                cursor.execute('select id, video_title, video_url, video_oss  from video_bilibili;')
                for item in cursor.fetchall():
                    if not item.get('video_oss'):
                        print('*' * 60)
                        # 下载视频
                        print(item.get('video_title'))
                        self.download(item.get('video_url'), item.get('video_title'))
                        # 上传视频
                        if video_oss := self.upload(item.get('video_title')):
                            # 编辑SQL语句
                            sql = f"update video_bilibili set video_oss='{video_oss}' where id={item.get('id')};"
                            # 执行SQL语句，更新数据
                            try:
                                cursor.execute(sql)
                                connection.commit()
                            except Exception as e:
                                print(e)
                                print(item)

    def download(self, url, filename):
        """
        下载视频
        """
        sys.argv = ['you-get', '-O', filename, url]
        you_get.main()

    def upload(self, title):
        """
        上传文件
        """
        files = [
            ('files', (
                f'{title}.mp4',  # 文件名称
                open(f'{title}.mp4', 'rb'),  # 读取文件
                'application/octet-stream'
            ))
        ]
        # 上传文件
        response = requests.post(url=self.upload_url, headers=self.headers, files=files)
        if response.json().get('code') == 200:
            video_oss = response.json().get('data')[0].get('urlPath')
            # 删除已下载的数据
            for file in os.listdir():
                if os.path.splitext(file)[1] == '.mp4' or os.path.splitext(file)[1] == '.xml':
                    os.remove(file)
            # 返回oss链接
            return video_oss


if __name__ == '__main__':
    upload = Upload()
    # upload.upload('上传测试文件')
    upload.update_data()

说明：该过程使用requests模拟了在Postman中使用form-data上传文件的过程。
流程：先使用you-get根据数据库中存储的第三方视频链接将视频下载下来，然后再使用upload方法将视频上传至oss，这时会获取到oss链接，最后将获取到的oss链接更新到对应的数据库中即可。
注意：具体操作根据实际情况而定，这里只提供思路。

参考链接

• 更加复杂的 POST 请求
• POST一个多部分编码(Multipart-Encoded)的文件
• requests toolbelt