#!/usr/bin/env python3 # -*- coding: utf-8 -*- import os import sys import yaml import logging import datetime try: from yaml import CLoader as Loader, CDumper as Dumper except ImportError: from yaml import Loader, Dumper import click from adapters import * logger = logging.getLogger(f'wp_cal') class BaseConfig(dict): def __init__(self, file, defaults, overrides, *args, **kwargs): self._ex = None self.file = file self.defaults = defaults self.overrides = overrides super().__init__(*args, **kwargs) try: self._load() except FileNotFoundError as ex: self._ex = ex def __repr__(self): return super().__repr__() def __str__(self): return super().__str__() def _load(self): try: if self.file == '-': config = yaml.load(sys.stdin, Loader=Loader) else: if os.stat(self.file).st_mode & 0o777 & ~0o600: raise Exception('refusing to load insecure configuration file, file must have permission 0o600') with open(self.file) as fp: config = yaml.load(fp, Loader=Loader) if config is None: config = {} for k, v in config.items(): self[k] = v finally: for keylist, value in self.defaults: d = self for i, key in enumerate(keylist): repl = value if (i == len(keylist) - 1) else {} d[key] = d.get(key, repl) d = d[key] for keylist, value in self.overrides: d = self for key in keylist[:-1]: d[key] = d.get(key, {}) d = d[key] d[keylist[-1]] = value def _save(self): with open(self.file, 'w') as fp: yaml.dump(self, fp, Dumper=Dumper) def exception(self): return self._ex def load(self): return self._load() def save(self): return self._save() class Config(BaseConfig): """ The default configuration. Keys: .google.calendar_id: the id of the calendar to sync .google.credentials: the json of the obtained credentials file .google.token_file: where to save the login token .wordpress.url: the base wordpress url .wordpress.calendar.id: the id of the (wp-booking-system) calendar .wordpress.calendar.name: the name of the calendar .wordpress.calendar.translations: a dictionary of language <-> translation pairs (example: {"en": "Reservations"}) .wordpress.credentials.user: the user as which to log into wordpress .wordpress.credentials.password: the users password .logging.level: one or more log levels of the form : seperated by a `,` (comma) .range: the range in the format , where n are digits and u is the unit; valid units: 'smhdw' <-> seconds minutes hours days weeks """ def __init__(self, file, defaults, overrides, converters, *args, **kwargs): defaults += [ ('google.calendar_id', '#TODO insert google calendar id'), ('google.credentials', {}), ('google.token_file', os.path.join(os.environ['HOME'], '.wp-cal-google-token')), ('wordpress.url', '#TODO insert url to wordpress site'), ('wordpress.calendar.id', '#TODO insert wp-booking-system calendar id'), ('wordpress.calendar.name', '#TODO insert calendar name'), ('wordpress.calendar.translations', {'en': '#TODO insert english translation'}), ('wordpress.credentials.user', '#TODO insert wordpress username'), ('wordpress.credentials.password', '#TODO insert wordpress password'), ('logging.level', ':WARNING,wp_cal:INFO'), ('range', '365d'), ] defaults = [(x[0].split('.'), x[1]) for x in defaults] overrides = [(x[0].split('.'), x[1]) for x in overrides] self.converters = converters super().__init__(file, defaults, overrides, *args, **kwargs) def value(self, k): if isinstance(k, str): sel = k.split('.') elif isinstance(k, list): sel = k k = '.'.join(k) else: raise TypeError('invalid type for parameter k') v = self for s in sel: v = v[s] if k in self.converters: return self.converters[k](v) return v def range_str_to_timedelta(value): valid_units = 'smhdw' current_int = 0 values = {} for c in value: if c in '0123456789': current_int *= 10 current_int += int(c) elif c in valid_units: if c in values: logger.warning('unit %s already in values, overwriting', c) values[c] = current_int current_int = 0 c = 's' if current_int != 0: if c in values: logger.warning('unit %s already in values, overwriting', c) values['s'] = current_int for valid_unit in valid_units: values.setdefault(valid_unit, 0) return datetime.timedelta( seconds=values['s'], minutes=values['m'], hours=values['h'], days=values['d'], weeks=values['w'] ) def init_logging(): logging.getLogger().addHandler( logging.StreamHandler(), ) def set_logging_level(level: str): allowed_values = {'NOTSET', 'DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'} log_levels = [x.split(':') for x in level.split(',')] for module, level in log_levels: module = module.strip() if module else None level = level.strip().upper() if level not in allowed_values: raise ValueError(f'invalid log level, allowed values: {repr(allowed_values)}') level = getattr(logging, level) logging.getLogger(module).setLevel(level) @click.command() @click.option('--config', '-c', envvar='WP_CAL_CONFIG', default='-', help='The configuration file') @click.option('--dryrun', '-d', envvar='WP_CAL_DRYRUN', is_flag=True, help="Don't actually post any data, just show it") @click.option('--level', '-l', envvar='WP_CAL_LEVEL', default=None, help='The log level for the application') @click.option('--range', '-r', envvar='WP_CAL_RANGE', default=None, help='The time range from start to start + range to synchronize events') def main(config, dryrun, level, range): init_logging() set_logging_level(':DEBUG') config_overrides = [('logging.level', level), ('range', range)] config_overrides = [x for x in config_overrides if x[1] is not None] converters = { 'range': range_str_to_timedelta, } config = Config(config, [], config_overrides, converters) if config.exception(): logger.info('config not found, trying to generate template') try: config.save() except Exception: logger.exception('failed to generate template') else: logger.info('generated config at "%s"', config.file) return set_logging_level(str(config.value('logging.level'))) g = Google( config.value('google.calendar_id'), credentials=config.value('google.credentials'), token_file=config.value('google.token_file'), ) w = Wordpress( config.value('wordpress.url'), calendar_metadata=CalendarMetadata( id=config.value('wordpress.calendar.id'), name=config.value('wordpress.calendar.name'), translations=config.value('wordpress.calendar.translations'), ), credentials=config.value('wordpress.credentials'), ) g.login() events = g.get_events(until=config.value('range')) logger.info("syncing %d events", len(events)) if not w.login(): logger.info('failed to login to wordpress') return 1 if dryrun: logger.info("dryrun; would post events: %s", events) else: if not w.post_events(events, until=config.value('range')): logger.info('failed to post event data') return 1 logger.info("done") return 0 if __name__ == '__main__': main()