KomSuite/KomGrabber
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

initial commit

Browse Source
This commit is contained in:
WorldTeacher
2025-02-27 22:41:38 +01:00
commit abb16938f8
59 changed files with 4169 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

546
src/logic/cli.py Normal file
View File

@@ -0,0 +1,546 @@
import os
import re
import shutil
import subprocess
import time
import zipfile
import jaro
from src.data.komga import KomgaAPI
from komgapi.schemas.Series import Series
from src.data.mangadex import MangadexAPI
from src.data.cache import ListCache
from src.data.Feeds.nyaasi import Nyaa
from src.logic.download import Download
from komconfig import KomConfig
import loguru
import sys
config = KomConfig()
logs = loguru.logger
logs.remove()
logs.add("komgrabber.log", level="INFO")
Komga = KomgaAPI()
md = MangadexAPI()
LINE_CLEAR = "\x1b[2K" # <-- ANSI sequence
failed_items = []
class utils:
def __init__(self) -> None:
self.dl = Download("/home/alexander/Downloads/torrents/Manga_test/")
self.file = None
self.serie = ""
self.serie_id = ""
self.series_data: Series = None
self.volumes = []
self.download_path = config.komgrabber.download_location
if "~" in self.download_path:
self.download_path = os.path.expanduser(self.download_path)
# self.allSeries = Komga.getAllSeries()
pass
def download(self, feed_url: str):
def __chapter_check(title: str) -> bool:
if title.endswith(".cbz") or title.endswith(".cbr"):
if not re.search(r"(v\d{1,3}(-\d{1,3})?)|(Vol\. \d{1,3})", title):
return True
else:
return False
def __epub_check(title: str) -> bool:
if title.endswith(".epub"):
return True
else:
return False
file: str
file = self.dl.get_file(feed_url)
if __chapter_check(file):
print(f"Skipping {file}, reason: no volume number, likely a chapter")
return False
if __epub_check(file):
print(f"Skipping {file}, reason: epub file")
return False
self.file = file
print(f"Filename: {file}")
file_move = False
if file.endswith(".cbz") or file.endswith(".cbr"):
new_folder = f"{self.download_path}{self.serie}"
os.makedirs(new_folder, exist_ok=True)
file_move = True
state = self.dl.add_torrent(feed_url.split("/")[-1])
if state is False:
print("Error adding torrent")
return False
gid = self.dl.api.get_downloads()[0].gid
# check if the download is complete usin the gid
dl_complete = True
check_done = False
while not self.dl.api.get_downloads(gids=[gid])[0].seeder:
# while not self.dl.api.get_downloads()[0].seeder:
progress = self.dl.check_progress()
progress = "{:.2f}".format(progress)
eta = self.dl.api.get_downloads()[0].eta_string()
print(end=LINE_CLEAR)
print("Progress: ", progress, "ETA: ", eta, end="\r")
# if progress remains the same for 30 seconds, stop the download
progress = self.dl.check_progress()
time.sleep(30)
n_progress = self.dl.check_progress()
if not check_done:
local_files = os.listdir(f"{self.download_path}")
for f in local_files:
# print(f)
if os.path.isdir(f"{self.download_path}/{f}"):
local_files.extend(
[
f"{self.download_path}/{f}/{file}"
for file in os.listdir(f"{self.download_path}/{f}")
]
)
local_files = [
file
for file in local_files
if file.endswith(".cbz") or file.endswith(".cbr")
]
local_volumes = Komga.getVolumes(self.series_data.id)
# if not local_files:
# dl_complete=False
# break
local_files_volumes = []
for file in local_files:
vol_regex = r"(v\d{1,3}(-\d{1,3})?)|(Vol\. \d{1,3})"
# if the file does not match the naming convention, skip it
if re.search(vol_regex, file):
match = re.search(vol_regex, file)
if match:
vol = match.group(0).replace("v", "").replace("Vol. ", "")
if "-" in vol:
local_files_volumes.extend(
[int(volume) for volume in vol.split("-")]
)
continue
vol = int(vol)
local_files_volumes.append(vol)
print(f"Grabbed volumes: {local_files_volumes}")
print(f"Komga volumes: {local_volumes}")
if local_files_volumes == []:
pass
# check íf any local_file_volumes are not in local_volumes
if all([vol in local_volumes for vol in local_files_volumes]):
print("all volumes downloaded, stopping...")
dl_complete = False
break
else:
print("not all volumes downloaded, continuing...")
check_done = True
if progress == n_progress:
print("Progress has not changed for 30 seconds, stopping the download")
self.dl.api.get_downloads()[0].remove(force=True)
dl_complete = False
break
else:
pass
# stop the download, remove the torrent files
try:
self.dl.api.get_downloads()[0].remove(force=True)
except:
pass
self.dl.remove_torrents()
print(end=LINE_CLEAR)
print("Download complete")
# self.dl.download(feed_url, file_rename=True)
if not dl_complete:
# remove everything from the download folder
data = os.listdir(f"{self.download_path}")
for file in data:
try:
os.remove(f"{self.download_path}{file}")
except IsADirectoryError:
shutil.rmtree(f"{self.download_path}{file}")
if dl_complete is True:
# for dfile in os.listdir(f'{self.download_path}{file}'):
# if __chapter_check(dfile):
# os.remove(f'{self.download_path}{file}{dfile}')
try:
if file_move is True:
shutil.move(
f"{self.download_path}{file}",
f"{new_folder}/{file}",
)
except Exception as e:
print(e)
return False
return True
return False
def tag_files(self, folder: str, interactive: bool = False):
"""Tag all files in the specified folder.
Args:
----
- folder (str): the path to the folder containing the files to tag
- interactive (bool, optional): if set to True, the shell will pause and await user input instead of not writing data to file. Defaults to False.
"""
def is_valid_cbz(file_path) -> bool:
try:
with zipfile.ZipFile(file_path, "r") as cbz_file:
# Check if the file is a valid ZIP archive
if cbz_file.testzip() is not None:
return False
# Check if the CBZ file contains at least one image file
for file_info in cbz_file.infolist():
if (
not file_info.is_dir()
and file_info.filename.lower().endswith(
(".jpg", ".jpeg", ".png")
)
):
return True
return False
except (zipfile.BadZipFile, FileNotFoundError):
return False
for file in os.listdir(f"{folder}"):
print(f"Checking {file}")
# if file is a not cbz file, skip
if not file.endswith(".cbz"):
print(f"Skipping {file}")
continue
try:
# if not is_valid_cbz(f"{folder}/{file}"):
# print(f"removing {file}, not a valid cbz file")
# os.remove(f"{folder}/{file}")
# continue
print(f"Tagging {file}")
regex = r"v(\d{2,3}) #(\d{2,3})"
match = re.search(regex, file)
if not match:
print(f"Skipping {file}, no match")
os.remove(f"{folder}/{file}")
continue
if interactive:
subprocess.call(
f'comictagger -s -t cr -f -o "{folder}/{file}" --nosummary --overwrite -i',
shell=True,
)
subprocess.call(
f'comictagger -s -t cr -f -o "{folder}/{file}" --nosummary --overwrite',
shell=True,
)
print(f"Tagged {file}")
except Exception as e:
print(e)
continue
def rename_folder_and_files(self, file: str, komga_data, remove=False):
logs.info(f"Renaming {file}")
# rename the folder to the komga name
series_id = komga_data.id
series_name = komga_data.name
new_folder = f"{self.download_path}{series_name}"
try:
os.rename(f"{self.download_path}{file}", new_folder)
except Exception as e:
print(e)
try:
files = os.listdir(new_folder)
except NotADirectoryError:
return
volumes = []
for file in files:
if not (file.endswith(".cbz") or file.endswith(".cbr")):
print(f"Skipping {file}, not a comicarchive file")
continue
ext = file.split(".")[-1]
# match = re.search(r"v\d{2,4}(-\d{2,4})*", file)
match = re.search(r"v\d{2,4} ", file)
if match:
# print(match)
split_start = match.start()
split_end = match.end()
# Split the filename between split_start and split_end
volume = file[split_start:split_end]
# Split the filename at the split index, but keep the "v" and digits in the title
title = file[:split_start].strip()
# add the volume number to the title as a suffix #nr
title = f"{title} {volume} #{volume.replace('v', '')}".strip().replace(
" ", " "
)
# print(title)
# rename the file
os.rename(f"{new_folder}/{file}", f"{new_folder}/{title}.{ext}")
volumes.append(int(volume.replace("v", "")))
logs.info(f"Renamed {file} to {title}")
if remove:
print("removing files that are already in komga")
# search komga_name in series
# get all volumes of the serie
local_volumes = Komga.getVolumes(series_id=series_id)
# remove the volumes that are already in komga
self.remove_if_alr_in_db(local_volumes, volumes, series_name)
self.tag_files(new_folder)
def process_serie(self, data: Series):
"""Pprocess a single serie based on its title.
The process is as follows:
1. get all volumes of the serie from komga using the api
2. get all feed entries from nyaa.si using the api
3. compare the volumes from komga with the volumes from nyaa.si
3.1 if the volumes from nyaa.si are greater than the volumes from komga, add the entry to the download list.
Args:
----
- data (dict): a dict containing the title of the serie at ["title"] and the id of the serie at ["id"]
Returns:
-------
- list[dict]: a list of dictionaries containing the entries to download
"""
serie = data.name
series_id = data.id
vols = Komga.getVolumes(series_id=series_id, unpaged=True)
feed_titles = Nyaa.search(keyword=serie, category=3, subcategory=1)
print(feed_titles)
f_d = []
if feed_titles == []:
failed_items.append(serie)
added_max_vols = vols if vols else [0]
# #print(len(added_max_vols))
for entry in feed_titles:
if entry.seeders > 0:
if (
serie.lower() in entry.name.lower()
or jaro.jaro_metric(entry.name.lower(), serie.lower()) > 0.7
):
volumes = entry["volumes"]
if isinstance(volumes, list):
volumes = volumes[
::-1
] # reverse the list to get the highest volume number quickly
for vol in volumes:
if vol not in added_max_vols:
f_d.append(entry)
added_max_vols.append(vol)
break
return f_d
def media_grabber(self, serie: Series):
result = self.process_serie(serie)
logs.info(f"Found {len(result)} new volumes for {serie.name}")
logs.info(f"Data: {result}")
print(
f"current volumes: {Komga.getVolumes(series_id=serie.id, unpaged=True)}, new volumes: {result}"
)
# print(result)
if len(result) != 0:
for entry in result:
# print(entry["link"])
if self.download(entry["link"]) is True:
print("renaming...")
self.rename_folder_and_files(
self.file, komga_data=serie, remove=True
)
# self.move_to_komga(serie=entry)
print("done")
return True
else:
# remove the folder
try:
folders = os.listdir(self.download_path)
for folder in folders:
os.remove(f"{self.download_path}{folder}")
except Exception as e:
print(e)
return False
def remove_if_alr_in_db(
self, present_volumes: list, downloaded_volumes: list, folder: str
):
"""Delete any file from the folder that is already in the database, or does not conform to the naming convention.
Args:
----
present_volumes (list): a list of volumes that are already in the database, retrieved from komga api
downloaded_volumes (list): the list of volumes that are downloaded from the corresponding feed/api
folder (str): relative path to the folder containing the downloaded files
"""
print(f"present_volumes: {present_volumes}")
print(f"downloaded_volumes: {downloaded_volumes}")
content_folder = f"{self.download_path}{folder}"
content_files = [file for file in os.listdir(content_folder)]
print(f"content_files: {content_files}")
duplicates = [any(file in content_files for file in present_volumes)]
for file in os.listdir(content_folder):
if "#" not in file:
try:
os.remove(os.path.join(content_folder, file))
if IsADirectoryError:
shutil.rmtree(os.path.join(content_folder, file))
if FileNotFoundError:
continue
except Exception as e:
# print(e)
continue
# print(f"removed {file}, Reason: not a valid file")
content_files.remove(file)
for vol in present_volumes:
if vol < 10:
vol = f"0{vol}"
for file in content_files:
if str(vol) in file:
# print(f"removing {vol}")
try:
os.remove(os.path.join(content_folder, file))
except:
print(f"could not remove {vol}")
def move_to_komga(self, serie: tuple[str, str] = None):
komga_path = f"{config.komga.media_path}{self.serie}"
# print(f"komga_path: {komga_path}")
# print("moving to komga")
# move files to komga
for file in os.listdir(f"{self.download_path}{self.serie}"):
file_path = os.path.join(f"{self.download_path}{self.serie}", file)
final_path = os.path.join(komga_path, file)
shutil.move(file_path, final_path)
print(f"moved {file} to {komga_path}")
# delete empty folder
try:
os.rmdir(f"{self.download_path}{self.serie}")
logs.info(f"moved {self.serie} to komga")
except:
print(f"could not remove {self.serie}")
logs.error(f"could not remove {self.serie}")
return self
def search_for_new_volumes(self):
series = Komga.series_controller.getAllSeries(
body={
"condition": {
"seriesStatus": {
"operator": "is",
"value": "HIATUS",
"value": "ENDED",
}
}
}
)
shutil.rmtree(self.download_path)
os.mkdir(self.download_path)
for serie in series:
position = series.index(serie)
print("Working on serie", position, "of ", len(series))
logs.info(f"searching for new volumes for {serie.name}")
print(serie.name)
self.series_data = serie
self.serie = serie.name
self.serie_id = serie.id
if self.media_grabber(serie) is True:
self.move_to_komga(serie)
time.sleep(5)
# print("done", serie.name)
return self
def add_missing_to_db(self):
database_series = ListCache("mangacache.db").get_all_series("name")
database_series = [serie[0] for serie in database_series]
database_set = set(database_series)
# print(database_series)
komga_series = Komga.series_controller.getAllSeries()
db_added = []
for serie in komga_series:
if serie.id not in database_set:
# print(serie.id)
db_added.append(serie)
ListCache("mangacache.db").add_series(
serie.id, serie.name, serie.metadata.status
)
else:
print(f"{serie.id} already in db")
print("added to db:", len(db_added))
# print(f"{serie[1]} has status {komga_series}")
@DeprecationWarning
def get_md_metadata(self, id: str):
data = md.get_metadata(id, lang="en")
db_data = ListCache("mangacache.db").get_series_by_id(id, "mangadex_id")
def automated(self, series_data: tuple[str, str]):
"""_summary_.
Args:
----
series_data (list[tuple[str,str]]): _description_
"""
if self.media_grabber(series_data) is True:
self.move_to_komga(series_data)
time.sleep(5)
def parallel_execution(series: list[tuple[str, str, str]]):
"""_summary_.
Args:
----
series (list[tuple[str,str,str]]): _description_
"""
th = utils()
for serie in series:
th.automated(serie)
@DeprecationWarning
def update_state():
database_series = ListCache("mangacache.db").get_all_series()
database_series = [serie for serie in database_series if serie[3] != "ENDED"]
for serie in database_series:
komga_series = Komga.getSeriesStatus(serie)
if komga_series == "ONGOING":
continue
else:
ListCache("mangacache.db").update_database(
komga_id=serie[2], complete=komga_series
)
def avail_check():
komga_avail = True
return (True, komga_avail)
def main():
utils().search_for_new_volumes()
# update_state()
print("Failed series:\n", failed_items)
if __name__ == "__main__":
utils().search_for_new_volumes()

231
src/logic/constants.py Normal file
View File

@@ -0,0 +1,231 @@
LINK_TRANSFORM = {"al":"https://anilist.co/manga/",
"ap":"https://www.anime-planet.com/manga/",
"bw":"https://bookwalker.jp/",
"kt":"https://kitsu.io/manga/",
"mu":"https://www.mangaupdates.com/series.html?id=",
"mal":"https://myanimelist.net/manga/"}
LANG_CODES = [
"ab",
"aa",
"af",
"ak",
"sq",
"am",
"ar",
"an",
"hy",
"as",
"av",
"ae",
"ay",
"az",
"bm",
"ba",
"eu",
"be",
"bn",
"bi",
"bs",
"br",
"bg",
"my",
"ca",
"ch",
"ce",
"ny",
"zh",
"cu",
"cv",
"kw",
"co",
"cr",
"hr",
"cs",
"da",
"dv",
"nl",
"dz",
"en",
"eo",
"et",
"ee",
"fo",
"fj",
"fi",
"fr",
"fy",
"ff",
"gd",
"gl",
"lg",
"ka",
"de",
"el",
"kl",
"gn",
"gu",
"ht",
"ha",
"he",
"hz",
"hi",
"ho",
"hu",
"is",
"io",
"ig",
"id",
"ia",
"ie",
"iu",
"ik",
"ga",
"it",
"ja",
"jv",
"kn",
"kr",
"ks",
"kk",
"km",
"ki",
"rw",
"ky",
"kv",
"kg",
"ko",
"kj",
"ku",
"lo",
"la",
"lv",
"li",
"ln",
"lt",
"lu",
"lb",
"mk",
"mg",
"ms",
"ml",
"mt",
"gv",
"mi",
"mr",
"mh",
"mn",
"na",
"nv",
"nd",
"nr",
"ng",
"ne",
"no",
"nb",
"nn",
"ii",
"oc",
"oj",
"or",
"om",
"os",
"pi",
"ps",
"fa",
"pl",
"pt",
"pa",
"qu",
"ro",
"rm",
"rn",
"ru",
"se",
"sm",
"sg",
"sa",
"sc",
"sr",
"sn",
"sd",
"si",
"sk",
"sl",
"so",
"st",
"es",
"su",
"sw",
"ss",
"sv",
"tl",
"ty",
"tg",
"ta",
"tt",
"te",
"th",
"bo",
"ti",
"to",
"ts",
"tn",
"tr",
"tk",
"tw",
"ug",
"uk",
"ur",
"uz",
"ve",
"vi",
"vo",
"wa",
"cy",
"wo",
"xh",
"yi",
"yo",
"za",
"zu",
]
READING_DIRECTIONS = ["Left to Right", "Right to Left", "Vertical", "Webtoon"]
READING_DIRECTONS_KOMGA = ["LEFT_TO_RIGHT", "RIGHT_TO_LEFT", "VERTICAL", "WEBTOON"]
READING_DIR_TRANSLATION = {
READING_DIRECTIONS[0]: READING_DIRECTONS_KOMGA[0],
READING_DIRECTIONS[1]: READING_DIRECTONS_KOMGA[1],
READING_DIRECTIONS[2]: READING_DIRECTONS_KOMGA[2],
READING_DIRECTIONS[3]: READING_DIRECTONS_KOMGA[3],
}
METADATA_PROVIDERS = ["MangaDex", "ComicVine", "AniList", "MyAnimeList", "Comics.org"]
SERIES_STATUS = ["---", "Ongoing", "Ended", "Hiatus", "Abandoned"]
SERIES_STATUS_KOMGA = ["UNKNOWN", "ONGOING", "ENDED", "HIATUS", "ABANDONED"]
SERIES_STATUS_TRANSLATION = {
SERIES_STATUS[0]: SERIES_STATUS_KOMGA[0],
SERIES_STATUS[1]: SERIES_STATUS_KOMGA[1],
SERIES_STATUS[2]: SERIES_STATUS_KOMGA[2],
SERIES_STATUS[3]: SERIES_STATUS_KOMGA[3],
SERIES_STATUS[4]: SERIES_STATUS_KOMGA[4],
}
def translate_series_status(status: str) -> str:
if status in SERIES_STATUS_TRANSLATION.keys():
return SERIES_STATUS_TRANSLATION[status]
else:
#get the key from the value
for key, value in SERIES_STATUS_TRANSLATION.items():
if value == status:
return key
def translate_reading_direction(direction: str) -> str:
if direction in READING_DIR_TRANSLATION.keys():
return READING_DIR_TRANSLATION[direction]
else:
#get the key from the value
for key, value in READING_DIR_TRANSLATION.items():
if value == direction:
return key

0
src/logic/data.py Normal file
View File

16
src/logic/detect_chapters.py Normal file
View File

@@ -0,0 +1,16 @@
import os
import re
def detect_chapters(src: str = "/home/alexander/Downloads/torrents/manga/"):
for folder in os.listdir(src):
if os.path.isdir(f"{src}/{folder}"):
files = os.listdir(f"{src}/{folder}")
for file in files:
# check for regex "v(d) #(d)" in the file name
regex = re.compile(r"^.* v(\d+) #(\d+(?:-\d+)?)\.cbz$")
if regex.search(file):
print(f"File {file} is a Volume")
else:
print(f"Deleting chapter {file}")
os.remove(f"{src}/{folder}/{file}")

99
src/logic/download.py Normal file
View File

@@ -0,0 +1,99 @@
import sys
import os
import time
import bencodepy
from .rename import rename
from aria2p import Client
from aria2p import API
class Download:
""" Download a file from a url and start the download using aria2"""
def __init__(self, download_location) -> None:
self.download_location=download_location
self.filename=None
self.torrent_file=None
self.progress=0
self.canceled=False
self.aria2_running=self.check_aria2()
self.api = API(
client=Client(
host="http://localhost",
port=6800,
secret="",
timeout=60,
)
)
self.api.set_global_options({"dir": self.download_location})
if not self.aria2_running:
print("Aria2 is not running")
sys.exit()
def check_aria2(self):
#check if aria2 is running
if os.system("ps -A | grep aria2c") == 0:
return True
else:
return False
def check_progress(self):
try:
current_progress=self.api.get_downloads()[0].progress
except:
return self.progress+0.01
if current_progress > self.progress:
self.progress=current_progress
return current_progress
def get_file(self, url, series_name=None):
#get the file name from the url
#use wget to download the file to the download location
name=url.split('/')[-1]
dl_url=f'{self.download_location}{name}'
while self.get_filename(dl_url) is None:
if not os.path.exists(dl_url):
os.system(f'wget -P {self.download_location} {url}')
filename = self.get_filename(dl_url)
self.torrent_file=url.split('/')[-1]
self.filename=filename
return filename
def remove_torrents(self):
tr_files=[file for file in os.listdir(self.download_location) if ".torrent" in file]
for file in tr_files:
os.remove(f'{self.download_location}{file}')
def add_torrent(self, torr_name):
try:
self.api.add_torrent(f'{self.download_location}{torr_name}')
print("Torrent added")
except Exception as e:
print(f"Error adding torrent: {e}")
return False
def rename_download(self):
filename=self.filename.replace(".aria2", "")
foldername=filename.replace(".cbz", "") if ".cbz" in filename else filename
print(f'Filename: {filename}')
print(f'Foldername: {foldername}')
if not os.path.exists(f'{self.download_location}{foldername}'):
os.mkdir(f'{self.download_location}{foldername}')
os.rename(f'{self.download_location}{filename}', f'{self.download_location}{foldername}/{filename}')
#rename the file
rename(f'{self.download_location}{foldername}')
def get_filename(self, torrent_file):
try:
with open(torrent_file, 'rb') as f:
torrent = bencodepy.decode(f.read())
#self.filename=torrent[b'info'][b'name'].decode('utf-8')
return torrent[b'info'][b'name'].decode('utf-8')
except FileNotFoundError:
return None

53
src/logic/manual.py Normal file
View File

@@ -0,0 +1,53 @@
import os
import re
def _vol_list() -> list[int]:
path = "/home/alexander/Downloads/torrents/Manga_test/"
# get all files in the dir and subdirs
files = os.listdir(path)
for f in files:
print(f)
if os.path.isdir(f"{path}/{f}"):
print("is dir")
files.extend([f"{path}/{f}/{file}" for file in os.listdir(f"{path}/{f}")])
return files
def isdircheck(path: str) -> bool:
if os.path.isdir(path):
return True
else:
return False
def __chapter_check(title: str) -> bool:
if title.endswith(".cbz") or title.endswith(".cbr"):
if not re.search(r"(v\d{1,3}(-\d{1,3})?)|(Vol\. \d{1,3})", title):
return True
else:
return False
def check_folder(folder):
files = os.listdir(folder)
for file in files:
if os.path.isdir(f"{folder}/{file}"):
print(f"{file} is a dir")
check_folder(f"{folder}/{file}")
else:
print(f"{file} is a file")
if __chapter_check(file):
print(f"{file} is a chapter")
os.remove(f"{folder}/{file}")
else:
print(f"{file} is a volume")
if __name__ == "__main__":
# # print(__chapter_check("Even Given the Worthless 'Appraiser' Class, I'm Actually the Strongest v01-09+073 (2023) (Digital) (danke-Empire).cbz"))
# __vol_list()
# # print(isdircheck("/home/alexander/Downloads/torrents/Manga_test/Peter Grill and the Philosopher's Time (Digital)/"))
check_folder("/home/alexander/Downloads/torrents/Manga_test/")

42
src/logic/move.py Normal file
View File

@@ -0,0 +1,42 @@
import os
import shutil
def move(src, dest: str = "/mnt/Media/Manga"):
"""Move the files in a folder to another folder.
Args:
----
- dest (str): the string to the destination folder
"""
# Get the files in the folder
# +move the folders from src to disc, if folder already exists, only move new files
folders = os.listdir(src)
for folder in folders:
if not os.path.exists(f"{dest}/{folder}"):
print(f"Moving {folder} to {dest}")
shutil.move(f"{src}/{folder}", dest)
else:
files = os.listdir(f"{src}/{folder}")
for file in files:
if not os.path.exists(f"{dest}/{folder}/{file}"):
print(f"Moving {file} to {dest}/{folder}")
shutil.move(f"{src}/{folder}/{file}", f"{dest}/{folder}")
# Remove empty folders
remove_empty_folders(src)
def remove_empty_folders(src):
"""Remove empty folders from a directory.
Args:
----
- src (str): the string to the source folder
"""
folders = os.listdir(src)
for folder in folders:
if os.path.isfile(f"{src}/{folder}"):
continue
if not os.listdir(f"{src}/{folder}"):
print(f"Removing {folder}")
os.rmdir(f"{src}/{folder}")
else:
remove_empty_folders(f"{src}/{folder}")

45
src/logic/move_test.py Normal file
View File

@@ -0,0 +1,45 @@
import os
import re
folder_path = "/home/alexander/Downloads/torrents/Manga_test/"
def rename(folder):
"""Rename the files in a folder according to the template.
Template: [Name] v[nr] #[nr].ext (e.g. "The Flash v1 #1.cbz").
Args:
----
- folder (str): the string to the folder
"""
# Get the files in the folder
files = os.listdir(folder)
for file in files:
if not file.endswith(".cbz"):
print(f"Skipping {file}, not a cbz file")
continue
ext = file.split(".")[-1]
match = re.search(r"v\d{2,4} ", file)
if match:
print(match)
split_start = match.start()
split_end = match.end()
# Split the filename between split_start and split_end
volume = file[split_start:split_end]
# Split the filename at the split index, but keep the "v" and digits in the title
title = file[:split_start].strip()
# add the volume number to the title as a suffix #nr
title = f"{title} {volume} #{volume.replace('v','')}"
print(title)
# rename the file
os.rename(f"{folder}/{file}", f"{folder}/{title}.{ext}")
for folder in os.listdir(folder_path):
if os.path.isdir(f"{folder_path}/{folder}"):
rename(f"{folder_path}/{folder}")
print(f"Renamed {folder}")
else:
print(f"{folder} is not a folder")
continue

7
src/logic/pickles.py Normal file
View File

@@ -0,0 +1,7 @@
import pickle
def make_pickle(obj):
return pickle.dumps(obj)
def load_pickle(pickled_obj):
return pickle.loads(pickled_obj)

51
src/logic/rename.py Normal file
View File

@@ -0,0 +1,51 @@
# Rename the downloaded files according to the template
# Template: [Name] v[nr] #[nr].ext (e.g. "The Flash v1 #1.cbz")
import os
import re
def rename(folder: str = "/home/alexander/Downloads/torrents/manga/") -> None:
"""Rename the files in a folder according to the template.
Template: [Name] v[nr] #[nr].ext (e.g. "The Flash v1 #1.cbz").
Args:
----
- folder (str): the string to the folder
"""
# Get the files in the folder
files = os.listdir(folder)
print(files)
for file in files:
if os.path.isdir(f"{folder}/{file}"):
rename(f"{folder}/{file}")
if not file.endswith(".cbz"):
print(f"Skipping {file}, not a cbz file")
continue
ext = file.split(".")[-1]
match = re.search(r"v\d{2,4}", file)
if match:
split_start = match.start()
split_end = match.end()
# Split the filename between split_start and split_end
volume = file[split_start:split_end]
# Split the filename at the split index, but keep the "v" and digits in the title
title = file[:split_start].strip()
# add the volume number to the title as a suffix #nr
title = f"{title} {volume} #{volume.replace('v', '')}"
print(title)
# rename the file
os.rename(f"{folder}/{file}", f"{folder}/{title}.{ext}")
# rename the folder
def rename_recursive(folder: str) -> None:
# get all directories in the folder and apply the rename function to them
for root, dirs, _files in os.walk(folder):
for dir in dirs:
rename(f"{root}/{dir}")
if __name__ == "__main__":
rename_recursive("/home/alexander/Downloads/torrents/manga/")

27
src/logic/tag.py Normal file
View File

@@ -0,0 +1,27 @@
import os
from pathlib import Path
import subprocess
def tag_folder(
folder: Path = Path("/home/alexander/Downloads/torrents/manga/"),
) -> None:
"""Tag the files in a folder according to the template.
Template: [Name] v[nr] #[nr].ext (e.g. "The Flash v1 #1.cbz").
Args:
----
- folder (Path): the string to the folder
"""
# Get the files in the folder
files = os.listdir(folder)
for file in files:
if os.path.isdir(f"{folder}/{file}"):
tag_folder(f"{folder}/{file}")
if not file.endswith(".cbz"):
continue
print("Trying to tag file", file)
subprocess.call(
f'comictagger -s -t cr -f -o "{folder}/{file}" --nosummary --overwrite -i',
shell=True,
)

190
src/logic/testing.py Normal file
View File

@@ -0,0 +1,190 @@
import json
import os
import re
# import a library to open zip files
import zipfile
from jellyfish import jaro_similarity
from APIs import KomgaAPI, MangadexAPI, NyaaFeed
from APIs.cache import ListCache
Komga = KomgaAPI()
config = json.load(open("config.json"))
cat = NyaaFeed()
# with open("compare1.json") as f:
# data = json.load(f)
# with open("compare2.json") as f:
# data2 = json.load(f)
def compare(data1, data2):
# compare the two data sets and return a list of differences
differences = []
for key in data1:
if key in data2:
if data1[key] != data2[key]:
differences.append(key)
else:
differences.append(key)
return differences
# diffs=compare(data, data2)
# #get the differences from the first data set
# for diff in diffs:
# print(diff, data[diff])
# #get the differences from the second data set
# for diff in diffs:
# print(diff, data2[diff])
def check_presence_of_xml_file(filename: str):
with zipfile.ZipFile(filename, "r") as zip_ref:
return "ComicInfo.xml" in zip_ref.namelist()
def create_xml_file(filename: str):
with zipfile.ZipFile(filename, "r") as zip_ref:
return zip_ref.read("ComicInfo.xml")
def rename_files(komga_data: str):
"""Rename the files in the folder to the komga name with the volume number.
Args:
----
file (str): pth to the folder
komga_data (str): series name
"""
# rename the folder to the komga name
new_folder = f'{config["download_location"]}/{komga_data}'
try:
files = os.listdir(new_folder)
except FileNotFoundError:
return
for file in files:
if not (file.endswith(".cbz") or file.endswith(".cbr")):
print(f"Skipping {file}, not a cbz file")
continue
ext = file.split(".")[-1]
match = re.search(r"v\d{2,4}(-\d{2,4})*", file)
if match:
print(match.group(0))
vol = match.group(0).replace("v", "")
title = file.split(match.group(0))[0]
title = title.lstrip().rstrip()
new_filename = f"{title} v{vol} #{vol}.{ext}"
print(new_filename)
os.rename(f"{new_folder}/{file}", f"{new_folder}/{new_filename}")
# try:
# os.rename(f"{new_folder}/{file}", f"{new_folder}/{filename} v{vol} #{vol}.{ext}")
# except FileNotFoundError:
# print(f"File not found: {file}")
# split_index = match.start()
# #split the title after the split_index
# title=f"{file[:split_index]}{file[:split_index+len(match.group(0)):]}"
# print(title)
# volumes = match.group(0).split("-")
# volumes = [volume.replace("v", "") for volume in volumes]
# volume_data="-".join(volumes)
# # print(volume)
# # volumes.append(int(volume))
# # #add the volume number to the title as a suffix #nr
# title=f"{title} #{volume_data}"
# #rename the file
# os.rename(f"{new_folder}/{file}", f"{new_folder}/{title}.{ext}")
def __chapter_check(title: str) -> bool:
if title.endswith(".cbz") or title.endswith(".cbr"):
if not re.search(r"(v\d{1,3}(-\d{1,3})?)|(Vol\. \d{1,3})", title):
return True
else:
return False
def check_folder(folder):
files = os.listdir(folder)
for file in files:
if os.path.isdir(f"{folder}/{file}"):
print(f"{file} is a dir")
check_folder(f"{folder}/{file}")
else:
print(f"{file} is a file")
if __chapter_check(file):
print(f"{file} is a chapter")
os.remove(f"{folder}/{file}")
else:
print(f"{file} is a volume")
def add_ids():
def __determine_similarity(search_string: str, given_string: str) -> float:
return jaro_similarity(search_string, given_string)
database = ListCache("mangacache.db")
Ma = MangadexAPI()
result = database.query_all_missing_id_type("mangadex")
print(len(result))
max_sim = 0
manga_id = None
for series in result:
title = series[1]
mangadex_id = Ma.get_manga_id(title)
if type(mangadex_id) == tuple:
print("result is a tuple")
similarity = __determine_similarity(mangadex_id["title"], title)
if similarity > max_sim:
max_sim = similarity
manga_id = mangadex_id["id"]
for alt_title in mangadex_id["alternate_titles"]:
similarity = __determine_similarity(alt_title, title)
if similarity > max_sim:
max_sim = similarity
manga_id = mangadex_id["id"]
# print(mangadex_id)
elif type(mangadex_id) == list:
print("result is a list")
# print(mangadex_id)
for res_title in mangadex_id:
similarity = __determine_similarity(res_title["title"], title)
if similarity > max_sim:
max_sim = similarity
manga_id = res_title["id"]
for alt_title in res_title["alternate_titles"]:
similarity = __determine_similarity(alt_title, title)
if similarity > max_sim:
max_sim = similarity
manga_id = res_title["id"]
else:
print(mangadex_id)
print(manga_id)
if __name__ == "__main__":
# series_names=Komga.get_all_series()
# for series in series_names:
# print(series[0])
# rename_files(series[0])
folders = os.listdir(config["download_location"])
for folder in folders:
print(folder)
check_folder(f'{config["download_location"]}/{folder}')
rename_files(f'{config["download_location"]}/{folder}')
# rename_files(komga_data="Hell's Paradise - Jigokuraku")
# add_ids()

28
src/logic/threads.py Normal file
View File

@@ -0,0 +1,28 @@
from gui import SeriesSelectDialog
from PySide6 import QtWidgets, QtCore, QtGui
#import thread capabilties
from PySide6.QtCore import QThread, Signal
class SeriesThread(QThread):
"""Thread to get the series from the api.
"""
#signal to send the series data to the main thread
series = Signal(list)
def __init__(self, api, series_name) -> None:
super().__init__()
self.api = api
self.series_name = series_name
def run(self):
#display the SeriesSelectDialog
print("running thread")
dialog = QtWidgets.QDialog()
ui = SeriesSelectDialog(self.series_name, self.api)
ui.setupUi(dialog)
dialog.show()
dialog.exec()
#send the data back to the main thread
self.series.emit(ui.data)