diff options
Diffstat (limited to 'rss_to_gitea')
| -rw-r--r-- | rss_to_gitea/config.py | 110 | ||||
| -rw-r--r-- | rss_to_gitea/gitea.py | 46 | ||||
| -rw-r--r-- | rss_to_gitea/main.py | 44 |
3 files changed, 174 insertions, 26 deletions
diff --git a/rss_to_gitea/config.py b/rss_to_gitea/config.py new file mode 100644 index 0000000..1737a39 --- /dev/null +++ b/rss_to_gitea/config.py @@ -0,0 +1,110 @@ +''' +''' + + +from __future__ import annotations + +import typing +import yaml + + +STRUCTURE={ + 'url':str, + 'token':str, + 'owner':str, + 'repo':str, + 'feeds':list, + 'label':str +} + +STRUCTURE_DICT_LIST={ + 'feeds': { + 'url':str, + 'name':str, + 'assign':str, + 'exclude':list, + 'include':list, + } +} + + + +class ConfigError(Exception): + pass + +class Config: + ''' + Imports and stores curvegenerator parameters + + Example: + ```yaml + # test.yaml + --- + testval: 4 + ``` + ```python + conf = Config("test.yaml") + print(conf.testval) + ``` + ''' + def __init__(self, _file): + ''' + Constructor + + Args: + _file (str): Filename to yaml configfile + ''' + + if _file is None: + return + + with open(_file, 'r') as f: + self.config = yaml.load(f.read(), Loader=yaml.FullLoader) + + self._validate() + + def __iter__(self): + self.n = 0 + pass + + def __next__(self): + pass + + def __getitem__(self, _key): + self.config[_key] + + @staticmethod + def _validate_dict(_dict, _spec, _context=''): + for e in _spec: + if e not in _dict: + if _spec[e] is not list: + raise ConfigError(f'{_context}Key {e} is not set.') + else: + pass + elif type(_dict[e]) is not _spec[e]: + raise ConfigError(f'{_context}Key {e} is {type(_dict[e])}. Should be {_spec[e]}') + + def _validate(self): + Config._validate_dict(self.config, STRUCTURE) + + for lst in STRUCTURE_DICT_LIST: + for e in self.config[lst]: + Config._validate_dict(e, STRUCTURE_DICT_LIST[lst], 'feeds: ') + + def load(self, _dict): + self.config = _dict + + def __getattr__(self, _attr) -> float | int | Config | None: + if _attr not in self.config: + return None + + if isinstance(self.config[_attr], dict): + ret = Config(None) + ret.load(self.config[_attr]) + return ret + + return self.config[_attr] + + def __str__(self): + return yaml.dump(self.config) + diff --git a/rss_to_gitea/gitea.py b/rss_to_gitea/gitea.py index 39ea763..fcc985c 100644 --- a/rss_to_gitea/gitea.py +++ b/rss_to_gitea/gitea.py @@ -32,11 +32,30 @@ class GiteaAPI: return result.json() + def _api_post(self, _endpoint, _data): + headers={ + 'Authorization':f'token {self.token}', + 'Content-Type': 'application/json', + 'accept': 'application/json' + } + result = requests.post(f'{self.address}/{_endpoint}',headers=headers, json=_data) + + return result.json() + + + def createIssue(self, _owner, _repo, _title, _content, _assign, _labels): + data={ + 'assignee':_assign, +# 'body':_content, + 'labels':_labels, + 'title':_title + } + + result = self._api_post(f'repos/{_owner}/{_repo}/issues', data ) - def createIssue(self, _owner, _repo, _title, _content): - pass + return result - def searchIssue(self, _owner, _repo, _title, _labels, _state='open'): + def searchIssue(self, _owner, _repo, _title, _labels, _state='all'): data= { 'state':_state, 'labels':_labels, @@ -47,8 +66,23 @@ class GiteaAPI: result = self._api_get(f'repos/{_owner}/{_repo}/issues', data ) for issue in result: - print(issue['title']) + if issue['title'] == _title: + return issue + + return None + + def getLabelId(self, _owner, _repo, _label): + data= {} + + result = self._api_get(f'repos/{_owner}/{_repo}/labels', data ) + + label_filtered = filter(lambda a: a['name']==_label, result) + label = list(label_filtered) + + if len(label) != 1: + print('No or more than one label found') + return None + + return label[0]['id'] - def updateIssue(self, _owner, _repo, _issueid): - pass diff --git a/rss_to_gitea/main.py b/rss_to_gitea/main.py index f4425a2..c1a4a11 100644 --- a/rss_to_gitea/main.py +++ b/rss_to_gitea/main.py @@ -6,29 +6,33 @@ import sys from .gitea import GiteaAPI from .atom import AtomFeed +from .config import Config +def print_help(): + print(''' +USAGE: + rsstogitea <config> + ''') -feeds = [ - { - 'name':'Gitea', - 'url':'https://github.com/go-gitea/gitea/releases.atom', - 'exclude':['dev', 'rc'], - 'assign':'' - } -] +def main(): + if len(sys.argv) <= 1: + return 1 -def load_yaml(_file: str): - required = ['feeds', 'token', 'url'] - config = {} + config = Config(sys.argv[1]) + api = GiteaAPI("https://gitea.my.cum.re", config.token) - with open(_file, 'r') as f: - config = yaml.load(f.read(), Loader=yaml.FullLoader) + label_id = api.getLabelId(config.owner, config.repo, config.label) - return config + for feed in config.feeds: + remote = AtomFeed(feed['url']) + latest = remote.get_latest(feed['exclude'], feed['include']) -def main(): - #token = sys.argv[1] - #api = GiteaAPI("https://gitea.my.cum.re", token) - #api.searchIssue('infra', 'ansible', '', 'update') - feed = AtomFeed(feeds[0]['url']) - print(feed.get_latest([])) + issue_title = f'{feed["name"]}: {latest["title"]}' + print("Title=", issue_title) + + ticket = api.searchIssue(config.owner, config.repo, issue_title, [config.label]) + if ticket is not None: + print(f'{issue_title} already exists. Skipping') + continue + + result = api.createIssue(config.owner, config.repo, issue_title, '', feed['assign'], [label_id]) |