diff --git a/backend/core/celery.py b/backend/core/celery.py index 4af78a7..5f09a95 100644 --- a/backend/core/celery.py +++ b/backend/core/celery.py @@ -1,7 +1,12 @@ # backend/core/celery.py import os +import gc +import logging from celery import Celery +from celery import Task + +logger = logging.getLogger(__name__) # Postavi Django settings modul os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'core.settings') @@ -11,5 +16,42 @@ app = Celery('core') # Koristi konfiguraciju iz settings.py s prefiksom 'CELERY_' app.config_from_object('django.conf:settings', namespace='CELERY') + +def _rss_memory_mb(): + try: + import resource + + usage_kb = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss + # Na Linuxu je ru_maxrss u KiB. + return round(float(usage_kb) / 1024.0, 2) + except (ImportError, AttributeError, OSError, ValueError): + return None + + +class MemoryAwareTask(Task): + abstract = True + + def __call__(self, *args, **kwargs): + gc.collect() + return super().__call__(*args, **kwargs) + + def after_return(self, status, retval, task_id, args, kwargs, einfo): + memory_before_gc = _rss_memory_mb() + gc.collect() + memory_after_gc = _rss_memory_mb() + if memory_before_gc is not None and memory_after_gc is not None: + logger.info( + "Task %s (%s) status=%s RSS prije/poslije GC: %.2fMB -> %.2fMB", + self.name, + task_id, + status, + memory_before_gc, + memory_after_gc, + ) + return super().after_return(status, retval, task_id, args, kwargs, einfo) + + +app.Task = MemoryAwareTask + # Automatski pronalazi taskove u svim 'modules.*' app.autodiscover_tasks() \ No newline at end of file diff --git a/backend/core/settings/base.py b/backend/core/settings/base.py index 8a72846..fb86453 100644 --- a/backend/core/settings/base.py +++ b/backend/core/settings/base.py @@ -154,6 +154,11 @@ REDIS_URL = os.environ.get('REDIS_URL', 'redis://localhost:6379/0') CELERY_BROKER_URL = REDIS_URL CELERY_RESULT_BACKEND = REDIS_URL +CELERY_WORKER_PREFETCH_MULTIPLIER = int(os.environ.get('CELERY_WORKER_PREFETCH_MULTIPLIER', '1')) +CELERY_WORKER_MAX_TASKS_PER_CHILD = int(os.environ.get('CELERY_WORKER_MAX_TASKS_PER_CHILD', '20')) +CELERY_WORKER_MAX_MEMORY_PER_CHILD = int(os.environ.get('CELERY_WORKER_MAX_MEMORY_PER_CHILD', '350000')) +CELERY_TASK_ACKS_LATE = True +CELERY_TASK_REJECT_ON_WORKER_LOST = True # Ako želiš koristiti Redis kao brzi cache sustav unutar Djanga (izvrsno za ERP performanse) # Za ovo ti je potreban paket 'django-redis' u requirements.txt @@ -171,10 +176,18 @@ CACHES = { CELERY_ACCEPT_CONTENT = ['json'] CELERY_TASK_SERIALIZER = 'json' CELERY_BEAT_SCHEDULE = { + 'cleanup-expired-generated-archives-hourly': { + 'task': 'modules.fleet.tasks.cleanup_expired_generated_archives_task', + 'schedule': crontab(minute=5), + }, 'cleanup-expired-generated-pdfs-hourly': { 'task': 'modules.fleet.tasks.cleanup_expired_generated_pdfs_task', 'schedule': crontab(minute=0), }, + 'cleanup-app-tmp-daily': { + 'task': 'modules.fleet.tasks.cleanup_tmp_files_task', + 'schedule': crontab(hour=3, minute=15), + }, 'notify-upcoming-tasks-daily': { 'task': 'modules.task_management.tasks.notify_upcoming_tasks', 'schedule': crontab(hour=8, minute=0), @@ -190,6 +203,13 @@ STATIC_URL = '/static/' # (Opcionalno, ali preporučeno za ERP) Ako koristiš i medije (dokumente) MEDIA_URL = '/media/' MEDIA_ROOT = BASE_DIR / 'media' +APP_TMP_CLEANUP_DIR = os.environ.get('APP_TMP_CLEANUP_DIR', '/tmp') +APP_TMP_CLEANUP_MAX_AGE_HOURS = int(os.environ.get('APP_TMP_CLEANUP_MAX_AGE_HOURS', '24')) +APP_TMP_CLEANUP_PREFIXES = [ + prefix.strip() + for prefix in os.environ.get('APP_TMP_CLEANUP_PREFIXES', 'erp-fleet-archive-').split(',') + if prefix.strip() +] # ============================================================================== diff --git a/backend/modules/fleet/tasks.py b/backend/modules/fleet/tasks.py index 5ee21d8..4d2cf4b 100644 --- a/backend/modules/fleet/tasks.py +++ b/backend/modules/fleet/tasks.py @@ -5,6 +5,7 @@ from django.core.mail import EmailMessage from django.conf import settings from django.utils import timezone from django.core.files.base import ContentFile +from django.core.files import File import logging from io import BytesIO from io import StringIO @@ -13,6 +14,9 @@ import base64 import re import csv import mimetypes +import os +import tempfile +import zipfile from decimal import Decimal, InvalidOperation from smtplib import SMTPSenderRefused from datetime import timedelta @@ -838,10 +842,9 @@ def process_work_order_invoice_ocr(invoice_id): @shared_task def build_monthly_archive_cached_task(generated_archive_id): from .models import GeneratedFleetArchive - from .services import NotificationService from .views import ( - _build_monthly_service_tasks_archive_content, - _build_monthly_work_orders_archive_content, + _write_monthly_service_tasks_archive_entries, + _write_monthly_work_orders_archive_entries, _notify_monthly_archive_request, ) @@ -854,25 +857,39 @@ def build_monthly_archive_cached_task(generated_archive_id): if generated is None: return {"status": "failed", "error": "Generated archive record not found"} + tmp_dir = str(getattr(settings, 'APP_TMP_CLEANUP_DIR', '/tmp')) + with tempfile.NamedTemporaryFile( + mode='w+b', + delete=False, + dir=tmp_dir if os.path.isdir(tmp_dir) else None, + prefix='erp-fleet-archive-', + suffix='.zip', + ) as temp_zip: + temp_zip_path = temp_zip.name try: - if generated.archive_type == 'service_tasks': - archive_content = _build_monthly_service_tasks_archive_content( - user=generated.requested_by, - year=generated.year, - month=generated.month, - ) - else: - archive_content = _build_monthly_work_orders_archive_content( - user=generated.requested_by, - year=generated.year, - month=generated.month, - ) - - if not archive_content: + with zipfile.ZipFile(temp_zip_path, mode='w', compression=zipfile.ZIP_DEFLATED) as archive: + if generated.archive_type == 'service_tasks': + entries_written = _write_monthly_service_tasks_archive_entries( + archive, + user=generated.requested_by, + year=generated.year, + month=generated.month, + ) + else: + entries_written = _write_monthly_work_orders_archive_entries( + archive, + user=generated.requested_by, + year=generated.year, + month=generated.month, + ) + if entries_written == 0: raise ValueError('ZIP arhiva je prazna.') + if os.path.getsize(temp_zip_path) <= 0: + raise ValueError('ZIP arhiva je prazna.') filename = generated.filename or f"{generated.archive_type}-{generated.year}-{generated.month}.zip" - generated.file.save(filename, ContentFile(archive_content), save=False) + with open(temp_zip_path, 'rb') as temp_file: + generated.file.save(filename, File(temp_file), save=False) generated.status = 'ready' generated.error_message = '' generated.save(update_fields=['file', 'status', 'error_message', 'updated_at']) @@ -902,6 +919,12 @@ def build_monthly_archive_cached_task(generated_archive_id): ) logger.exception("Greška kod build_monthly_archive_cached_task: %s", exc) return {"status": "failed", "error": str(exc)} + finally: + try: + if os.path.exists(temp_zip_path): + os.remove(temp_zip_path) + except OSError: + logger.warning("Ne mogu obrisati privremenu ZIP datoteku: %s", temp_zip_path) @shared_task @@ -944,6 +967,46 @@ def cleanup_expired_generated_pdfs_task(): return {"deleted": deleted} +@shared_task +def cleanup_tmp_files_task(): + tmp_dir = str(getattr(settings, 'APP_TMP_CLEANUP_DIR', '/tmp')) + max_age_hours = int(getattr(settings, 'APP_TMP_CLEANUP_MAX_AGE_HOURS', 24)) + prefixes = tuple(getattr(settings, 'APP_TMP_CLEANUP_PREFIXES', ['erp-fleet-archive-'])) + + if max_age_hours <= 0: + raise ValueError('APP_TMP_CLEANUP_MAX_AGE_HOURS mora biti > 0.') + if not prefixes: + raise ValueError('APP_TMP_CLEANUP_PREFIXES ne smije biti prazan.') + if not os.path.isdir(tmp_dir): + return {'deleted': 0, 'tmp_dir': tmp_dir, 'reason': 'tmp-dir-not-found'} + + now = timezone.now().timestamp() + cutoff = now - (max_age_hours * 3600) + deleted = 0 + + for name in os.listdir(tmp_dir): + if not any(name.startswith(prefix) for prefix in prefixes): + continue + full_path = os.path.join(tmp_dir, name) + try: + stat_info = os.stat(full_path) + except FileNotFoundError: + continue + if stat_info.st_mtime > cutoff: + continue + if os.path.isdir(full_path): + try: + os.rmdir(full_path) + except OSError: + logger.warning("Preskačem %s: direktorij nije prazan ili je nedostupan.", full_path) + continue + else: + os.remove(full_path) + deleted += 1 + + return {'deleted': deleted, 'tmp_dir': tmp_dir, 'max_age_hours': max_age_hours} + + @shared_task def build_work_order_pdf_cached_task(generated_pdf_id): from .views import _build_work_order_pdf, _build_work_order_service_records_pdf diff --git a/backend/modules/fleet/views.py b/backend/modules/fleet/views.py index 24e77f7..b13d883 100644 --- a/backend/modules/fleet/views.py +++ b/backend/modules/fleet/views.py @@ -2450,6 +2450,26 @@ def _file_attachment(file_field, fallback_name): return (filename, content, _guess_content_type(filename)) +def _write_file_field_to_zip(archive, *, file_field, entry_name, chunk_size=64 * 1024): + if not file_field: + return False + file_field.open('rb') + try: + first_chunk = file_field.read(chunk_size) + if not first_chunk: + return False + with archive.open(entry_name, mode='w') as destination: + destination.write(first_chunk) + while True: + chunk = file_field.read(chunk_size) + if not chunk: + break + destination.write(chunk) + finally: + file_field.close() + return True + + def _build_image_attachments_for_work_order(work_order): attachments = [] photos = WorkOrderPhoto.objects.filter(is_active=True, work_order=work_order).order_by('created_at') @@ -3008,7 +3028,7 @@ def _parse_year_month_params(request): return year, month -def _build_monthly_service_tasks_archive_content(*, user, year, month): +def _write_monthly_service_tasks_archive_entries(archive, *, user, year, month): from modules.task_management.models import Task tasks = list( @@ -3028,24 +3048,37 @@ def _build_monthly_service_tasks_archive_content(*, user, year, month): raise DRFValidationError({'detail': 'Nema servisnih taskova za odabrani mjesec.'}) used_names = set() + entries_written = 0 + for task in tasks: + work_order = task.work_order + if work_order is None: + continue + docx_bytes = _build_work_order_service_records_docx_bytes(work_order, related_tasks=[task]) + base_name = _service_records_docx_filename(work_order, task) + entry_name = _unique_zip_entry_name(base_name, used_names) + archive.writestr(entry_name, docx_bytes) + entries_written += 1 + if entries_written == 0: + raise DRFValidationError({'detail': 'Nije moguće kreirati ZIP za odabrani mjesec.'}) + return entries_written + + +def _build_monthly_service_tasks_archive_content(*, user, year, month): archive_buffer = BytesIO() with zipfile.ZipFile(archive_buffer, mode='w', compression=zipfile.ZIP_DEFLATED) as archive: - for task in tasks: - work_order = task.work_order - if work_order is None: - continue - docx_bytes = _build_work_order_service_records_docx_bytes(work_order, related_tasks=[task]) - base_name = _service_records_docx_filename(work_order, task) - entry_name = _unique_zip_entry_name(base_name, used_names) - archive.writestr(entry_name, docx_bytes) - + _write_monthly_service_tasks_archive_entries( + archive, + user=user, + year=year, + month=month, + ) archive_content = archive_buffer.getvalue() if not archive_content: raise DRFValidationError({'detail': 'Nije moguće kreirati ZIP za odabrani mjesec.'}) return archive_content -def _build_monthly_work_orders_archive_content(*, user, year, month): +def _write_monthly_work_orders_archive_entries(archive, *, user, year, month): from modules.task_management.models import Task monthly_tasks = ( @@ -3084,23 +3117,34 @@ def _build_monthly_work_orders_archive_content(*, user, year, month): raise DRFValidationError({'detail': 'Nema putnih naloga za odabrani mjesec.'}) used_names = set() + entries_written = 0 + for work_order in work_orders: + pdf_bytes = _build_work_order_pdf(work_order) + work_order_pdf_name = _unique_zip_entry_name(_pdf_filename(work_order, 'work_order'), used_names) + archive.writestr(work_order_pdf_name, pdf_bytes) + entries_written += 1 + + display_code = _work_order_display_code(work_order) + invoices = work_order.invoices.filter(is_active=True).order_by('datum', 'created_at') + for index, invoice in enumerate(invoices, start=1): + file_name = Path(str(getattr(getattr(invoice, 'image', None), 'name', '') or f"invoice-{index}.bin")).name + archive_path = _unique_zip_entry_name(f"Racuni/{display_code}/{file_name}", used_names) + if _write_file_field_to_zip(archive, file_field=invoice.image, entry_name=archive_path): + entries_written += 1 + if entries_written == 0: + raise DRFValidationError({'detail': 'Nije moguće kreirati ZIP za odabrani mjesec.'}) + return entries_written + + +def _build_monthly_work_orders_archive_content(*, user, year, month): archive_buffer = BytesIO() with zipfile.ZipFile(archive_buffer, mode='w', compression=zipfile.ZIP_DEFLATED) as archive: - for work_order in work_orders: - pdf_bytes = _build_work_order_pdf(work_order) - work_order_pdf_name = _unique_zip_entry_name(_pdf_filename(work_order, 'work_order'), used_names) - archive.writestr(work_order_pdf_name, pdf_bytes) - - display_code = _work_order_display_code(work_order) - invoices = work_order.invoices.filter(is_active=True).order_by('datum', 'created_at') - for index, invoice in enumerate(invoices, start=1): - attachment = _file_attachment(invoice.image, fallback_name=f"invoice-{index}.bin") - if not attachment: - continue - invoice_filename, content, _content_type = attachment - archive_path = f"Racuni/{display_code}/{invoice_filename}" - archive.writestr(_unique_zip_entry_name(archive_path, used_names), content) - + _write_monthly_work_orders_archive_entries( + archive, + user=user, + year=year, + month=month, + ) archive_content = archive_buffer.getvalue() if not archive_content: raise DRFValidationError({'detail': 'Nije moguće kreirati ZIP za odabrani mjesec.'}) @@ -3411,12 +3455,18 @@ def generated_archive_download(request, archive_id): if generated_archive is None: raise DRFValidationError({'detail': 'ZIP arhiva nije dostupna ili je istekla.'}) - generated_archive.file.open('rb') + file_name = getattr(generated_archive.file, 'name', '') + storage = getattr(generated_archive.file, 'storage', None) + if not file_name or storage is None: + raise DRFValidationError({'detail': 'ZIP arhiva je prazna ili nedostupna.'}) try: - archive_bytes = generated_archive.file.read() - finally: - generated_archive.file.close() - if not archive_bytes: + if not storage.exists(file_name): + raise DRFValidationError({'detail': 'ZIP arhiva je prazna ili nedostupna.'}) + except (FileNotFoundError, OSError, ValueError): + raise DRFValidationError({'detail': 'ZIP arhiva je prazna ili nedostupna.'}) + try: + generated_archive.file.open('rb') + except (FileNotFoundError, OSError, ValueError): raise DRFValidationError({'detail': 'ZIP arhiva je prazna ili nedostupna.'}) filename = generated_archive.filename or _generated_archive_filename_for_user( request.user, @@ -3424,10 +3474,13 @@ def generated_archive_download(request, archive_id): month=generated_archive.month, archive_type=generated_archive.archive_type, ) - response = HttpResponse(archive_bytes, content_type='application/zip') + response = FileResponse(generated_archive.file, content_type='application/zip') response['Content-Disposition'] = f'attachment; filename="{filename}"' response['Cache-Control'] = 'private, max-age=3600' - response['Content-Length'] = str(len(archive_bytes)) + try: + response['Content-Length'] = str(generated_archive.file.size) + except (OSError, ValueError, TypeError): + pass return response diff --git a/backend/scripts/cleanup_tmp_files.sh b/backend/scripts/cleanup_tmp_files.sh new file mode 100644 index 0000000..e4c7be6 --- /dev/null +++ b/backend/scripts/cleanup_tmp_files.sh @@ -0,0 +1,24 @@ +#!/bin/sh +set -eu + +TMP_DIR="${APP_TMP_CLEANUP_DIR:-/tmp}" +MAX_AGE_DAYS="${APP_TMP_CLEANUP_MAX_AGE_DAYS:-1}" +PREFIXES="${APP_TMP_CLEANUP_PREFIXES:-erp-fleet-archive-}" + +if [ ! -d "$TMP_DIR" ]; then + echo "TMP dir ne postoji: $TMP_DIR" + exit 0 +fi + +OLD_IFS="$IFS" +IFS=',' +for prefix in $PREFIXES; do + prefix_trimmed="$(echo "$prefix" | xargs)" + if [ -z "$prefix_trimmed" ]; then + continue + fi + find "$TMP_DIR" -maxdepth 1 -mindepth 1 -name "${prefix_trimmed}*" -mtime "+${MAX_AGE_DAYS}" -print -delete +done +IFS="$OLD_IFS" + +echo "TMP cleanup dovršen za $TMP_DIR (older than ${MAX_AGE_DAYS} day(s))." diff --git a/docker-compose.prod.yml b/docker-compose.prod.yml index 03e110e..30449f6 100644 --- a/docker-compose.prod.yml +++ b/docker-compose.prod.yml @@ -37,7 +37,13 @@ services: environment: DEBUG: "False" DJANGO_SETTINGS_MODULE: core.settings.production - command: celery -A core worker --loglevel=info + CELERY_WORKER_PREFETCH_MULTIPLIER: "1" + CELERY_WORKER_MAX_TASKS_PER_CHILD: "20" + CELERY_WORKER_MAX_MEMORY_PER_CHILD: "350000" + APP_TMP_CLEANUP_DIR: "/tmp" + APP_TMP_CLEANUP_MAX_AGE_HOURS: "24" + APP_TMP_CLEANUP_PREFIXES: "erp-fleet-archive-" + command: celery -A core worker --loglevel=info --concurrency=${CELERY_WORKER_CONCURRENCY:-2} volumes: - media_volume:/app/media depends_on: