scan-vid/scan-vid.py

#!/usr/local/bin/python

# выбирает информацию по видео файлу.

import os
import subprocess
import json
import zlib
import datetime

# правильный перевод!
def humanbytes(B):
    """Return the given bytes as a human friendly KB, MB, GB, or TB string."""
    B = float(B)
    KB = float(1024)
    MB = float(KB ** 2) # 1,048,576
    GB = float(KB ** 3) # 1,073,741,824
    TB = float(KB ** 4) # 1,099,511,627,776

    if B < KB:
        return '{0} {1}'.format(B,'Bytes' if 0 == B > 1 else 'Byte')
    elif KB <= B < MB:
        return '{0:.2f} KB'.format(B / KB)
    elif MB <= B < GB:
        return '{0:.2f} MB'.format(B / MB)
    elif GB <= B < TB:
        return '{0:.2f} GB'.format(B / GB)
    elif TB <= B:
        return '{0:.2f} TB'.format(B / TB)

# file_info
def file_info(fn):
	meta = {}
	fp = open(fn, 'rb')

	filename, file_extension = os.path.splitext(fn)
	created = os.path.getmtime(fn)
	dt_m = datetime.datetime.fromtimestamp(created)
	size = os.path.getsize(fn)
	data = fp.read()

	meta["name"] = fn
	meta["path"] = os.path.splitdrive(fn)[1]
	meta["ext"] = file_extension.lower()
	meta["date"] = dt_m
	meta["type"] = "meta"
	meta["crc"] = zlib.crc32(data)
	meta["size"] = size
	return meta	

def humanbitrate(s):
	"""
	The function "humanbitrate" takes a string representing a bitrate in the format "a/b" and returns
	the bitrate as a float with one decimal place.
	
	:param s: The parameter "s" is a string that represents a fraction in the format "a/b", where "a"
	and "b" are numbers
	:return: the bitrate as a string.
	"""
	items = s.split("/")
	a = float(items[0])
	b = float(items[1])
	bitrate = "%8.1f" % (a / b)
	return  bitrate.strip()

# крманда ffmpeg для получения постера
def poster(fn):
	"""
	The function "poster" takes a file name as input and returns a command to extract a poster image
	from a video file using ffmpeg.
	
	:param fn: The parameter `fn` is a string that represents the file name or path of the video file
	for which you want to generate a poster image
	:return: a string that represents a command to be executed using ffmpeg.
	"""
	seconds = 5
	path = fn
	image = postername(fn)
	ffmpeg = "ffmpeg -ss %s -i \"%s\" -f image2 -vframes 1 -y \"%s\"" % (seconds, path, image)
	return ffmpeg

def postername(file_path):
	file_name, file_extension = os.path.splitext(file_path)
	return file_name + ".jpg"

def check_file(file):
	meta = file_info(file)
	#print(meta)

	# cmd = "ffprobe -v quiet -print_format json -show_format -show_streams \"%s\"" % file
	cmd = "ffprobe -v quiet -print_format json -show_format -show_streams \"%s\" >ffprobe.info" % file
	# output = subprocess.run(cmd, capture_output=True, text=True, shell=True)

	# обычный вывод
	output = subprocess.run(cmd, text=True, shell=True)
	#print(output)

	info = {}
	
	# data = json.loads(output.stdout)
	f = open('ffprobe.info', "r", encoding='utf-8') 
	data = json.load(f)
	f.close
	
	info["audio"] = len(data["streams"]) > 1
	total_seconds = datetime.timedelta(seconds=float(data["format"]["duration"]))
	info["duration"] = total_seconds
	#print("seconds", total_seconds, file)
	info["hduration"] = "00:00:00"
	try:
		info["hduration"] = datetime.datetime.strptime(str(total_seconds), "%H:%M:%S.%f").strftime("%H:%M:%S")
	except:
		info["hduration"] = '0' + str(total_seconds)
		# print("не удалось распарсить время:", info["hduration"])
	info["width"] = data["streams"][0]["width"]
	info["height"] = data["streams"][0]["height"]
	info["bit_rate"] = humanbitrate(data["streams"][0]["avg_frame_rate"])
	info["codec"] = data["streams"][0]["codec_name"]
	info["filename"] = os.path.basename(meta["name"])
	info["ext"] = meta["ext"]
	info["crc"] = meta["crc"]
	info["created"] = meta["date"]
	info["hcreated"] = meta["date"].strftime("%d.%m.%Y")
	info["size"] = data["format"]["size"]
	info["hsize"] = humanbytes(data["format"]["size"])
	info["poster"] = poster(file)
	return info


def valid_file(file):
	"""
	The function checks if a file has a valid extension by comparing it to a list of valid extensions.
	
	:param file: The "file" parameter is a string that represents the name of a file, including its
	extension
	:return: a boolean value. It returns True if the file extension is valid (i.e., it matches one of
	the extensions in the valid_ext list), and False otherwise.
	"""
	valid_ext = ["mp4", "mpeg", "webm"]
	ext = file.split(".")[-1]
	for e in valid_ext:
		if ext == e:
			return True
	return False

def generate_sql(m):
	cmd = """
	('%s', %s, '%s', 
	 '%s', '%s', %s, %s,
	 %s, '%s', %s, '%s', '%s',
	 '%s',
	 '%s', '%s', '%s', '%s'),\n""" % (
		m["filename"], m["size"], m["hsize"], 
		m["duration"], m["hduration"], m["width"], m["height"],
		m["bit_rate"], m["ext"], m["crc"], m["created"], m["hcreated"],
		m["poster"],
		m["audio"], m["codec"], m["tags"], m["type"] )
	print(cmd)
	return cmd

# =================== начало программы ====================

dir = "d:\\vids\\ero"

log_errors = open("errors.log", "w", encoding="utf-8")

# подготовка sql 
sql_file = open("meta.sql", "w", encoding="utf-8")
values = "\t(filename, size, hsize,"
values += " duration, hduration, width, height,"
values += " bitrate, ext, crc, created, hcreated,"
values += " poster, audio, codec, tags, type"
values += ")\n  values"
cmd = "insert\n  into filemeta\n%s" % values
sql_file.write(cmd)

cmd = ""
files = os.listdir(dir)
for file in files:
	path = os.path.join(dir, file)
	# path = os.path.join(dir, "Бандиты насилуют и убивают двух сестер Изнасилования в ретро.mp4")
	if valid_file(file):
		try:
			meta = check_file(path)
			if meta != None:
				meta["type"] = "video"
				meta["tags"] = ""
				cmd += generate_sql(meta)
				print(meta["filename"])
			else:
				log_errors.write("ошибка ffprobe : %s\n" % path)
			
			# предупреждение о маленьком файле
			width = int(meta["width"])
			if width < 400:
				print("[%sx%s] %s" % (meta["width"], meta["height"], path))
		except:
			log_errors.write("не удалось обработать файл: %s\n" % path)
	else:
		log_errors.write("пропускаем файл: %s\n" % path)

# меняем последнюю запятую на точку с запятой
cmd = cmd[:-3] + ";\n"
sql_file.write(cmd)		
sql_file.close()

log_errors.close()