You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
777 lines
30 KiB
777 lines
30 KiB
import os
|
|
import secrets
|
|
import shutil
|
|
import mimetypes
|
|
import re
|
|
from urllib.parse import urlparse, unquote
|
|
|
|
from django.core.files.storage import default_storage
|
|
from django.core.files.base import ContentFile
|
|
|
|
from pathlib import Path
|
|
from django.utils.text import get_valid_filename
|
|
from django.conf import settings
|
|
from django.core.files import File
|
|
from django.http import HttpRequest
|
|
from django.core.mail import send_mail
|
|
|
|
from rest_framework import serializers, status
|
|
from rest_framework.generics import GenericAPIView
|
|
from rest_framework.parsers import MultiPartParser, FormParser
|
|
from rest_framework.response import Response
|
|
from unidecode import unidecode
|
|
from django.utils.text import slugify
|
|
import random
|
|
import string
|
|
import logging
|
|
|
|
from django.conf import settings
|
|
|
|
from django.utils.translation import gettext_lazy as _
|
|
|
|
from cachetools.func import lru_cache
|
|
from django.http import HttpRequest
|
|
from django.contrib import admin
|
|
from utils.image_compression import maybe_compress_uploaded_file
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def absolute_https_url(url, request=None):
|
|
"""Build an absolute URL and always expose it over HTTPS."""
|
|
if request:
|
|
url = request.build_absolute_uri(url)
|
|
|
|
if isinstance(url, str) and url.startswith('http://'):
|
|
return f'https://{url[7:]}'
|
|
|
|
return url
|
|
|
|
# Moved filer imports to avoid circular imports
|
|
# These will be imported when needed in functions
|
|
|
|
@lru_cache
|
|
def qs_thumbs():
|
|
from filer.models import ThumbnailOption
|
|
return ThumbnailOption.objects.all()
|
|
|
|
|
|
def get_thumbs(obj, request: HttpRequest = None) -> dict:
|
|
# print(f'----> {obj}')
|
|
if not obj:
|
|
return {}
|
|
|
|
try:
|
|
# تعریف سه سایز ثابت
|
|
sizes = ['sm', 'md', 'lg']
|
|
thumbnail_object = {}
|
|
|
|
# گرفتن URL اصلی تصویر
|
|
if hasattr(obj, 'url'):
|
|
original_url = obj.url
|
|
else:
|
|
return {}
|
|
|
|
# برای هر سه سایز، همان URL اصلی را برگردان
|
|
for size in sizes:
|
|
thumbnail_object[size] = absolute_https_url(original_url, request)
|
|
|
|
return thumbnail_object
|
|
|
|
except Exception as p:
|
|
print(p)
|
|
return {}
|
|
|
|
|
|
def environment_callback(request):
|
|
if settings.DEBUG:
|
|
return [_("Development"), "primary"]
|
|
|
|
return [_("Production"), "primary"]
|
|
|
|
|
|
def send_email(recipient, code):
|
|
import requests
|
|
from django.conf import settings
|
|
|
|
api_key = getattr(settings, "RESEND_API_KEY", None)
|
|
if not api_key:
|
|
logger.warning("RESEND_API_KEY is not set in settings.")
|
|
return False
|
|
|
|
url = "https://api.resend.com/emails"
|
|
headers = {
|
|
"Authorization": f"Bearer {api_key}",
|
|
"Content-Type": "application/json",
|
|
}
|
|
|
|
# Normalize recipient into a clean list of email strings
|
|
if isinstance(recipient, str):
|
|
to_emails = [recipient.strip()]
|
|
elif isinstance(recipient, (list, tuple, set)):
|
|
to_emails = [str(r).strip() for r in recipient if r and str(r).strip()]
|
|
else:
|
|
to_emails = [str(recipient).strip()]
|
|
|
|
if not to_emails:
|
|
logger.error(f"Cannot send email: recipient list is empty ({recipient})")
|
|
return False
|
|
|
|
from_email = getattr(settings, "RESEND_FROM_EMAIL", "Dovodi <[email protected]>")
|
|
if "<" not in from_email and "@" in from_email:
|
|
from_email = f"Dovodi <{from_email}>"
|
|
|
|
site_domain = getattr(settings, "SITE_DOMAIN", "https://dovodi.newhorizonco.uk/").rstrip('/')
|
|
subject = "Код подтверждения | Verification Code"
|
|
|
|
html_content = f"""
|
|
<!DOCTYPE html>
|
|
<html lang="ru">
|
|
<head>
|
|
<meta charset="UTF-8">
|
|
<meta name="viewport" content="width=device-width, initial-scale=1.0">
|
|
<title>Код подтверждения | Verification Code</title>
|
|
</head>
|
|
<body style="margin: 0; padding: 0; background-color: #F4F6F9; font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, Helvetica, Arial, sans-serif; color: #15171C; -webkit-font-smoothing: antialiased;">
|
|
<table border="0" cellpadding="0" cellspacing="0" width="100%" style="table-layout: fixed; background-color: #F4F6F9; padding: 40px 16px;">
|
|
<tr>
|
|
<td align="center" valign="top">
|
|
<table border="0" cellpadding="0" cellspacing="0" width="100%" style="max-width: 540px; background-color: #FFFFFF; border-radius: 16px; border: 1px solid #E2E8F0; box-shadow: 0 4px 20px -2px rgba(21, 23, 28, 0.06); overflow: hidden;">
|
|
<!-- Top Gradient Accent Line -->
|
|
<tr>
|
|
<td style="height: 4px; background: linear-gradient(90deg, #5172E1 0%, #1C458C 100%); line-height: 4px; font-size: 4px;"> </td>
|
|
</tr>
|
|
<!-- Header with Logo -->
|
|
<tr>
|
|
<td style="padding: 32px 32px 20px 32px; text-align: center; background-color: #FAFAFC; border-bottom: 1px solid #F1F3F7;">
|
|
<img src="{site_domain}/static/images/dovoodi_logo.svg" alt="Dovodi" style="height: 38px; width: auto; max-width: 160px; display: inline-block;" />
|
|
</td>
|
|
</tr>
|
|
<!-- Main Body Section -->
|
|
<tr>
|
|
<td style="padding: 36px 36px 28px 36px;">
|
|
<!-- Security Badge -->
|
|
<div style="text-align: center; margin-bottom: 16px;">
|
|
<span style="display: inline-block; background-color: #EEF2FF; color: #3B66DE; border: 1px solid #D5DEFF; padding: 4px 12px; border-radius: 9999px; font-size: 11px; font-weight: 700; text-transform: uppercase; letter-spacing: 0.5px;">
|
|
Безопасность / Security
|
|
</span>
|
|
</div>
|
|
|
|
<!-- Main Title -->
|
|
<h1 style="margin: 0 0 12px 0; font-size: 22px; font-weight: 800; color: #15171C; text-align: center; line-height: 1.3;">
|
|
Код подтверждения
|
|
</h1>
|
|
|
|
<!-- Intro Message -->
|
|
<p style="margin: 0 0 24px 0; font-size: 14px; color: #646A75; text-align: center; line-height: 1.6;">
|
|
Используйте следующий одноразовый код для входа یا подтверждения учетной записи в <strong>Dovodi</strong>:
|
|
</p>
|
|
|
|
<!-- OTP Box -->
|
|
<div style="text-align: center; margin: 28px 0; padding: 22px; background-color: #F8FAFD; border: 1.5px dashed #5172E1; border-radius: 14px;">
|
|
<span style="font-family: 'SF Pro Mono', Monaco, Consolas, 'Liberation Mono', 'Courier New', monospace; font-size: 38px; font-weight: 800; color: #15171C; letter-spacing: 10px; display: inline-block;">
|
|
{code}
|
|
</span>
|
|
</div>
|
|
|
|
<!-- Expiry & Security Notice -->
|
|
<div style="background-color: #FFFBEB; border: 1px solid #FDE68A; border-radius: 10px; padding: 12px 16px; margin-bottom: 24px;">
|
|
<table border="0" cellpadding="0" cellspacing="0" width="100%">
|
|
<tr>
|
|
<td style="font-size: 12px; color: #92400E; line-height: 1.5;">
|
|
⏱ <strong>Внимание:</strong> Срок действия кода истекает через 5 минут. Если вы не запрашивали этот код, просто проигнорируйте это письмо.
|
|
</td>
|
|
</tr>
|
|
</table>
|
|
</div>
|
|
|
|
<p style="margin: 0; font-size: 13px; color: #8C93A0; text-align: center; line-height: 1.5;">
|
|
Никому не передавайте этот код в целях безопасности вашего аккаунта.
|
|
</p>
|
|
</td>
|
|
</tr>
|
|
<!-- Footer Section -->
|
|
<tr>
|
|
<td style="padding: 24px 36px; background-color: #F8F9FB; border-top: 1px solid #EEF0F4; text-align: center;">
|
|
<p style="margin: 0 0 6px 0; font-size: 13px; font-weight: 700; color: #15171C;">
|
|
Dovodi
|
|
</p>
|
|
<p style="margin: 0 0 12px 0; font-size: 12px; color: #646A75;">
|
|
<a href="https://dovodi.newhorizonco.uk/" style="color: #5172E1; text-decoration: none; font-weight: 600;">dovodi.newhorizonco.uk</a>
|
|
•
|
|
<a href="mailto:[email protected]" style="color: #5172E1; text-decoration: none; font-weight: 600;">[email protected]</a>
|
|
</p>
|
|
<p style="margin: 0; font-size: 11px; color: #9AA0AC; line-height: 1.4;">
|
|
© 2026 Dovodi. Все права защищены.
|
|
</p>
|
|
</td>
|
|
</tr>
|
|
</table>
|
|
</td>
|
|
</tr>
|
|
</table>
|
|
</body>
|
|
</html>
|
|
"""
|
|
|
|
text_content = (
|
|
f"Код подтверждения Dovodi: {code}\n\n"
|
|
f"Your Dovodi verification code is: {code}\n\n"
|
|
f"Срок действия кода истекает через 5 минут. Если вы не запрашивали этот код, проигнорируйте это письмо.\n"
|
|
f"Никому не передавайте этот код в целях безопасности вашего аккаунта."
|
|
)
|
|
|
|
payload = {
|
|
"from": from_email,
|
|
"to": to_emails,
|
|
"subject": subject,
|
|
"html": html_content,
|
|
"text": text_content,
|
|
}
|
|
|
|
try:
|
|
response = requests.post(url, headers=headers, json=payload, timeout=15)
|
|
if response.status_code >= 400:
|
|
logger.error(f"Failed to send email via Resend: HTTP {response.status_code} - {response.text}")
|
|
print(f"Failed to send email via Resend: HTTP {response.status_code} - {response.text}")
|
|
return False
|
|
logger.info(f"Email OTP sent successfully via Resend to {to_emails}")
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"Failed to send email via Resend: {str(e)}", exc_info=True)
|
|
print(f"Failed to send email via Resend: {str(e)}")
|
|
return False
|
|
|
|
|
|
def is_valid_email(email):
|
|
# تعریف الگوی regex برای یک ایمیل معتبر
|
|
email_regex = r'^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$'
|
|
|
|
# بررسی اینکه آیا ایمیل با regex مطابقت دارد یا خیر
|
|
if re.match(email_regex, email):
|
|
return True
|
|
return False
|
|
|
|
|
|
def generate_slug_for_model(model, value: str, recycled_count: int = 0):
|
|
from slugify import slugify
|
|
try:
|
|
|
|
base_slug = slugify(unidecode(value))
|
|
slug = base_slug
|
|
if recycled_count > 0:
|
|
slug = f"{base_slug}-{recycled_count}"
|
|
|
|
if model.objects.filter(slug=slug).exists():
|
|
return generate_slug_for_model(model, value, recycled_count + 1)
|
|
|
|
return slug[:50]
|
|
except Exception as exp:
|
|
letters = string.ascii_lowercase
|
|
result_str = ''.join(random.choice(letters) for i in range(8))
|
|
return result_str
|
|
|
|
|
|
def generate_slugen_for_model(model, value_en, value_pk):
|
|
try:
|
|
unique_slug = value_en
|
|
if not value_pk or not value_en:
|
|
base_slug = slugify(unidecode(value_en))
|
|
unique_slug = base_slug
|
|
num = 1
|
|
while model.objects.filter(slug=unique_slug).exists():
|
|
unique_slug = f"{base_slug}-{num}"
|
|
num += 1
|
|
|
|
return unique_slug
|
|
except Exception as exp:
|
|
letters = string.ascii_lowercase
|
|
result_str = ''.join(random.choice(letters) for i in range(8))
|
|
return result_str
|
|
|
|
|
|
def exclude_host_from_url(url):
|
|
# Parse the URL
|
|
parsed_url = urlparse(url)
|
|
|
|
# Extract the path and query parameters
|
|
path_with_query = parsed_url.path + parsed_url.query
|
|
|
|
return path_with_query
|
|
|
|
|
|
def generate_slug_for_model(model, value: str, recycled_count: int = 0):
|
|
from slugify import slugify
|
|
|
|
slug = slugify(value)
|
|
if model.objects.filter(slug=slug).exists():
|
|
recycled_count += 1
|
|
if value.endswith(f'-{recycled_count - 1}'):
|
|
value = value.replace(f'-{recycled_count - 1}', f'-{recycled_count}')
|
|
else:
|
|
value = f"{value}-{recycled_count}"
|
|
return generate_slug_for_model(model, value, recycled_count)
|
|
|
|
return slug[:50]
|
|
|
|
|
|
def generate_language_slugs(translations):
|
|
"""
|
|
Build a list of {language_code, title} where title is a slugified string
|
|
from provided multilingual translations list.
|
|
Expected input shape:
|
|
- list[dict]: [{'language_code': 'fa', 'title': 'متن'}, ...]
|
|
Fallback keys supported: code/lang/language for language, value/text for content.
|
|
"""
|
|
try:
|
|
result = []
|
|
if isinstance(translations, list):
|
|
for tr in translations:
|
|
if isinstance(tr, dict):
|
|
language_code = tr.get('language_code') or tr.get('code') or tr.get('lang') or tr.get('language')
|
|
text = tr.get('title') or tr.get('text') or tr.get('value')
|
|
if language_code and text:
|
|
slug_text = slugify(text, allow_unicode=True)
|
|
result.append({'language_code': str(language_code), 'title': slug_text})
|
|
return result
|
|
except Exception as e:
|
|
print(f"Error generating slugs: {e}")
|
|
return []
|
|
|
|
def absolute_url(req, url):
|
|
"""
|
|
can either be a file instance or a URL string
|
|
"""
|
|
try:
|
|
return req.build_absolute_uri(url.url if hasattr(url, 'url') else url)
|
|
except Exception:
|
|
return None
|
|
|
|
def sizeof_fmt(num, suffix="B"):
|
|
for unit in ["", "K", "M", "G"]:
|
|
if abs(num) < 1024.0:
|
|
return f"{num:3.1f} {unit}{suffix}"
|
|
num /= 1024.0
|
|
return f"{num:.1f} Yi{suffix}"
|
|
|
|
def file_location_media(path: str):
|
|
"""
|
|
Resolve a media URL/relative path to absolute filesystem path under MEDIA_ROOT.
|
|
"""
|
|
from django.conf import settings
|
|
|
|
media_url = (getattr(settings, "MEDIA_URL", "/media/") or "/media/").rstrip("/")
|
|
media_root = settings.MEDIA_ROOT
|
|
|
|
if path.startswith("http"):
|
|
path = exclude_host_from_url(path)
|
|
|
|
path = unquote(path)
|
|
|
|
if path.startswith(media_url + "/"):
|
|
path = path[len(media_url):]
|
|
|
|
if path.startswith("/"):
|
|
path = path[1:]
|
|
|
|
return os.path.join(media_root, path)
|
|
|
|
|
|
def file_location(path):
|
|
from django.conf import settings
|
|
import os
|
|
|
|
if path.startswith('http'):
|
|
path = exclude_host_from_url(path)
|
|
|
|
if path.startswith("/static"):
|
|
path = path[7:]
|
|
|
|
if path.startswith('/'):
|
|
path = path[1:]
|
|
|
|
return os.path.join(settings.STATIC_ROOT, path)
|
|
|
|
|
|
def guess_file_type(filename):
|
|
try:
|
|
mimetype = mimetypes.guess_type(filename)[0].split('/')[0]
|
|
return mimetype
|
|
|
|
except Exception:
|
|
return False
|
|
|
|
class FileFieldSerializer(serializers.CharField):
|
|
"""
|
|
a field to handle uploaded file
|
|
"""
|
|
|
|
def get_rpath(self, p):
|
|
if not p:
|
|
return p
|
|
if not isinstance(p, str):
|
|
return p
|
|
|
|
p = unquote(p)
|
|
|
|
# If /media/ is in the path, extract the relative path after the last /media/
|
|
if "/media/" in p:
|
|
idx = p.rfind("/media/")
|
|
return p[idx + 7:].lstrip('/')
|
|
|
|
# If /static/ is in the path, extract the relative path after the last /static/
|
|
if "/static/" in p:
|
|
idx = p.rfind("/static/")
|
|
return p[idx + 8:].lstrip('/')
|
|
|
|
# If it's a URL but doesn't contain /media/ or /static/
|
|
if "://" in p:
|
|
try:
|
|
parsed = urlparse(p)
|
|
path = parsed.path
|
|
if path.startswith('/media/'):
|
|
return path[7:].lstrip('/')
|
|
elif path.startswith('/static/'):
|
|
return path[8:].lstrip('/')
|
|
return path.lstrip('/')
|
|
except Exception:
|
|
pass
|
|
|
|
return p.lstrip('/')
|
|
|
|
def to_representation(self, value):
|
|
request = self.context.get('request', None)
|
|
if value:
|
|
if isinstance(value, str):
|
|
# If value is a string, assume it's a file path
|
|
return value
|
|
elif hasattr(value, 'url'):
|
|
if 'http://' in str(value) or 'https://' in str(value):
|
|
return str(value)
|
|
|
|
return absolute_url(request, value.url) if request else value.url
|
|
return None
|
|
|
|
def to_internal_value(self, data):
|
|
if not data:
|
|
return None
|
|
|
|
if "/tmp/" not in data:
|
|
# value not changed and here we simply return old file path
|
|
return self.get_rpath(data)
|
|
|
|
# if data.startswith('http'):
|
|
# data = self.get_rpath(data)
|
|
|
|
fpath = file_location_media(data)
|
|
if not os.path.exists(fpath):
|
|
raise serializers.ValidationError(f"File: '{fpath}' Does not exist")
|
|
|
|
rel = os.path.basename(unquote(data))
|
|
return File(open(fpath, "rb"), rel)
|
|
# return File(open(fpath, 'rb'), os.path.basename(data))
|
|
|
|
|
|
class UploadChatMediaSerializer(serializers.Serializer):
|
|
"""Upload files permanently to /media/chat/"""
|
|
file = serializers.FileField()
|
|
url = serializers.URLField(read_only=True)
|
|
name = serializers.CharField(read_only=True)
|
|
size = serializers.CharField(read_only=True)
|
|
mime_type = serializers.CharField(read_only=True)
|
|
thumbnail_url = serializers.URLField(read_only=True, required=False)
|
|
|
|
def to_representation(self, instance):
|
|
data = super(UploadChatMediaSerializer, self).to_representation(instance)
|
|
data['file'] = instance['file']
|
|
return data
|
|
|
|
# def store_file(self, file):
|
|
# from django.conf import settings
|
|
# from utils.image_utils import (
|
|
# create_thumbnail,
|
|
# is_image_file,
|
|
# is_video_file,
|
|
# extract_video_thumbnail
|
|
# )
|
|
# media_path = settings.MEDIA_ROOT
|
|
|
|
# os.makedirs(f'{media_path}/chat/uploads', exist_ok=True)
|
|
# fpath = f"/chat/uploads/{secrets.token_urlsafe(4)}-{file.name}"
|
|
# full_path = media_path + fpath
|
|
# if hasattr(file, 'read'):
|
|
# default_storage.save(str(full_path), ContentFile(file.read()))
|
|
# else:
|
|
# default_storage.save(str(full_path), file)
|
|
# os.chmod(full_path, 0o644)
|
|
|
|
# result = {
|
|
# 'file': fpath,
|
|
# 'url': absolute_url(self.context['request'], f"/media{fpath}"),
|
|
# 'name': file.name,
|
|
# 'size': sizeof_fmt(file.size),
|
|
# 'mime_type': guess_file_type(fpath)
|
|
# }
|
|
|
|
# # Generate thumbnail if file is an image (low quality for preview)
|
|
# if is_image_file(full_path):
|
|
# try:
|
|
# thumbnail_path = create_thumbnail(full_path, size=(200, 200), quality=60)
|
|
# thumbnail_relative = thumbnail_path.replace(media_path, '')
|
|
# result['thumbnail_url'] = absolute_url(
|
|
# self.context['request'],
|
|
# f"/media{thumbnail_relative}"
|
|
# )
|
|
# except Exception as e:
|
|
# print(f"Failed to generate image thumbnail: {e}")
|
|
# result['thumbnail_url'] = None
|
|
# # Generate thumbnail if file is a video (low quality for preview)
|
|
# elif is_video_file(full_path):
|
|
# try:
|
|
# thumbnail_path = extract_video_thumbnail(
|
|
# full_path,
|
|
# time_offset='00:00:01',
|
|
# size=(200, 200),
|
|
# quality=60
|
|
# )
|
|
# thumbnail_relative = thumbnail_path.replace(media_path, '')
|
|
# result['thumbnail_url'] = absolute_url(
|
|
# self.context['request'],
|
|
# f"/media{thumbnail_relative}"
|
|
# )
|
|
# except Exception as e:
|
|
# print(f"Failed to generate video thumbnail: {e}")
|
|
# result['thumbnail_url'] = None
|
|
# else:
|
|
# result['thumbnail_url'] = None
|
|
|
|
# return result
|
|
def store_file(self, file):
|
|
from django.conf import settings
|
|
from utils.image_utils import (
|
|
create_thumbnail,
|
|
is_image_file,
|
|
is_video_file,
|
|
extract_video_thumbnail,
|
|
)
|
|
|
|
media_root = Path(settings.MEDIA_ROOT) # Path object
|
|
chat_dir = media_root / "chat" / "uploads"
|
|
chat_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
safe_name = get_valid_filename(os.path.basename(file.name))
|
|
rel_path = Path("chat") / "uploads" / f"{secrets.token_urlsafe(4)}-{safe_name}"
|
|
full_path = media_root / rel_path # Path object
|
|
|
|
# Save via default_storage (path must be relative to MEDIA_ROOT)
|
|
compressed_bytes = maybe_compress_uploaded_file(file)
|
|
if compressed_bytes is not None:
|
|
print(
|
|
f"[upload-chat-media] saving file={safe_name} "
|
|
f"original={getattr(file, 'size', 'unknown')} final={len(compressed_bytes)}"
|
|
)
|
|
default_storage.save(str(rel_path), ContentFile(compressed_bytes))
|
|
elif hasattr(file, "read"):
|
|
print(f"[upload-chat-media] saving original file={safe_name}")
|
|
default_storage.save(str(rel_path), ContentFile(file.read()))
|
|
else:
|
|
print(f"[upload-chat-media] saving original stream file={safe_name}")
|
|
default_storage.save(str(rel_path), file)
|
|
|
|
# Optional: chmod only if local filesystem and OS supports it
|
|
try:
|
|
os.chmod(full_path, 0o644)
|
|
except PermissionError:
|
|
pass
|
|
|
|
result = {
|
|
"file": f"/{rel_path.as_posix()}",
|
|
"url": absolute_url(
|
|
self.context["request"],
|
|
f"{settings.MEDIA_URL.rstrip('/')}/{rel_path.as_posix()}",
|
|
),
|
|
"name": safe_name,
|
|
"size": sizeof_fmt(len(compressed_bytes) if compressed_bytes is not None else file.size),
|
|
"mime_type": guess_file_type(rel_path.name),
|
|
}
|
|
|
|
# For images
|
|
if is_image_file(str(full_path)):
|
|
try:
|
|
thumbnail_path = create_thumbnail(str(full_path), size=(200, 200), quality=60)
|
|
thumb_rel = str(thumbnail_path).replace(str(media_root), "").lstrip("\\/")
|
|
result["thumbnail_url"] = absolute_url(
|
|
self.context["request"],
|
|
f"{settings.MEDIA_URL.rstrip('/')}/{thumb_rel.replace(os.sep, '/')}",
|
|
)
|
|
except Exception as e:
|
|
print(f"Failed to generate image thumbnail: {e}")
|
|
result["thumbnail_url"] = None
|
|
# For videos
|
|
elif is_video_file(str(full_path)):
|
|
try:
|
|
thumbnail_path = extract_video_thumbnail(
|
|
str(full_path),
|
|
time_offset="00:00:01",
|
|
size=(200, 200),
|
|
quality=60,
|
|
)
|
|
thumb_rel = str(thumbnail_path).replace(str(media_root), "").lstrip("\\/")
|
|
result["thumbnail_url"] = absolute_url(
|
|
self.context["request"],
|
|
f"{settings.MEDIA_URL.rstrip('/')}/{thumb_rel.replace(os.sep, '/')}",
|
|
)
|
|
except Exception as e:
|
|
print(f"Skipping video thumbnail generation: {e}")
|
|
result["thumbnail_url"] = None
|
|
else:
|
|
result["thumbnail_url"] = None
|
|
return result
|
|
|
|
def validate(self, attrs):
|
|
file_details = self.store_file(attrs['file'])
|
|
return file_details
|
|
|
|
|
|
class UploadTmpSerializer(serializers.Serializer):
|
|
file = serializers.FileField()
|
|
url = serializers.URLField(read_only=True)
|
|
name = serializers.CharField(read_only=True)
|
|
size = serializers.CharField(read_only=True)
|
|
mime_type = serializers.CharField(read_only=True)
|
|
thumbnail_url = serializers.URLField(read_only=True, required=False)
|
|
|
|
def to_representation(self, instance):
|
|
data = super(UploadTmpSerializer, self).to_representation(instance)
|
|
data['file'] = instance['file']
|
|
return data
|
|
|
|
def store_file(self, file):
|
|
from django.conf import settings
|
|
from utils.image_utils import (
|
|
create_thumbnail,
|
|
is_image_file,
|
|
is_video_file,
|
|
extract_video_thumbnail,
|
|
)
|
|
|
|
media_root = Path(settings.MEDIA_ROOT)
|
|
tmp_dir = media_root / "tmp"
|
|
tmp_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
safe_name = get_valid_filename(os.path.basename(file.name))
|
|
rel_path = Path("tmp") / f"{secrets.token_urlsafe(4)}-{safe_name}"
|
|
|
|
# Save the file using Django storage (safe on Windows)
|
|
# If the file is an InMemoryFile or TemporaryUploadedFile, read its content
|
|
compressed_bytes = maybe_compress_uploaded_file(file)
|
|
if compressed_bytes is not None:
|
|
print(
|
|
f"[upload-tmp-media] saving file={safe_name} "
|
|
f"original={getattr(file, 'size', 'unknown')} final={len(compressed_bytes)}"
|
|
)
|
|
default_storage.save(str(rel_path), ContentFile(compressed_bytes))
|
|
elif hasattr(file, 'read'):
|
|
print(f"[upload-tmp-media] saving original file={safe_name}")
|
|
default_storage.save(str(rel_path), ContentFile(file.read()))
|
|
else:
|
|
print(f"[upload-tmp-media] saving original stream file={safe_name}")
|
|
default_storage.save(str(rel_path), file)
|
|
|
|
full_path = str(media_root / rel_path)
|
|
|
|
result = {
|
|
"file": f"/{rel_path.as_posix()}",
|
|
"url": absolute_url(
|
|
self.context["request"],
|
|
f"{settings.MEDIA_URL.rstrip('/')}/{rel_path.as_posix()}",
|
|
),
|
|
"name": safe_name,
|
|
"size": sizeof_fmt(len(compressed_bytes) if compressed_bytes is not None else file.size),
|
|
"mime_type": guess_file_type(rel_path.name),
|
|
}
|
|
|
|
# Generate thumbnail if image
|
|
if is_image_file(full_path):
|
|
try:
|
|
thumb_path = create_thumbnail(full_path, size=(200, 200), quality=60)
|
|
thumb_rel = thumb_path.replace(str(media_root), "").lstrip("\\/")
|
|
result["thumbnail_url"] = absolute_url(
|
|
self.context["request"],
|
|
f"{settings.MEDIA_URL.rstrip('/')}/{thumb_rel.replace(os.sep, '/')}",
|
|
)
|
|
except Exception as e:
|
|
print("Failed to generate image thumbnail:", e)
|
|
result["thumbnail_url"] = None
|
|
# Generate thumbnail if video
|
|
elif is_video_file(full_path):
|
|
try:
|
|
thumb_path = extract_video_thumbnail(
|
|
full_path,
|
|
time_offset="00:00:01",
|
|
size=(200, 200),
|
|
quality=60,
|
|
)
|
|
thumb_rel = thumb_path.replace(str(media_root), "").lstrip("\\/")
|
|
result["thumbnail_url"] = absolute_url(
|
|
self.context["request"],
|
|
f"{settings.MEDIA_URL.rstrip('/')}/{thumb_rel.replace(os.sep, '/')}",
|
|
)
|
|
except Exception as e:
|
|
print("Skipping video thumbnail generation:", e)
|
|
result["thumbnail_url"] = None
|
|
else:
|
|
result["thumbnail_url"] = None
|
|
|
|
return result
|
|
|
|
def validate(self, attrs):
|
|
file_details = self.store_file(attrs['file'])
|
|
return file_details
|
|
|
|
|
|
class UploadChatMedia(GenericAPIView):
|
|
"""
|
|
Upload files permanently to /media/chat/
|
|
Files are stored permanently and will NOT be removed
|
|
"""
|
|
parser_classes = (FormParser, MultiPartParser)
|
|
serializer_class = UploadChatMediaSerializer
|
|
|
|
def post(self, request: HttpRequest, *args, **kwargs):
|
|
serializer = UploadChatMediaSerializer(data=request.FILES, context={'request': request})
|
|
is_valid = serializer.is_valid(raise_exception=True)
|
|
if not is_valid:
|
|
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
|
|
|
|
return Response(serializer.data)
|
|
|
|
|
|
class UploadTmpMedia(GenericAPIView):
|
|
"""
|
|
Upload files temporarily to /static/tmp/
|
|
Files will be removed every 1 hour
|
|
"""
|
|
parser_classes = (FormParser, MultiPartParser)
|
|
serializer_class = UploadTmpSerializer
|
|
|
|
def post(self, request: HttpRequest, *args, **kwargs):
|
|
serializer = UploadTmpSerializer(data=request.FILES, context={'request': request})
|
|
is_valid = serializer.is_valid(raise_exception=True)
|
|
if not is_valid:
|
|
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
|
|
|
|
return Response(serializer.data)
|
|
|
|
# Configure filer admin after Django is fully loaded
|
|
def configure_filer_admin():
|
|
try:
|
|
from filer.admin.fileadmin import FileAdmin
|
|
from filer.apps import FilerConfig
|
|
|
|
FileAdmin.readonly_fields += ('owner',)
|
|
FilerConfig.icon = 'icon-folder'
|
|
except ImportError:
|
|
pass
|
|
|
|
# This will be executed when this module is imported after Django is fully loaded
|