lysis/pyuserconfig
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Compare commits

base: lysis:875e6a4831449f4613dd4fcc2fbe62b311f2f69b
Branches Tags
lysis:main
..
compare: lysis:a986a708000ca1948e3654abc5a71157d5a8b93e
Branches Tags
lysis:main

No commits in common. "875e6a4831449f4613dd4fcc2fbe62b311f2f69b" and "a986a708000ca1948e3654abc5a71157d5a8b93e" have entirely different histories.
875e6a4831 ... a986a70800

13 changed files with 420 additions and 380 deletions
Show Stats Expand all files Collapse all files

19
.idea/inspectionProfiles/Project_Default.xml generated Normal file
View File

@ -0,0 +1,19 @@
<component name="InspectionProjectProfileManager">
<profile version="1.0">
<option name="myName" value="Project Default" />
<inspection_tool class="PyPep8Inspection" enabled="true" level="WEAK WARNING" enabled_by_default="true">
<option name="ignoredErrors">
<list>
<option value="E501" />
</list>
</option>
</inspection_tool>
<inspection_tool class="PyShadowingBuiltinsInspection" enabled="true" level="WEAK WARNING" enabled_by_default="true">
<option name="ignoredNames">
<list>
<option value="format" />
</list>
</option>
</inspection_tool>
</profile>
</component>

7
.idea/misc.xml generated Normal file
View File

@ -0,0 +1,7 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="Black">
<option name="sdkName" value="Python 3.6 (pyuserconfig)" />
</component>
<component name="ProjectRootManager" version="2" project-jdk-name="Python 3.12 (userconfig)" project-jdk-type="Python SDK" />
</project>

8
.idea/modules.xml generated Normal file
View File

@ -0,0 +1,8 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="ProjectModuleManager">
<modules>
<module fileurl="file://$PROJECT_DIR$/.idea/pyuserconfig.iml" filepath="$PROJECT_DIR$/.idea/pyuserconfig.iml" />
</modules>
</component>
</project>

13
.idea/pyuserconfig.iml generated Normal file
View File

@ -0,0 +1,13 @@
<?xml version="1.0" encoding="UTF-8"?>
<module type="PYTHON_MODULE" version="4">
<component name="NewModuleRootManager">
<content url="file://$MODULE_DIR$">
<excludeFolder url="file://$MODULE_DIR$/venv" />
</content>
<orderEntry type="inheritedJdk" />
<orderEntry type="sourceFolder" forTests="false" />
</component>
<component name="TestRunnerService">
<option name="PROJECT_TEST_RUNNER" value="Unittests" />
</component>
</module>

40
.idea/shelf/Uncommitted_changes_before_Checkout_at_29_03_2024_21_55_[Changes]/shelved.patch generated Normal file
View File

@ -0,0 +1,40 @@
Index: .idea/pyuserconfig.iml
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.BaseRevisionTextPatchEP
<+><?xml version=\"1.0\" encoding=\"UTF-8\"?>\n<module type=\"PYTHON_MODULE\" version=\"4\">\n <component name=\"NewModuleRootManager\">\n <content url=\"file://$MODULE_DIR$\" />\n <orderEntry type=\"jdk\" jdkName=\"Python 3.7 (pyuserconfig)\" jdkType=\"Python SDK\" />\n <orderEntry type=\"sourceFolder\" forTests=\"false\" />\n </component>\n <component name=\"TestRunnerService\">\n <option name=\"PROJECT_TEST_RUNNER\" value=\"Unittests\" />\n </component>\n</module>
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/.idea/pyuserconfig.iml b/.idea/pyuserconfig.iml
--- a/.idea/pyuserconfig.iml (revision cba08629cbfd9427213cb8e1719d2ce1e601bdfc)
+++ b/.idea/pyuserconfig.iml (date 1711709021136)
@@ -1,8 +1,10 @@
<?xml version="1.0" encoding="UTF-8"?>
<module type="PYTHON_MODULE" version="4">
<component name="NewModuleRootManager">
- <content url="file://$MODULE_DIR$" />
- <orderEntry type="jdk" jdkName="Python 3.7 (pyuserconfig)" jdkType="Python SDK" />
+ <content url="file://$MODULE_DIR$">
+ <excludeFolder url="file://$MODULE_DIR$/venv" />
+ </content>
+ <orderEntry type="inheritedJdk" />
<orderEntry type="sourceFolder" forTests="false" />
</component>
<component name="TestRunnerService">
Index: .idea/misc.xml
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.BaseRevisionTextPatchEP
<+><?xml version=\"1.0\" encoding=\"UTF-8\"?>\n<project version=\"4\">\n <component name=\"ProjectRootManager\" version=\"2\" project-jdk-name=\"Python 3.7 (pyuserconfig)\" project-jdk-type=\"Python SDK\" />\n</project>
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/.idea/misc.xml b/.idea/misc.xml
--- a/.idea/misc.xml (revision cba08629cbfd9427213cb8e1719d2ce1e601bdfc)
+++ b/.idea/misc.xml (date 1711709021140)
@@ -1,4 +1,4 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
- <component name="ProjectRootManager" version="2" project-jdk-name="Python 3.7 (pyuserconfig)" project-jdk-type="Python SDK" />
+ <component name="ProjectRootManager" version="2" project-jdk-name="Python 3.6 (pyuserconfig)" project-jdk-type="Python SDK" />
</project>
\ No newline at end of file

4
.idea/shelf/Uncommitted_changes_before_Checkout_at_29_03_2024_21_55__Changes_.xml generated Normal file
View File

@ -0,0 +1,4 @@
<changelist name="Uncommitted_changes_before_Checkout_at_29_03_2024_21_55_[Changes]" date="1711745731399" recycled="true" deleted="true">
<option name="PATH" value="$PROJECT_DIR$/.idea/shelf/Uncommitted_changes_before_Checkout_at_29_03_2024_21_55_[Changes]/shelved.patch" />
<option name="DESCRIPTION" value="Uncommitted changes before Checkout at 29/03/2024 21:55 [Changes]" />
</changelist>

6
.idea/vcs.xml generated Normal file
View File

@ -0,0 +1,6 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="VcsDirectoryMappings">
<mapping directory="$PROJECT_DIR$" vcs="Git" />
</component>
</project>

163
cli/__init__.py
View File

@ -1,69 +1,11 @@
import os
import argparse
from userconfig.cfgfile import Conf
from userconfig.tools import Debug
from userconfig.checks import classes_for_host
from userconfig import Userconfig
import sys
import subprocess
class Debug:
_verbose = 0
_COLOR_RED = '\033[91m'
_COLOR_BLUE = '\33[34m'
_COLOR_GREEN = '\33[32m'
_COLOR_YELLOW = '\033[93m'
_COLOR_END = '\33[0m'
_FORMAT = {'STANDARD': 'INFO: ',
'ERROR': f'{_COLOR_RED}ERROR:{_COLOR_END} ',
'NOTICE': f'{_COLOR_BLUE}NOTICE:{_COLOR_END} ',
'WARNING': f'{_COLOR_YELLOW}WARNING:{_COLOR_END} ',
'SUCCESS': f'{_COLOR_GREEN}SUCCESS:{_COLOR_END} '
}
def __init__(self, verbose=0):
self.set_verbose(verbose)
def set_verbose(self, verbose):
self._verbose = verbose
def add_verbose(self):
self._verbose += 1
def get_verbose(self):
return self._verbose
def stdout(self, out, verbose_level=0, category='STANDARD'):
spaces = ''
if verbose_level > 1:
spaces = ' '*(verbose_level-1)
if self._verbose >= verbose_level:
if category in self._FORMAT:
print(f'{spaces}{self._FORMAT[category]}{out}')
else:
print(f'{spaces}[category {category} unknown]:{out}')
def stderr(self, out, verbose_level=0, category='STANDARD'):
if self._verbose >= verbose_level:
if category in self._FORMAT:
print(f'{self._FORMAT[category]}{out}', file=sys.stderr)
else:
print(f'[category {category} unknown]:{out}', file=sys.stderr)
def red(self, out):
return f'{self._COLOR_RED}{out}{self._COLOR_END}'
def green(self, out):
return f'{self._COLOR_GREEN}{out}{self._COLOR_END}'
def blue(self, out):
return f'{self._COLOR_BLUE}{out}{self._COLOR_END}'
def yellow(self, out):
return f'{self._COLOR_YELLOW}{out}{self._COLOR_END}'
def main():
parser = argparse.ArgumentParser(prog='userconfig',
description='Manages configuration files, usually in the user home directory')
@ -80,87 +22,34 @@ def main():
cfg = Conf(filename=cmdline.file, debug=debug)
uc = Userconfig(cfg)
cfg.debug.stdout(f'Verbose level: {cfg.debug.get_verbose()}', 1)
configdir = cfg.get('configdir')
# configdir is the root of the userconfig files
# Directory structure:
# configdir/
# |-- userconfig2.conf
# |-- package1/
# |- package.conf
# |- 001_Arch_Linux/
# |- file1
# |- file2
# |- 002_Host_glitters/
# |- file1
# |- 003_all/
# |- file2
#
# Terminology:
# directories below configdir are packages, packages contain directories categorizing for which host/arch they
# are fitted, below that are files which are installed at the destination
#
# directory names in packages:
# [Number]_[category]_[value]
# Number: can have leading zeros for better sorting in directory structure, will be used for sorting and priority
# category: which kind of match we are looking for. currently: Arch for `uname -s`, Host for `hostname -s`
# value: optional, if category requires an input value, for example: Arch_Linux
# can be empty for example if category is all (no match needed, we want this always to be applied)
cfg.debug.stdout(f'configdir: {configdir}', 1)
for package in os.scandir(configdir):
# Skip on non-production files
if not package.is_dir():
cfg.debug.stdout(f'{package.path} is not a directory, skipping', 2, 'WARNING')
cfg.debug.stdout(f"Verbose level is {cfg.debug.get_verbose()}", 1)
cfg.debug.stdout("================ main ===============", 1)
temp_classes = ""
for h in classes_for_host():
temp_classes = temp_classes + str(h) + ","
cfg.debug.stdout("+++ Current host is in classes %s" % temp_classes, 1)
configdir = cfg.get("configdir")
for d in os.listdir(configdir):
name = configdir+"/"+d
cfg.debug.stdout("+++ Working in %s" % name, 1)
if not os.path.isdir(name):
cfg.debug.stdout("--- %s is not a directory, skipping." % name, 3)
continue
if package.name in ['.svn', '.git']:
cfg.debug.stdout(f'{package.path} is a svn or git data directory, skipping', 2, 'WARNING')
elif d.startswith(".svn") or d.startswith(".git"):
cfg.debug.stdout("--- %s is .svn or .git, skipping." % name, 3)
continue
if os.path.isfile(f'{package.path}/.ignore'):
cfg.debug.stdout(f'{package.path} contains .ignore, skipping', 2, 'WARNING')
elif os.path.isfile(name+"/.ignore"):
cfg.debug.stdout("--- %s contains file .ignore, skipping." % name, 3)
continue
# Start processing
cfg.debug.stdout(f'Start Package {cfg.debug.green(package.path)}', 1, 'NOTICE')
# Get all category directories for package
(category_dirs, dir_config) = uc.process_package_dir(package.path)
if not category_dirs:
cfg.debug.stdout(f'Could not get category_dirs for package {package.name}, skipping package.', 0, 'ERROR')
continue
if not dir_config:
cfg.debug.stdout(f'Could not get dir_config for package {package.name}, skipping package.', 0, 'ERROR')
continue
cfg.debug.stdout(f'Got categories: {category_dirs}', 2)
host_category_dirs = uc.filter_categories(category_dirs)
cfg.debug.stdout('Host uses categories: %s' % ", ".join([f'{v[1]}_{v[2]}' for v in host_category_dirs]), 2)
file_list = dict()
for c in host_category_dirs:
file_list = uc.process_category_dir(c, file_list)
cfg.debug.stdout(f'Found files: {file_list}', 2)
if len(file_list) and os.access(f'{package.path}/install.sh', os.X_OK):
cfg.debug.stdout(f'Execute {package.path}/install.sh', 2)
subprocess.call([f'{package.path}/install.sh'])
for file in file_list:
dest = f'{dir_config.get(section="Main", option="dest")}/{file}'
cfg.debug.stdout('Generating %s from:\n %s' % (dest, "\n ".join(file_list[file])), 1)
try:
comment_string = dir_config.get(section="Main", option="commentstring")
except ValueError:
cfg.debug.stdout(f'commentstring does not exist in config file {dir_config._cfgfiles}', 0, 'ERROR')
sys.exit(1)
# Make sure all directories for destination file exist
if uc.create_destination_directories(dir_config.get(section="Main", option="dest")):
cfg.debug.stdout(f'Created target directories {cfg.debug.green(dir_config.get(section="Main", option="dest"))}',
0, 'SUCCESS')
else:
cfg.debug.stdout("+++ Processing files in %s" % name, 2)
(destfiles, dirConfig) = uc.workdir(name)
if isinstance(destfiles, dict):
if len(destfiles.keys()) > 0:
cfg.debug.stdout("+++ Building %d files: %s" % (len(destfiles.keys()), destfiles.keys()), 3)
uc.process_all_files(destfiles, dirConfig)
else:
cfg.debug.stdout(f'All target directories exist', 2)
temp_filename = uc.build_file(file_list[file], dest, comment_string)
if uc.diff_and_copy_file(temp_filename, dest, comment_string):
cfg.debug.stdout(f'Copy {temp_filename} -> {cfg.debug.green(dest)} (changed)\n', 0, 'SUCCESS')
else:
cfg.debug.stdout(f'Generated file and destination are the same.\n', 1)
cfg.debug.stdout(f'End Package {cfg.debug.green(package.path)}\n\n', 1, 'NOTICE')
cfg.debug.stdout("--- No files found for %s, skipping." % name, 1)
if __name__ == '__main__':

2
pyproject.toml
View File

@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "userconfig"
version = "2.0"
version = "2.0b-1"
authors = [ {name = "Marcus Stoegbauer", email = "marcus@grmpf.org"} ]
maintainers = [ {name = "Marcus Stoegbauer", email = "marcus@grmpf.org"} ]
description = "Generate config files for user home"

2
test.py
View File

@ -1,2 +0,0 @@
import cli
cli.main()

370
userconfig/__init__.py
View File

@ -1,12 +1,11 @@
from userconfig.checks import check_class
from userconfig.cfgfile import Conf
from userconfig.tools import get_config
import subprocess
from userconfig.checks import classes_for_host
import os
from userconfig.tools import diff, user_config_generated, backup_file, copy_file
import time
import tempfile
from pathlib import Path
import shutil
import re
import sys
import tempfile
class Userconfig:
@ -15,239 +14,160 @@ class Userconfig:
def __init__(self, cfg):
self._cfg = cfg
def process_package_dir(self, package_dir):
"""
Walk through package_dir, collect all directories inside, parse them according to our specs and return
a sorted list with entries (prio, category, value, path)
:param package_dir: root of package directory
:return: list of tuples (prio, category, value, path), sorted with prio
"""
self._cfg.debug.stdout(f'process package_dir {package_dir}', 3)
package_config_file = f'{package_dir}/{self._cfg.get("configfile")}'
if not os.path.isfile(package_config_file):
self._cfg.debug.stdout(f'No config file {self._cfg.get("configfile")} in {package_dir}, skipping',
0, 'ERROR')
return None, None
dir_config = self.get_config(package_config_file)
if not dir_config:
self._cfg.debug.stdout(f'Cannot read config file {package_config_file}, skipping', 0, 'ERROR')
return None, None
classes = []
for category in os.scandir(package_dir):
if category.path == package_config_file:
continue
if not category.is_dir():
self._cfg.debug.stdout(f'{category.path} is not a directory, skipping', 3)
continue
self._cfg.debug.stdout(f'process category {category.path}', 3)
content = category.name.split('_')
# Format: <number>_<category>_<value>
if len(content) == 3:
prio_string = content[0]
category_name = content[1]
value = content[2]
elif len(content) == 2:
prio_string = content[0]
category_name = content[1]
value = ''
else:
self._cfg.debug.stdout(f'Format of package directory {category.path} wrong, skipping', 0, 'ERROR')
continue
try:
prio = int(prio_string)
except ValueError:
self._cfg.debug.stdout(f'Cannot convert prio to integer ({category.path}, skipping', 0, 'ERROR')
continue
# self._cfg.debug.stdout(f'Got class: {(prio, category_name, value)}', 3)
classes.append((prio, category_name, value, category.path))
classes.sort(key=lambda k: k[0])
return classes, dir_config
def filter_categories(self, categories):
"""
Get classes list of tuples from process_package_dir and filter it for host running the command
:param categories: list of tuples from process_package_dir
:return: matching classes for this host
"""
ret = []
for c in categories:
if not c[1]:
self._cfg.debug.stdout(f'category not set for {c[3]}, skipping', 0, 'ERROR')
continue
if check_class(c):
ret.append(c)
return ret
def process_category_dir(self, category_dir_tuple, file_list):
"""
Walk through category_dir, collect all files
:param category_dir_tuple:
:param file_list:
:return:
"""
(prio, category, value, category_dir) = category_dir_tuple
self._cfg.debug.stdout(f'process category {category_dir}', 3)
for file in os.scandir(category_dir):
if file.name not in file_list:
file_list[file.name] = []
file_list[file.name].append(file.path)
return file_list
def build_file(self, files, dest_file, comment_string):
"""
merge all files in files[] and write contents into a temporary file
:param files:
:param dest_file:
:param comment_string:
:return: temporary filename
"""
def build_file(self, classfiles, destfile, commentstring):
"""open all classfiles, assemble them and write the contents into a tempfile
returns the name of tempfile
:param classfiles:
:param destfile:
:param commentstring:
:return: str
"""
content = []
self._cfg.debug.stdout(f'building file for {dest_file}', 3)
if comment_string:
content.append(f'{comment_string} {self._cfg.get("stamp")} {time.strftime("%+")}\n')
for file in files:
self._cfg.debug.stdout(f'Merging {file}', 3)
fp = open(file, "r")
file_content = fp.read()
fp.close()
# if comment_string exists, add comment with category dir and filename
if comment_string:
content.append(f'{comment_string} {"/".join(file.split("/")[-2:])}')
content.append(file_content)
self._cfg.debug.stdout(f" ================ build_file {destfile} ===============", 1)
if commentstring != "":
self._cfg.debug.stdout(" +++ commentstring found, adding header.", 3)
content.append(commentstring + " " + self._cfg.get("stamp") + " " + time.strftime("%+") + "\n")
for f in classfiles:
self._cfg.debug.stdout(" +++ Merging %s." % f, 4)
fp = open(f, "r")
filecontent = fp.read()
fp.close()
if commentstring == "":
# look for stamp in content, replace with real stamp
if re.search(re.escape(self._cfg.get("stampreplace")), filecontent):
self._cfg.debug.stdout(" +++ commentstring empty, replacing stamp in file", 3)
filecontent = re.sub(re.escape(self._cfg.get("stampreplace")), self._cfg.get("stamp"),
filecontent)
content.append(filecontent)
(tempfd, tempfilename) = tempfile.mkstemp(prefix=os.path.basename(destfile), dir="/tmp")
(temp_fd, temp_filename) = tempfile.mkstemp(prefix=os.path.basename(dest_file), dir="/tmp")
try:
fp = os.fdopen(temp_fd, "w")
fp = os.fdopen(tempfd, "w")
except Exception as e:
self._cfg.debug.stderr(f'Cannot write to temporary file {temp_filename}: {e}', 0, 'ERROR')
os.remove(temp_filename)
self._cfg.debug.stderr(f"Cannot write to temporary file {tempfilename}: {e}")
os.remove(tempfilename)
return False
self._cfg.debug.stdout(f'Writing merged files into temporary file {temp_filename}', 3)
self._cfg.debug.stdout(" +++ Writing merged files into tempfile %s." % tempfilename, 3)
for block in content:
fp.write(block)
fp.write("\n")
fp.close()
return temp_filename
self._cfg.debug.stdout(f" ================ done: build_file {destfile} ===============", 1)
return tempfilename
def diff_and_copy_file(self, temp_filename, dest_filename, comment_string):
ret = False
if self.diff(dest_filename, temp_filename, comment_string):
if not self.user_config_generated(dest_filename):
self._cfg.debug.stdout(f'{dest_filename} not generated by userconfig, running back_up.', 3)
self.backup_file(dest_filename)
self.copy_file(temp_filename, dest_filename)
ret = True
os.remove(temp_filename)
self._cfg.debug.stdout(f'Removed temporary file {temp_filename}', 3, 'SUCCESS')
def process_all_files(self, destfiles, dir_config):
"""processes all files in destfiles, generate files from classes, compare and copy if necessary"""
self._cfg.debug.stdout(" ================ process_all_files ===============", 1)
for df in destfiles.keys():
self._cfg.debug.stdout(" ??? Processing source files for %s." % df, 2)
if not os.path.exists(os.path.dirname(df)):
self._cfg.debug.stdout(" +++ Directory %s does not exist, creating" % os.path.dirname(df), 1)
os.mkdir(os.path.dirname(df))
if not os.path.isdir(os.path.dirname(df)):
self._cfg.debug.stdout(f" --- Destination directory {os.path.dirname(df)} does not exist, skipping.",
1)
return False
# assemble file to tmp
commentstring = ""
if dir_config.check(section="Main", option="commentstring"):
commentstring = dir_config.get(section="Main", option="commentstring")
self._cfg.debug.stdout(" +++ Found commentstring %s in %s" % (commentstring, df), 3)
tempfilename = self.build_file(destfiles[df], df, commentstring)
if not tempfilename:
self._cfg.debug.stdout(" --- Error while creating temp file for %s, skipping." % df, 1)
continue
self._cfg.debug.stdout(" +++ Merged files %s for %s into %s" % (str(destfiles[df]), df, tempfilename), 2)
# diff assembled file and config file
if diff(df, tempfilename, commentstring, self._cfg):
self._cfg.debug.stdout("File %s has changed" % df, 0)
if not user_config_generated(df, self._cfg):
self._cfg.debug.stdout(" +++ %s not generated by userconfig, backing up." % df, 2)
# file not generated from userconfig -> back up
backup_file(df, self._cfg)
# copy tmp file to real location
self._cfg.debug.stdout("Copy %s to %s." % (tempfilename, df), 0)
copy_file(tempfilename, df, self._cfg)
# remove tmp
self._cfg.debug.stdout(" +++ Removing temporary file %s." % tempfilename, 2)
os.remove(tempfilename)
self._cfg.debug.stdout(" ================ process_all_files ===============", 1)
def workconf(self, directory, depth=2):
"""walks through directory, collecting all filenames, returns list of all filenames"""
dirs = os.listdir(directory)
ret = []
self._cfg.debug.stdout(f" ================ workconf {directory} ===============", 1)
for d in dirs:
name = directory + "/" + d
if os.path.isdir(name):
self.workconf(name, depth + 1)
if name.endswith(".swp"):
continue
if d == ".svn":
continue
ret.append(name)
self._cfg.debug.stdout(" +++ Found file %s in directory %s" % (name, directory), 4)
self._cfg.debug.stdout(f" ================ done: workconf {directory} ===============", 1)
return ret
def create_destination_directories(self, dest_directory):
path = Path(dest_directory)
if not path.exists():
path.mkdir(parents=True)
return True
else:
return False
def get_config(self, filename):
"""reads filename as config, checks for DEST parameter and returns cfgfile object"""
try:
ret = Conf(filename=filename, debug=self._cfg.debug, force_filename=True)
except ValueError:
self._cfg.debug.stderr(f'Error reading config file {filename}', 0, 'ERROR')
return False
# check for DEST parameter
if not ret.check(section="Main", option="dest"):
self._cfg.debug.stderr(f'No dest in config file {filename}', 0, 'ERROR')
return False
# make sure DEST ends with /
if not ret.get(section="Main", option="dest").endswith("/"):
ret.set(section="Main", option="dest", value=ret.get(section="Main", option="dest")+"/")
return ret
@staticmethod
def read_skip_comment(fp, comment_string):
"""Read line from filehandle fp and skip all empty (whitespace) lines and lines starting with comment_string
def workdir(self, directory):
"""walks through all host classes, checking if the classes directory exists in directory
then collect all filenames within this directory and return a dict of all files for this
directory
"""
for line in fp:
line = line[:-1]
if ((comment_string != "" and not re.match("^"+re.escape(comment_string), line)) and line != "" and
not re.match(r"^\s+$", line)):
yield line
self._cfg.debug.stdout(f" ================ workdir {directory} ===============", 1)
# skip directory if no CONFIGFILE present
if not os.path.isfile(directory+"/"+self._cfg.get("configfile")):
self._cfg.debug.stdout(f' --- No file {self._cfg.get("configfile")} in {directory}, skipping.', 1)
return {}, None
def diff(self, dest_file, temp_file, comment_string):
"""diff dest_file and temp_file, returns True if files differ, False if they are the same"""
self._cfg.debug.stdout(f'Diffing {dest_file} and {temp_file}, comment: {comment_string}', 3)
if not os.path.isfile(dest_file):
self._cfg.debug.stdout(f'dest_file {dest_file} does not exist.', 3)
return True
if not os.path.isfile(temp_file):
self._cfg.debug.stderr(f'Temporary file {temp_file} does not exist, this should not happen.', 0, 'ERROR')
sys.exit(1)
# get config file for directory
dir_config = get_config(directory + "/" + self._cfg.get("configfile"), self._cfg)
if not dir_config:
self._cfg.debug.stdout(f' --- Cannot read config file {self._cfg.get("configfile")} in {directory}, '
f'skipping.', 1)
return {}, None
fp1 = open(temp_file)
fp2 = open(dest_file)
destdir = dir_config.get(section="Main", option="dest")
# destfiles is a dict of all files that will be created from the classes config
# key is the destination filename, values are all classes filenames that are used to
# build the file
destfiles = {}
try:
reverse_sort = dir_config.get(section="Main", option="reverse", boolean=True)
except ValueError:
reverse_sort = False
self._cfg.debug.stdout(f' +++ reverse_sort is {reverse_sort}', 3)
if os.access(directory + "/install.sh", os.X_OK):
subprocess.call([directory + "/install.sh"])
for line1, line2 in zip(self.read_skip_comment(fp1, comment_string), self.read_skip_comment(fp2, comment_string)):
if line1 != line2:
fp1.close()
fp2.close()
self._cfg.debug.stdout(f'{dest_file} differs from generated config', 3)
return True
fp1.close()
fp2.close()
self._cfg.debug.stdout(f'{dest_file} is the same as generated config', 3)
return False
def user_config_generated(self, filename):
"""returns True if filename has been generated by userconfig, False else"""
if not os.path.isfile(filename):
# filename does not exist, so it was not generated by userconfig
return False
if not self._cfg.check("stamp"):
# no STAMP in userconfig.cfg, so no way to check if file was generated by userconfig
return False
fp = open(filename, "r")
for line in fp:
if re.search(re.escape(self._cfg.get("stamp")), line):
return True
return False
def backup_file(self, filename):
"""make backup of filename, returns True if backup is successful, False else"""
if os.path.isfile(filename):
self._cfg.debug.stdout(f'{filename} exists, finding backup name.', 3)
backup_name = filename+".userconfig."+time.strftime("%F")
test_backup_name = backup_name
counter = 0
while os.path.isfile(test_backup_name):
counter += 1
test_backup_name = backup_name+"."+str(counter)
os.rename(filename, test_backup_name)
self._cfg.debug.stdout(f'Renamed {filename} to {test_backup_name}', 1, 'SUCCESS')
return True
else:
self._cfg.debug.stdout(f'{filename} does not exist, do not need backup.', 3)
return False
def copy_file(self, sourcefile, dest_file):
"""copy sourcefile to dest_file, returns True if successful, False else"""
if os.path.isfile(sourcefile):
# sourcefile exists
self._cfg.debug.stdout(f'Source file {sourcefile} exists, proceeding with copy.', 3)
if not os.path.isfile(dest_file) or os.access(dest_file, os.W_OK):
shutil.copy(sourcefile, dest_file)
return True
# walk through all know classes in directory and find filenames
for h in classes_for_host(reverse_sort):
# build classes directory
if h[0] != "":
classdir = directory+"/"+h[0]+"_"+h[1]
else:
self._cfg.debug.stdout('Destination {dest_file} is not a file or not writable.', 0, 'ERROR')
return False
classdir = directory+"/"+h[1]
self._cfg.debug.stdout(" ??? Looking for directory %s." % classdir, 4)
# if class directory exists
if os.path.isdir(classdir):
self._cfg.debug.stdout(" +++ Found directory %s, getting files." % classdir, 4)
# get list of files within this class directory
tempfiles = self.workconf(classdir)
self._cfg.debug.stdout(" +++ Got %d files: %s." % (len(tempfiles), str(tempfiles)), 4)
# put files into dict
for f in tempfiles:
destname = destdir+os.path.basename(f) # destination filename
if destname not in destfiles:
destfiles[destname] = []
destfiles[destname].append(f) # append each file to dict
self._cfg.debug.stdout(f" +++ Added file to {destname}, now {len(destfiles[destname])} files: "
f"{destfiles[destname]}", 4)
self._cfg.debug.stdout(" === workdir: %s, Files: %s" % (directory, str(destfiles)), 3)
self._cfg.debug.stdout(f" ================ done: workdir {directory} ===============", 1)
return destfiles, dir_config

25
userconfig/checks.py
View File

@ -1,24 +1,19 @@
import platform
def get_hostname():
def get_short_hostname():
hostname = platform.node()
if hostname.count("."):
hostname = hostname.split(".")[0]
return hostname
def get_arch():
return platform.system()
def check_class(class_tuple):
(prio, category, value, path) = class_tuple
if category == 'Arch':
return get_arch() == value
elif category == 'Host':
return get_hostname() == value
elif value == '': # if value is empty, we cannot filter anything, so it matches always
return True
else:
return False
def classes_for_host(reverse=False):
classes = [(0, "", "header"),
(1000, "", "footer"),
(998, "", "all"),
(800, "Arch", platform.system()),
(10, "Host", get_short_hostname())
]
classes.sort(key=lambda k: k[0], reverse=reverse)
return [(k[1], k[2]) for k in classes]

141
userconfig/tools.py Normal file
View File

@ -0,0 +1,141 @@
import sys
import os
from userconfig.cfgfile import Conf
import re
import time
import shutil
class Debug:
_verbose = 0
def __init__(self, verbose=0):
self.set_verbose(verbose)
def set_verbose(self, verbose):
self._verbose = verbose
def add_verbose(self):
self._verbose += 1
def get_verbose(self):
return self._verbose
def stdout(self, out, level=0):
if self._verbose >= level:
print(out)
def stderr(self, out, level=0):
if self._verbose >= level:
print(str(out)+"\n", file=sys.stderr)
def get_config(filename, cfg):
"""reads filename as config, checks for DEST parameter and returns cfgfile object"""
try:
ret = Conf(filename=filename, debug=cfg.debug, force_filename=True)
except ValueError:
cfg.debug.stderr("Error reading config file %s" % filename)
return False
# check for DEST parameter
if not ret.check(section="Main", option="dest"):
cfg.debug.stderr("No dest in config file %s" % filename)
return False
# make sure DEST ends with /
if not ret.get(section="Main", option="dest").endswith("/"):
ret.set(section="Main", option="dest", value=ret.get(section="Main", option="dest")+"/")
return ret
def read_skip_comment(fp, commentstring):
"""Read line from filehandle fp and skip all empty (whitespace) lines and lines starting with commentstring
"""
for line in fp:
line = line[:-1]
if ((commentstring != "" and not re.match("^"+re.escape(commentstring), line)) and line != "" and
not re.match(r"^\s+$", line)):
yield line
def diff(destfile, tempfile, commentstring, cfg):
"""diff destfile and tempfile, returns True if files differ, False if they are the same"""
cfg.debug.stdout("Diffing %s and %s, comment: %s" % (destfile, tempfile, commentstring), 3)
if not os.path.isfile(destfile):
cfg.debug.stdout("Destfile %s does not exist, returning True." % destfile, 3)
# destfile does not exist -> copy tempfile over
return True
# if not destfile
if not os.path.isfile(tempfile):
# tempfile does not exist, this should never happen
cfg.debug.stderr("Temporary file %s does not exist, this should not happen." % tempfile)
sys.exit(1)
# if not tempfile
fp1 = open(tempfile)
fp2 = open(destfile)
for line1, line2 in zip(read_skip_comment(fp1, commentstring), read_skip_comment(fp2, commentstring)):
if line1 != line2:
fp1.close()
fp2.close()
cfg.debug.stdout("%s differs, return true" % destfile, 3)
return True
# if differ
# for line
fp1.close()
fp2.close()
cfg.debug.stdout("%s is the same, return false" % destfile, 3)
return False
def user_config_generated(filename, cfg):
"""returns True if filename has been generated by userconfig, False else"""
if not os.path.isfile(filename):
# filename does not exist, so it was not generated by userconfig
return False
if not cfg.check("stamp"):
# no STAMP in userconfig.cfg, so no way to check if file was generated by userconfig
return False
fp = open(filename, "r")
for line in fp:
if re.search(re.escape(cfg.get("stamp")), line):
return True
return False
def backup_file(filename, cfg):
"""make backup of filename, returns True if backup is successful, False else"""
if os.path.isfile(filename):
cfg.debug.stdout("%s exists, finding backup name." % filename, 3)
backupname = filename+".userconfig."+time.strftime("%F")
testbackupname = backupname
counter = 0
while os.path.isfile(testbackupname):
counter += 1
testbackupname = backupname+"."+str(counter)
cfg.debug.stdout("Renaming %s to %s" % (filename, testbackupname), 1)
os.rename(filename, testbackupname)
return True
else:
cfg.debug.stdout("%s does not exist, do not need backup." % filename, 3)
return False
def copy_file(sourcefile, destfile, cfg):
"""copy sourcefile to destfile, returns True if successful, False else"""
if os.path.isfile(sourcefile):
# sourcefile exists
cfg.debug.stdout("Source file %s exists, proceeding with copy." % sourcefile, 3)
if not os.path.isfile(destfile) or os.access(destfile, os.W_OK):
cfg.debug.stdout("Copying %s to %s" % (sourcefile, destfile), 1)
shutil.copy(sourcefile, destfile)
return True
# destfile is writable
else:
cfg.debug.stdout("Destination file %s does not exist or is not writable." % destfile, 3)
return False