Files
Local-Booru/system_module/tasks.py
T
2026-06-06 22:19:53 +03:00

224 lines
9.5 KiB
Python

import os
import json
from pathlib import Path
def find_file(filename: str, file_folders: list[str], is_folder: bool = True):
for folder in file_folders:
folder_path = Path(folder)
if not folder_path.exists():
continue
for file_path in folder_path.rglob(filename):
return str(folder_path) if is_folder else str(file_path)
return None
def get_size_format(b: int, factor=1024, suffix="B"):
for unit in ["", "K", "M", "G", "T"]:
if b < factor:
return f"{b:.2f} {unit}{suffix}"
b /= factor
return f"{b:.2f} P{suffix}"
def get_dir_size(path: str):
total = 0
if os.path.exists(path):
for f in os.listdir(path):
fp = os.path.join(path, f)
if os.path.isfile(fp):
total += os.path.getsize(fp)
return total
def get_file_list_size(file_list: list[str]):
total = 0
for fp in file_list:
if os.path.isfile(fp):
total += os.path.getsize(fp)
return total
def get_size_edge(type_search: bool, file_list: list[str]):
if type_search:
result = min(file_list, key=lambda f: Path(f).stat().st_size)
else:
result = max(file_list, key=lambda f: Path(f).stat().st_size)
return str(result)
def get_files_not_found(database, config):
all_db = database.get_all()
list_md5 = []
for one_post in all_db:
file_path = find_file(one_post["local_filename"], config["file"]["repos"], is_folder=False)
if file_path == None:
list_md5.append(one_post["md5"])
return list_md5
def calculate_analytic(database, config):
all_db = database.get_all()
explicit_db = database.get_search(["e"], [])
sensitive_db = database.get_search(["s"], [])
questionable_db = database.get_search(["q"], [])
general_db = database.get_search(["g"], [])
rating_list_file_path = {"explicit":[], "sensitive":[], "questionable":[], "general":[]}
type_list_file_path = {"images":[], "videos":[], "gifs":[]}
files_not_found = 0
for one_post in all_db:
file_path = find_file(one_post["local_filename"], config["file"]["repos"], is_folder=False)
if file_path != None:
file_path = str(file_path)
rating_list_file_path[one_post["rating"]].append(file_path)
if one_post["is_gif"] == 1:
type_list_file_path["gifs"].append(file_path)
elif one_post["is_video"] == 1:
type_list_file_path["videos"].append(file_path)
else:
type_list_file_path["images"].append(file_path)
else:
files_not_found += 1
file_size_dict = {"explicit":get_file_list_size(rating_list_file_path["explicit"]),
"sensitive":get_file_list_size(rating_list_file_path["sensitive"]),
"questionable":get_file_list_size(rating_list_file_path["questionable"]),
"general":get_file_list_size(rating_list_file_path["general"])
}
type_size_dict = {"images":get_file_list_size(type_list_file_path["images"]),
"videos":get_file_list_size(type_list_file_path["videos"]),
"gifs":get_file_list_size(type_list_file_path["gifs"])}
ratings_stats = {
"explicit": {
"count": len(explicit_db),
"size_formatted": get_size_format(file_size_dict["explicit"])
},
"sensitive": {
"count": len(sensitive_db),
"size_formatted": get_size_format(file_size_dict["sensitive"])
},
"questionable": {
"count": len(questionable_db),
"size_formatted": get_size_format(file_size_dict["questionable"])
},
"general": {
"count": len(general_db),
"size_formatted": get_size_format(file_size_dict["general"])
}
}
total_file_data_size = file_size_dict["general"]+file_size_dict["questionable"]+file_size_dict["sensitive"]+file_size_dict["explicit"]
total_thumb_size = get_dir_size(config["file"]["thumb"])
total_log_size = get_file_list_size(["logs.log"])
total_db_size = get_file_list_size([os.path.join(config["file"]["data"], "db.sqlite")])
weight_stats_smallest_file = get_size_edge(True, type_list_file_path["gifs"]+type_list_file_path["images"]+type_list_file_path["videos"])
weight_stats_largest_file = get_size_edge(False, type_list_file_path["gifs"]+type_list_file_path["images"]+type_list_file_path["videos"])
weight_stats_data = {
"smallest":{
"file":"",
"md5":"",
"size":0
},
"largest":{
"file":"",
"md5":"",
"size":0
}
}
resolution_stats_data = {
"smallest":{
"file":"",
"md5":"",
"res":{
"width":0,
"height":0,
"total":99999999999999999999999999
}
},
"largest":{
"file":"",
"md5":"",
"res":{
"width":0,
"height":0,
"total":0
}
}
}
for one_post_scan in all_db:
if one_post_scan["local_filename"] == os.path.basename(weight_stats_smallest_file):
weight_stats_data["smallest"] = {
"file":os.path.basename(weight_stats_smallest_file),
"md5":one_post_scan["md5"],
"size":get_file_list_size([weight_stats_smallest_file])
}
if one_post_scan["local_filename"] == os.path.basename(weight_stats_largest_file):
weight_stats_data["largest"] = {
"file":os.path.basename(weight_stats_largest_file),
"md5":one_post_scan["md5"],
"size":get_file_list_size([weight_stats_largest_file])
}
print(get_file_list_size([os.path.basename(weight_stats_largest_file)]))
one_post_res_total = one_post_scan["width"]*one_post_scan["height"]
if resolution_stats_data["largest"]["res"]["total"] < one_post_res_total:
resolution_stats_data["largest"] = {
"file":one_post_scan["local_filename"],
"md5":one_post_scan["md5"],
"res":{
"width":one_post_scan["width"],
"height":one_post_scan["height"],
"total":one_post_res_total
}
}
if resolution_stats_data["smallest"]["res"]["total"] > one_post_res_total:
resolution_stats_data["smallest"] = {
"file":one_post_scan["local_filename"],
"md5":one_post_scan["md5"],
"res":{
"width":one_post_scan["width"],
"height":one_post_scan["height"],
"total":one_post_res_total
}
}
result = {
"total_images_analyzed": len(all_db),
"files_not_found":files_not_found,
"average_weight": {
"bytes": round(total_file_data_size/(len(rating_list_file_path["explicit"])+len(rating_list_file_path["sensitive"])+len(rating_list_file_path["questionable"])+len(rating_list_file_path["general"])), 2),
"formatted": get_size_format(round(total_file_data_size/(len(rating_list_file_path["explicit"])+len(rating_list_file_path["sensitive"])+len(rating_list_file_path["questionable"])+len(rating_list_file_path["general"])), 2))
},
"storage_stats": {
"folders": {
"data": get_size_format(total_file_data_size),
"db": get_size_format(total_db_size),
"thumb": get_size_format(total_thumb_size),
"logs": get_size_format(total_log_size)
},
"types": {
"images": get_size_format(type_size_dict["images"]),
"videos": get_size_format(type_size_dict["videos"]),
"gifs": get_size_format(type_size_dict["gifs"])
},
"grand_total": get_size_format(total_file_data_size+total_thumb_size+total_log_size+total_db_size)
},
"ratings_stats": ratings_stats,
"resolution_stats": {
"largest": {
"filename": resolution_stats_data["largest"]["file"],
"md5": resolution_stats_data["largest"]["md5"],
"resolution": str(resolution_stats_data["largest"]["res"]["width"]) + " x " + str(resolution_stats_data["largest"]["res"]["height"])
},
"smallest": {
"filename": resolution_stats_data["smallest"]["file"],
"md5": resolution_stats_data["smallest"]["md5"],
"resolution": str(resolution_stats_data["smallest"]["res"]["width"]) + " x " + str(resolution_stats_data["smallest"]["res"]["height"])
}
},
"weight_stats": {
"largest": {
"filename": os.path.basename(weight_stats_data["largest"]["file"]),
"md5": weight_stats_data["largest"]["md5"],
"size_formatted": get_size_format(weight_stats_data["largest"]["size"])
},
"smallest": {
"filename": os.path.basename(weight_stats_data["smallest"]["file"]),
"md5": weight_stats_data["smallest"]["md5"],
"size_formatted": get_size_format(weight_stats_data["smallest"]["size"])
}
},
"top_tag":database.get_top_tag("")
}
with open(os.path.join(config["file"]["data"], "analytic.json"), 'w', encoding='utf-8') as out_f:
json.dump(result, out_f, indent=4, ensure_ascii=False)
return True