fix: apply Phase 1-2 correctness and constant extraction from PR #273
- Replace datetime.utcnow() with datetime.now(timezone.utc) in 4 files - Extract duplicate literals to constants in 5 files - models.py: "files.id" → _FILES_ID_FK - upload_to_email.py: "logo.png" → _LOGO_FILENAME - general.py: "%B %d, %Y" → _DATE_DISPLAY_FORMAT - files.py: "File not found" → _FILE_NOT_FOUND - upload_to_google_drive.py: Google token URL → _GOOGLE_TOKEN_URL Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
@@ -6,7 +6,7 @@ that have been stuck in "in_progress" state for too long and mark them as failed
|
||||
"""
|
||||
|
||||
import logging
|
||||
from datetime import datetime
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from app.celery_app import celery
|
||||
from app.database import SessionLocal
|
||||
@@ -37,12 +37,12 @@ def monitor_stalled_steps():
|
||||
|
||||
if stalled_count > 0:
|
||||
logger.warning(
|
||||
f"[{datetime.utcnow().isoformat()}] "
|
||||
f"[{datetime.now(timezone.utc).isoformat()}] "
|
||||
f"Recovered {stalled_count} stalled step(s). "
|
||||
f"Marked as failed due to timeout."
|
||||
)
|
||||
else:
|
||||
logger.debug(f"[{datetime.utcnow().isoformat()}] No stalled steps found.")
|
||||
logger.debug(f"[{datetime.now(timezone.utc).isoformat()}] No stalled steps found.")
|
||||
|
||||
return {"recovered": stalled_count}
|
||||
|
||||
|
||||
@@ -20,6 +20,9 @@ from app.utils import log_task_progress
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Constants
|
||||
_LOGO_FILENAME = "logo.png"
|
||||
|
||||
|
||||
def get_email_template(template_name="default.html"):
|
||||
"""
|
||||
@@ -87,16 +90,16 @@ def attach_logo(msg):
|
||||
"""Attach the DocuElevate logo to the email with proper Content-ID."""
|
||||
try:
|
||||
# Try to find logo in workdir first (for customization)
|
||||
custom_logo_path = os.path.join(settings.workdir, "templates", "email", "logo.png")
|
||||
custom_logo_path = os.path.join(settings.workdir, "templates", "email", _LOGO_FILENAME)
|
||||
if os.path.exists(custom_logo_path):
|
||||
logo_path = custom_logo_path
|
||||
else:
|
||||
# Use built-in logo
|
||||
app_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
logo_path = os.path.join(app_dir, "static", "logo.png")
|
||||
logo_path = os.path.join(app_dir, "static", _LOGO_FILENAME)
|
||||
# Fallback to logo in frontend/static if app/static doesn't exist
|
||||
if not os.path.exists(logo_path):
|
||||
logo_path = os.path.join(app_dir, "..", "frontend", "static", "logo.png")
|
||||
logo_path = os.path.join(app_dir, "..", "frontend", "static", _LOGO_FILENAME)
|
||||
|
||||
if os.path.exists(logo_path):
|
||||
with open(logo_path, "rb") as img:
|
||||
@@ -106,7 +109,7 @@ def attach_logo(msg):
|
||||
mimetype = "image/svg+xml" if logo_path.endswith(".svg") else "image/png"
|
||||
logo_attach = MIMEImage(logo_data, mimetype)
|
||||
logo_attach.add_header("Content-ID", "<logo>")
|
||||
logo_attach.add_header("Content-Disposition", "inline", filename="logo.png")
|
||||
logo_attach.add_header("Content-Disposition", "inline", filename=_LOGO_FILENAME)
|
||||
msg.attach(logo_attach)
|
||||
logger.info(f"Logo attached from {logo_path}")
|
||||
return True
|
||||
|
||||
@@ -20,6 +20,9 @@ from app.utils import log_task_progress
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Google OAuth constants
|
||||
_GOOGLE_TOKEN_URL = "https://oauth2.googleapis.com/token"
|
||||
|
||||
|
||||
def get_drive_service_oauth():
|
||||
"""
|
||||
@@ -40,7 +43,7 @@ def get_drive_service_oauth():
|
||||
credentials = OAuthCredentials(
|
||||
None, # No access token initially, will be refreshed
|
||||
refresh_token=settings.google_drive_refresh_token,
|
||||
token_uri="https://oauth2.googleapis.com/token",
|
||||
token_uri=_GOOGLE_TOKEN_URL,
|
||||
client_id=settings.google_drive_client_id,
|
||||
client_secret=settings.google_drive_client_secret,
|
||||
# Use only drive.file scope
|
||||
|
||||
Reference in New Issue
Block a user