mirror of
https://github.com/aaPanel/aaPanel.git
synced 2026-08-20 06:35:47 +02:00
238 lines
11 KiB
Python
238 lines
11 KiB
Python
# coding: utf-8
|
|
# -------------------------------------------------------------------
|
|
# 宝塔Linux面板
|
|
# -------------------------------------------------------------------
|
|
# Copyright (c) 2015-2099 宝塔软件(http://bt.cn) All rights reserved.
|
|
# -------------------------------------------------------------------
|
|
# Author: sww <cjxin@bt.cn>
|
|
# -------------------------------------------------------------------
|
|
import json
|
|
import os
|
|
import sys
|
|
import time
|
|
import traceback
|
|
|
|
os.chdir('/www/server/panel')
|
|
if not 'class/' in sys.path:
|
|
sys.path.insert(0, 'class/')
|
|
# 文件格式转化
|
|
# ------------------------------
|
|
from filesModel.base import filesBase
|
|
import public
|
|
|
|
|
|
|
|
class main(filesBase):
|
|
def __init__(self):
|
|
self.convert_type = {
|
|
'audio': ['mp3', 'wav', 'flac'],
|
|
'video': ['mp4', 'avi', 'mov', 'mkv'],
|
|
'image': ['png', 'jpeg', 'jpg', 'bmp', 'tiff', 'webp'],
|
|
'all': ['png', 'jpeg', 'jpg', 'bmp', 'tiff', 'webp', 'mp4', 'avi', 'mov', 'mkv', 'mp3', 'wav', 'flac']
|
|
}
|
|
|
|
def convert_audio(self, input_file, output_file):
|
|
try:
|
|
# 读取输入文件
|
|
try:
|
|
import librosa
|
|
except:
|
|
os.system('btpip install librosa')
|
|
import librosa
|
|
|
|
audio, sr = librosa.load(input_file)
|
|
# 将音频转换为 WAV 格式
|
|
try:
|
|
import soundfile as sf
|
|
except:
|
|
os.system('btpip install soundfile')
|
|
import soundfile as sf
|
|
sf.write(output_file, audio, sr)
|
|
if os.path.exists(output_file):
|
|
self.write_log('{}; audio ;{}->{};Conversion successful!'.format(int(time.time()), input_file, output_file))
|
|
else:
|
|
self.write_log('{}; audio ;{}->{};Conversion failed. No converted file was detected!'.format(int(time.time()), input_file, output_file))
|
|
return False
|
|
if self.is_del:
|
|
os.remove(input_file)
|
|
self.write_log('{}; audio ;The source file from {} has been deleted.'.format(int(time.time()), input_file))
|
|
return True
|
|
except:
|
|
print(traceback.format_exc())
|
|
self.write_log('{}; audio ;{}->{};Conversion failed!'.format(int(time.time()), input_file, output_file))
|
|
if os.path.exists(output_file):
|
|
os.remove(output_file)
|
|
return False
|
|
|
|
def convert_video(self, input_file, output_file):
|
|
try:
|
|
|
|
try:
|
|
import imageio
|
|
except:
|
|
os.system('btpip install imageio')
|
|
import imageio
|
|
|
|
import imageio
|
|
# 读取输入文件
|
|
reader = imageio.get_reader(input_file)
|
|
# 创建输出文件
|
|
try:
|
|
meta_data = reader.get_meta_data()
|
|
writer = imageio.get_writer(output_file, fps=meta_data["fps"])
|
|
except ValueError:
|
|
os.system("btpip install imageio[ffmpeg]")
|
|
os.system("btpip install imageio[pyav]")
|
|
meta_data = reader.get_meta_data()
|
|
writer = imageio.get_writer(output_file, fps=meta_data["fps"])
|
|
|
|
# 读取每一帧并写入输出文件
|
|
for frame in reader:
|
|
writer.append_data(frame)
|
|
|
|
# 关闭文件
|
|
reader.close()
|
|
writer.close()
|
|
|
|
if os.path.exists(output_file):
|
|
self.write_log('{}; video ;{}->{};Conversion successful!'.format(int(time.time()), input_file, output_file))
|
|
else:
|
|
self.write_log('{}; video ;{}->{};Conversion failed. No converted file was detected!'.format(int(time.time()), input_file, output_file))
|
|
print('1111')
|
|
return False
|
|
if self.is_del:
|
|
os.remove(input_file)
|
|
self.write_log('{}; video ;The source file from {} has been deleted.'.format(int(time.time()), input_file))
|
|
return True
|
|
except:
|
|
print(traceback.format_exc())
|
|
self.write_log('{}; video ;{}->{};Conversion failed!'.format(int(time.time()), input_file, output_file))
|
|
if os.path.exists(output_file):
|
|
os.remove(output_file)
|
|
return False
|
|
|
|
def convert_image(self, input_file, output_file):
|
|
try:
|
|
|
|
try:
|
|
import cv2
|
|
except:
|
|
os.system('btpip install opencv-python')
|
|
import cv2
|
|
|
|
if output_file.split('.')[-1] in ['ico', 'svg'] or input_file.split('.')[-1] in ['ico', 'svg']:
|
|
from PIL import Image
|
|
img = Image.open(input_file)
|
|
img.save(output_file, format=input_file.split('.')[-1])
|
|
import cv2
|
|
img = cv2.imread(input_file)
|
|
cv2.imwrite(output_file, img)
|
|
if os.path.exists(output_file):
|
|
self.write_log('{}; picture ;{}->{};Conversion successful!'.format(int(time.time()), input_file, output_file))
|
|
else:
|
|
self.write_log('{}; picture ;{}->{};Conversion failed. No converted file was detected!'.format(int(time.time()), input_file, output_file))
|
|
return False
|
|
if self.is_del:
|
|
os.remove(input_file)
|
|
self.write_log('{}; picture ;The source file from {} has been deleted.'.format(int(time.time()), input_file))
|
|
return True
|
|
except:
|
|
print(traceback.format_exc())
|
|
self.write_log('{}; picture ;{}->{};Conversion failed!'.format(int(time.time()), input_file, output_file))
|
|
if os.path.exists(output_file):
|
|
os.remove(output_file)
|
|
return False
|
|
|
|
def run(self, get):
|
|
try:
|
|
public.set_module_logs('conversion', 'run_conversion', 1)
|
|
if not (hasattr(get, 'pdata') and get.pdata != '[]' and get.pdata):
|
|
return public.return_message(-1, 0, 'The conversion list is empty!')
|
|
self.is_del = int(get.get('is_save', 0))
|
|
pdata = json.loads(get.pdata)
|
|
if not pdata:
|
|
return public.return_message(-1, 0, 'parameter error!')
|
|
result = []
|
|
for i in pdata:
|
|
# 判断参数是否正确
|
|
type = [k for k, v in self.convert_type.items() if i['input_file'].split('.')[-1] in v]
|
|
if not type:
|
|
result.append({'name': i['input_file'].split('/')[-1], 'status': False})
|
|
self.write_log('{};{};{}->{}; Conversion failed. Unsupported format!'.format(int(time.time()), 'unknown', i['input_file'], i['output_file']))
|
|
continue
|
|
if os.path.exists(i['output_file']):
|
|
self.write_log('{};{};{}->{};Conversion failed. The output address already has a file with the same name!'.format(int(time.time()), 'unknown', i['input_file'], i['output_file']))
|
|
result.append({'name': i['input_file'].split('/')[-1] + '->' + i['output_file'].split('/')[-1], 'status': False})
|
|
continue
|
|
type = type[0]
|
|
if not i['input_file'] or not i['output_file']:
|
|
return public.return_message(-1, 0, 'parameter error!')
|
|
# 判断源文件和目标文件是否相同
|
|
if i['input_file'] == i['output_file']:
|
|
return public.return_message(-1, 0, 'The source file and the target file cannot be the same!')
|
|
# 判断源文件是否存在
|
|
if not os.path.exists(i['input_file']):
|
|
return public.return_message(-1, 0, 'Source file does not exist!')
|
|
# 音频格式转换
|
|
if type == 'audio':
|
|
if i['input_file'].split('.')[-1] not in self.convert_type['audio'] or i['output_file'].split('.')[-1] not in self.convert_type['audio']:
|
|
return public.return_message(-1, 0, 'Unsupported format conversion!')
|
|
if self.convert_audio(i['input_file'], i['output_file']):
|
|
result.append({'name': i['input_file'].split('/')[-1] + '->' + i['output_file'].split('/')[-1], 'status': True})
|
|
else:
|
|
result.append({'name': i['input_file'].split('/')[-1] + '->' + i['output_file'].split('/')[-1], 'status': False})
|
|
# 视频格式转换
|
|
elif type == 'video':
|
|
if i['input_file'].split('.')[-1] not in self.convert_type['video'] or i['output_file'].split('.')[-1] not in self.convert_type['video']:
|
|
return public.return_message(-1, 0, 'Unsupported format conversion!')
|
|
if self.convert_video(i['input_file'], i['output_file']):
|
|
result.append({'name': i['input_file'].split('/')[-1] + '->' + i['output_file'].split('/')[-1], 'status': True})
|
|
else:
|
|
result.append({'name': i['input_file'].split('/')[-1] + '->' + i['output_file'].split('/')[-1], 'status': False})
|
|
# 图片格式转换
|
|
elif type == 'image':
|
|
if i['input_file'].split('.')[-1] not in self.convert_type['image'] or i['output_file'].split('.')[-1] not in self.convert_type['image']:
|
|
return public.return_message(-1, 0, 'Unsupported format conversion!')
|
|
if self.convert_image(i['input_file'], i['output_file']):
|
|
result.append({'name': i['input_file'].split('/')[-1] + '->' + i['output_file'].split('/')[-1], 'status': True})
|
|
else:
|
|
result.append({'name': i['input_file'].split('/')[-1] + '->' + i['output_file'].split('/')[-1], 'status': False})
|
|
return public.return_message(0, 0, result)
|
|
except:
|
|
return traceback.format_exc()
|
|
return public.return_message(-1, 0, 'Conversion failed!')
|
|
|
|
def get_convert_liet(self, get=None):
|
|
# 获取转换支持列表
|
|
try:
|
|
return public.return_message(0, 0, self.convert_type)
|
|
except:
|
|
return public.return_message(-1, 0, public.lang('Failed to obtain the list of conversion support options!'))
|
|
|
|
def write_log(self, log):
|
|
# 写入日志
|
|
path = '/www/server/panel/logs/convert.log'
|
|
if not os.path.exists(path):
|
|
public.writeFile(path, '')
|
|
public.writeFile(path, log + '\n', 'a+')
|
|
|
|
def get_log(self, get=None):
|
|
try:
|
|
# 获取日志
|
|
p = int(get.p) if hasattr(get, 'p') else 1
|
|
row = int(get.row) if hasattr(get, 'row') else 10
|
|
path = '/www/server/panel/logs/convert.log'
|
|
if not os.path.exists(path):
|
|
return public.return_message(0, 0, 'No log.')
|
|
logs = public.readFile(path)
|
|
logs = logs.strip().split('\n')
|
|
logs.reverse()
|
|
logs = [i.split(';') for i in logs]
|
|
logs = [{'time': i[0], 'type': i[1], 'operation': i[2], 'status': i[3]} for i in logs if len(i) == 4]
|
|
data = public.get_page(len(logs), p, row)
|
|
data['data'] = logs[p * row - row: p * row]
|
|
return public.return_message(0, 0, data)
|
|
except:
|
|
return traceback.format_exc()
|
|
return public.return_message(-1, 0, '获取日志失败!')
|