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
|
# Copyright 2009-2011 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
"""Utilities for the update-bugzilla-remote-components cronscript"""
__metaclass__ = type
__all__ = [
'BugzillaRemoteComponentFinder',
'BugzillaRemoteComponentScraper',
]
import re
from urllib2 import (
HTTPError,
urlopen,
)
from BeautifulSoup import BeautifulSoup
from canonical.launchpad.scripts.logger import log as default_log
from zope.component import getUtility
from lp.bugs.interfaces.bugtracker import (
BugTrackerType,
IBugTrackerSet,
)
from lp.bugs.model.bugtracker import (
BugTrackerComponent,
)
from canonical.launchpad.interfaces.lpstorm import IStore
def dictFromCSV(line):
items_dict = {}
for item in line.split(","):
item = item.strip()
item = item.replace("'", "")
item = item.replace("\\", "")
items_dict[item] = {
'name': item,
}
return items_dict
class BugzillaRemoteComponentScraper:
"""Scrapes Bugzilla query.cgi page for lists of products and components"""
re_cpts = re.compile(r'cpts\[(\d+)\] = \[(.*)\]')
re_vers = re.compile(r'vers\[(\d+)\] = \[(.*)\]')
def __init__(self, base_url=None):
self.base_url = re.sub(r'/$', '', base_url)
self.url = "%s/query.cgi?format=advanced" % (self.base_url)
self.products = {}
def getPage(self):
return urlopen(self.url).read()
def parsePage(self, page_text):
soup = BeautifulSoup(page_text)
if soup is None:
return None
# Load products into a list since Bugzilla references them
# by index number
products = []
for product in soup.find(
name='select',
onchange="doOnSelectProduct(2);").contents:
if product.string != "\n":
products.append({
'name': product.string,
'components': {},
'versions': None,
})
for script_text in soup.findAll(name="script"):
if script_text is None or script_text.string is None:
continue
for line in script_text.string.split(";"):
m = self.re_cpts.search(line)
if m:
num = int(m.group(1))
products[num]['components'] = dictFromCSV(m.group(2))
m = self.re_vers.search(line)
if m:
num = int(m.group(1))
products[num]['versions'] = dictFromCSV(m.group(2))
# Re-map list into dict for easier lookups
for product in products:
product_name = product['name']
self.products[product_name] = product
return True
class BugzillaRemoteComponentFinder:
"""Updates remote components for all Bugzillas registered in Launchpad"""
# Names of bug trackers we should not pull data from
_BLACKLIST = [
u"ubuntu-bugzilla",
u"mozilla.org",
]
def __init__(self, logger=None, static_bugzilla_text=None):
"""Instantiates object, without performing any parsing.
:param logger: A logger object
:param static_bugzilla_text: Instead of retrieving the remote
web page for a bug tracker, act as if this static text was
returned. This is intended for testing purposes to avoid
needing to make remote web connections.
"""
self.logger = logger
if logger is None:
self.logger = default_log
self.static_bugzilla_text = static_bugzilla_text
def getRemoteProductsAndComponents(self, bugtracker_name=None):
lp_bugtrackers = getUtility(IBugTrackerSet)
if bugtracker_name is not None:
lp_bugtrackers = [
lp_bugtrackers.getByName(bugtracker_name),
]
if not lp_bugtrackers or len(lp_bugtrackers) != 1:
self.logger.warning(
"Could not find specified bug tracker %s",
bugtracker_name)
for lp_bugtracker in lp_bugtrackers:
if lp_bugtracker.bugtrackertype != BugTrackerType.BUGZILLA:
continue
if lp_bugtracker.name in self._BLACKLIST:
continue
self.logger.info("%s: %s" % (
lp_bugtracker.name, lp_bugtracker.baseurl))
bz_bugtracker = BugzillaRemoteComponentScraper(
base_url=lp_bugtracker.baseurl)
if self.static_bugzilla_text is not None:
self.logger.debug("Using static bugzilla text")
page_text = self.static_bugzilla_text
else:
try:
self.logger.debug("...Fetching page")
page_text = bz_bugtracker.getPage()
except HTTPError, error:
self.logger.error("Error fetching %s: %s" % (
lp_bugtracker.baseurl, error))
continue
self.logger.debug("...Parsing html")
bz_bugtracker.parsePage(page_text)
self.logger.debug("...Storing new data to Launchpad")
self.storeRemoteProductsAndComponents(
bz_bugtracker, lp_bugtracker)
def storeRemoteProductsAndComponents(self, bz_bugtracker, lp_bugtracker):
components_to_add = []
for product in bz_bugtracker.products.itervalues():
# Look up the component group id from Launchpad for the product
# if it already exists. Otherwise, add it.
lp_component_group = lp_bugtracker.getRemoteComponentGroup(
product['name'])
if lp_component_group is None:
lp_component_group = lp_bugtracker.addRemoteComponentGroup(
product['name'])
if lp_component_group is None:
self.logger.warning("Failed to add new component group")
continue
else:
for component in lp_component_group.components:
if (component.name in product['components'] or
component.is_visible == False or
component.is_custom == True):
# We already know something about this component,
# or a user has configured it, so ignore it
del product['components'][component.name]
else:
# Component is now missing from Bugzilla,
# so drop it here too
component.remove()
# The remaining components in the collection will need to be
# added to launchpad. Record them for now.
for component in product['components'].values():
components_to_add.append(
"('%s', %d, 'True', 'False')" % (
component['name'], lp_component_group.id))
if len(components_to_add) > 0:
sqltext = """
INSERT INTO BugTrackerComponent
(name, component_group, is_visible, is_custom)
VALUES %s""" % ",\n ".join(components_to_add)
self.logger.debug("...Inserting components into database")
store = IStore(BugTrackerComponent)
store.execute(sqltext)
store.commit()
store.flush()
self.logger.debug("...Done")
|