cowtransfer(奶牛快传)上传文件脚本实现

序：本文仅用于交流学习，请勿用于非法用途。如有侵权请私信通知删除。

在前面我们分析上传文件的流程，以及发送单个文件夹和文件的区别于联系。本文就基于此基础实现文件上传的脚本。在这就以上传单个文件举例。

文件路径：C:\Users\HASEE\Desktop\book

文件名：绿野仙踪.txt

（一）流程分析

根据我们前面可知上传文件大致分为五个步骤：

1.上传文件夹或文件相关权限 （是否能被下载等）

2.获取关键参数 （服务器存储名，位置等一些关键参数）

3.上传文件夹或文件 （这一步最关键，也最难）

4.上传文件信息 （文件大小，文件路径等）

5.从服务器获取上传文件（下载链接等）

分别对应：

1.transfer（上传相关权限）

2.tokens（获取关键参数）

3)folders(上传文件夹信息)

3.xxxx.xxx(上传文件)

4.flies（上传文件信息）

5.uploaded(获取已上传文件)

（二）代码实现

此代码都是基于登录状态下进行，服务器会校验header里的authorization参数，直接复制即可

一、transfer 请求实现

enableDownload:是否能够被下载，默认为ture

enablePreview:是否能够被浏览,默认为ture

enableSaveTo: 是否能够被转存，默认为ture

message: 文件描述（非必须）

name:文件名（非必须）

validDays:存储时间（1,7,30天,-1代表永久有效）

1.实现

# -*- coding:utf-8 -*- import json import requests def transfer(): url = 'https://cowtransfer.com/core/api/transfer' data = { 'enableDownload': 'true', # 是否能够被下载(默认为true) 'enablePreview': 'true', # 是否能够被浏览(默认为true) 'enableSaveTo': 'true', # 是否能够被转存(默认为true) 'message': '', # 补充描述 'name': '', # 上传文件名字(相当于文件夹的名字),默认为上传文件名 'validDays': '-1' # 有效期,分别有1,7,30,-1,失效传输(1,7,30)代表天数,-1代表永久有效，失效传输未知 } res = requests.post(url=url, headers=headers, data=json.dumps(data), timeout=30) if res.json()['code'] == '0000' and res.json()['message'] == 'success': return res.json()['data'] else: print(f'文件权限设置失败 错误信息: 请求状态:{res.status_code} msg:{res.json()["message"]}') if __name__ == '__main__': AUTHORIZATION = 'eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJndWlkIjoiYzZjM2ViOGItZDQ4NS00ZWI2LThlNWEtZTNhZGNiZjkyNzRiIiwiZXhwIjoxNjc5MjIwMTc3fQ.zGYPNlGewGqk0UrjHBy8t8X2cH6UlCd4tx-bCdetCKQ' headers = { # authorization 账号身份标识,每个账号每次登录都不同 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36 Edg/108.0.1462.76', 'authorization': AUTHORIZATION, 'cookie': f'cow-auth-token={AUTHORIZATION};', 'content-type': 'application/json' } print(transfer())

2.效果

其中uniqueUrl 就是我们下载文件地址的一部分

二、tokens请求实现

可以看到此请求只有一个参数file_format,其实就是文件后缀名，我上传的是txt文件那么后缀自然是txt

1. 实现

# -*- coding:utf-8 -*- import json import requests def get_tokens(): url = 'https://cowtransfer.com/core/api/filems/front/upload/tokens' data = { 'file_format': 'txt' # 文件后缀 } res = requests.post(url=url, headers=headers, data=json.dumps(data), timeout=30) if 'code' not in res.json(): return res.json() else: print(f'获取参数失败 错误信息: 请求状态:{res.status_code} msg:{res.json()["message"]}') if __name__ == '__main__': AUTHORIZATION = 'eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJndWlkIjoiYzZjM2ViOGItZDQ4NS00ZWI2LThlNWEtZTNhZGNiZjkyNzRiIiwiZXhwIjoxNjc5MjIwMTc3fQ.zGYPNlGewGqk0UrjHBy8t8X2cH6UlCd4tx-bCdetCKQ' headers = { # authorization 账号身份标识,每个账号每次登录都不同 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36 Edg/108.0.1462.76', 'authorization': AUTHORIZATION, 'cookie': f'cow-auth-token={AUTHORIZATION};', 'content-type': 'application/json' } print(get_tokens())

2. 效果

返回过来一堆数据，其中region,security_token,access_key_secret,accexx_key_id,object_name一些参数都是后面需要的

三、xxxx.txt 请求实现

此请求是将我们本地的文件上传的服务器，但是这却不是我们常见的get,post请求，而是put请求。

其次可以发现我们的Authorization并不是我们自定义的，其实这里的authorization是用AWS S3 V4 加密过的，这是一种签名算法，感兴趣的小伙伴可看看下面的这两篇文章

AWS S3 V4签名实现（nodejs） 对象存储-Python实现AWS鉴权

1. 实现

# 此代码不能直接运行,如需完整代码请私信作者 def put_file(tokens): front_url = f'https://{tokens["bucket_name"]}.{tokens["region"]}.aliyuncs.com/' latter_url = tokens['object_name'] really_url = front_url + latter_url # 服务器存储文件路径 # 获取autorization header(关键) security_token = tokens['security_token'] accesskeyid = tokens['access_key_id'] secretaccesskey = tokens['access_key_secret'] amz_ua = 'aws-sdk-js/2.828.0 callback' autorization = '' # AWS S3 v4 签名后的autorization         # headers的顺序不能改变 headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36 Edg/108.0.1462.76', 'Authorization': autorization, 'Host': 'gd-cowtransfer.oss-cn-hangzhou.aliyuncs.com', 'X-Amz-Content-Sha256': 'UNSIGNED-PAYLOAD', 'X-Amz-Date': self.amzdate, 'x-amz-security-token': tokens['security_token'], 'X-Amz-User-Agent': 'aws-sdk-js/2.828.0 callback' } if self.content_type is not None: headers.update({'Content-Type': self.content_type}) with open(self.file_path + f'/{self.file_name}', 'rb') as f: file_data = f.read() # 文件对象 respones = requsts.put(url=really_url, data=file_data, headers=headers) if respones.status_code == 200: print(f'文件{self.file_name} 上传成功') else: print(f'{self.file_name} 上传失败 错误信息:请求状态:{respones.status_code}')

四、flies 请求实现

origin_url : 服务器存储文件位置

folder_id: 文件夹id（上传单个文件默认为0）

second_transmission: 是否第二次上传（默认为ture）

1.实现

# -*- coding:utf-8 -*- # 此代码不能直接运行，完整代码请私信作者 import json import requests import os def get_tokens(): url = 'https://cowtransfer.com/core/api/filems/front/upload/tokens' data = { 'file_format': 'txt' # 文件后缀 } res = requests.post(url=url, headers=headers, data=json.dumps(data), timeout=30) if 'code' not in res.json(): return res.json() else: print(f'获取参数失败 错误信息: 请求状态:{res.status_code} msg:{res.json()["message"]}') def files_info(tokens, folder_id=None): if folder_id is None: folder_id = '0' url = 'https://cowtransfer.com/core/api/dam/asset/files' data = { 'file_info': { 'origin_url': tokens['host'] + '/' + tokens['object_name'], 'size': str(os.path.getsize(r'C:\Users\HASEE\Desktop\book' + '\绿野仙踪.txt')), # 获取文件大小 'title': '绿野仙踪.txt' # 文件名 }, 'file_md5': '', 'file_sha1': '', 'folder_id': folder_id, # 文件夹标识,上传至哪个文件夹 'second_transmission': 'true' # 第二次传输(默认为ture) } r_content_id = requests.post(url=url, headers=headers, data=json.dumps(data)).json() return r_content_id if __name__ == '__main__': AUTHORIZATION = 'eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJndWlkIjoiYzZjM2ViOGItZDQ4NS00ZWI2LThlNWEtZTNhZGNiZjkyNzRiIiwiZXhwIjoxNjc5MjIwMTc3fQ.zGYPNlGewGqk0UrjHBy8t8X2cH6UlCd4tx-bCdetCKQ' headers = { # authorization 账号身份标识,每个账号每次登录都不同 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36 Edg/108.0.1462.76', 'authorization': AUTHORIZATION, 'cookie': f'cow-auth-token={AUTHORIZATION};', 'content-type': 'application/json' } res = get_tokens() print(files_info(res)) 

2.效果

content_id就是我们链接文件的id,后面请求文件会用到

五、uploaded请求实现

files : 文件id,前面files请求返回的content_id,每个文件都会返回一个content_id

folders : 文件夹id, 这里上传的单个文件不需要

guid: 未知，transfer 请求返回

一、实现

# 此代码不能直接运行，完整代码请私信作者 def get_uploaded_file(l_content_id, r_guid):     # 更新本地文件 url = 'https://cowtransfer.com/core/api/transfer/uploaded' data = { 'files': l_content_id, 'folders': [], 'guid': r_guid } res = requests.post(url=url, headers=headers, data=json.dumps(data), timeout=30) if res.json()['code'] == '0000' and res.json()['message'] == 'success': pass else: print(f'文件请求失败 错误信息: 请求状态:{res.status_code} msg:{res.json()["message"]}')

（三）最终效果

代码运行效果：

需要完整代码请私信作者