aboutsummaryrefslogtreecommitdiff
path: root/rss_to_gitea/gitea.py
blob: dbfba8e9985e457c66748504583b14ace15550ce (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
import requests
import json

APIBASE="/api/v1"

class GiteaAPIAuthException (Exception):
    pass

class GiteaAPIException (Exception):
    pass

class GiteaAPI:
    def __init__(self, _url, _token):
        self.token = _token
        self.address = _url.strip('/') + APIBASE

        headers={'Authorization':f'token {self.token}'}
        result = requests.get(f'{self.address}/user',headers=headers)

        if result.status_code != 200:
            raise GiteaAPIAuthException(result.json()['message'])

        self.username = result.json()['login']

    def _api_get(self, _endpoint, _params):
        headers={
                'Authorization':f'token {self.token}',
                'Content-Type': 'application/json',
                'accept': 'application/json'
            }
        result = requests.get(f'{self.address}/{_endpoint}',headers=headers, params=_params)

        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 _api_patch(self, _endpoint, _data):
        headers={
                'Authorization':f'token {self.token}',
                'Content-Type': 'application/json',
                'accept': 'application/json'
            }
        result = requests.patch(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 )

        return result

    def changeIssueState(self, _owner, _repo, _index, _state):
        data= {
            'state':_state,
        }

        result = self._api_patch(f'repos/{_owner}/{_repo}/issues/{_index}', data )


    def getFirstExactIssue(self, _owner, _repo, _title, _labels, _state='all'):
        data= {
            'state':_state,
            'labels':_labels,
            'created_by':self.username,
            'q':_title
        }

        result = self._api_get(f'repos/{_owner}/{_repo}/issues', data )

        for issue in result:
            if issue['title'] == _title:
                return issue

        return None

    def getAllIssuesStartingWith(self, _owner, _repo, _title, _labels, _state='all'):
        ret = []

        data= {
            'state':_state,
            'labels':_labels,
            'created_by':self.username,
            'q':_title
        }

        result = self._api_get(f'repos/{_owner}/{_repo}/issues', data )

        for issue in result:
            if issue['title'].startswith(_title):
                ret.append(issue)

        return ret

    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']