1from builtins import str 2import debianbts 3import requests 4 5from bugwarrior.config import die, asbool 6from bugwarrior.services import Issue, IssueService, ServiceClient 7 8import logging 9log = logging.getLogger(__name__) 10 11UDD_BUGS_SEARCH = "https://udd.debian.org/bugs/" 12 13 14class BTSIssue(Issue): 15 SUBJECT = 'btssubject' 16 URL = 'btsurl' 17 NUMBER = 'btsnumber' 18 PACKAGE = 'btspackage' 19 SOURCE = 'btssource' 20 FORWARDED = 'btsforwarded' 21 STATUS = 'btsstatus' 22 23 UDAS = { 24 SUBJECT: { 25 'type': 'string', 26 'label': 'Debian BTS Subject', 27 }, 28 URL: { 29 'type': 'string', 30 'label': 'Debian BTS URL', 31 }, 32 NUMBER: { 33 'type': 'numeric', 34 'label': 'Debian BTS Number', 35 }, 36 PACKAGE: { 37 'type': 'string', 38 'label': 'Debian BTS Package', 39 }, 40 SOURCE: { 41 'type': 'string', 42 'label': 'Debian BTS Source Package', 43 }, 44 FORWARDED: { 45 'type': 'string', 46 'label': 'Debian BTS Forwarded URL', 47 }, 48 STATUS: { 49 'type': 'string', 50 'label': 'Debian BTS Status', 51 } 52 } 53 UNIQUE_KEY = (URL, ) 54 55 PRIORITY_MAP = { 56 'wishlist': 'L', 57 'minor': 'L', 58 'normal': 'M', 59 'important': 'M', 60 'serious': 'H', 61 'grave': 'H', 62 'critical': 'H', 63 } 64 65 def to_taskwarrior(self): 66 return { 67 'priority': self.get_priority(), 68 'annotations': self.extra.get('annotations', []), 69 70 self.URL: self.record['url'], 71 self.SUBJECT: self.record['subject'], 72 self.NUMBER: self.record['number'], 73 self.PACKAGE: self.record['package'], 74 self.SOURCE: self.record['source'], 75 self.FORWARDED: self.record['forwarded'], 76 self.STATUS: self.record['status'], 77 } 78 79 def get_default_description(self): 80 81 return self.build_default_description( 82 title=self.record['subject'], 83 url=self.get_processed_url(self.record['url']), 84 number=self.record['number'], 85 cls='issue' 86 ) 87 88 def get_priority(self): 89 return self.PRIORITY_MAP.get( 90 self.record.get('severity'), 91 self.origin['default_priority'] 92 ) 93 94 95class BTSService(IssueService, ServiceClient): 96 ISSUE_CLASS = BTSIssue 97 CONFIG_PREFIX = 'bts' 98 99 def __init__(self, *args, **kw): 100 super(BTSService, self).__init__(*args, **kw) 101 self.email = self.config.get('email', default=None) 102 self.packages = self.config.get('packages', default=None) 103 self.udd = self.config.get( 104 'udd', default=False, to_type=asbool) 105 self.udd_ignore_sponsor = self.config.get( 106 'udd_ignore_sponsor', default=True, to_type=asbool) 107 self.ignore_pkg = self.config.get('ignore_pkg', default=None) 108 self.ignore_src = self.config.get('ignore_src', default=None) 109 self.ignore_pending = self.config.get( 110 'ignore_pending', default=True, to_type=asbool) 111 112 @classmethod 113 def validate_config(cls, service_config, target): 114 if ('udd' in service_config and 115 asbool(service_config.get('udd')) and 116 'email' not in service_config): 117 die("[%s] has no 'bts.email' but UDD search was requested" % 118 (target,)) 119 120 if 'packages' not in service_config and 'email' not in service_config: 121 die("[%s] has neither 'bts.email' or 'bts.packages'" % (target,)) 122 123 if ('udd_ignore_sponsor' in service_config and 124 (not asbool(service_config.get('udd')))): 125 die("[%s] defines settings for UDD search without enabling" 126 " UDD search" % (target,)) 127 128 IssueService.validate_config(service_config, target) 129 130 def _record_for_bug(self, bug): 131 return {'number': bug.bug_num, 132 'url': 'https://bugs.debian.org/' + str(bug.bug_num), 133 'package': bug.package, 134 'subject': bug.subject, 135 'severity': bug.severity, 136 'source': bug.source, 137 'forwarded': bug.forwarded, 138 'status': bug.pending, 139 } 140 141 def _get_udd_bugs(self): 142 request_params = { 143 'format': 'json', 144 'dmd': 1, 145 'email1': self.email, 146 } 147 if self.udd_ignore_sponsor: 148 request_params['nosponsor1'] = "on" 149 resp = requests.get(UDD_BUGS_SEARCH, request_params) 150 return self.json_response(resp) 151 152 def annotations(self, issue, issue_obj): 153 return self.build_annotations( 154 [], 155 issue_obj.get_processed_url(issue['url']) 156 ) 157 158 def issues(self): 159 # Initialise empty list of bug numbers 160 collected_bugs = [] 161 162 # Search BTS for bugs owned by email address 163 if self.email: 164 owned_bugs = debianbts.get_bugs("owner", self.email, 165 "status", "open") 166 collected_bugs.extend(owned_bugs) 167 168 # Search BTS for bugs related to specified packages 169 if self.packages: 170 packages = self.packages.split(",") 171 for pkg in packages: 172 pkg_bugs = debianbts.get_bugs("package", pkg, 173 "status", "open") 174 for bug in pkg_bugs: 175 if bug not in collected_bugs: 176 collected_bugs.append(bug) 177 178 # Search UDD bugs search for bugs belonging to packages that 179 # are maintained by the email address 180 if self.udd: 181 udd_bugs = self._get_udd_bugs() 182 for bug in udd_bugs: 183 if bug not in collected_bugs: 184 collected_bugs.append(bug['id']) 185 186 issues = [self._record_for_bug(bug) 187 for bug in debianbts.get_status(collected_bugs)] 188 189 log.debug(" Found %i total.", len(issues)) 190 191 if self.ignore_pkg: 192 ignore_pkg = self.ignore_pkg.split(",") 193 for pkg in ignore_pkg: 194 issues = [issue 195 for issue in issues if not issue['package'] == pkg] 196 197 if self.ignore_src: 198 ignore_src = self.ignore_src.split(",") 199 for src in ignore_src: 200 issues = [issue 201 for issue in issues if not issue['source'] == src] 202 203 if self.ignore_pending: 204 issues = [issue 205 for issue in issues 206 if not issue['status'] == 'pending-fixed'] 207 208 issues = [issue 209 for issue in issues 210 if not (issue['status'] == 'done' or 211 issue['status'] == 'fixed')] 212 213 log.debug(" Pruned down to %i.", len(issues)) 214 215 for issue in issues: 216 issue_obj = self.get_issue_for_record(issue) 217 extra = { 218 'annotations': self.annotations(issue, issue_obj) 219 } 220 issue_obj.update_extra(extra) 221 yield issue_obj 222