0.1.4s

system_module/db.py

@@ -1,8 +1,10 @@
-import sqlite3
-import os
 import json
+import os
+import sqlite3
 from datetime import datetime
-from system_module import gelbooru
+
+from system_module.gelbooru import gelbooru_date_parse
+
 
 class DB:
     def __init__(self, debug: bool):
@@ -117,7 +119,7 @@ class DB:
             "score": data_add["score"],
             "timestamp": data_add["created_at"],
             "parsed_time": parsed_time,
-            "display_date": gelbooru.gelbooru_date_parse(data_add["created_at"]),
+            "display_date": gelbooru_date_parse(data_add["created_at"]),
             "source": data_add["source"],
             "width": data_add["width"],
             "height": data_add["height"]
@@ -165,15 +167,15 @@ class DB:
         cursor = self.conn.execute(query, (md5,))
         return cursor.fetchone() is not None
     
-    def get_autocomplete(self, tag_slise: str):
+    def get_top_tag(self, tag_slise: str, count_tag_get: int = 9999999):
         query = """
             SELECT value AS tag, COUNT(*) AS count 
             FROM files, json_each(files.tags) 
             WHERE value LIKE ? 
             GROUP BY value 
             ORDER BY count DESC 
-            LIMIT 10
-        """
+            LIMIT {count_tag_get}
+        """.format(count_tag_get=count_tag_get)
         cursor = self.conn.execute(query, (f"{tag_slise.lower()}%",))
         return [{"tag": row["tag"], "count": row["count"]} for row in cursor.fetchall()]
     

system_module/gelbooru.py

@@ -1,8 +1,10 @@
-import requests
-from datetime import datetime
 import os
+from datetime import datetime
 
-def gelbooru_date_parse(date_str):
+import requests
+
+
+def gelbooru_date_parse(date_str: str):
     if not date_str:
         return 0, "Unknown"
     try:
@@ -13,14 +15,14 @@ def gelbooru_date_parse(date_str):
         return 0, "Unknown"
 
 class GB:
-    def __init__(self, api_id, api_hash, headers, dw_folder):
+    def __init__(self, api_id: str, api_hash: str, headers: dict[str, str], dw_folder: str):
         self.api_id = api_id
         self.api_hash = api_hash
         self.dw_folder = dw_folder
         self.root_url = "https://gelbooru.com/index.php"
         self.HEADERS = headers
         
-    def get_from_id(self, id_post):
+    def get_from_id(self, id_post: int):
         params = {"page": "dapi", "s": "post", "q": "index", "json": 1, "id": id_post, "api_key":self.api_hash, "user_id":self.api_id}
         response = requests.get(self.root_url, params=params, headers=self.HEADERS, timeout=15)
         print(response.url)
@@ -28,13 +30,10 @@ class GB:
         file_name = response.json()["post"][0]["image"]
         return response.json()["post"][0], url_file, file_name
     
-    def get_from_md5(self, md5):
+    def get_from_md5(self, md5: str):
         params = {"page": "dapi", "s": "post", "q": "index", "json": 1, "tags":"md5:"+md5, "api_key":self.api_hash, "user_id":self.api_id}
         response = requests.get(self.root_url, params=params, headers=self.HEADERS, timeout=15)
         print(response.url)
         url_file = response.json()["post"][0]["file_url"]
         file_name = response.json()["post"][0]["image"]
-        return response.json()["post"][0], url_file, file_name
-    
-    def get_binary_file(self, id_post):
-        pass
+        return response.json()["post"][0], url_file, file_name

system_module/image_processor.py

@@ -1,6 +1,7 @@
-import subprocess
-from PIL import Image
 import os
+import subprocess
+
+from PIL import Image
 
 EXTENSIONS_IMG = ('jpg', 'jpeg', 'png', 'gif', 'webp', 'bmp')
 EXTENSIONS_VID = ('mp4', 'webm')
@@ -15,7 +16,7 @@ def check_ffmpeg():
     except (FileNotFoundError, subprocess.CalledProcessError):
         return False
 
-def video_gen(input_path_file, output_path_file, TARGET_MIN_SIZE):
+def video_gen(input_path_file: str, output_path_file: str, TARGET_MIN_SIZE: int):
     vf_scale = f"scale='if(gt(iw,ih),-1,{TARGET_MIN_SIZE})':'if(gt(iw,ih),{TARGET_MIN_SIZE},-1)'"
     cmd = ["ffmpeg", "-y", "-i", input_path_file, "-ss", "00:00:00", "-vframes", "1", "-vf", vf_scale, "-q:v", "2", output_path_file]
     if os.name == 'nt': creationflags = subprocess.CREATE_NO_WINDOW
@@ -24,7 +25,7 @@ def video_gen(input_path_file, output_path_file, TARGET_MIN_SIZE):
     if result.returncode != 0:
         raise Exception(f"FFmpeg error: {result.stderr.decode('utf-8', errors='ignore')}")
 
-def image_and_gif_gen(input_path_file, output_path_file, TARGET_MIN_SIZE):
+def image_and_gif_gen(input_path_file: str, output_path_file: str, TARGET_MIN_SIZE: int):
     with Image.open(input_path_file) as img:
         img_rgba = img.convert("RGBA")
         background = Image.new("RGB", img_rgba.size, (255, 255, 255))
@@ -43,7 +44,7 @@ def image_and_gif_gen(input_path_file, output_path_file, TARGET_MIN_SIZE):
             resized_img = final_img.resize((new_width, new_height), Image.Resampling.LANCZOS)
             resized_img.save(output_path_file, "JPEG", quality=85)
 
-def generate_thumb(input_path_file, THUMB_FOLDER, TARGET_MIN_SIZE):
+def generate_thumb(input_path_file: str, THUMB_FOLDER: str, TARGET_MIN_SIZE: int):
     ext_file = input_path_file.split(".")[-1]
     if ext_file not in VALID_EXTENSIONS:
         return {"status":False, "message":"Расширение не поддерживаеться!"}

system_module/logger.py

@@ -18,7 +18,7 @@ class Logger:
         self.logger_cli.setLevel(logging.INFO)
         self.logger_cli.addHandler(console_handler)
 
-    def send(self, level_log, type_log, message):
+    def send(self, level_log: str, type_log: str, message: str):
         if type_log == "file":
             getattr(self.logger_file, level_log)(message)
         elif type_log == "cli":