308 lines
8.7 KiB
Python
308 lines
8.7 KiB
Python
from __future__ import annotations
|
|
|
|
import datetime
|
|
import os
|
|
import uuid
|
|
from typing import Protocol, Self
|
|
|
|
import httpx
|
|
import pydantic
|
|
import state
|
|
from PyQt6.QtCore import QPoint, Qt
|
|
from PyQt6.QtGui import (
|
|
QAction,
|
|
QDragEnterEvent,
|
|
QDragMoveEvent,
|
|
QDropEvent,
|
|
QIcon,
|
|
)
|
|
from PyQt6.QtWidgets import (
|
|
QHBoxLayout,
|
|
QListWidget,
|
|
QListWidgetItem,
|
|
QMenu,
|
|
QMessageBox,
|
|
QPushButton,
|
|
QVBoxLayout,
|
|
QWidget,
|
|
QLabel,
|
|
QFileDialog,
|
|
)
|
|
from request_client import RequestClient
|
|
|
|
|
|
class DisplayProtocol(Protocol):
|
|
def name(self) -> str:
|
|
raise NotImplementedError
|
|
|
|
def delete(self) -> None:
|
|
raise NotImplementedError
|
|
|
|
def details(self, list: FileListWidget) -> QWidget:
|
|
raise NotImplementedError
|
|
|
|
def icon(self) -> QIcon:
|
|
raise NotImplementedError
|
|
|
|
def double_click(self, list: FileListWidget) -> None:
|
|
raise NotImplementedError
|
|
|
|
|
|
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: FileListWidget) -> QWidget:
|
|
del list
|
|
file_size = self._format_bytes(self.file_size)
|
|
file_size_text = f"{file_size[0]:.2f} {file_size[1]}"
|
|
details = (
|
|
f"file id: {self.file_id}\nfile_name: {self.file_name}\n"
|
|
+ f"file_size: {file_size_text}\n"
|
|
+ f"created at: {self.created_at}\nupdated at: {self.updated_at}"
|
|
)
|
|
label = QLabel()
|
|
label.setWindowTitle("File info")
|
|
label.setText(details)
|
|
return label
|
|
|
|
@staticmethod
|
|
def _format_bytes(size: int):
|
|
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"
|
|
|
|
def icon(self) -> QIcon:
|
|
return QIcon("assets/file.png")
|
|
|
|
def double_click(self, list: 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)
|
|
|
|
|
|
class Folder(pydantic.BaseModel):
|
|
folder_id: uuid.UUID
|
|
owner_id: int
|
|
folder_name: str
|
|
created_at: datetime.datetime
|
|
|
|
def name(self) -> str:
|
|
return self.folder_name
|
|
|
|
def delete(self) -> None:
|
|
RequestClient().client.delete(
|
|
"/folders", params={"folder_id": self.folder_id}
|
|
)
|
|
|
|
def details(self, list: FileListWidget) -> QWidget:
|
|
# TODO
|
|
raise NotImplementedError
|
|
|
|
def icon(self) -> QIcon:
|
|
return QIcon("assets/folder.png")
|
|
|
|
def double_click(self, list: FileListWidget) -> None:
|
|
list.responses.append(ListResponse.get(self.folder_id))
|
|
list.update()
|
|
|
|
|
|
class ResponseProtocol(Protocol):
|
|
def items(self) -> list[DisplayProtocol]:
|
|
raise NotImplementedError
|
|
|
|
def update(self) -> ResponseProtocol:
|
|
raise NotImplementedError
|
|
|
|
|
|
class ListResponse(pydantic.BaseModel):
|
|
folder_id: uuid.UUID
|
|
files: list[File]
|
|
folders: list[Folder]
|
|
|
|
def items(self) -> list[DisplayProtocol]:
|
|
return self.files + self.folders
|
|
|
|
@classmethod
|
|
def get(cls, folder_id: uuid.UUID | None = None) -> Self:
|
|
params = {}
|
|
if folder_id:
|
|
params["folder_id"] = folder_id
|
|
return ListResponse.model_validate_json(
|
|
RequestClient().client.get("/folders", params=params).text
|
|
)
|
|
|
|
def update(self) -> ResponseProtocol:
|
|
return self.get(self.folder_id)
|
|
|
|
|
|
class FileListWidget(QListWidget):
|
|
def __init__(self, state: state.State):
|
|
super().__init__()
|
|
self.state = state
|
|
self.setAcceptDrops(True)
|
|
self.setContextMenuPolicy(Qt.ContextMenuPolicy.CustomContextMenu)
|
|
self.customContextMenuRequested.connect(self.show_context_menu)
|
|
self.itemDoubleClicked.connect(self.show_item)
|
|
self.details_widget = None
|
|
|
|
self.responses: list[ResponseProtocol] = [ListResponse.get()]
|
|
self.update()
|
|
|
|
def show_item(self, qitem: QListWidgetItem) -> None:
|
|
row = self.row(qitem)
|
|
item = self.current_response().items()[row]
|
|
item.double_click(self)
|
|
|
|
def current_response(self) -> ListResponse:
|
|
if not self.responses:
|
|
self.update_response()
|
|
return self.responses[-1]
|
|
|
|
def update_response(self):
|
|
self.responses[-1] = self.responses[-1].update()
|
|
self.update()
|
|
|
|
def dragEnterEvent(self, event: QDragEnterEvent):
|
|
if event.mimeData().hasUrls():
|
|
event.accept()
|
|
else:
|
|
event.ignore()
|
|
|
|
def dragMoveEvent(self, e: QDragMoveEvent | None) -> None:
|
|
return self.dragEnterEvent(e)
|
|
|
|
def dropEvent(self, event: QDropEvent):
|
|
event.accept()
|
|
print("hi")
|
|
for url in event.mimeData().urls():
|
|
file_path = url.toLocalFile()
|
|
self.upload_file(file_path)
|
|
|
|
def upload_file(self, file_path):
|
|
file_name = os.path.basename(file_path)
|
|
try:
|
|
with open(file_path, "rb") as f:
|
|
files = {"file": (file_name, f)}
|
|
response = RequestClient().client.post(
|
|
"http://localhost:3000/files",
|
|
files=files,
|
|
params={
|
|
"parent_folder": self.current_response().folder_id
|
|
},
|
|
)
|
|
if response.is_success:
|
|
QMessageBox.information(
|
|
self, "Success", "File uploaded successfully"
|
|
)
|
|
self.update_response()
|
|
else:
|
|
QMessageBox.warning(
|
|
self, "Error", f"Upload failed: {response.text}"
|
|
)
|
|
except httpx.HTTPError as e:
|
|
QMessageBox.critical(self, "HTTP Error", str(e))
|
|
|
|
def add_item(self, item: DisplayProtocol):
|
|
widget = QListWidgetItem(item.name())
|
|
widget.setIcon(item.icon())
|
|
self.addItem(widget)
|
|
|
|
def show_context_menu(self, pos: QPoint):
|
|
item = self.itemAt(pos)
|
|
if item:
|
|
menu = QMenu(self)
|
|
details_action = QAction("Details", self)
|
|
details_action.triggered.connect(lambda: self.show_details(item))
|
|
delete_action = QAction("Delete", self)
|
|
delete_action.triggered.connect(lambda: self.delete_item(item))
|
|
menu.addAction(details_action)
|
|
menu.addAction(delete_action)
|
|
menu.exec(self.mapToGlobal(pos))
|
|
|
|
def show_details(self, item):
|
|
row = self.row(item)
|
|
item = self.current_response().items()[row]
|
|
self.details_widget = item.details(self)
|
|
self.details_widget.show()
|
|
|
|
def delete_item(self, item):
|
|
row = self.row(item)
|
|
item = self.current_response().items()[row]
|
|
item.delete()
|
|
self.update_response()
|
|
QMessageBox.information(self, "Delete", f"{item.name()} deleted")
|
|
|
|
def update(self) -> None:
|
|
self.clear()
|
|
last = self.responses[-1]
|
|
for item in last.items():
|
|
self.add_item(item)
|
|
|
|
|
|
class Sidebar(QWidget):
|
|
def __init__(self, state: state.State, file_list: FileListWidget):
|
|
super().__init__()
|
|
self.state = state
|
|
self.file_list = file_list
|
|
layout = QVBoxLayout()
|
|
self.setLayout(layout)
|
|
# Add your sidebar buttons here
|
|
for i in range(5): # Example buttons
|
|
btn = QPushButton(f"Button {i+1}")
|
|
layout.addWidget(btn)
|
|
layout.addStretch()
|
|
|
|
def get_user(self): ...
|
|
|
|
def logout(self): ...
|
|
|
|
def go_back(self): ...
|
|
|
|
def go_root(self): ...
|
|
|
|
def get_tlp(self):
|
|
"""Get top level permitted folders"""
|
|
|
|
def sync(self): ...
|
|
|
|
|
|
class MainFileWidget(QWidget):
|
|
def __init__(self, state: state.State):
|
|
super().__init__()
|
|
self.state = state
|
|
layout = QHBoxLayout()
|
|
self.file_list = FileListWidget(state)
|
|
self.sidebar = Sidebar(state, self.file_list)
|
|
layout.addWidget(self.sidebar)
|
|
layout.addWidget(self.file_list)
|
|
self.setLayout(layout)
|