mirror of
https://github.com/BlueSkyXN/WorkerJS_CloudFlare_ImageBed.git
synced 2024-11-16 11:42:33 +08:00
94 lines
3.1 KiB
Python
94 lines
3.1 KiB
Python
|
import requests
|
|||
|
|
import json
|
|||
|
|
import argparse
|
|||
|
|
import os
|
|||
|
|
import mimetypes
|
|||
|
|
from requests_toolbelt.multipart.encoder import MultipartEncoder
|
|||
|
|
|
|||
|
|
# 硬编码JWT令牌
|
|||
|
|
JWT = ""
|
|||
|
|
|
|||
|
|
def pin_file_to_ipfs(file_path):
|
|||
|
|
# 解析绝对路径
|
|||
|
|
file_path = os.path.abspath(file_path)
|
|||
|
|
|
|||
|
|
if not os.path.isfile(file_path):
|
|||
|
|
print(f"错误: 文件 '{file_path}' 不存在或不是一个文件。")
|
|||
|
|
return
|
|||
|
|
|
|||
|
|
# 获取文件名
|
|||
|
|
file_name = os.path.basename(file_path)
|
|||
|
|
|
|||
|
|
# 手动添加对较新格式的 MIME 类型支持（如 .avif）
|
|||
|
|
mimetypes.add_type('image/avif', '.avif')
|
|||
|
|
mimetypes.add_type('video/webm', '.webm') # 其他可能需要的类型
|
|||
|
|
|
|||
|
|
# 使用 mimetypes 通过扩展名识别 MIME 类型
|
|||
|
|
mime_type, _ = mimetypes.guess_type(file_path)
|
|||
|
|
|
|||
|
|
# 如果 MIME 类型无法识别，设置为'application/octet-stream'
|
|||
|
|
if mime_type is None:
|
|||
|
|
mime_type = 'application/octet-stream'
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
with open(file_path, 'rb') as file:
|
|||
|
|
# 使用 multipart form data 传递文件和 MIME 类型
|
|||
|
|
form_data = MultipartEncoder(
|
|||
|
|
fields={
|
|||
|
|
'file': (file_name, file, mime_type),
|
|||
|
|
'pinataMetadata': json.dumps({
|
|||
|
|
'name': file_name
|
|||
|
|
}),
|
|||
|
|
'pinataOptions': json.dumps({
|
|||
|
|
'cidVersion': 0
|
|||
|
|
})
|
|||
|
|
}
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
# 设置请求头，确保 JWT 和 Content-Type 正确传递
|
|||
|
|
headers = {
|
|||
|
|
'Authorization': f'Bearer {JWT}',
|
|||
|
|
'Content-Type': form_data.content_type
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
print(f"正在上传文件 '{file_name}'，MIME类型: {mime_type}...")
|
|||
|
|
|
|||
|
|
# 向 Pinata 发起 POST 请求上传文件
|
|||
|
|
response = requests.post(
|
|||
|
|
"https://api.pinata.cloud/pinning/pinFileToIPFS",
|
|||
|
|
data=form_data,
|
|||
|
|
headers=headers
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
if response.status_code == 200:
|
|||
|
|
response_json = response.json()
|
|||
|
|
ipfs_hash = response_json["IpfsHash"]
|
|||
|
|
ipfs_url = f"https://gateway.pinata.cloud/ipfs/{ipfs_hash}"
|
|||
|
|
|
|||
|
|
print("文件成功上传至IPFS！")
|
|||
|
|
print(json.dumps(response_json, indent=4))
|
|||
|
|
print(f"可访问的IPFS URL: {ipfs_url}")
|
|||
|
|
else:
|
|||
|
|
print(f"上传失败。状态码: {response.status_code}")
|
|||
|
|
print(f"响应内容: {response.text}")
|
|||
|
|
|
|||
|
|
except requests.exceptions.RequestException as e:
|
|||
|
|
print(f"请求错误: {e}")
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f"发生错误: {e}")
|
|||
|
|
|
|||
|
|
def main():
|
|||
|
|
parser = argparse.ArgumentParser(description="将文件上传到Pinata的IPFS。")
|
|||
|
|
parser.add_argument('file_path', type=str, help='要上传的文件的路径')
|
|||
|
|
|
|||
|
|
args = parser.parse_args()
|
|||
|
|
|
|||
|
|
# 处理中文路径和空格，确保编码正确
|
|||
|
|
try:
|
|||
|
|
pin_file_to_ipfs(args.file_path)
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f"上传时发生错误: {e}")
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
main()
|