from django.db import models from django.conf import settings from functools import cached_property from pathlib import Path from urllib.parse import quote import mimetypes import uuid import shlex class Torrent(models.Model): id = models.CharField(max_length=40, primary_key=True) name = models.CharField(max_length=255) date_created = models.DateTimeField(auto_now_add=True) date_modified = models.DateTimeField(auto_now=True) user = models.ForeignKey("user.User", on_delete=models.CASCADE, related_name="torrents") shared_users = models.ManyToManyField("user.User", related_name="torrents_shares", blank=True, through="SharedUser") size = models.PositiveBigIntegerField() transmission_data = models.JSONField(default=dict) @cached_property def len_files(self): if hasattr(self, "_len_files"): return self._len_files else: return File.objects.filter(torrent_id=self.id).count() @property async def alen_files(self): if hasattr(self, "_len_files"): return self._len_files else: return await File.objects.filter(torrent_id=self.id).acount() @cached_property def related_users(self): return [ self.user_id, *self.shared_users.values_list("id", flat=True) ] class SharedUser(models.Model): user = models.ForeignKey("user.User", models.CASCADE) torrent = models.ForeignKey("Torrent", models.CASCADE) date_created = models.DateTimeField(auto_now_add=True) class Meta: unique_together = ("user", "torrent") class File(models.Model): id = models.UUIDField(primary_key=True, default=uuid.uuid4) torrent = models.ForeignKey("Torrent", models.CASCADE, related_name="files") rel_name = models.TextField() size = models.BigIntegerField() @property def pathname(self): return Path(self.rel_name) @property def filename(self): return self.pathname.name @property def abs_pathname(self): return settings.DOWNLOAD_BASE_DIR / self.pathname @cached_property def mime_types(self): mime = mimetypes.guess_type(self.pathname) if mime: return mime[0] or mime[1] or "application/octet-stream" else: return "application/octet-stream" @property def is_stream_video(self): video_extensions = ["mp4", "flv", "webm"] return self.pathname.suffix.lower() in video_extensions @property def is_video(self): if self.mime_types.startswith("video/"): return True video_extensions = ['.mp4', '.flv', '.webm', '.avi', '.mkv', '.mov', '.wmv'] return self.pathname.suffix.lower() in video_extensions @property def accel_redirect(self): # Encode chaque partie du chemin séparément pour préserver la structure encoded_parts = [] for part in self.pathname.parts: # Ignorer un slash initial si présent if part == '/' or part == '\\': continue encoded_parts.append(quote(part)) # Construction du chemin final avec le préfixe Nginx if settings.NGINX_ACCEL_BASE.endswith('/'): base = settings.NGINX_ACCEL_BASE.rstrip('/') else: base = settings.NGINX_ACCEL_BASE return f"{base}/{'/'.join(encoded_parts)}" @property def disposition(self): return f'attachment; filename="{quote(self.filename)}"; filename*="{quote(self.filename)}"'