Simple example
curl https://server3.cloudkp.com/bin/raw/kpil8 -s | python - file path
Kpcloud API Documentation
Parameter | Required | Type | Description |
---|---|---|---|
dzuuid | Yes | String | A unique identifier for the current file upload session. Generate a random 20 digit string and use it as the value for this parameter. |
dzchunkindex | Yes | Number | The index of the current chunk being uploaded. The first chunk has an index of 0. |
dztotalchunkcount | Yes | Number | The total number of chunks that make up the file being uploaded. |
dztotalfilesize | Yes | Number | The total size of the file being uploaded in bytes. |
file | Yes | File | The actual file chunk being uploaded. |
complete (last req) | Yes | String | Set to "yes" with the last chunk of the file. |
file_name (last req) | Yes | String | The name of the file being uploaded. |
Chunks Logic
The initial chunk size is 131072 bytes (128 KB). If the size of the file is greater than or equal to 786432000 bytes (approximately 750 MB), then the chunk size is multiplied by 4, resulting in a chunk size of 524288 bytes (512 KB). Otherwise, if the size of the file is greater than or equal to 104857600 bytes (approximately 100 MB), then the chunk size is multiplied by 2, resulting in a chunk size of 262144 bytes (256 KB).
Response
The response text from the server contains the following properties:
Property | Type | Description |
---|---|---|
error | String | If there was an error during the upload process, this property will contain the error message. Otherwise, it will be null. |
file_id | String | If the upload was successful, this property will contain the publicly accessible file id of the uploaded file hosted on |
Python Example Usage:
Click here to get kpbin linkimport requests
import random
import requests
import os
import math
import sys
from concurrent.futures import ThreadPoolExecutor
def m_req(data, chunk):
url = "https://cdn.cloudkp.com/upload"
resp = requests.post(url, data=data, files={'file': chunk})
return resp.text
class CloudKP:
def __init__(self):
self.id = ''.join(str(random.randint(0, 9)) for _ in range(20))
self.up = 0
def do_upload(self, part_index, part_count, file_size, chunk):
data = {
'dzuuid': self.id,
'dzchunkindex': part_index,
'dztotalchunkcount': part_count,
'dztotalfilesize': file_size
}
aa = m_req(data, chunk)
self.up += 1
print(f"Uploaded {math.ceil((self.up / part_count) * 100)}%", end='\r')
return aa
def finish_upload(self, part_count, file_name):
data = {
'dzuuid': self.id,
'complete': "yes",
'file_name': file_name,
'dztotalchunkcount': part_count
}
return m_req(data, None)
def read_in_chunks(file_object, chunk_size):
while True:
data = file_object.read(chunk_size)
if not data:
break
yield data
def upload_file(file_path):
file_info = os.stat(file_path)
file_size = file_info.st_size
chunk_size = 131072
if file_size >= 786432000:
chunk_size *= 4
elif file_size >= 104857600:
chunk_size *= 2
part_count = math.ceil(file_size / chunk_size)
part_index = 0
print(f"Uploading {file_size} bytes as {part_count} parts...")
cloud_kp = CloudKP()
pool = ThreadPoolExecutor(max_workers=5)
with open(file_path, 'rb') as f:
for chunk in read_in_chunks(f, chunk_size):
pool.submit(cloud_kp.do_upload, part_index, part_count, file_size, chunk)
part_index += 1
pool.shutdown(wait=True)
file_name = os.path.basename(file_path)
resp2 = cloud_kp.finish_upload(part_count, file_name)
if "error" in resp2:
return resp2
else:
return "https://cloudkp.com?c="+resp2
if len(sys.argv) > 1:
file_path = sys.argv[1]
uploaded_file_url = upload_file(file_path)
print("File link: ", uploaded_file_url)
else:
print("usage: python upload.py filename")