gitea-admin/gh-christianlouis-docuelevate
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix failing dependencies check due to temporary issue

Browse Source 
Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
google-labs-jules[bot]
2026-03-23 16:29:22 +00:00
parent 084171395d
commit 1fd3e745cf
24 changed files with 468 additions and 217 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/api/billing.py
+1 -1
View File
@@ -260,7 +260,7 @@ async def stripe_webhook(request: Request, db: Session = Depends(get_db)) -> dic
@require_login
async def billing_success(request: Request) -> Any:
"""Show a success page after a completed Stripe Checkout."""
return _templates.TemplateResponse("billing_success.html", {"request": request})
return _templates.TemplateResponse(request, "billing_success.html")
# ---------------------------------------------------------------------------
app/api/local_auth.py
+9 -9
View File
@@ -101,9 +101,9 @@ async def signup_page(request: Request) -> Any:
if not settings.allow_local_signup:
return RedirectResponse(url="/login?error=Registration+is+not+enabled", status_code=302)
return templates.TemplateResponse(
request,
"signup.html",
{
"request": request,
context={
"csrf_token": getattr(request.state, "csrf_token", ""),
"app_version": settings.version,
},
@@ -113,16 +113,16 @@ async def signup_page(request: Request) -> Any:
@router.get("/verify-email-sent", include_in_schema=False)
async def verify_email_sent_page(request: Request) -> Any:
"""Render the verify-email-sent confirmation page."""
return templates.TemplateResponse("verify_email_sent.html", {"request": request})
return templates.TemplateResponse(request, "verify_email_sent.html")
@router.get("/forgot-username", include_in_schema=False)
async def forgot_username_page(request: Request) -> Any:
"""Render the forgot-username page where users can request a username reminder email."""
return templates.TemplateResponse(
request,
"forgot_username.html",
{
"request": request,
context={
"csrf_token": getattr(request.state, "csrf_token", ""),
"app_version": settings.version,
},
@@ -133,9 +133,9 @@ async def forgot_username_page(request: Request) -> Any:
async def forgot_password_page(request: Request) -> Any:
"""Render the forgot-password page where users can request a reset email."""
return templates.TemplateResponse(
request,
"forgot_password.html",
{
"request": request,
context={
"csrf_token": getattr(request.state, "csrf_token", ""),
"app_version": settings.version,
},
@@ -147,9 +147,9 @@ async def reset_password_page(request: Request) -> Any:
"""Render the password reset form page."""
token = request.query_params.get("token", "")
return templates.TemplateResponse(
request,
"password_reset_form.html",
{
"request": request,
context={
"token": token,
"csrf_token": getattr(request.state, "csrf_token", ""),
"app_version": settings.version,
app/auth.py
+2 -2
View File
@@ -536,9 +536,9 @@ async def login(request: Request):
return RedirectResponse(url="/oauth-login", status_code=status.HTTP_302_FOUND)
return templates.TemplateResponse(
request,
"login.html",
{
"request": request,
context={
"error": error,
"message": message,
"show_oauth": show_oauth,
app/main.py
+4 -5
View File
@@ -424,15 +424,13 @@ async def http_exception_handler(request: Request, exc: HTTPException):
# For frontend routes, return appropriate HTML templates
# Handle 404 errors with a custom template
if exc.status_code == 404:
return _error_templates.TemplateResponse(
"404.html", {"request": request}, status_code=status.HTTP_404_NOT_FOUND
)
return _error_templates.TemplateResponse(request, "404.html", status_code=status.HTTP_404_NOT_FOUND)
# For other HTTP errors, we could create specific templates or use a generic one
# For now, return a simple error page
return _error_templates.TemplateResponse(
request,
"404.html", # Reuse 404 template for other errors, or create a generic error template
{"request": request},
status_code=exc.status_code,
)
@@ -452,8 +450,9 @@ async def custom_500_handler(request: Request, exc: Exception):
# Serve the 500 template for non-API routes
return _error_templates.TemplateResponse(
request,
"500.html",
{"request": request, "exc": exc},
context={"exc": exc},
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
)
app/tasks/upload_to_nextcloud.py
+141 -157
View File
@@ -1,157 +1,141 @@
#!/usr/bin/env python3
import logging
import os
import requests
from requests.auth import HTTPBasicAuth
from app.celery_app import celery
from app.config import settings
from app.tasks.retry_config import UploadTaskWithRetry
from app.utils import log_task_progress
from app.utils.filename_utils import extract_remote_path, get_unique_filename
logger = logging.getLogger(__name__)
@celery.task(base=UploadTaskWithRetry, bind=True)
def upload_to_nextcloud(self, file_path: str, file_id: int = None, folder_override: str = None):
"""
Upload a file to Nextcloud WebDAV.
Args:
file_path: Path to the file to upload
file_id: Optional file ID to associate with logs
"""
task_id = self.request.id
logger.info(f"[{task_id}] Starting Nextcloud upload: {file_path}")
log_task_progress(
task_id,
"upload_to_nextcloud",
"in_progress",
f"Uploading to Nextcloud: {os.path.basename(file_path)}",
file_id=file_id,
)
if not os.path.exists(file_path):
error_msg = f"File not found: {file_path}"
logger.error(f"[{task_id}] {error_msg}")
log_task_progress(task_id, "upload_to_nextcloud", "failure", error_msg, file_id=file_id)
raise FileNotFoundError(error_msg)
# For Nextcloud, we need to check for 'nextcloud_upload_url' instead of 'nextcloud_url'
# This is what's shown in your env view
if not (
getattr(settings, "nextcloud_upload_url", None)
and getattr(settings, "nextcloud_username", None)
and getattr(settings, "nextcloud_password", None)
):
logger.info(f"[{task_id}] Nextcloud upload skipped: Missing configuration")
log_task_progress(task_id, "upload_to_nextcloud", "success", "Skipped: Not configured", file_id=file_id)
return {"status": "Skipped", "reason": "Nextcloud settings not configured"}
filename = os.path.basename(file_path)
try:
# Prepare WebDAV URL - use nextcloud_upload_url instead of nextcloud_url
webdav_url = settings.nextcloud_upload_url
if not webdav_url.endswith("/"):
webdav_url += "/"
# Calculate remote path based on local file structure
remote_base = (
folder_override if folder_override is not None else (getattr(settings, "nextcloud_folder", "") or "")
)
remote_path = extract_remote_path(file_path, settings.workdir, remote_base)
full_url = f"{webdav_url}/{remote_path}"
# Remove any double slashes (except in http://)
full_url = full_url.replace("://", "$PLACEHOLDER$")
while "//" in full_url:
full_url = full_url.replace("//", "/")
full_url = full_url.replace("$PLACEHOLDER$", "://")
# Function to check if file exists in Nextcloud
def check_exists_in_nextcloud(path):
check_url = f"{webdav_url}{os.path.dirname(path)}"
try:
response = requests.request(
"PROPFIND",
check_url,
auth=HTTPBasicAuth(settings.nextcloud_username, settings.nextcloud_password),
headers={"Depth": "1"},
timeout=10,
)
return path in response.text
except Exception:
# If we can't check, assume it doesn't exist
return False
# Check for potential file collision and get a unique name if needed
remote_path = get_unique_filename(remote_path, check_exists_in_nextcloud)
full_url = f"{webdav_url}/{remote_path}"
# Fix double slashes again
full_url = full_url.replace("://", "$PLACEHOLDER$")
while "//" in full_url:
full_url = full_url.replace("//", "/")
full_url = full_url.replace("$PLACEHOLDER$", "://")
# Create necessary parent folders
parent_dirs = os.path.dirname(remote_path)
if parent_dirs:
current_path = ""
for folder in parent_dirs.split("/"):
if not folder:
continue
current_path += f"{folder}/"
mkdir_url = f"{webdav_url}/{current_path}"
# Fix double slashes
mkdir_url = mkdir_url.replace("://", "$PLACEHOLDER$")
while "//" in mkdir_url:
mkdir_url = mkdir_url.replace("//", "/")
mkdir_url = mkdir_url.replace("$PLACEHOLDER$", "://")
requests.request(
"MKCOL",
mkdir_url,
auth=HTTPBasicAuth(settings.nextcloud_username, settings.nextcloud_password),
timeout=10,
)
# Upload the file
logger.info(f"[{task_id}] Uploading {filename} to Nextcloud at {full_url}")
log_task_progress(task_id, "upload_file", "in_progress", f"Uploading to {remote_path}", file_id=file_id)
with open(file_path, "rb") as file_data:
response = requests.put(
full_url,
data=file_data,
auth=HTTPBasicAuth(settings.nextcloud_username, settings.nextcloud_password),
headers={"Content-Type": "application/octet-stream"},
timeout=settings.http_request_timeout, # Use configured timeout for large files
)
if response.status_code in (201, 204): # Created or No Content
logger.info(f"[{task_id}] Successfully uploaded {filename} to Nextcloud at {remote_path}")
log_task_progress(
task_id, "upload_to_nextcloud", "success", f"Uploaded to Nextcloud: {remote_path}", file_id=file_id
)
return {
"status": "Completed",
"file_path": file_path,
"nextcloud_path": remote_path,
"response_code": response.status_code,
}
else:
error_msg = f"Failed to upload {filename} to Nextcloud: {response.status_code} - {response.text}"
logger.error(f"[{task_id}] {error_msg}")
log_task_progress(task_id, "upload_to_nextcloud", "failure", error_msg, file_id=file_id)
raise Exception(error_msg)
except Exception as e:
error_msg = f"Failed to upload {filename} to Nextcloud: {str(e)}"
logger.error(f"[{task_id}] {error_msg}")
log_task_progress(task_id, "upload_to_nextcloud", "failure", error_msg, file_id=file_id)
raise Exception(error_msg)
#!/usr/bin/env python3
import logging
import os
import requests
from requests.auth import HTTPBasicAuth
from app.celery_app import celery
from app.config import settings
from app.tasks.retry_config import UploadTaskWithRetry
from app.utils import log_task_progress
from app.utils.filename_utils import extract_remote_path, get_unique_filename
from app.utils.network import join_url
logger = logging.getLogger(__name__)
@celery.task(base=UploadTaskWithRetry, bind=True)
def upload_to_nextcloud(self, file_path: str, file_id: int = None, folder_override: str = None):
"""
Upload a file to Nextcloud WebDAV.
Args:
file_path: Path to the file to upload
file_id: Optional file ID to associate with logs
"""
task_id = self.request.id
logger.info(f"[{task_id}] Starting Nextcloud upload: {file_path}")
log_task_progress(
task_id,
"upload_to_nextcloud",
"in_progress",
f"Uploading to Nextcloud: {os.path.basename(file_path)}",
file_id=file_id,
)
if not os.path.exists(file_path):
error_msg = f"File not found: {file_path}"
logger.error(f"[{task_id}] {error_msg}")
log_task_progress(task_id, "upload_to_nextcloud", "failure", error_msg, file_id=file_id)
raise FileNotFoundError(error_msg)
# For Nextcloud, we need to check for 'nextcloud_upload_url' instead of 'nextcloud_url'
# This is what's shown in your env view
if not (
getattr(settings, "nextcloud_upload_url", None)
and getattr(settings, "nextcloud_username", None)
and getattr(settings, "nextcloud_password", None)
):
logger.info(f"[{task_id}] Nextcloud upload skipped: Missing configuration")
log_task_progress(task_id, "upload_to_nextcloud", "success", "Skipped: Not configured", file_id=file_id)
return {"status": "Skipped", "reason": "Nextcloud settings not configured"}
filename = os.path.basename(file_path)
try:
# Prepare WebDAV URL - use nextcloud_upload_url instead of nextcloud_url
webdav_url = settings.nextcloud_upload_url
if not webdav_url.endswith("/"):
webdav_url += "/"
# Calculate remote path based on local file structure
remote_base = (
folder_override if folder_override is not None else (getattr(settings, "nextcloud_folder", "") or "")
)
remote_path = extract_remote_path(file_path, settings.workdir, remote_base)
full_url = join_url(webdav_url, remote_path)
# Function to check if file exists in Nextcloud
def check_exists_in_nextcloud(path):
check_url = join_url(webdav_url, os.path.dirname(path))
try:
response = requests.request(
"PROPFIND",
check_url,
auth=HTTPBasicAuth(settings.nextcloud_username, settings.nextcloud_password),
headers={"Depth": "1"},
timeout=10,
)
return path in response.text
except Exception:
# If we can't check, assume it doesn't exist
return False
# Check for potential file collision and get a unique name if needed
remote_path = get_unique_filename(remote_path, check_exists_in_nextcloud)
full_url = join_url(webdav_url, remote_path)
# Create necessary parent folders
parent_dirs = os.path.dirname(remote_path)
if parent_dirs:
current_path = ""
for folder in parent_dirs.split("/"):
if not folder:
continue
current_path += f"{folder}/"
mkdir_url = join_url(webdav_url, current_path)
requests.request(
"MKCOL",
mkdir_url,
auth=HTTPBasicAuth(settings.nextcloud_username, settings.nextcloud_password),
timeout=10,
)
# Upload the file
logger.info(f"[{task_id}] Uploading {filename} to Nextcloud at {full_url}")
log_task_progress(task_id, "upload_file", "in_progress", f"Uploading to {remote_path}", file_id=file_id)
with open(file_path, "rb") as file_data:
response = requests.put(
full_url,
data=file_data,
auth=HTTPBasicAuth(settings.nextcloud_username, settings.nextcloud_password),
headers={"Content-Type": "application/octet-stream"},
timeout=settings.http_request_timeout, # Use configured timeout for large files
)
if response.status_code in (201, 204): # Created or No Content
logger.info(f"[{task_id}] Successfully uploaded {filename} to Nextcloud at {remote_path}")
log_task_progress(
task_id, "upload_to_nextcloud", "success", f"Uploaded to Nextcloud: {remote_path}", file_id=file_id
)
return {
"status": "Completed",
"file_path": file_path,
"nextcloud_path": remote_path,
"response_code": response.status_code,
}
else:
error_msg = f"Failed to upload {filename} to Nextcloud: {response.status_code} - {response.text}"
logger.error(f"[{task_id}] {error_msg}")
log_task_progress(task_id, "upload_to_nextcloud", "failure", error_msg, file_id=file_id)
raise Exception(error_msg)
except Exception as e:
error_msg = f"Failed to upload {filename} to Nextcloud: {str(e)}"
logger.error(f"[{task_id}] {error_msg}")
log_task_progress(task_id, "upload_to_nextcloud", "failure", error_msg, file_id=file_id)
raise Exception(error_msg)
app/utils/network.py
+18 -5
View File
@@ -27,8 +27,21 @@ def is_private_ip(hostname: str) -> bool:
return True
return False
except (socket.gaierror, socket.error):
# Cannot resolve - allow for testing/development
# In production, DNS should work properly
# Log this for debugging
logger.warning(f"Could not resolve hostname: {hostname}")
return False # Changed from True to False to allow external domains in tests
# Cannot resolve.
# Fail securely: block unresolved domains to prevent DNS rebinding
# and SSRF bypasses via unresolvable addresses.
logger.warning(f"Could not resolve hostname (blocking securely): {hostname}")
return True
def join_url(base: str, *parts: str) -> str:
"""
Safely join a base URL and multiple path parts.
Handles double slashes while preserving the protocol '://'.
"""
url = "/".join([base, *parts])
url = url.replace("://", "$PLACEHOLDER$")
while "//" in url:
url = url.replace("//", "/")
url = url.replace("$PLACEHOLDER$", "://")
return url
app/views/base.py
+29 -5
View File
@@ -162,12 +162,36 @@ def _inject_global_context(ctx: dict) -> None:
def template_response_with_version(*args, **kwargs):
"""Wrapper for TemplateResponse to include version and CSRF token in all templates"""
# If context dict is provided, add version to it
if len(args) >= 2 and isinstance(args[1], dict):
_inject_global_context(args[1])
elif "context" in kwargs and isinstance(kwargs["context"], dict):
"""Wrapper for TemplateResponse to include version and CSRF token in all templates.
Handles both old-style and new-style Starlette TemplateResponse calls:
- Old-style (Starlette <1.0): TemplateResponse(name, {"request": req, ...}, ...)
- New-style (Starlette 1.0+): TemplateResponse(request, name, context={...}, ...)
"""
if len(args) >= 1 and isinstance(args[0], str):
# Old-style call: first positional arg is the template name (string).
# Convert to new-style: (request, name, context=..., ...)
name = args[0]
if len(args) >= 2 and isinstance(args[1], dict):
context = args[1]
# Old-style may have status_code as 3rd positional arg
if len(args) >= 3 and "status_code" not in kwargs:
kwargs["status_code"] = args[2]
else:
context = kwargs.pop("context", {})
request_obj = context.pop("request", None)
if request_obj is not None:
context["request"] = request_obj
_inject_global_context(context)
if request_obj is not None:
return original_template_response(request_obj, name, context=context, **kwargs)
return original_template_response(name, context=context, **kwargs)
# New-style call: (request, name, context=..., ...)
if "context" in kwargs and isinstance(kwargs["context"], dict):
_inject_global_context(kwargs["context"])
elif len(args) >= 3 and isinstance(args[2], dict):
_inject_global_context(args[2])
return original_template_response(*args, **kwargs)
app/views/share.py
+2 -1
View File
@@ -23,6 +23,7 @@ templates = Jinja2Templates(directory=str(_templates_dir))
async def shared_link_view(request: Request, token: str):
"""Render the public share landing page for a given token."""
return templates.TemplateResponse(
request,
"shared_link_view.html",
{"request": request, "token": token},
context={"token": token},
)