utils and event logging

This commit is contained in:
2025-05-29 15:19:42 +02:00
parent 6d959ac253
commit 5dbdd43785
23 changed files with 657 additions and 101 deletions

Binary file not shown.

Binary file not shown.

View File

@@ -1,6 +1,6 @@
from flask import render_template, Blueprint, redirect, url_for, request, flash, Response
from flask import render_template, Blueprint, redirect, url_for, request, flash, Response, jsonify
from flask_login import current_user, login_required
from models import User, db, Room, RoomFile, RoomMemberPermission, SiteSettings
from models import User, db, Room, RoomFile, RoomMemberPermission, SiteSettings, Event
from routes.auth import require_password_change
import os
from werkzeug.utils import secure_filename
@@ -9,6 +9,7 @@ from datetime import datetime, timedelta
import logging
import sys
import time
from flask import session
# Set up logging to show in console
logging.basicConfig(
@@ -531,3 +532,79 @@ def init_routes(main_bp):
logger.info(f"[Dynamic Colors] Cache version: {site_settings.updated_at.timestamp()}")
return Response(css, mimetype='text/css')
@main_bp.route('/settings/events')
@login_required
def events():
if not current_user.is_admin:
flash('Only administrators can access event logs.', 'error')
return redirect(url_for('main.dashboard'))
# Get filter parameters
event_type = request.args.get('event_type')
date_range = request.args.get('date_range', '7d')
user_id = request.args.get('user_id')
page = request.args.get('page', 1, type=int)
per_page = 50
# Calculate date range
end_date = datetime.utcnow()
if date_range == '24h':
start_date = end_date - timedelta(days=1)
elif date_range == '7d':
start_date = end_date - timedelta(days=7)
elif date_range == '30d':
start_date = end_date - timedelta(days=30)
else:
start_date = None
# Build query
query = Event.query
if event_type:
query = query.filter_by(event_type=event_type)
if start_date:
query = query.filter(Event.timestamp >= start_date)
if user_id:
query = query.filter_by(user_id=user_id)
# Get total count for pagination
total_events = query.count()
total_pages = (total_events + per_page - 1) // per_page
# Get paginated events
events = query.order_by(Event.timestamp.desc()).paginate(page=page, per_page=per_page)
# Get all users for filter dropdown
users = User.query.order_by(User.username).all()
return render_template('settings/tabs/events.html',
events=events.items,
total_pages=total_pages,
current_page=page,
event_type=event_type,
date_range=date_range,
user_id=user_id,
users=users,
csrf_token=session.get('csrf_token'))
@main_bp.route('/api/events/<int:event_id>')
@login_required
def get_event_details(event_id):
if not current_user.is_admin:
return jsonify({'error': 'Unauthorized'}), 403
event = Event.query.get_or_404(event_id)
return jsonify({
'id': event.id,
'event_type': event.event_type,
'user': {
'id': event.user.id,
'username': event.user.username,
'last_name': event.user.last_name
},
'timestamp': event.timestamp.isoformat(),
'details': event.details,
'ip_address': event.ip_address,
'user_agent': event.user_agent
})

95
static/js/events.js Normal file
View File

@@ -0,0 +1,95 @@
document.addEventListener('DOMContentLoaded', function() {
// Initialize variables
let currentPage = 1;
const totalPages = parseInt(document.getElementById('totalPages').textContent);
// Get filter elements
const eventTypeFilter = document.getElementById('eventTypeFilter');
const dateRangeFilter = document.getElementById('dateRangeFilter');
const userFilter = document.getElementById('userFilter');
const applyFiltersBtn = document.getElementById('applyFilters');
// Get pagination elements
const prevPageBtn = document.getElementById('prevPage');
const nextPageBtn = document.getElementById('nextPage');
const currentPageSpan = document.getElementById('currentPage');
// Event details modal
const eventDetailsModal = document.getElementById('eventDetailsModal');
const eventDetailsContent = document.getElementById('eventDetailsContent');
// Function to update URL with current filters
function updateURL() {
const params = new URLSearchParams(window.location.search);
params.set('page', currentPage);
if (eventTypeFilter.value) params.set('event_type', eventTypeFilter.value);
if (dateRangeFilter.value) params.set('date_range', dateRangeFilter.value);
if (userFilter.value) params.set('user_id', userFilter.value);
window.history.replaceState({}, '', `${window.location.pathname}?${params.toString()}`);
}
// Function to load events with current filters
function loadEvents() {
const params = new URLSearchParams();
params.set('page', currentPage);
if (eventTypeFilter.value) params.set('event_type', eventTypeFilter.value);
if (dateRangeFilter.value) params.set('date_range', dateRangeFilter.value);
if (userFilter.value) params.set('user_id', userFilter.value);
window.location.href = `${window.location.pathname}?${params.toString()}`;
}
// Function to load event details
function loadEventDetails(eventId) {
fetch(`/api/events/${eventId}`)
.then(response => response.json())
.then(data => {
const formattedDetails = JSON.stringify(data.details, null, 2);
eventDetailsContent.textContent = formattedDetails;
})
.catch(error => {
console.error('Error loading event details:', error);
eventDetailsContent.textContent = 'Error loading event details';
});
}
// Event listeners for filters
applyFiltersBtn.addEventListener('click', function() {
currentPage = 1;
loadEvents();
});
// Event listeners for pagination
prevPageBtn.addEventListener('click', function() {
if (currentPage > 1) {
currentPage--;
loadEvents();
}
});
nextPageBtn.addEventListener('click', function() {
if (currentPage < totalPages) {
currentPage++;
loadEvents();
}
});
// Event listener for event details modal
eventDetailsModal.addEventListener('show.bs.modal', function(event) {
const button = event.relatedTarget;
const eventId = button.getAttribute('data-event-id');
loadEventDetails(eventId);
});
// Initialize filters from URL parameters
const urlParams = new URLSearchParams(window.location.search);
if (urlParams.has('event_type')) eventTypeFilter.value = urlParams.get('event_type');
if (urlParams.has('date_range')) dateRangeFilter.value = urlParams.get('date_range');
if (urlParams.has('user_id')) userFilter.value = urlParams.get('user_id');
if (urlParams.has('page')) currentPage = parseInt(urlParams.get('page'));
// Update pagination buttons state
prevPageBtn.disabled = currentPage === 1;
nextPageBtn.disabled = currentPage === totalPages;
currentPageSpan.textContent = currentPage;
});

View File

@@ -4,6 +4,7 @@
{% from "settings/tabs/company_info.html" import company_info_tab %}
{% from "settings/tabs/security.html" import security_tab %}
{% from "settings/tabs/debugging.html" import debugging_tab %}
{% from "settings/tabs/events.html" import events_tab %}
{% from "settings/components/reset_colors_modal.html" import reset_colors_modal %}
{% block title %}Settings - DocuPulse{% endblock %}
@@ -41,6 +42,11 @@
<i class="fas fa-shield-alt me-2"></i>Security
</button>
</li>
<li class="nav-item" role="presentation">
<button class="nav-link {% if active_tab == 'events' %}active{% endif %}" id="events-tab" data-bs-toggle="tab" data-bs-target="#events" type="button" role="tab" aria-controls="events" aria-selected="{{ 'true' if active_tab == 'events' else 'false' }}">
<i class="fas fa-history me-2"></i>Event Log
</button>
</li>
<li class="nav-item" role="presentation">
<button class="nav-link {% if active_tab == 'debugging' %}active{% endif %}" id="debugging-tab" data-bs-toggle="tab" data-bs-target="#debugging" type="button" role="tab" aria-controls="debugging" aria-selected="{{ 'true' if active_tab == 'debugging' else 'false' }}">
<i class="fas fa-bug me-2"></i>Debugging
@@ -65,6 +71,11 @@
{{ security_tab() }}
</div>
<!-- Events Tab -->
<div class="tab-pane fade {% if active_tab == 'events' %}show active{% endif %}" id="events" role="tabpanel" aria-labelledby="events-tab">
{{ events_tab(events, csrf_token) }}
</div>
<!-- Debugging Tab -->
<div class="tab-pane fade {% if active_tab == 'debugging' %}show active{% endif %}" id="debugging" role="tabpanel" aria-labelledby="debugging-tab">
{{ debugging_tab() }}

View File

@@ -0,0 +1,165 @@
{% macro events_tab(events, csrf_token) %}
<div class="card shadow-sm">
<div class="card-body">
<div class="d-flex justify-content-between align-items-center mb-4">
<h5 class="card-title mb-0">Event Log</h5>
<div class="d-flex gap-2">
<select id="eventTypeFilter" class="form-select form-select-sm">
<option value="">All Event Types</option>
<option value="user_login">User Login</option>
<option value="user_logout">User Logout</option>
<option value="user_register">User Registration</option>
<option value="user_update">User Update</option>
<option value="file_upload">File Upload</option>
<option value="file_delete">File Delete</option>
<option value="file_download">File Download</option>
<option value="file_restore">File Restore</option>
<option value="file_move">File Move</option>
<option value="file_rename">File Rename</option>
<option value="file_star">File Star</option>
<option value="file_unstar">File Unstar</option>
<option value="file_delete_permanent">File Delete Permanent</option>
<option value="room_create">Room Create</option>
<option value="room_delete">Room Delete</option>
<option value="room_update">Room Update</option>
<option value="room_open">Room Open</option>
<option value="room_list_view">Room List View</option>
<option value="room_search">Room Search</option>
<option value="room_join">Room Join</option>
<option value="room_leave">Room Leave</option>
<option value="room_member_permissions_update">Room Member Permissions Update</option>
<option value="conversation_create">Conversation Create</option>
<option value="conversation_delete">Conversation Delete</option>
<option value="message_sent">Message Sent</option>
<option value="attachment_download">Attachment Download</option>
</select>
<select id="dateRangeFilter" class="form-select form-select-sm">
<option value="24h">Last 24 Hours</option>
<option value="7d">Last 7 Days</option>
<option value="30d">Last 30 Days</option>
<option value="all">All Time</option>
</select>
<select id="userFilter" class="form-select form-select-sm">
<option value="">All Users</option>
</select>
<button id="applyFilters" class="btn btn-primary btn-sm">Apply Filters</button>
</div>
</div>
<div class="table-responsive">
<table class="table table-hover">
<thead>
<tr>
<th>Timestamp</th>
<th>Event Type</th>
<th>User</th>
<th>Details</th>
<th>IP Address</th>
</tr>
</thead>
<tbody id="eventsTableBody">
{% for event in events %}
<tr>
<td>{{ event.timestamp.strftime('%Y-%m-%d %H:%M:%S') }}</td>
<td>
{% if event.event_type == 'user_login' %}
<span class="badge bg-success">User Login</span>
{% elif event.event_type == 'user_logout' %}
<span class="badge bg-secondary">User Logout</span>
{% elif event.event_type == 'user_register' %}
<span class="badge bg-info">User Registration</span>
{% elif event.event_type == 'user_update' %}
<span class="badge bg-primary">User Update</span>
{% elif event.event_type == 'file_upload' %}
<span class="badge bg-success">File Upload</span>
{% elif event.event_type == 'file_delete' %}
<span class="badge bg-danger">File Delete</span>
{% elif event.event_type == 'file_download' %}
<span class="badge bg-info">File Download</span>
{% elif event.event_type == 'file_restore' %}
<span class="badge bg-warning">File Restore</span>
{% elif event.event_type == 'file_move' %}
<span class="badge bg-primary">File Move</span>
{% elif event.event_type == 'file_rename' %}
<span class="badge bg-info">File Rename</span>
{% elif event.event_type == 'file_star' %}
<span class="badge bg-warning">File Star</span>
{% elif event.event_type == 'file_unstar' %}
<span class="badge bg-secondary">File Unstar</span>
{% elif event.event_type == 'file_delete_permanent' %}
<span class="badge bg-danger">File Delete Permanent</span>
{% elif event.event_type == 'room_create' %}
<span class="badge bg-success">Room Create</span>
{% elif event.event_type == 'room_delete' %}
<span class="badge bg-danger">Room Delete</span>
{% elif event.event_type == 'room_update' %}
<span class="badge bg-primary">Room Update</span>
{% elif event.event_type == 'room_open' %}
<span class="badge bg-info">Room Open</span>
{% elif event.event_type == 'room_list_view' %}
<span class="badge bg-secondary">Room List View</span>
{% elif event.event_type == 'room_search' %}
<span class="badge bg-info">Room Search</span>
{% elif event.event_type == 'room_join' %}
<span class="badge bg-info">Room Join</span>
{% elif event.event_type == 'room_leave' %}
<span class="badge bg-secondary">Room Leave</span>
{% elif event.event_type == 'room_member_permissions_update' %}
<span class="badge bg-primary">Room Member Permissions Update</span>
{% elif event.event_type == 'conversation_create' %}
<span class="badge bg-success">Conversation Create</span>
{% elif event.event_type == 'conversation_delete' %}
<span class="badge bg-danger">Conversation Delete</span>
{% elif event.event_type == 'message_sent' %}
<span class="badge bg-primary">Message Sent</span>
{% elif event.event_type == 'attachment_download' %}
<span class="badge bg-info">Attachment Download</span>
{% else %}
<span class="badge bg-secondary">{{ event.event_type }}</span>
{% endif %}
</td>
<td>{{ event.user.username }} {% if event.user.last_name %}{{ event.user.last_name }}{% endif %}</td>
<td>
<button class="btn btn-sm btn-outline-secondary"
data-bs-toggle="modal"
data-bs-target="#eventDetailsModal"
data-event-id="{{ event.id }}">
<i class="fas fa-info-circle"></i> View Details
</button>
</td>
<td>{{ event.ip_address or '-' }}</td>
</tr>
{% endfor %}
</tbody>
</table>
</div>
<div class="d-flex justify-content-between align-items-center mt-3">
<div>
<button id="prevPage" class="btn btn-outline-primary btn-sm">Previous</button>
<span class="mx-2">Page <span id="currentPage">1</span> of <span id="totalPages">1</span></span>
<button id="nextPage" class="btn btn-outline-primary btn-sm">Next</button>
</div>
</div>
</div>
</div>
<!-- Event Details Modal -->
<div class="modal fade" id="eventDetailsModal" tabindex="-1">
<div class="modal-dialog">
<div class="modal-content">
<div class="modal-header">
<h5 class="modal-title">Event Details</h5>
<button type="button" class="btn-close" data-bs-dismiss="modal"></button>
</div>
<div class="modal-body">
<pre id="eventDetailsContent" class="bg-light p-3 rounded"></pre>
</div>
</div>
</div>
</div>
{% endmacro %}
{% block extra_js %}
<script src="{{ url_for('static', filename='js/events.js', v=config.JS_VERSION) }}"></script>
{% endblock %}

21
utils/__init__.py Normal file
View File

@@ -0,0 +1,21 @@
# Utils package initialization
from .permissions import user_has_permission, get_user_permissions
from .event_logger import log_event, get_user_events, get_room_events, get_recent_events, get_events_by_type, get_events_by_date_range
from .path_utils import clean_path, secure_file_path
from .time_utils import timeago, format_datetime, parse_datetime
__all__ = [
'user_has_permission',
'get_user_permissions',
'log_event',
'get_user_events',
'get_room_events',
'get_recent_events',
'get_events_by_type',
'get_events_by_date_range',
'clean_path',
'secure_file_path',
'timeago',
'format_datetime',
'parse_datetime'
]

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

View File

@@ -2,26 +2,28 @@ from flask import request
from models import Event, EventType, db
from typing import Optional, Dict, Any, List
from datetime import datetime
from flask_login import current_user
from sqlalchemy import desc
def log_event(
event_type: EventType,
user_id: int,
details: Optional[Dict[str, Any]] = None
) -> Event:
def log_event(event_type: str, details: Optional[Dict[str, Any]] = None, user_id: Optional[int] = None) -> Event:
"""
Log an event in the system.
Log an event to the database.
Args:
event_type: The type of event from EventType enum
user_id: The ID of the user performing the action
details: Optional dictionary containing additional event-specific data
event_type: The type of event (must match EventType enum)
details: Optional dictionary containing event details
user_id: Optional user ID (defaults to current user)
Returns:
The created Event object
"""
if user_id is None and current_user.is_authenticated:
user_id = current_user.id
event = Event(
event_type=event_type.value,
event_type=event_type,
user_id=user_id,
timestamp=datetime.utcnow(),
details=details or {},
ip_address=request.remote_addr if request else None,
user_agent=request.user_agent.string if request and request.user_agent else None
@@ -31,93 +33,27 @@ def log_event(
db.session.commit()
return event
def get_user_events(
user_id: int,
event_type: Optional[EventType] = None,
start_date: Optional[datetime] = None,
end_date: Optional[datetime] = None,
limit: int = 100
) -> List[Event]:
"""
Retrieve events for a specific user with optional filtering.
def get_user_events(user_id: int, limit: int = 50) -> List[Event]:
"""Get recent events for a specific user"""
return Event.query.filter_by(user_id=user_id).order_by(desc(Event.timestamp)).limit(limit).all()
Args:
user_id: The ID of the user to get events for
event_type: Optional event type to filter by
start_date: Optional start date to filter events
end_date: Optional end date to filter events
limit: Maximum number of events to return
def get_room_events(room_id: int, limit: int = 50) -> List[Event]:
"""Get recent events for a specific room"""
return Event.query.filter(
Event.details['room_id'].astext == str(room_id)
).order_by(desc(Event.timestamp)).limit(limit).all()
Returns:
List of Event objects matching the criteria
"""
query = Event.query.filter_by(user_id=user_id)
def get_recent_events(limit: int = 50) -> List[Event]:
"""Get most recent events across all types"""
return Event.query.order_by(desc(Event.timestamp)).limit(limit).all()
if event_type:
query = query.filter_by(event_type=event_type.value)
if start_date:
query = query.filter(Event.timestamp >= start_date)
if end_date:
query = query.filter(Event.timestamp <= end_date)
def get_events_by_type(event_type: str, limit: int = 50) -> List[Event]:
"""Get recent events of a specific type"""
return Event.query.filter_by(event_type=event_type).order_by(desc(Event.timestamp)).limit(limit).all()
return query.order_by(Event.timestamp.desc()).limit(limit).all()
def get_room_events(
room_id: int,
event_type: Optional[EventType] = None,
start_date: Optional[datetime] = None,
end_date: Optional[datetime] = None,
limit: int = 100
) -> List[Event]:
"""
Retrieve events related to a specific room with optional filtering.
Args:
room_id: The ID of the room to get events for
event_type: Optional event type to filter by
start_date: Optional start date to filter events
end_date: Optional end date to filter events
limit: Maximum number of events to return
Returns:
List of Event objects matching the criteria
"""
query = Event.query.filter(Event.details['room_id'].astext.cast(Integer) == room_id)
if event_type:
query = query.filter_by(event_type=event_type.value)
if start_date:
query = query.filter(Event.timestamp >= start_date)
if end_date:
query = query.filter(Event.timestamp <= end_date)
return query.order_by(Event.timestamp.desc()).limit(limit).all()
def get_recent_events(
event_type: Optional[EventType] = None,
start_date: Optional[datetime] = None,
end_date: Optional[datetime] = None,
limit: int = 100
) -> List[Event]:
"""
Retrieve recent events across the system with optional filtering.
Args:
event_type: Optional event type to filter by
start_date: Optional start date to filter events
end_date: Optional end date to filter events
limit: Maximum number of events to return
Returns:
List of Event objects matching the criteria
"""
query = Event.query
if event_type:
query = query.filter_by(event_type=event_type.value)
if start_date:
query = query.filter(Event.timestamp >= start_date)
if end_date:
query = query.filter(Event.timestamp <= end_date)
return query.order_by(Event.timestamp.desc()).limit(limit).all()
def get_events_by_date_range(start_date: datetime, end_date: datetime, limit: int = 50) -> List[Event]:
"""Get events within a date range"""
return Event.query.filter(
Event.timestamp >= start_date,
Event.timestamp <= end_date
).order_by(desc(Event.timestamp)).limit(limit).all()

59
utils/path_utils.py Normal file
View File

@@ -0,0 +1,59 @@
import os
from typing import Optional
from werkzeug.utils import secure_filename
def clean_path(path: str, base_dir: Optional[str] = None) -> str:
"""
Clean and secure a file path, ensuring it's safe and within allowed directories.
Args:
path: The path to clean
base_dir: Optional base directory to ensure path stays within
Returns:
str: Cleaned and secured path
"""
# Remove any leading/trailing slashes and normalize path separators
path = path.strip('/\\')
path = os.path.normpath(path)
# If base_dir is provided, ensure path stays within it
if base_dir:
base_dir = os.path.normpath(base_dir)
# Get absolute paths
abs_path = os.path.abspath(path)
abs_base = os.path.abspath(base_dir)
# Check if path is within base directory
if not abs_path.startswith(abs_base):
raise ValueError("Path is outside of allowed directory")
# Return path relative to base directory
return os.path.relpath(abs_path, abs_base)
# If no base_dir, just return the cleaned path
return path
def secure_file_path(filename: str) -> str:
"""
Secure a filename using Werkzeug's secure_filename and add a timestamp.
Args:
filename: The original filename
Returns:
str: Secured filename with timestamp
"""
from datetime import datetime
# Get file extension
_, ext = os.path.splitext(filename)
# Secure the base filename
secure_name = secure_filename(filename)
# Add timestamp to prevent filename collisions
timestamp = datetime.utcnow().strftime('%Y%m%d_%H%M%S')
# Combine parts
return f"{os.path.splitext(secure_name)[0]}_{timestamp}{ext}"

83
utils/permissions.py Normal file
View File

@@ -0,0 +1,83 @@
from models import RoomMemberPermission, Room
from flask_login import current_user
from typing import Optional
def user_has_permission(room_id: int, permission_type: str, user_id: Optional[int] = None) -> bool:
"""
Check if a user has a specific permission in a room.
Args:
room_id: The ID of the room to check permissions for
permission_type: The type of permission to check (e.g., 'can_upload', 'can_download')
user_id: Optional user ID (defaults to current user)
Returns:
bool: True if the user has the permission, False otherwise
"""
if user_id is None:
if not current_user.is_authenticated:
return False
user_id = current_user.id
# Admins have all permissions
if current_user.is_authenticated and current_user.is_admin:
return True
# Check room membership and permissions
permission = RoomMemberPermission.query.filter_by(
room_id=room_id,
user_id=user_id
).first()
if not permission:
return False
# Check if the specific permission is granted
return getattr(permission, permission_type, False)
def get_user_permissions(room_id: int, user_id: Optional[int] = None) -> dict:
"""
Get all permissions for a user in a room.
Args:
room_id: The ID of the room to get permissions for
user_id: Optional user ID (defaults to current user)
Returns:
dict: Dictionary containing all permissions for the user
"""
if user_id is None:
if not current_user.is_authenticated:
return {}
user_id = current_user.id
# Admins have all permissions
if current_user.is_authenticated and current_user.is_admin:
return {
'can_upload': True,
'can_download': True,
'can_delete': True,
'can_rename': True,
'can_move': True,
'can_share': True,
'can_manage_members': True
}
# Get user's permissions
permission = RoomMemberPermission.query.filter_by(
room_id=room_id,
user_id=user_id
).first()
if not permission:
return {}
return {
'can_upload': permission.can_upload,
'can_download': permission.can_download,
'can_delete': permission.can_delete,
'can_rename': permission.can_rename,
'can_move': permission.can_move,
'can_share': permission.can_share,
'can_manage_members': permission.can_manage_members
}

109
utils/time_utils.py Normal file
View File

@@ -0,0 +1,109 @@
from datetime import datetime, timedelta
from typing import Union, Optional
def timeago(dt: Union[datetime, str], now: Optional[datetime] = None) -> str:
"""
Convert a datetime to a human-readable relative time string.
Args:
dt: The datetime to convert (can be datetime object or ISO format string)
now: Optional reference time (defaults to current time)
Returns:
str: Human-readable relative time string (e.g., "2 hours ago", "3 days ago")
"""
if isinstance(dt, str):
dt = datetime.fromisoformat(dt.replace('Z', '+00:00'))
if now is None:
now = datetime.utcnow()
diff = now - dt
# Less than a minute
if diff < timedelta(minutes=1):
return "just now"
# Less than an hour
if diff < timedelta(hours=1):
minutes = int(diff.total_seconds() / 60)
return f"{minutes} minute{'s' if minutes != 1 else ''} ago"
# Less than a day
if diff < timedelta(days=1):
hours = int(diff.total_seconds() / 3600)
return f"{hours} hour{'s' if hours != 1 else ''} ago"
# Less than a week
if diff < timedelta(days=7):
days = diff.days
return f"{days} day{'s' if days != 1 else ''} ago"
# Less than a month
if diff < timedelta(days=30):
weeks = int(diff.days / 7)
return f"{weeks} week{'s' if weeks != 1 else ''} ago"
# Less than a year
if diff < timedelta(days=365):
months = int(diff.days / 30)
return f"{months} month{'s' if months != 1 else ''} ago"
# More than a year
years = int(diff.days / 365)
return f"{years} year{'s' if years != 1 else ''} ago"
def format_datetime(dt: Union[datetime, str], format: str = "%Y-%m-%d %H:%M:%S") -> str:
"""
Format a datetime object or ISO string to a specified format.
Args:
dt: The datetime to format (can be datetime object or ISO format string)
format: The format string to use (defaults to "YYYY-MM-DD HH:MM:SS")
Returns:
str: Formatted datetime string
"""
if isinstance(dt, str):
dt = datetime.fromisoformat(dt.replace('Z', '+00:00'))
return dt.strftime(format)
def parse_datetime(dt_str: str) -> datetime:
"""
Parse a datetime string in various formats to a datetime object.
Args:
dt_str: The datetime string to parse
Returns:
datetime: Parsed datetime object
Raises:
ValueError: If the string cannot be parsed
"""
# Try ISO format first
try:
return datetime.fromisoformat(dt_str.replace('Z', '+00:00'))
except ValueError:
pass
# Try common formats
formats = [
"%Y-%m-%d %H:%M:%S",
"%Y-%m-%d %H:%M",
"%Y-%m-%d",
"%d/%m/%Y %H:%M:%S",
"%d/%m/%Y %H:%M",
"%d/%m/%Y",
"%m/%d/%Y %H:%M:%S",
"%m/%d/%Y %H:%M",
"%m/%d/%Y"
]
for fmt in formats:
try:
return datetime.strptime(dt_str, fmt)
except ValueError:
continue
raise ValueError(f"Could not parse datetime string: {dt_str}")