Quico/kodi-nfo-feeder
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Compare commits

base: Quico:7bcf3665817108728d248b3cbe30387654ad6b07
Branches Tags
Quico:master
..
compare: Quico:ec05611ee349b79038ba32714db73cc9d098633b
Branches Tags
Quico:master

No commits in common. "7bcf3665817108728d248b3cbe30387654ad6b07" and "ec05611ee349b79038ba32714db73cc9d098633b" have entirely different histories.
7bcf366581 ... ec05611ee3

4 changed files with 18 additions and 206 deletions
Show Stats Expand all files Collapse all files

14
config.ini
View File

@ -1,15 +1,3 @@
[DEFAULT]
self_name = kodi-nfo-feeder
ignored_target_file_exts = .jpg, .jpeg, .png, .nfo
title_regex_search =
title_regex_replace =
[maus]
watch_dir = /var/lib/kodi-nfo-feeder/watch
output_dir = /var/lib/kodi-nfo-feeder/output
# title_regex_search = (\w)-(\s)
# title_regex_replace = \1 -\2
[otto]
watch_dir = /var/lib/kodi-nfo-feeder/test/watch
output_dir = /var/lib/kodi-nfo-feeder/test/output
output_dir = /var/lib/kodi-nfo-feeder/output

205
kodi-nfo-feeder.py
View File

@ -2,26 +2,16 @@ import os
import logging
import sys
import time
import re
import shutil
from rich.logging import RichHandler
from rich.traceback import install
import configparser
import inflect
from inotify_simple import INotify, flags
import lxml.etree
import lxml.builder
# TODO Create season subdir if it doesn't exist
# TODO Thread config sections
# Exit codes
# 1: Config file invalid, it has no sections
# 2: Config file invalid, sections must define at least CONST.CFG_MANDATORY
# 3: Watch directory does not exist and unable to create
# 4: Unable to create output directory
class CONST(object):
@ -32,15 +22,11 @@ class CONST(object):
CFG_DEFAULT_ABS_PATH = os.path.join(CFG_THIS_FILE_DIRNAME, CFG_DEFAULT_FILENAME)
CFG_KNOWN_DEFAULTS = [
{"key": "self_name", "value": "kodi-nfo-feeder"},
{"key": "ignored_target_file_exts", "value": ".jpg, .jpeg, .png, .nfo"},
{"key": "title_regex_search", "value": ""},
{"key": "title_regex_replace", "value": ""}
{"key": "watch_dir", "value": os.path.join(CFG_THIS_FILE_DIRNAME, "data/var/lib/%(self_name)s/watch")}
]
CFG_KNOWN_SECTION = [
{"key": "watch_dir", "is_mandatory": True},
{"key": "output_dir", "is_mandatory": True},
{"key": "title_regex_search", "is_mandatory": False},
{"key": "title_regex_replace", "is_mandatory": False}
{"key": "output_dir", "is_mandatory": True}
]
CFG_MANDATORY = [section_cfg["key"] for section_cfg in CFG_KNOWN_SECTION if section_cfg["is_mandatory"]]
@ -85,8 +71,6 @@ ini_defaults = []
internal_defaults = {default["key"]: default["value"] for default in CONST.CFG_KNOWN_DEFAULTS}
config = ConfigParser(defaults=internal_defaults)
config.read(CONST.CFG_DEFAULT_FILENAME)
ignored_target_file_exts_str = config.get(config.default_section, "ignored_target_file_exts")
ignored_target_file_exts = re.split(r""",\s?|\s""", ignored_target_file_exts_str)
def print_section_header(
@ -164,151 +148,11 @@ def setup_watch(
log.debug(f"Watching for files moved to '{watch_this}' ...")
except FileNotFoundError:
log.error(f"Watch directory '{watch_this}' does not exist. Please create it. Exiting 3 ...")
sys.exit(3)
sys.exit(4)
else:
return inotify
def generate_nfo(
title: str,
raw_file_name: str) -> lxml.builder.ElementMaker:
season_ep_str = raw_file_name.split(" - ")[0]
data = lxml.builder.ElementMaker()
ep_details_tag = data.episodedetails
title_tag = data.title
id_tag = data.id
nfo_data = ep_details_tag(
title_tag(title),
id_tag(f"{season_ep_str} - {title}")
)
return nfo_data
def get_basic_cleaned_title(
section_name: str,
config_obj: configparser.ConfigParser(),
dirty_title: str) -> str:
regex_search_pattern = config_obj.get(section_name, "title_regex_search")
regex_replace_pattern = config_obj.get(section_name, "title_regex_replace")
if regex_search_pattern:
log.debug(regex_search_pattern)
log.debug(f"Doing basic title cleaning ...")
pattern = re.compile(regex_search_pattern)
clean_title = re.sub(pattern, regex_replace_pattern, dirty_title)
log.debug(f"""Title's now "{clean_title}".""")
quit()
return clean_title
else:
return dirty_title
def get_season_and_episode(
section_name: str,
config_obj: configparser.ConfigParser(),
raw_file_name: str) -> dict:
file_name_ext_split = os.path.splitext(raw_file_name)
season_ep_str = file_name_ext_split[0].split(" - ", 1)
ext = file_name_ext_split[1]
season_episode = re.split("[S|E]", season_ep_str[0])
season = f"Season {season_episode[1]}"
title = season_ep_str[1]
basic_cleaned_title = get_basic_cleaned_title(section_name, config_obj, title)
got_season_and_episode = {
"season_str": season,
"title_str": basic_cleaned_title,
"season_ep_list": season_ep_str,
"ext": ext
}
log.debug(f"""Identified {got_season_and_episode["season_str"]}, """
f"""title "{got_season_and_episode["title_str"]}" """
f"and episode object {season_ep_str} "
f"with extension '{ext}'.")
return got_season_and_episode
def get_target_file_list(
target_dir: str) -> list:
log.debug(f"Generating list of files in '{target_dir}' ...")
onlyfiles = [f for f in os.listdir(target_dir) if os.path.isfile(os.path.join(target_dir, f))]
filtered_files = [f for f in onlyfiles if not f.endswith(tuple(ignored_target_file_exts))]
log.debug(f"Files in '{target_dir}' filtered for extensions we're ignoring "
f"{ignored_target_file_exts}: {filtered_files}")
return filtered_files
def move_file_to_target_dir(
section_name: str,
config_obj: configparser.ConfigParser(),
raw_file_name: str,
season_ep_str: dict) -> str:
this_watch_dir = config_obj.get(section_name, "watch_dir")
source_abs_path = os.path.join(this_watch_dir, raw_file_name)
target_dir = config_obj.get(section_name, "output_dir")
target_file_list = get_target_file_list(target_dir)
target_file_name = season_ep_str["season_ep_list"][0]
target_ext = season_ep_str["ext"]
target_file_name_plus_ext = f"{target_file_name}{target_ext}"
if target_file_name_plus_ext in target_file_list:
log.debug(f"File name already exists in target dir, incrementing counter ...")
episode_minus_counter = target_file_name[:-2]
counter = target_file_name[-2:]
counter_length = len(counter)
counter_stripped = int(counter.lstrip("0"))
counter_stripped += 1
target_file_name = f"{episode_minus_counter}{str(counter_stripped).zfill(counter_length)}"
target_abs_path = os.path.join(target_dir, f"{target_file_name}{target_ext}")
try:
log.debug(f"Moving '{source_abs_path}' to '{target_abs_path}' ...")
# shutil.move(source_abs_path, target_abs_path)
except OSError as ose:
log.error(f"Failed moving file with an OSError:\n"
f"{ose}\n"
f"Continuing file watch ...")
return ""
else:
return target_file_name
def write_nfo_to_disk(
nfo_data: lxml.builder.ElementMaker,
target_file_name: str,
output_dir_name: str) -> bool:
target_dir = output_dir_name
target_nfo_name = f"{target_file_name}.nfo"
target_abs_path = os.path.join(target_dir, target_nfo_name)
nfo_str = lxml.etree.tostring(
nfo_data,
pretty_print=True,
xml_declaration=True,
standalone=True,
encoding="UTF-8")
try:
with open(target_abs_path, 'wb') as nfo_file:
log.debug(f"Writing NFO data to '{target_abs_path}':\n"
f"""{nfo_str.decode("UTF-8")}""")
nfo_file.write(nfo_str)
except OSError as ose:
log.error(f"Failed writing NFO file '{target_abs_path}' with an OSError:\n"
f"{ose}\n"
f"Continuing file watch ...")
return False
else:
return True
if __name__ == '__main__':
validate_default_section(config)
if config_has_valid_section(config):
@ -316,8 +160,8 @@ if __name__ == '__main__':
else:
log.error(f"No valid config section found. A valid config section has at least the mandatory "
f"""{p.plural("option", len(CONST.CFG_MANDATORY))} """
f"{CONST.CFG_MANDATORY} set. Exiting 2 ...")
sys.exit(2)
f"{CONST.CFG_MANDATORY} set. Exiting 1 ...")
sys.exit(1)
log.debug(f"Iterating over config sections ...")
for section in config.sections():
@ -325,32 +169,15 @@ if __name__ == '__main__':
watch_dir = config.get(section, "watch_dir")
inotify_watch = setup_watch(watch_dir)
output_dir = config.get(section, "output_dir")
try:
os.makedirs(output_dir, exist_ok=True)
except OSError as ose:
log.error(f"Unable to create section '[{section}]' output dir '{output_dir}' with an OSError:\n"
f"{ose}\n"
f"Exiting 4 ...")
sys.exit(4)
else:
while True:
time.sleep(0.2)
for event in inotify_watch.read():
events = [str(flags) for flags in flags.from_mask(event.mask)]
if "flags.MOVED_TO" in events:
file_name = event.name
log.info(f"File '{file_name}' was moved to watch directory '{watch_dir}', processing ...")
season_and_episode = get_season_and_episode(section, config, file_name)
nfo = generate_nfo(season_and_episode["title_str"], file_name)
file_moved_to_target_dir = move_file_to_target_dir(
section,
config,
file_name,
season_and_episode)
if file_moved_to_target_dir:
write_nfo_to_disk(
nfo,
file_moved_to_target_dir,
output_dir)
while True:
time.sleep(0.2)
for event in inotify_watch.read():
events = [str(flags) for flags in flags.from_mask(event.mask)]
if "flags.MOVED_TO" in events:
file_name = event.name
log.info(f"File '{file_name}' was moved to watch directory '{watch_dir}', processing ...")
# TODO https://docs.python.org/3/library/xml.etree.elementtree.html
# TODO generate xml
# TODO: Basic string manipulation via regex in options file

3
requirements.in
View File

@ -1,4 +1,3 @@
rich
inflect
inotify_simple
lxml
inotify_simple

2
requirements.txt
View File

@ -10,8 +10,6 @@ inflect==5.4.0
# via -r requirements.in
inotify-simple==1.3.5
# via -r requirements.in
lxml==4.8.0
# via -r requirements.in
pygments==2.11.2
# via rich
rich==12.0.0