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
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
|
import itertools
import time
import six
import requests
from bugwarrior.services import IssueService, Issue, ServiceClient
from bugwarrior.config import die
import logging
log = logging.getLogger(__name__)
class ActiveCollab2Client(ServiceClient):
def __init__(self, url, key, user_id, projects, target):
self.url = url
self.key = key
self.user_id = user_id
self.projects = projects
self.target = target
def get_task_dict(self, project, key, task):
assigned_task = {
'project': project
}
if task[u'type'] == 'Ticket':
# Load Ticket data
# @todo Implement threading here.
ticket_data = self.call_api(
"/projects/" + six.text_type(task[u'project_id']) +
"/tickets/" + six.text_type(task[u'ticket_id']))
assignees = ticket_data[u'assignees']
for assignee in assignees:
if (
(assignee[u'is_owner'] is True)
and (assignee[u'user_id'] == int(self.user_id))
):
assigned_task.update(ticket_data)
return assigned_task
elif task[u'type'] == 'Task':
# Load Task data
assigned_task.update(task)
return assigned_task
def get_issue_generator(self, user_id, project_id, project_name):
"""
Approach:
1. Get user ID from bugwarriorrc file
2. Get list of tickets from /user-tasks for a given project
3. For each ticket/task returned from #2, get ticket/task info and
check if logged-in user is primary (look at `is_owner` and
`user_id`)
"""
user_tasks_data = self.call_api(
"/projects/" + six.text_type(project_id) + "/user-tasks")
for key, task in enumerate(user_tasks_data):
assigned_task = self.get_task_dict(project_id, key, task)
if assigned_task:
log.debug(
" Adding '" + assigned_task['description'] +
"' to task list.")
yield assigned_task
def call_api(self, uri):
url = self.url.rstrip("/")
params = {
'token': self.key,
'path_info': uri,
'format': 'json'}
return self.json_response(requests.get(url, params=params))
class ActiveCollab2Issue(Issue):
BODY = 'ac2body'
NAME = 'ac2name'
PERMALINK = 'ac2permalink'
TICKET_ID = 'ac2ticketid'
PROJECT_ID = 'ac2projectid'
TYPE = 'ac2type'
CREATED_ON = 'ac2createdon'
CREATED_BY_ID = 'ac2createdbyid'
UDAS = {
BODY: {
'type': 'string',
'label': 'ActiveCollab2 Body'
},
NAME: {
'type': 'string',
'label': 'ActiveCollab2 Name'
},
PERMALINK: {
'type': 'string',
'label': 'ActiveCollab2 Permalink'
},
TICKET_ID: {
'type': 'string',
'label': 'ActiveCollab2 Ticket ID'
},
PROJECT_ID: {
'type': 'string',
'label': 'ActiveCollab2 Project ID'
},
TYPE: {
'type': 'string',
'label': 'ActiveCollab2 Task Type'
},
CREATED_ON: {
'type': 'date',
'label': 'ActiveCollab2 Created On'
},
CREATED_BY_ID: {
'type': 'string',
'label': 'ActiveCollab2 Created By'
},
}
UNIQUE_KEY = (PERMALINK, )
PRIORITY_MAP = {
-2: 'L',
-1: 'L',
0: 'M',
1: 'H',
2: 'H',
}
def to_taskwarrior(self):
record = {
'project': self.record['project'],
'priority': self.get_priority(),
'due': self.parse_date(self.record.get('due_on')),
self.PERMALINK: self.record['permalink'],
self.TICKET_ID: self.record['ticket_id'],
self.PROJECT_ID: self.record['project_id'],
self.TYPE: self.record['type'],
self.CREATED_ON: self.parse_date(self.record.get('created_on')),
self.CREATED_BY_ID: self.record['created_by_id'],
self.BODY: self.record.get('body'),
self.NAME: self.record.get('name'),
}
return record
def get_default_description(self):
record_type = self.record['type'].lower()
record_type = 'issue' if record_type == 'ticket' else record_type
return self.build_default_description(
title=(
self.record['name']
if self.record['name']
else self.record['body']
),
url=self.get_processed_url(self.record['permalink']),
number=self.record['ticket_id'],
cls=record_type,
)
class ActiveCollab2Service(IssueService):
ISSUE_CLASS = ActiveCollab2Issue
CONFIG_PREFIX = 'activecollab2'
def __init__(self, *args, **kw):
super(ActiveCollab2Service, self).__init__(*args, **kw)
self.url = self.config.get('url').rstrip('/')
self.key = self.config.get('key')
self.user_id = self.config.get('user_id')
projects_raw = self.config.get('projects')
projects_list = projects_raw.split(',')
projects = []
for k, v in enumerate(projects_list):
project_data = v.strip().split(":")
project = dict([(project_data[0], project_data[1])])
projects.append(project)
self.projects = projects
self.client = ActiveCollab2Client(
self.url, self.key, self.user_id, self.projects, self.target
)
@classmethod
def validate_config(cls, service_config, target):
for k in (
'url',
'key',
'projects',
'user_id'
):
if k not in service_config:
die("[%s] has no 'activecollab2.%s'" % (target, k))
super(ActiveCollab2Service, cls).validate_config(service_config, target)
def issues(self):
# Loop through each project
start = time.time()
issue_generators = []
projects = self.projects
for project in projects:
for project_id, project_name in project.items():
log.debug(
" Getting tasks for #" + project_id +
" " + project_name + '"')
issue_generators.append(
self.client.get_issue_generator(
self.user_id, project_id, project_name
)
)
log.debug(" Elapsed Time: %s" % (time.time() - start))
for record in itertools.chain(*issue_generators):
yield self.get_issue_for_record(record)
|
