204 lines
6.5 KiB
Python
204 lines
6.5 KiB
Python
"""
|
|
Conversion logging module for tracking video optimization jobs
|
|
Logs conversion details to daily JSON files for audit trail and analytics
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
from datetime import datetime
|
|
from typing import Optional
|
|
|
|
|
|
class ConversionLogger:
|
|
"""Handles logging of video conversion operations to JSON files"""
|
|
|
|
def __init__(self, logs_folder: str):
|
|
"""
|
|
Initialize the conversion logger
|
|
|
|
Args:
|
|
logs_folder: Path to the folder where log files will be stored
|
|
"""
|
|
self.logs_folder = logs_folder
|
|
os.makedirs(logs_folder, exist_ok=True)
|
|
|
|
def _get_log_file_path(self, date: Optional[str] = None) -> str:
|
|
"""
|
|
Get the path to the log file for a specific date
|
|
|
|
Args:
|
|
date: Date string in YYYY-MM-DD format. If None, uses today's date
|
|
|
|
Returns:
|
|
Full path to the log file
|
|
"""
|
|
if date is None:
|
|
date = datetime.now().strftime('%Y-%m-%d')
|
|
|
|
filename = f"{date}_conversions.json"
|
|
return os.path.join(self.logs_folder, filename)
|
|
|
|
def _read_log_file(self, file_path: str) -> list:
|
|
"""
|
|
Read existing log entries from a file
|
|
|
|
Args:
|
|
file_path: Path to the log file
|
|
|
|
Returns:
|
|
List of log entries, or empty list if file doesn't exist
|
|
"""
|
|
if not os.path.exists(file_path):
|
|
return []
|
|
|
|
try:
|
|
with open(file_path, 'r', encoding='utf-8') as f:
|
|
return json.load(f)
|
|
except (json.JSONDecodeError, IOError):
|
|
# If file is corrupted or empty, return empty list
|
|
return []
|
|
|
|
def _write_log_file(self, file_path: str, entries: list):
|
|
"""
|
|
Write log entries to a file
|
|
|
|
Args:
|
|
file_path: Path to the log file
|
|
entries: List of log entries to write
|
|
"""
|
|
with open(file_path, 'w', encoding='utf-8') as f:
|
|
json.dump(entries, f, indent=2, ensure_ascii=False)
|
|
|
|
def log_conversion(
|
|
self,
|
|
user_email: str,
|
|
platform: str,
|
|
aspect_ratio: str,
|
|
input_file_size: int,
|
|
output_file_size: int,
|
|
conversion_duration: float,
|
|
status: str,
|
|
file_id: str,
|
|
error_message: Optional[str] = None
|
|
):
|
|
"""
|
|
Log a video conversion operation
|
|
|
|
Args:
|
|
user_email: Email of the user who performed the conversion
|
|
platform: Target platform (e.g., 'tiktok', 'meta')
|
|
aspect_ratio: Video aspect ratio (e.g., '9:16', '16:9')
|
|
input_file_size: Size of input file in bytes
|
|
output_file_size: Size of output file in bytes
|
|
conversion_duration: Time taken for conversion in seconds
|
|
status: Status of conversion ('success' or 'failure')
|
|
file_id: Unique identifier for the file
|
|
error_message: Error message if conversion failed (optional)
|
|
"""
|
|
# Calculate size reduction percentage
|
|
if input_file_size > 0:
|
|
size_reduction = ((input_file_size - output_file_size) / input_file_size) * 100
|
|
else:
|
|
size_reduction = 0.0
|
|
|
|
# Create log entry
|
|
log_entry = {
|
|
'timestamp': datetime.now().isoformat(),
|
|
'user_email': user_email,
|
|
'platform': platform,
|
|
'aspect_ratio': aspect_ratio,
|
|
'input_file_size': input_file_size,
|
|
'output_file_size': output_file_size,
|
|
'size_reduction_percent': round(size_reduction, 2),
|
|
'conversion_duration_seconds': round(conversion_duration, 2),
|
|
'status': status,
|
|
'file_id': file_id,
|
|
'error_message': error_message
|
|
}
|
|
|
|
# Get log file path for today
|
|
log_file_path = self._get_log_file_path()
|
|
|
|
# Read existing entries
|
|
entries = self._read_log_file(log_file_path)
|
|
|
|
# Append new entry
|
|
entries.append(log_entry)
|
|
|
|
# Write back to file
|
|
self._write_log_file(log_file_path, entries)
|
|
|
|
return log_entry
|
|
|
|
def get_logs(self, date: Optional[str] = None) -> list:
|
|
"""
|
|
Retrieve logs for a specific date
|
|
|
|
Args:
|
|
date: Date string in YYYY-MM-DD format. If None, returns today's logs
|
|
|
|
Returns:
|
|
List of log entries for the specified date
|
|
"""
|
|
log_file_path = self._get_log_file_path(date)
|
|
return self._read_log_file(log_file_path)
|
|
|
|
def get_user_logs(self, user_email: str, date: Optional[str] = None) -> list:
|
|
"""
|
|
Retrieve logs for a specific user on a specific date
|
|
|
|
Args:
|
|
user_email: Email of the user
|
|
date: Date string in YYYY-MM-DD format. If None, uses today's date
|
|
|
|
Returns:
|
|
List of log entries for the specified user and date
|
|
"""
|
|
all_logs = self.get_logs(date)
|
|
return [log for log in all_logs if log.get('user_email') == user_email]
|
|
|
|
def get_statistics(self, date: Optional[str] = None) -> dict:
|
|
"""
|
|
Get statistics for conversions on a specific date
|
|
|
|
Args:
|
|
date: Date string in YYYY-MM-DD format. If None, uses today's date
|
|
|
|
Returns:
|
|
Dictionary containing statistics
|
|
"""
|
|
logs = self.get_logs(date)
|
|
|
|
if not logs:
|
|
return {
|
|
'total_conversions': 0,
|
|
'successful_conversions': 0,
|
|
'failed_conversions': 0,
|
|
'total_input_size': 0,
|
|
'total_output_size': 0,
|
|
'average_size_reduction': 0.0,
|
|
'average_duration': 0.0
|
|
}
|
|
|
|
successful = [log for log in logs if log['status'] == 'success']
|
|
failed = [log for log in logs if log['status'] == 'failure']
|
|
|
|
total_input = sum(log['input_file_size'] for log in logs)
|
|
total_output = sum(log['output_file_size'] for log in logs)
|
|
|
|
if successful:
|
|
avg_reduction = sum(log['size_reduction_percent'] for log in successful) / len(successful)
|
|
avg_duration = sum(log['conversion_duration_seconds'] for log in successful) / len(successful)
|
|
else:
|
|
avg_reduction = 0.0
|
|
avg_duration = 0.0
|
|
|
|
return {
|
|
'total_conversions': len(logs),
|
|
'successful_conversions': len(successful),
|
|
'failed_conversions': len(failed),
|
|
'total_input_size': total_input,
|
|
'total_output_size': total_output,
|
|
'average_size_reduction': round(avg_reduction, 2),
|
|
'average_duration': round(avg_duration, 2)
|
|
}
|