2024-08-11 10:03:43 +03:00
|
|
|
from __future__ import annotations
|
|
|
|
|
|
|
|
|
|
import base64
|
|
|
|
|
import datetime
|
|
|
|
|
import hashlib
|
|
|
|
|
import os
|
|
|
|
|
import uuid
|
|
|
|
|
|
2024-08-21 05:22:00 +03:00
|
|
|
import dateutil
|
|
|
|
|
import dateutil.tz
|
2024-08-11 10:03:43 +03:00
|
|
|
import file_widgets
|
|
|
|
|
import pydantic
|
|
|
|
|
from PyQt6.QtGui import QIcon
|
|
|
|
|
from PyQt6.QtWidgets import QFileDialog, QLabel, QMessageBox, QWidget
|
|
|
|
|
from request_client import RequestClient
|
2024-08-21 04:05:36 +03:00
|
|
|
from utils import resource_path
|
2024-08-11 10:03:43 +03:00
|
|
|
|
|
|
|
|
|
|
|
|
|
class File(pydantic.BaseModel):
|
|
|
|
|
file_id: uuid.UUID
|
|
|
|
|
file_name: str
|
|
|
|
|
file_size: int
|
|
|
|
|
sha512: str
|
|
|
|
|
created_at: datetime.datetime
|
|
|
|
|
updated_at: datetime.datetime
|
|
|
|
|
|
|
|
|
|
def name(self) -> str:
|
|
|
|
|
return self.file_name
|
|
|
|
|
|
|
|
|
|
def delete(self) -> None:
|
|
|
|
|
RequestClient().client.delete(
|
|
|
|
|
"/files", params={"file_id": self.file_id}
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
def details(self, list: file_widgets.FileListWidget) -> QWidget:
|
|
|
|
|
del list
|
2024-08-21 05:22:00 +03:00
|
|
|
file_size = self._format_size(self.file_size)
|
2024-08-11 10:03:43 +03:00
|
|
|
file_size_text = f"{file_size[0]:.2f} {file_size[1]}"
|
2024-08-21 05:22:00 +03:00
|
|
|
created_at = self._format_date(self.created_at)
|
|
|
|
|
updated_at = self._format_date(self.updated_at)
|
2024-08-11 10:03:43 +03:00
|
|
|
details = (
|
|
|
|
|
f"file id: {self.file_id}\nfile_name: {self.file_name}\n"
|
|
|
|
|
+ f"file_size: {file_size_text}\n"
|
2024-08-21 05:22:00 +03:00
|
|
|
+ f"created at: {created_at}\nupdated at: {updated_at}"
|
2024-08-11 10:03:43 +03:00
|
|
|
)
|
|
|
|
|
label = QLabel()
|
|
|
|
|
label.setWindowTitle("File info")
|
|
|
|
|
label.setText(details)
|
|
|
|
|
return label
|
|
|
|
|
|
|
|
|
|
@staticmethod
|
2024-08-21 05:22:00 +03:00
|
|
|
def _format_size(size: int):
|
2024-08-11 10:03:43 +03:00
|
|
|
power = 2**10
|
|
|
|
|
n = 0
|
|
|
|
|
power_labels = {0: "", 1: "kibi", 2: "mebi", 3: "gibi", 4: "tebi"}
|
|
|
|
|
while size > power and n < 4:
|
|
|
|
|
size /= power
|
|
|
|
|
n += 1
|
|
|
|
|
return size, power_labels[n] + "bytes"
|
|
|
|
|
|
2024-08-21 05:22:00 +03:00
|
|
|
@staticmethod
|
|
|
|
|
def _format_date(date: datetime.datetime) -> str:
|
|
|
|
|
date = date.replace(tzinfo=dateutil.tz.tzutc())
|
|
|
|
|
date = date.astimezone(dateutil.tz.tzlocal())
|
|
|
|
|
return date.strftime("%Y-%m-%d %H:%M:%S")
|
|
|
|
|
|
2024-08-11 10:03:43 +03:00
|
|
|
def icon(self) -> QIcon:
|
2024-08-21 04:05:36 +03:00
|
|
|
return QIcon(resource_path("assets/file.png"))
|
2024-08-11 10:03:43 +03:00
|
|
|
|
|
|
|
|
def double_click(self, list: file_widgets.FileListWidget) -> None:
|
|
|
|
|
location = QFileDialog.getExistingDirectory(
|
|
|
|
|
list, caption="Select save location"
|
|
|
|
|
)
|
|
|
|
|
if not location:
|
|
|
|
|
return
|
|
|
|
|
with open(
|
|
|
|
|
os.path.join(location, self.file_name), "wb"
|
|
|
|
|
) as f, RequestClient().client.stream(
|
|
|
|
|
"GET", "/files", params={"file_id": self.file_id}
|
|
|
|
|
) as stream:
|
|
|
|
|
if not stream.is_success:
|
|
|
|
|
QMessageBox.warning(list, "Error downloading the file")
|
|
|
|
|
return
|
|
|
|
|
for data in stream.iter_bytes():
|
|
|
|
|
f.write(data)
|
|
|
|
|
|
|
|
|
|
def create(path: str, parent_id: uuid.UUID):
|
|
|
|
|
"""Upload the file"""
|
2024-08-21 05:22:00 +03:00
|
|
|
print(path)
|
2024-08-11 10:03:43 +03:00
|
|
|
file_name = os.path.basename(path)
|
|
|
|
|
try:
|
|
|
|
|
with open(path, "rb") as f:
|
|
|
|
|
files = {"file": (file_name, f)}
|
|
|
|
|
response = RequestClient().client.post(
|
|
|
|
|
"/files",
|
|
|
|
|
files=files,
|
|
|
|
|
params={"parent_folder": parent_id},
|
|
|
|
|
)
|
|
|
|
|
if not response.is_success:
|
|
|
|
|
QMessageBox.warning(
|
|
|
|
|
None, "Error", f"Upload failed: {response.text}"
|
|
|
|
|
)
|
|
|
|
|
print(response.text)
|
|
|
|
|
except Exception as e:
|
|
|
|
|
QMessageBox.critical(None, "HTTP Error", str(e))
|
|
|
|
|
|
|
|
|
|
def download(self, path: str):
|
2024-08-21 05:22:00 +03:00
|
|
|
try:
|
|
|
|
|
with open(path, "wb") as f, RequestClient().client.stream(
|
|
|
|
|
"GET", "/files", params={"file_id": self.file_id}
|
|
|
|
|
) as stream:
|
|
|
|
|
if not stream.is_success:
|
|
|
|
|
QMessageBox.warning(
|
|
|
|
|
None,
|
|
|
|
|
"Error downloading the file",
|
|
|
|
|
"Error downloading the file",
|
|
|
|
|
)
|
|
|
|
|
return
|
|
|
|
|
for data in stream.iter_bytes():
|
|
|
|
|
f.write(data)
|
|
|
|
|
except Exception as e:
|
|
|
|
|
QMessageBox.warning(None, "Error downloading the file", str(e))
|
2024-08-11 10:03:43 +03:00
|
|
|
|
|
|
|
|
def modify(self, path: str):
|
|
|
|
|
"""Upload the file"""
|
|
|
|
|
file_name = os.path.basename(path)
|
|
|
|
|
hash = hashlib.file_digest(open(path, "rb"), "sha512").digest()
|
|
|
|
|
if base64.b64decode(self.sha512) == hash:
|
|
|
|
|
return
|
|
|
|
|
try:
|
|
|
|
|
with open(path, "rb") as f:
|
|
|
|
|
files = {"file": (file_name, f)}
|
|
|
|
|
response = RequestClient().client.patch(
|
|
|
|
|
"/files",
|
|
|
|
|
files=files,
|
|
|
|
|
params={"file_id": self.file_id},
|
|
|
|
|
)
|
|
|
|
|
if not response.is_success:
|
|
|
|
|
QMessageBox.warning(
|
|
|
|
|
None, "Error", f"Upload failed: {response.text}"
|
|
|
|
|
)
|
|
|
|
|
except Exception as e:
|
|
|
|
|
QMessageBox.critical(None, "HTTP Error", str(e))
|