my-dev
/
uploader-bot
4
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix someth

This commit is contained in:
user 2025-04-26 13:01:10 +03:00
parent 0e4268fb4d
commit 2476d3b3b6
2 changed files with 183 additions and 116 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

298
app/core/background/convert_service.py
View File

@ -4,24 +4,24 @@ import os
import uuid import uuid
import json import json
import shutil import shutil
import magic # python-magic import magic # python-magic for MIME detection
from base58 import b58decode, b58encode from base58 import b58decode, b58encode
from sqlalchemy import and_, or_ from sqlalchemy import and_, or_
from app.core.models.node_storage import StoredContent from app.core.models.node_storage import StoredContent
from app.core.models._telegram import Wrapped_CBotChat from app.core.models._telegram import Wrapped_CBotChat
from app.core._utils.send_status import send_status from app.core._utils.send_status import send_status
from app.core.logger import make_log from app.core.logger import make_log
from app.core.models.user import User
from app.core.models import WalletConnection from app.core.models import WalletConnection
from app.core.storage import db_session from app.core.storage import db_session
from app.core._config import UPLOADS_DIR from app.core._config import UPLOADS_DIR
from app.core.content.content_id import ContentId from app.core.content.content_id import ContentId
import converter_module # наш модуль для конвертации
async def convert_loop(memory): async def convert_loop(memory):
with db_session() as session: with db_session() as session:
# 1) Найти несанкционированный контент # Query for unprocessed encrypted content
item = session.query(StoredContent).filter( unprocessed_encrypted_content = session.query(StoredContent).filter(
and_( and_(
StoredContent.type == "onchain/content", StoredContent.type == "onchain/content",
or_( or_(
@ -30,50 +30,51 @@ async def convert_loop(memory):
) )
) )
).first() ).first()
if not item: if not unprocessed_encrypted_content:
make_log("ConvertProcess", "No content to convert", level="debug") make_log("ConvertProcess", "No content to convert", level="debug")
return return
# 2) Достать расшифрованный файл # Достаем расшифрованный файл
decrypted = session.query(StoredContent).filter( decrypted_content = session.query(StoredContent).filter(
StoredContent.id == item.decrypted_content_id StoredContent.id == unprocessed_encrypted_content.decrypted_content_id
).first() ).first()
if not decrypted: if not decrypted_content:
make_log("ConvertProcess", "Decrypted content not found", level="error") make_log("ConvertProcess", "Decrypted content not found", level="error")
return return
input_path = f"/Storage/storedContent/{decrypted.hash}" # Определяем путь и расширение входного файла
filename = item.filename input_file_path = f"/Storage/storedContent/{decrypted_content.hash}"
ext = filename.split('.')[-1] if '.' in filename else "" input_ext = (unprocessed_encrypted_content.filename.split('.')[-1]
if '.' in unprocessed_encrypted_content.filename else "mp4")
# 3) Определяем MIME-тип через python-magic # ==== Новая логика: определение MIME-тип через python-magic ====
try: try:
mime = magic.from_file(input_path, mime=True) mime_type = magic.from_file(input_file_path, mime=True)
except Exception as e: except Exception as e:
make_log("ConvertProcess", f"magic probe failed: {e}", level="warning") make_log("ConvertProcess", f"magic probe failed: {e}", level="warning")
mime = "" mime_type = ""
if mime.startswith("video/"): if mime_type.startswith("video/"):
kind = "video" content_kind = "video"
elif mime.startswith("audio/"): elif mime_type.startswith("audio/"):
kind = "audio" content_kind = "audio"
else: else:
kind = "other" content_kind = "other"
make_log("ConvertProcess", f"Detected kind={kind}, mime={mime}", level="info") make_log("ConvertProcess", f"Detected content_kind={content_kind}, mime={mime_type}", level="info")
# 4) Если не видео и не аудио — сохраняем raw copy и выходим # Для прочих типов сохраняем raw копию и выходим
if kind == "other": if content_kind == "other":
raw_hash = item.hash raw_hash = unprocessed_encrypted_content.hash
raw = StoredContent( raw_content = StoredContent(
type="local/content_raw", type="local/content_raw",
hash=raw_hash, hash=raw_hash,
user_id=item.user_id, user_id=unprocessed_encrypted_content.user_id,
filename=filename, filename=unprocessed_encrypted_content.filename,
meta={'source': 'raw_copy'}, meta={'source': 'raw_copy'},
created=datetime.now(), created=datetime.now(),
) )
session.add(raw) session.add(raw_content)
session.commit() session.commit()
# Копируем файл в UPLOADS_DIR # Копируем файл в UPLOADS_DIR
@ -82,127 +83,192 @@ async def convert_loop(memory):
os.remove(dst) os.remove(dst)
except FileNotFoundError: except FileNotFoundError:
pass pass
shutil.copy2(input_path, dst) shutil.copy2(input_file_path, dst)
# Обновляем оригинальный объект # Обновляем оригинальный объект
item.btfs_cid = ContentId(version=2, content_hash=b58decode(raw_hash)).serialize_v2() unprocessed_encrypted_content.btfs_cid = ContentId(
item.ipfs_cid = ContentId(version=2, content_hash=b58decode(raw_hash)).serialize_v2() version=2, content_hash=b58decode(raw_hash)
item.meta = {**item.meta, 'converted_content': {'raw': raw_hash}} ).serialize_v2()
unprocessed_encrypted_content.ipfs_cid = ContentId(
version=2, content_hash=b58decode(raw_hash)
).serialize_v2()
unprocessed_encrypted_content.meta = {
**unprocessed_encrypted_content.meta,
'converted_content': {'raw': raw_hash}
}
session.commit() session.commit()
make_log("ConvertProcess", f"Raw content saved and CIDs set for {raw_hash}", level="info") make_log("ConvertProcess", f"Raw content saved for {raw_hash}", level="info")
return return
# 5) Задаём опции конвертации для видео/аудио # ==== Конвертация для видео или аудио: оригинальная логика ====
if kind == "video": # Static preview interval in seconds
options = ['high', 'low', 'low_preview']
else: # audio
options = ['high_audio', 'low_audio']
# Preview interval
preview_interval = [0, 30] preview_interval = [0, 30]
if item.onchain_index == 2: if unprocessed_encrypted_content.onchain_index in [2]:
preview_interval = [0, 60] preview_interval = [0, 60]
converted = {} make_log(
for opt in options: "ConvertProcess",
# quality и trim f"Processing content {unprocessed_encrypted_content.id} as {content_kind} with preview interval {preview_interval}",
if opt.endswith("_preview"): level="info"
quality = opt.replace("_preview", "") )
trim = preview_interval
# Выбираем опции конвертации для видео и аудио
if content_kind == "video":
REQUIRED_CONVERT_OPTIONS = ['high', 'low', 'low_preview']
else:
REQUIRED_CONVERT_OPTIONS = ['high', 'low'] # no preview for audio
converted_content = {}
logs_dir = "/Storage/logs/converter"
for option in REQUIRED_CONVERT_OPTIONS:
# Set quality parameter and trim option (only for preview)
if option == "low_preview":
quality = "low"
trim_value = f"{preview_interval[0]}-{preview_interval[1]}"
else: else:
quality = opt.replace("_audio", "") quality = option
trim = None trim_value = None
# уникальная папка вывода # Generate a unique output directory for docker container
uid = str(uuid.uuid4()) output_uuid = str(uuid.uuid4())
out_dir = f"/Storage/storedContent/converter-output/{uid}" output_dir = f"/Storage/storedContent/converter-output/{output_uuid}"
os.makedirs(out_dir, exist_ok=True)
# ==== Вызов конвертера ==== # Build the docker command
# converter_module.convert(input_path, out_dir, ext=ext, cmd = [
# quality=quality, trim=trim, audio_only=(kind=="audio")) "docker", "run", "--rm",
# Здесь предполагаем, что convert возвращает имя файла результата. "-v", f"{input_file_path}:/app/input",
try: "-v", f"{output_dir}:/app/output",
result_fname = await converter_module.convert( "-v", f"{logs_dir}:/app/logs",
input=input_path, "media_converter",
output_dir=out_dir, "--ext", input_ext,
ext=ext, "--quality", quality
quality=quality, ]
trim=trim, if trim_value:
audio_only=(kind == "audio") cmd.extend(["--trim", trim_value])
) if content_kind == "audio":
except Exception as e: cmd.append("--audio-only") # audio-only flag
make_log("ConvertProcess", f"Conversion failed {opt}: {e}", level="error")
return
out_file = os.path.join(out_dir, result_fname) process = await asyncio.create_subprocess_exec(
*cmd,
# 6) Считаем sha256 и b58
proc = await asyncio.create_subprocess_exec(
"sha256sum", out_file,
stdout=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE stderr=asyncio.subprocess.PIPE
) )
so, se = await proc.communicate() stdout, stderr = await process.communicate()
if proc.returncode != 0: if process.returncode != 0:
make_log("ConvertProcess", f"sha256sum error: {se.decode()}", level="error") make_log("ConvertProcess", f"Docker conversion failed for option {option}: {stderr.decode()}", level="error")
return return
sha_hex = so.decode().split()[0]
h58 = b58encode(bytes.fromhex(sha_hex)).decode()
# 7) Добавляем StoredContent, если нужно # List files in output dir
if not session.query(StoredContent).filter(StoredContent.hash == h58).first(): try:
new = StoredContent( files = os.listdir(output_dir.replace("/Storage/storedContent", "/app/data"))
except Exception as e:
make_log("ConvertProcess", f"Error reading output directory {output_dir}: {e}", level="error")
return
media_files = [f for f in files if f != "output.json"]
if len(media_files) != 1:
make_log("ConvertProcess", f"Expected one media file, found {len(media_files)} for option {option}", level="error")
return
output_file = os.path.join(
output_dir.replace("/Storage/storedContent", "/app/data"),
media_files[0]
)
# Compute SHA256 hash of the output file
hash_process = await asyncio.create_subprocess_exec(
"sha256sum", output_file,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
hash_stdout, hash_stderr = await hash_process.communicate()
if hash_process.returncode != 0:
make_log("ConvertProcess", f"Error computing sha256sum for option {option}: {hash_stderr.decode()}", level="error")
return
file_hash = hash_stdout.decode().split()[0]
file_hash = b58encode(bytes.fromhex(file_hash)).decode()
# Save new StoredContent if not exists
if not session.query(StoredContent).filter(
StoredContent.hash == file_hash
).first():
new_content = StoredContent(
type="local/content_bin", type="local/content_bin",
hash=h58, hash=file_hash,
user_id=item.user_id, user_id=unprocessed_encrypted_content.user_id,
filename=result_fname, filename=media_files[0],
meta={'encrypted_file_hash': item.hash}, meta={'encrypted_file_hash': unprocessed_encrypted_content.hash},
created=datetime.now(), created=datetime.now(),
) )
session.add(new) session.add(new_content)
session.commit() session.commit()
# 8) Перемещаем в UPLOADS_DIR save_path = os.path.join(UPLOADS_DIR, file_hash)
dst = os.path.join(UPLOADS_DIR, h58)
try: try:
os.remove(dst) os.remove(save_path)
except FileNotFoundError: except FileNotFoundError:
pass pass
shutil.move(out_file, dst)
converted[opt] = h58 try:
shutil.move(output_file, save_path)
except Exception as e:
make_log("ConvertProcess", f"Error moving output file {output_file} to {save_path}: {e}", level="error")
return
# 9) optional: metadata from converter_module converted_content[option] = file_hash
meta_info = converter_module.get_meta(out_dir) # допустим, так
if meta_info and 'ffprobe_meta' not in item.meta:
item.meta['ffprobe_meta'] = meta_info
# 10) очистка # Process output.json for ffprobe_meta
shutil.rmtree(out_dir, ignore_errors=True) output_json_path = os.path.join(
output_dir.replace("/Storage/storedContent", "/app/data"),
"output.json"
)
if os.path.exists(output_json_path) and unprocessed_encrypted_content.meta.get('ffprobe_meta') is None:
try:
with open(output_json_path, "r") as f:
ffprobe_meta = json.load(f)
unprocessed_encrypted_content.meta = {
**unprocessed_encrypted_content.meta,
'ffprobe_meta': ffprobe_meta
}
except Exception as e:
make_log("ConvertProcess", f"Error handling output.json for option {option}: {e}", level="error")
# 11) Обновляем оригинальный объект после всех опций # Cleanup output directory
make_log("ConvertProcess", f"Converted: {converted}", level="info") try:
main_high = 'high' if kind=='video' else 'high_audio' shutil.rmtree(output_dir.replace("/Storage/storedContent", "/app/data"))
main_low = 'low' if kind=='video' else 'low_audio' except Exception as e:
item.btfs_cid = ContentId(version=2, make_log("ConvertProcess", f"Error removing output dir {output_dir}: {e}", level="warning")
content_hash=b58decode(converted[main_high])).serialize_v2()
item.ipfs_cid = ContentId(version=2, # Finalize original record
content_hash=b58decode(converted[main_low])).serialize_v2() make_log("ConvertProcess", f"Content {unprocessed_encrypted_content.id} processed. Converted content: {converted_content}", level="info")
item.meta = {**item.meta, 'converted_content': converted} unprocessed_encrypted_content.btfs_cid = ContentId(
version=2, content_hash=b58decode(converted_content['high' if content_kind=='video' else 'low'])
).serialize_v2()
unprocessed_encrypted_content.ipfs_cid = ContentId(
version=2, content_hash=b58decode(converted_content['low'])
).serialize_v2()
unprocessed_encrypted_content.meta = {
**unprocessed_encrypted_content.meta,
'converted_content': converted_content
}
session.commit() session.commit()
# 12) Отправляем уведомление пользователю # Notify user if needed
if not item.meta.get('upload_notify_msg_id'): if not unprocessed_encrypted_content.meta.get('upload_notify_msg_id'):
wc = session.query(WalletConnection).filter( wallet_owner_connection = session.query(WalletConnection).filter(
WalletConnection.wallet_address == item.owner_address WalletConnection.wallet_address == unprocessed_encrypted_content.owner_address
).order_by(WalletConnection.id.desc()).first() ).order_by(WalletConnection.id.desc()).first()
if wc: if wallet_owner_connection:
bot = Wrapped_CBotChat(memory._client_telegram_bot, wallet_owner_user = wallet_owner_connection.user
chat_id=wc.user.telegram_id, bot = Wrapped_CBotChat(
user=wc.user, db_session=session) memory._client_telegram_bot,
item.meta['upload_notify_msg_id'] = await bot.send_content(session, item) chat_id=wallet_owner_user.telegram_id,
user=wallet_owner_user,
db_session=session
)
unprocessed_encrypted_content.meta['upload_notify_msg_id'] = await bot.send_content(session, unprocessed_encrypted_content)
session.commit() session.commit()

1
requirements.txt
View File

@ -17,3 +17,4 @@ pillow==10.2.0
ffmpeg-python==0.2.0 ffmpeg-python==0.2.0
python-magic==0.4.27 python-magic==0.4.27