194 lines
6.3 KiB
Python
Executable file
194 lines
6.3 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
|
|
import argparse
|
|
import dataclasses
|
|
import logging
|
|
import os
|
|
import sys
|
|
|
|
|
|
from string import Template
|
|
from typing import List, Tuple, Union
|
|
|
|
|
|
configuration_template = Template(
|
|
'''
|
|
cwd = $cwd
|
|
|
|
# Generated by link-tool.py
|
|
# $link_tool_cmd
|
|
|
|
$link_rules
|
|
''')
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
@dataclasses.dataclass(repr=True)
|
|
class Link:
|
|
src: str
|
|
dst: str
|
|
|
|
@dataclasses.dataclass(repr=True)
|
|
class Conf:
|
|
cwd: Union[str, None]
|
|
|
|
def assign(self, var_name, var_value):
|
|
if hasattr(self, var_name):
|
|
logger.debug('assign config: {} = {}'.format(var_name, var_value))
|
|
setattr(self, var_name, var_value)
|
|
else:
|
|
raise Exception('Unknown config: {}'.format(var_name))
|
|
|
|
def validate(self):
|
|
missing_fields = []
|
|
if self.cwd is None:
|
|
missing_fields.append('cwd')
|
|
if len(missing_fields) > 0:
|
|
logger.error('there are missing field: {}'.format(missing_fields))
|
|
return len(missing_fields) == 0
|
|
|
|
def load_config(conf_path: str) -> Tuple[Conf, List[Link]]:
|
|
with open(conf_path, 'r') as f:
|
|
lines = f.readlines()
|
|
|
|
lines = [ line.strip() for line in lines ]
|
|
lines = [ line for line in lines if len(line) > 0 ]
|
|
lines = [ line for line in lines if not line.startswith('#') ]
|
|
|
|
links = []
|
|
conf = Conf(cwd=None)
|
|
for line in lines:
|
|
if '=' in line:
|
|
line_fields = line.split('=')
|
|
logger.debug('[config ] line fields: {}'.format(line_fields))
|
|
var_name = line_fields[0].strip()
|
|
var_value = line_fields[1].strip()
|
|
|
|
conf.assign(var_name, var_value)
|
|
|
|
elif '->' in line:
|
|
line_fields = line.split('->')
|
|
logger.debug('[link-rule] line fields: {}'.format(line_fields))
|
|
src = line_fields[0].strip()
|
|
dst = line_fields[1].strip()
|
|
|
|
links.append(Link(src, dst))
|
|
|
|
else:
|
|
raise Exception('Unknown config line: {}'.format(line))
|
|
|
|
return conf, links
|
|
|
|
def make_link(links: List[Link], dry_run=False, overwrite=False):
|
|
for link in links:
|
|
if os.path.exists(link.dst):
|
|
if not overwrite:
|
|
logger.info('dst exists, skipping: {}'.format(link.dst))
|
|
continue
|
|
|
|
if os.path.exists(link.dst):
|
|
logger.warning('dst exists, removing: {}'.format(link.dst))
|
|
if not dry_run:
|
|
os.remove(link.dst)
|
|
|
|
dst_dir = os.path.dirname(link.dst)
|
|
if not os.path.exists(dst_dir):
|
|
logger.info('dst directory not exists, creating: {}'.format(dst_dir))
|
|
if not dry_run:
|
|
os.makedirs(dst_dir)
|
|
|
|
logger.info('creating link: {}'.format(link.dst))
|
|
if not dry_run:
|
|
os.link(link.src, link.dst)
|
|
|
|
def generate_configuration(src: str, dst: str, config_path: str):
|
|
config_path_depth = config_path.strip('/').count('/')
|
|
cwd = '.' if config_path_depth == 0 else '..' + '/..' * (config_path_depth - 1)
|
|
|
|
logger.info('configuration: cwd: {}'.format(cwd))
|
|
|
|
link_rules = []
|
|
for dirname, _, files in os.walk(src):
|
|
for file in files:
|
|
link_rules.append(Link(src=os.path.join(dirname, file),
|
|
dst=os.path.join(dst, os.path.basename(file))))
|
|
|
|
max_src_file_path_len = max([len(l.src) for l in link_rules])
|
|
formater = '{:%d} -> {}' % max_src_file_path_len
|
|
link_rules_str = ''
|
|
for l in link_rules:
|
|
link_rules_str += formater.format(l.src, l.dst) + '\n'
|
|
|
|
configuration_str = configuration_template.safe_substitute(
|
|
cwd=cwd,
|
|
link_rules=link_rules_str,
|
|
link_tool_cmd="'{}'".format("' '".join(sys.argv)))
|
|
|
|
# create parent dir for configuration file
|
|
configuration_dir = os.path.dirname(config_path)
|
|
if not os.path.exists(configuration_dir):
|
|
os.makedirs(configuration_dir)
|
|
|
|
# write configuration file
|
|
with open(config_path, 'w', encoding='utf-8') as f:
|
|
f.write(configuration_str)
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument('-c', '--config', required=True,
|
|
help='configuration file')
|
|
parser.add_argument('-o', '--overwrite', action='store_true',
|
|
help='overwrite dst when exists')
|
|
parser.add_argument('-n', '--dry-run', action='store_true',
|
|
help='dry run without actually create link')
|
|
parser.add_argument('-v', '--verbose', action='store_true',
|
|
help='more verbose log')
|
|
|
|
parser.add_argument('-g', '--generate', action='store_true',
|
|
help='generate configuration with a predefined template, '
|
|
'need other arguments. When this argument is used, '
|
|
'the generated configuration will be place to the path of --config')
|
|
parser.add_argument('--src', default=None,
|
|
help='src directory, link rule will be generate for '
|
|
'every file in this (sub)directory')
|
|
parser.add_argument('--dst', default=None,
|
|
help='dst directory, link rule\'target directory')
|
|
|
|
args = parser.parse_args()
|
|
|
|
logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO)
|
|
|
|
if args.generate:
|
|
if args.src is None or args.dst is None:
|
|
logger.error('invalid argument, --src or --dst is not specified')
|
|
exit(1)
|
|
if '..' in args.config:
|
|
logger.error('configuration file path must not contain any \'..\'')
|
|
else:
|
|
generate_configuration(args.src, args.dst, args.config)
|
|
else:
|
|
dry_run = args.dry_run
|
|
overwrite = args.overwrite
|
|
|
|
config, links = load_config(args.config)
|
|
|
|
if not config.validate():
|
|
logger.error('invalid configuration, exiting...')
|
|
exit(1)
|
|
|
|
for link in links:
|
|
logger.debug('loaded link: {}'.format(link))
|
|
|
|
assert(config.cwd is not None)
|
|
config_path_dir = os.path.dirname(args.config)
|
|
chdir_target = os.path.join(config_path_dir, config.cwd)
|
|
logger.info('changing CWD: {}/{}'.format(config_path_dir, config.cwd))
|
|
logger.info('the CWD realpath: {}'.format(os.path.realpath(chdir_target)))
|
|
os.chdir(chdir_target)
|
|
|
|
make_link(links, dry_run=dry_run, overwrite=overwrite)
|
|
|
|
if __name__ == '__main__':
|
|
main()
|