@ישראללוי
from uuid import uuid4
import requests
class Yemot:
BASE_URL = "https://www.call2all.co.il/ym/api/"
def __init__(
self,
user_name: str,
password: str | int,
) -> None:
self.token = f"{user_name}:{password}"
self.params = {
"token": self.token
}
def upload_file(self, path: str, blob: bytes, file_name: str, base_path: str = "ivr2:/") -> bool:
total_size = len(blob)
chunk_size = 49 * 1024 * 1024
chunks = [blob[offset:offset + chunk_size] for offset in range(0, total_size, chunk_size)]
if len(chunks) == 1:
return self._upload_small_file(path, blob, file_name, base_path)
else:
return self._upload_large_file(path, chunks, file_name, total_size, base_path)
def _upload_small_file(self, path: str, blob: bytes, file_name: str, base_path: str) -> bool:
data = {
"path": f"{base_path}{path}",
"convertAudio": "1",
"autoNumbering": "true",
}
data.update(self.params)
files = {"file": (file_name, blob)}
url = f"{self.BASE_URL}UploadFile"
response = requests.post(url, files=files, data=data).json()
return response["responseStatus"] == "OK"
def _upload_large_file(self, path: str, chunks: list[bytes], file_name: str, content_size: int, base_path: str) -> bool:
qquuid = str(uuid4())
offset = 0
for index, chunk in enumerate(chunks):
data = {
"path": f"{base_path}{path}",
"qquuid": qquuid,
"convertAudio": "1",
"autoNumbering": "true",
"uploader": "yemot-admin",
"qqfilename": file_name,
"qqtotalfilesize": content_size,
"qqtotalparts": len(chunks),
"qqchunksize": len(chunk),
"qqpartbyteoffset": offset,
"qqpartindex": index,
}
data.update(self.params)
files = {
"qqfile": chunk,
}
url = f"{self.BASE_URL}UploadFile"
response = requests.post(url, data=data, files=files).json()
if not response["success"]:
print(f"Error uploading chunk {index}: {response}")
return False
offset += len(chunk)
data = {
"path": f"{base_path}{path}",
"uploader": "yemot-admin",
"convertAudio": "1",
"autoNumbering": "true",
"qquuid": qquuid,
"qqfilename": file_name,
"qqtotalfilesize": content_size,
"qqtotalparts": len(chunks),
}
data.update(self.params)
response = requests.post(f"{self.BASE_URL}UploadFile?done", data=data).json()
return response["responseStatus"] == "OK"