from aiohttp import web
import aiohttp_jinja2
import os
from pathlib import Path
from aiohttp.web_response import json_response
from ..helpers.auth import login_required
from .auth import CustomPydanticView
# PROJECT_DIR is no longer directly used for user files, as FileService manages them
# PROJECT_DIR = Path(__file__).parent.parent.parent / "project"
class SiteView(web.View):
async def get(self):
if self.request.path == "/dashboard":
return await self.dashboard()
elif self.request.path == "/solutions":
return await self.solutions()
elif self.request.path == "/pricing":
return await self.pricing()
elif self.request.path == "/security":
return await self.security()
elif self.request.path == "/support":
return await self.support()
elif self.request.path == "/use_cases":
return await self.use_cases()
elif self.request.path == "/terms":
return await self.terms()
elif self.request.path == "/privacy":
return await self.privacy()
elif self.request.path == "/shared":
return await self.shared()
elif self.request.path == "/recent":
return await self.recent()
elif self.request.path == "/favorites":
return await self.favorites()
elif self.request.path == "/trash":
return await self.trash()
elif self.request.path == "/users":
return await self.users()
return aiohttp_jinja2.render_template(
"pages/index.html", self.request, {"request": self.request, "errors": {}, "user": self.request.get("user")}
)
@login_required
async def dashboard(self):
return web.HTTPFound(self.request.app.router["file_browser"].url_for())
async def solutions(self):
return aiohttp_jinja2.render_template(
"pages/solutions.html", self.request, {"request": self.request, "errors": {}, "user": self.request.get("user")}
)
async def pricing(self):
return aiohttp_jinja2.render_template(
"pages/pricing.html", self.request, {"request": self.request, "errors": {}, "user": self.request.get("user")}
)
async def security(self):
return aiohttp_jinja2.render_template(
"pages/security.html", self.request, {"request": self.request, "errors": {}, "user": self.request.get("user")}
)
@login_required
async def support(self):
return aiohttp_jinja2.render_template(
"pages/support.html", self.request, {"request": self.request, "errors": {}, "user": self.request["user"], "active_page": "support"}
)
async def use_cases(self):
return aiohttp_jinja2.render_template(
"pages/use_cases.html", self.request, {"request": self.request, "errors": {}, "user": self.request.get("user")}
)
async def terms(self):
return aiohttp_jinja2.render_template(
"pages/terms.html", self.request, {"request": self.request, "errors": {}, "user": self.request.get("user")}
)
async def privacy(self):
return aiohttp_jinja2.render_template(
"pages/privacy.html", self.request, {"request": self.request, "errors": {}, "user": self.request.get("user")}
)
@login_required
async def shared(self):
return aiohttp_jinja2.render_template(
"pages/shared.html", self.request, {"request": self.request, "errors": {}, "user": self.request["user"], "active_page": "shared"}
)
@login_required
async def recent(self):
return aiohttp_jinja2.render_template(
"pages/recent.html", self.request, {"request": self.request, "errors": {}, "user": self.request["user"], "active_page": "recent"}
)
@login_required
async def favorites(self):
return aiohttp_jinja2.render_template(
"pages/favorites.html", self.request, {"request": self.request, "errors": {}, "user": self.request["user"], "active_page": "favorites"}
)
@login_required
async def trash(self):
return aiohttp_jinja2.render_template(
"pages/trash.html", self.request, {"request": self.request, "errors": {}, "user": self.request["user"], "active_page": "trash"}
)
@login_required
async def users(self):
success_message = self.request.query.get("success")
error_message = self.request.query.get("error")
return aiohttp_jinja2.render_template(
"pages/users.html", self.request, {
"request": self.request,
"errors": {},
"user": self.request["user"],
"active_page": "users",
"success_message": success_message,
"error_message": error_message
}
)
class FileBrowserView(web.View):
@login_required
async def get(self):
if self.request.match_info.get("file_path") is not None:
return await self.get_download_file()
user_email = self.request["user"]["email"]
file_service = self.request.app["file_service"]
path = self.request.query.get("path", "")
files = await file_service.list_files(user_email, path)
success_message = self.request.query.get("success")
error_message = self.request.query.get("error")
return aiohttp_jinja2.render_template(
"pages/file_browser.html",
self.request,
{
"request": self.request,
"files": files,
"user": self.request["user"],
"current_path": path,
"success_message": success_message,
"error_message": error_message,
"active_page": "files",
},
)
@login_required
async def post(self):
user_email = self.request["user"]["email"]
file_service = self.request.app["file_service"]
route_name = self.request.match_info.route.name
if route_name == "new_folder":
data = await self.request.post()
folder_name = data.get("folder_name")
if not folder_name:
return web.HTTPFound(
self.request.app.router["file_browser"].url_for().with_query(
error="Folder name is required"
)
)
success = await file_service.create_folder(user_email, folder_name)
if success:
return web.HTTPFound(
self.request.app.router["file_browser"].url_for().with_query(
success=f"Folder '{folder_name}' created successfully"
)
)
else:
return web.HTTPFound(
self.request.app.router["file_browser"].url_for().with_query(
error=f"Folder '{folder_name}' already exists or could not be created"
)
)
elif route_name == "share_file":
file_path = self.request.match_info.get("file_path")
if not file_path:
return json_response({"error": "File path is required for sharing"}, status=400)
share_id = await file_service.generate_share_link(user_email, file_path)
if share_id:
share_link = f"{self.request.scheme}://{self.request.host}/shared_file/{share_id}"
return json_response({"share_link": share_link})
else:
return json_response({"error": "Failed to generate share link"}, status=500)
elif route_name == "delete_item":
item_path = self.request.match_info.get("file_path")
if not item_path:
return web.HTTPFound(
self.request.app.router["file_browser"].url_for().with_query(
error="Item path is required for deletion"
)
)
success = await file_service.delete_item(user_email, item_path)
if success:
return web.HTTPFound(
self.request.app.router["file_browser"].url_for().with_query(
success=f"Item deleted successfully"
)
)
else:
return web.HTTPFound(
self.request.app.router["file_browser"].url_for().with_query(
error="Failed to delete item - it may not exist"
)
)
return web.HTTPBadRequest(text="Unknown file action")
@login_required
async def get_download_file(self):
user_email = self.request["user"]["email"]
file_service = self.request.app["file_service"]
file_path = self.request.match_info.get("file_path")
if not file_path:
return web.HTTPBadRequest(text="File path is required for download")
result = await file_service.download_file(user_email, file_path)
if result:
content, filename = result
response = web.Response(body=content)
response.headers["Content-Disposition"] = f"attachment; filename={filename}"
response.headers["Content-Type"] = "application/octet-stream"
return response
else:
return web.HTTPNotFound(text="File not found")
class OrderView(CustomPydanticView):
template_name = "pages/order.html"
@login_required
async def get(self):
config_service = self.request.app["config_service"]
price_per_gb = config_service.get_price_per_gb()
return aiohttp_jinja2.render_template(
self.template_name, self.request, {
"request": self.request,
"errors": {},
"user": self.request["user"],
"price_per_gb": price_per_gb,
"active_page": "order"
}
)
@login_required
async def post(self):
config_service = self.request.app["config_service"]
price_per_gb = config_service.get_price_per_gb()
return aiohttp_jinja2.render_template(
self.template_name, self.request, {
"request": self.request,
"errors": {},
"user": self.request["user"],
"price_per_gb": price_per_gb,
"active_page": "order"
}
)
class UserManagementView(web.View):
@login_required
async def get(self):
route_name = self.request.match_info.route.name
if route_name == "add_user":
return await self.add_user_page()
elif route_name == "edit_user":
return await self.edit_user_page()
elif route_name == "user_details":
return await self.user_details_page()
return web.HTTPNotFound()
@login_required
async def post(self):
route_name = self.request.match_info.route.name
if route_name == "add_user":
return await self.add_user_submit()
elif route_name == "edit_user":
return await self.edit_user_submit()
elif route_name == "delete_user_page":
return await self.delete_user_submit()
return web.HTTPNotFound()
async def add_user_page(self):
return aiohttp_jinja2.render_template(
"pages/add_user.html",
self.request,
{
"request": self.request,
"user": self.request["user"],
"errors": {},
"form_data": {},
"active_page": "users"
}
)
async def add_user_submit(self):
data = await self.request.post()
full_name = data.get("full_name", "").strip()
email = data.get("email", "").strip()
password = data.get("password", "")
confirm_password = data.get("confirm_password", "")
storage_quota_gb = data.get("storage_quota_gb", "10")
errors = {}
if not full_name:
errors["full_name"] = "Full name is required"
if not email:
errors["email"] = "Email is required"
if not password:
errors["password"] = "Password is required"
if password != confirm_password:
errors["confirm_password"] = "Passwords do not match"
try:
storage_quota_gb = int(storage_quota_gb)
if storage_quota_gb < 1:
errors["storage_quota_gb"] = "Storage quota must be at least 1 GB"
except ValueError:
errors["storage_quota_gb"] = "Invalid storage quota value"
if errors:
return aiohttp_jinja2.render_template(
"pages/add_user.html",
self.request,
{
"request": self.request,
"user": self.request["user"],
"errors": errors,
"form_data": data,
"active_page": "users"
}
)
user_service = self.request.app["user_service"]
parent_email = self.request["user"]["email"]
try:
new_user = user_service.create_user(
full_name=full_name,
email=email,
password=password,
parent_email=parent_email
)
user_service.update_user_quota(email, float(storage_quota_gb))
return web.HTTPFound(
self.request.app.router["users"].url_for().with_query(
success=f"User {email} added successfully"
)
)
except ValueError as e:
errors["email"] = str(e)
return aiohttp_jinja2.render_template(
"pages/add_user.html",
self.request,
{
"request": self.request,
"user": self.request["user"],
"errors": errors,
"form_data": data,
"active_page": "users"
}
)
async def edit_user_page(self):
email = self.request.match_info.get("email")
user_service = self.request.app["user_service"]
user_data = user_service.get_user_by_email(email)
if not user_data:
return web.HTTPNotFound(text="User not found")
success_message = self.request.query.get("success")
return aiohttp_jinja2.render_template(
"pages/edit_user.html",
self.request,
{
"request": self.request,
"user": self.request["user"],
"user_data": user_data,
"errors": {},
"success_message": success_message,
"active_page": "users"
}
)
async def edit_user_submit(self):
email = self.request.match_info.get("email")
data = await self.request.post()
storage_quota_gb = data.get("storage_quota_gb", "")
errors = {}
try:
storage_quota_gb = float(storage_quota_gb)
if storage_quota_gb < 1:
errors["storage_quota_gb"] = "Storage quota must be at least 1 GB"
except ValueError:
errors["storage_quota_gb"] = "Invalid storage quota value"
user_service = self.request.app["user_service"]
user_data = user_service.get_user_by_email(email)
if not user_data:
return web.HTTPNotFound(text="User not found")
if errors:
return aiohttp_jinja2.render_template(
"pages/edit_user.html",
self.request,
{
"request": self.request,
"user": self.request["user"],
"user_data": user_data,
"errors": errors,
"active_page": "users"
}
)
user_service.update_user_quota(email, storage_quota_gb)
return web.HTTPFound(
self.request.app.router["edit_user"].url_for(email=email).with_query(
success="User quota updated successfully"
)
)
async def user_details_page(self):
email = self.request.match_info.get("email")
user_service = self.request.app["user_service"]
user_data = user_service.get_user_by_email(email)
if not user_data:
return web.HTTPNotFound(text="User not found")
return aiohttp_jinja2.render_template(
"pages/user_details.html",
self.request,
{
"request": self.request,
"user": self.request["user"],
"user_data": user_data,
"active_page": "users"
}
)
async def delete_user_submit(self):
email = self.request.match_info.get("email")
user_service = self.request.app["user_service"]
user_data = user_service.get_user_by_email(email)
if not user_data:
return web.HTTPNotFound(text="User not found")
user_service.delete_user(email)
return web.HTTPFound(
self.request.app.router["users"].url_for().with_query(
success=f"User {email} deleted successfully"
)
)
Reference in New Issue View Git Blame Copy Permalink