295 lines
14 KiB
Python
295 lines
14 KiB
Python
import asyncio
|
||
import hashlib
|
||
import os
|
||
from datetime import datetime
|
||
from mimetypes import guess_type
|
||
|
||
import aiofiles
|
||
import traceback
|
||
from base58 import b58encode
|
||
from sanic import response
|
||
import json
|
||
|
||
from app.core._config import UPLOADS_DIR
|
||
from sqlalchemy import select
|
||
from app.core._utils.resolve_content import resolve_content
|
||
from app.core.logger import make_log
|
||
from app.core.models.node_storage import StoredContent
|
||
from pydub import AudioSegment
|
||
from PIL import Image
|
||
from uuid import uuid4
|
||
import subprocess
|
||
|
||
|
||
# Производится загрузка любого контента одним запросом с определением mime_type по расширению
|
||
# file_mimetype audio/video
|
||
# extension_encoding file encode container
|
||
# Файл сохраняется под sha256(file_content) !!, очень тяжело
|
||
# генерируется CID с учетом типа контента и его декодирования
|
||
# Загрузка происходит только от пользователя либо если наш же бэкенд просит загрузить что-то
|
||
# Создание расшифрованного (local/content_bin) StoredContent
|
||
|
||
async def s_api_v1_storage_post(request):
|
||
if not request.files:
|
||
return response.json({"error": "No file provided"}, status=400)
|
||
|
||
file_param = list(request.files.values())[0][0] if request.files else None
|
||
# file_name_json = request.json.get("filename") if request.json else None
|
||
|
||
if file_param:
|
||
file_content = file_param.body
|
||
file_name = file_param.name
|
||
else:
|
||
return response.json({"error": "No file provided"}, status=400)
|
||
|
||
file_meta = {}
|
||
file_mimetype, file_encoding = guess_type(file_name)
|
||
if file_mimetype:
|
||
file_meta["content_type"] = file_mimetype
|
||
|
||
if file_encoding:
|
||
file_meta["extension_encoding"] = file_encoding
|
||
|
||
try:
|
||
file_hash_bin = hashlib.sha256(file_content).digest()
|
||
file_hash = b58encode(file_hash_bin).decode()
|
||
stored_content = (await request.ctx.db_session.execute(
|
||
select(StoredContent).where(StoredContent.hash == file_hash)
|
||
)).scalars().first()
|
||
if stored_content:
|
||
stored_cid = stored_content.cid.serialize_v1()
|
||
stored_cid_v2 = stored_content.cid.serialize_v2()
|
||
return response.json({
|
||
"content_sha256": file_hash,
|
||
"content_id_v1": stored_cid,
|
||
"content_id": stored_cid_v2,
|
||
"content_url": f"dmy://storage?cid={stored_cid_v2}"
|
||
})
|
||
|
||
if request.ctx.user:
|
||
pass
|
||
elif request.ctx.verified_hash:
|
||
assert request.ctx.verified_hash == file_hash_bin, "Invalid service request hash"
|
||
else:
|
||
return response.json({"error": "Unauthorized"}, status=401)
|
||
|
||
new_content = StoredContent(
|
||
type="local/content_bin",
|
||
user_id=request.ctx.user.id if request.ctx.user else None,
|
||
hash=file_hash,
|
||
filename=file_name,
|
||
meta=file_meta,
|
||
created=datetime.now(),
|
||
key_id=None,
|
||
)
|
||
request.ctx.db_session.add(new_content)
|
||
await request.ctx.db_session.commit()
|
||
|
||
file_path = os.path.join(UPLOADS_DIR, file_hash)
|
||
async with aiofiles.open(file_path, "wb") as file:
|
||
await file.write(file_content)
|
||
|
||
new_content_id = new_content.cid
|
||
new_cid_v1 = new_content_id.serialize_v1()
|
||
new_cid = new_content_id.serialize_v2()
|
||
|
||
return response.json({
|
||
"content_sha256": file_hash,
|
||
"content_id": new_cid,
|
||
"content_id_v1": new_cid_v1,
|
||
"content_url": f"dmy://storage?cid={new_cid}",
|
||
})
|
||
except BaseException as e:
|
||
make_log("Storage", f"sid={getattr(request.ctx, 'session_id', None)} Error: {e}" + '\n' + traceback.format_exc(), level="error")
|
||
return response.json({"error": f"Error: {e}"}, status=500)
|
||
|
||
|
||
# Получение контента с использованием seconds_limit по file_hash
|
||
|
||
async def s_api_v1_storage_get(request, file_hash=None):
|
||
seconds_limit = int(request.args.get("seconds_limit", 0))
|
||
|
||
content_id = file_hash
|
||
cid, errmsg = resolve_content(content_id)
|
||
if errmsg:
|
||
return response.json({"error": errmsg}, status=400)
|
||
|
||
content_sha256 = b58encode(cid.content_hash).decode()
|
||
content = (await request.ctx.db_session.execute(
|
||
select(StoredContent).where(StoredContent.hash == content_sha256)
|
||
)).scalars().first()
|
||
if not content:
|
||
return response.json({"error": "File not found"}, status=404)
|
||
|
||
make_log("Storage", f"sid={getattr(request.ctx, 'session_id', None)} File {content_sha256} requested by user={getattr(getattr(request.ctx, 'user', None), 'id', None)}")
|
||
file_path = os.path.join(UPLOADS_DIR, content_sha256)
|
||
if not os.path.exists(file_path):
|
||
make_log("Storage", f"sid={getattr(request.ctx, 'session_id', None)} File {content_sha256} not found locally", level="error")
|
||
return response.json({"error": "File not found"}, status=404)
|
||
|
||
async with aiofiles.open(file_path, "rb") as file:
|
||
content_file_bin = await file.read()
|
||
|
||
# query_id = str(uuid4().hex())
|
||
tempfile_path = os.path.join(UPLOADS_DIR, f"tmp_{content_sha256}")
|
||
|
||
accept_type = cid.accept_type or content.meta.get("content_type")
|
||
if accept_type:
|
||
if accept_type == "application/json":
|
||
return response.json(
|
||
json.loads(content_file_bin.decode())
|
||
)
|
||
content_type, content_encoding = accept_type.split("/")
|
||
if content_type == 'audio':
|
||
tempfile_path += "_mpeg" + (f"_{seconds_limit}" if seconds_limit else "")
|
||
if not os.path.exists(tempfile_path):
|
||
try:
|
||
# Resolve cover content by CID (async)
|
||
from app.core.content.content_id import ContentId
|
||
try:
|
||
_cid = ContentId.deserialize(content.meta.get('cover_cid'))
|
||
_cover_hash = _cid.content_hash_b58
|
||
cover_content = (await request.ctx.db_session.execute(
|
||
select(StoredContent).where(StoredContent.hash == _cover_hash)
|
||
)).scalars().first()
|
||
except Exception:
|
||
cover_content = None
|
||
cover_tempfile_path = os.path.join(UPLOADS_DIR, f"tmp_{cover_content.hash}_jpeg")
|
||
if not os.path.exists(cover_tempfile_path):
|
||
cover_image = Image.open(cover_content.filepath)
|
||
cover_image = cover_image.convert('RGB')
|
||
quality = 95
|
||
while quality > 10:
|
||
cover_image.save(cover_tempfile_path, 'JPEG', quality=quality)
|
||
if os.path.getsize(cover_tempfile_path) <= 200 * 1024:
|
||
break
|
||
quality -= 5
|
||
|
||
assert os.path.exists(cover_tempfile_path), "Cover image not found"
|
||
except:
|
||
cover_content = None
|
||
cover_tempfile_path = None
|
||
|
||
try:
|
||
file_ext = content.filename.split('.')[-1]
|
||
if file_ext == 'mp3':
|
||
audio = AudioSegment.from_mp3(file_path)
|
||
elif file_ext == 'wav':
|
||
audio = AudioSegment.from_wav(file_path)
|
||
elif file_ext == 'ogg':
|
||
audio = AudioSegment.from_ogg(file_path)
|
||
elif file_ext == 'flv':
|
||
audio = AudioSegment.from_flv(file_path)
|
||
else:
|
||
audio = None
|
||
|
||
if not audio:
|
||
try:
|
||
audio = AudioSegment.from_file(file_path)
|
||
except BaseException as e:
|
||
make_log("Storage", f"sid={getattr(request.ctx, 'session_id', None)} Error loading audio from file: {e}", level="debug")
|
||
|
||
if not audio:
|
||
try:
|
||
audio = AudioSegment(content_file_bin)
|
||
except BaseException as e:
|
||
make_log("Storage", f"sid={getattr(request.ctx, 'session_id', None)} Error loading audio from binary: {e}", level="debug")
|
||
|
||
audio = audio[:seconds_limit * 1000] if seconds_limit else audio
|
||
audio.export(tempfile_path, format="mp3", cover=cover_tempfile_path)
|
||
except BaseException as e:
|
||
make_log("Storage", f"sid={getattr(request.ctx, 'session_id', None)} Error converting audio: {e}" + '\n' + traceback.format_exc(), level="error")
|
||
|
||
if os.path.exists(tempfile_path):
|
||
async with aiofiles.open(tempfile_path, "rb") as file:
|
||
content_file_bin = await file.read()
|
||
|
||
accept_type = 'audio/mpeg'
|
||
make_log("Storage", f"sid={getattr(request.ctx, 'session_id', None)} Audio {content_sha256} converted successfully", level='debug')
|
||
else:
|
||
tempfile_path = tempfile_path[:-5]
|
||
|
||
elif content_type == 'image':
|
||
tempfile_path += "_jpeg"
|
||
if not os.path.exists(tempfile_path):
|
||
try:
|
||
image = Image.open(file_path)
|
||
image = image.convert('RGB')
|
||
quality = 95
|
||
while quality > 10:
|
||
image.save(tempfile_path, 'JPEG', quality=quality)
|
||
if os.path.getsize(tempfile_path) <= 200 * 1024:
|
||
break
|
||
quality -= 5
|
||
except BaseException as e:
|
||
make_log("Storage", f"sid={getattr(request.ctx, 'session_id', None)} Error converting image: {e}" + '\n' + traceback.format_exc(), level="error")
|
||
|
||
if os.path.exists(tempfile_path):
|
||
async with aiofiles.open(tempfile_path, "rb") as file:
|
||
content_file_bin = await file.read()
|
||
|
||
make_log("Storage", f"sid={getattr(request.ctx, 'session_id', None)} Image {content_sha256} converted successfully", level='debug')
|
||
accept_type = 'image/jpeg'
|
||
else:
|
||
tempfile_path = tempfile_path[:-5]
|
||
|
||
elif content_type == 'video':
|
||
# Build a temp path for the video
|
||
tempfile_path += "_mp4" + (f"_{seconds_limit}" if seconds_limit else "") + ".mp4"
|
||
if not os.path.exists(tempfile_path):
|
||
try:
|
||
# Use ffmpeg to cut or convert to mp4
|
||
if seconds_limit > 0:
|
||
# Cut the video to the specified seconds_limit
|
||
subprocess.run([
|
||
"ffmpeg",
|
||
"-y",
|
||
"-ss", "0", # Set start time (fast seeking)
|
||
"-i", file_path,
|
||
"-t", str(seconds_limit), # Set duration of the output
|
||
"-c:v", "libx264", # Encode video with libx264
|
||
"-profile:v", "baseline", # Set baseline profile for compatibility with Telegram
|
||
"-level", "3.0", # Set level to 3.0 for compatibility
|
||
"-pix_fmt", "yuv420p", # Set pixel format for maximum compatibility
|
||
"-c:a", "aac", # Encode audio with AAC
|
||
"-b:a", "128k", # Set audio bitrate
|
||
"-movflags", "+faststart", # Enable fast start for streaming
|
||
tempfile_path
|
||
], check=True)
|
||
else:
|
||
# Just convert to mp4 (no cutting)
|
||
subprocess.run([
|
||
"ffmpeg",
|
||
"-y",
|
||
"-ss", "0", # Set start time (fast seeking)
|
||
"-i", file_path,
|
||
# "-t", str(seconds_limit), # Set duration of the output
|
||
"-c:v", "libx264", # Encode video with libx264
|
||
"-profile:v", "baseline", # Set baseline profile for compatibility with Telegram
|
||
"-level", "3.0", # Set level to 3.0 for compatibility
|
||
"-pix_fmt", "yuv420p", # Set pixel format for maximum compatibility
|
||
"-c:a", "aac", # Encode audio with AAC
|
||
"-b:a", "128k", # Set audio bitrate
|
||
"-movflags", "+faststart", # Enable fast start for streaming
|
||
tempfile_path
|
||
], check=True)
|
||
except BaseException as e:
|
||
make_log("Storage", f"Error converting video: {e}" + '\n' + traceback.format_exc(), level="error")
|
||
|
||
if os.path.exists(tempfile_path):
|
||
async with aiofiles.open(tempfile_path, "rb") as file:
|
||
content_file_bin = await file.read()
|
||
make_log("Storage", f"Video {content_sha256} processed successfully")
|
||
accept_type = 'video/mp4'
|
||
else:
|
||
tempfile_path = tempfile_path[:-4] # remove _mp4 or similar suffix
|
||
|
||
return response.raw(body=content_file_bin, **({'content_type': accept_type} if accept_type else {}))
|
||
|
||
async def s_api_v1_storage_decode_cid(request, content_id=None):
|
||
cid, errmsg = resolve_content(content_id)
|
||
if errmsg:
|
||
return response.json({"error": errmsg}, status=400)
|
||
|
||
return response.json(cid.json_format())
|