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
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
|
# Copyright 2009 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
__metaclass__ = type
__all__ = [
'ProductReleaseFinder'
]
from datetime import datetime
import mimetypes
import os
import re
import urllib
import urlparse
from cscvs.dircompare import path
import pytz
from zope.component import getUtility
from lp.app.validators.name import invalid_name_pattern
from lp.app.validators.version import sane_version
from lp.registry.interfaces.product import IProductSet
from lp.registry.interfaces.productrelease import UpstreamFileType
from lp.registry.interfaces.series import SeriesStatus
from lp.registry.scripts.productreleasefinder.filter import FilterPattern
from lp.registry.scripts.productreleasefinder.hose import Hose
processors = '|'.join([
'all',
'amd64',
'arm',
'armel',
'i386',
'intel',
'hppa',
'hurd-i386',
'ia64',
'm68k',
'mips',
'mipsel',
'powerpc',
's390',
'sparc',
])
flavor_pattern = re.compile(r"""
(~ # Packaging target
|_[a-z][a-z]_[A-Z][A-Z] # or language version
|_(%s) # or processor version
|[\.-](win32|OSX) # or OS version
|\.(deb|noarch|rpm|dmg|exe) # or packaging version
).* # to the end of the string
""" % processors, re.VERBOSE)
def extract_version(filename):
"""Return the release version of the file, or None.
Ensure the version is compatible with Launchpad. None is returned
if a version could not be extracted.
"""
version = path.split_version(path.name(filename))[1]
if version is None:
return None
# Tarballs pulled from a Debian-style archive often have
# ".orig" appended to the version number. We don't want this.
if version.endswith('.orig'):
version = version[:-len('.orig')]
# Remove processor and language flavors from the version:
# eg. _de_DE, _all, _i386.
version = flavor_pattern.sub('', version)
# Bug #599250. If there is no file extension after extracting
# the version number, we have added an unknown file extension to the
# version. Ignore this dud match.
if filename.endswith(version):
return None
# Launchpad requires all versions to be lowercase. They may contain
# letters, numbers, dots, underscores, and hyphens (a-z0-9._-).
version = version.lower()
version = invalid_name_pattern.sub('-', version)
version = version.replace('+', '-')
return version
class ProductReleaseFinder:
def __init__(self, ztm, log):
self.ztm = ztm
self.log = log
def findReleases(self):
"""Scan for new releases in all products."""
for product_name, filters in self.getFilters():
self.handleProduct(product_name, filters)
def getFilters(self):
"""Build the list of products and filters.
Returns a list of (product_name, filters) for each product in
the database, where the filter keys are series names.
"""
todo = []
self.ztm.begin()
products = getUtility(IProductSet)
for product in products.get_all_active(eager_load=False):
filters = []
for series in product.series:
if (series.status == SeriesStatus.OBSOLETE
or not series.releasefileglob):
continue
filters.append(FilterPattern(series.name,
series.releasefileglob))
if not len(filters):
continue
self.log.info("%s has %d series with information", product.name,
len(filters))
todo.append((product.name, filters))
self.ztm.abort()
return todo
def handleProduct(self, product_name, filters):
"""Scan for tarballs and create ProductReleases for the given product.
"""
hose = Hose(filters, log_parent=self.log)
for series_name, url in hose:
if series_name is not None:
try:
self.handleRelease(product_name, series_name, url)
except (KeyboardInterrupt, SystemExit):
raise
except:
self.log.exception("Could not successfully process "
"URL %s for %s/%s",
url, product_name, series_name)
else:
self.log.debug("File in %s found that matched no glob: %s",
product_name, url)
def hasReleaseFile(self, product_name, series_name,
release_name, filename):
"""Return True if we have a tarball for the given product release."""
has_file = False
self.ztm.begin()
try:
product = getUtility(IProductSet).getByName(product_name)
if product is not None:
series = product.getSeries(series_name)
if series is not None:
release = series.getRelease(release_name)
if release is not None:
for fileinfo in release.files:
if filename == fileinfo.libraryfile.filename:
has_file = True
break
finally:
self.ztm.abort()
return has_file
def addReleaseTarball(self, product_name, series_name, release_name,
filename, size, file, content_type):
"""Create a ProductRelease (if needed), and attach tarball"""
# Get the series.
self.ztm.begin()
try:
product = getUtility(IProductSet).getByName(product_name)
# XXX: This might match a milestone on a product series that was
# not intended, since product release used to have unique
# names per product series, but are now dependent on the milestone
# name which is unique per product. The series_name method
# parameter can be removed.
milestone = product.getMilestone(release_name)
if milestone is None:
series = product.getSeries(series_name)
milestone = series.newMilestone(release_name)
# Normally, a milestone is deactived when that version is
# released. This is only safe to do in an automated script
# if we are not using a pre-existing milestone.
milestone.active = False
release = milestone.product_release
if release is None:
release = milestone.createProductRelease(
owner=product.owner, datereleased=datetime.now(pytz.UTC))
self.log.info("Created new release %s for %s/%s",
release_name, product_name, series_name)
release.addReleaseFile(
filename, file, content_type, uploader=product.owner)
self.ztm.commit()
except:
self.ztm.abort()
raise
def handleRelease(self, product_name, series_name, url):
"""If the given URL looks like a release tarball, download it
and create a corresponding ProductRelease."""
filename = urlparse.urlsplit(url)[2]
slash = filename.rfind("/")
if slash != -1:
filename = filename[slash+1:]
self.log.debug("Filename portion is %s", filename)
version = extract_version(filename)
if version is None:
self.log.info("Unable to parse version from %s", url)
return
self.log.debug("Version is %s", version)
if not sane_version(version):
self.log.error("Version number '%s' for '%s' is not sane",
version, url)
return
if self.hasReleaseFile(product_name, series_name, version, filename):
self.log.debug("Already have a tarball for release %s", version)
return
mimetype, encoding = mimetypes.guess_type(url)
self.log.debug("Mime type is %s", mimetype)
if mimetype is None:
mimetype = 'application/octet-stream'
self.log.info("Downloading %s", url)
try:
local, headers = urllib.urlretrieve(url)
stat = os.stat(local)
except IOError:
self.log.error("Download of %s failed", url)
raise
except OSError:
self.log.error("Unable to stat downloaded file")
raise
fp = open(local, 'r')
os.unlink(local)
self.addReleaseTarball(product_name, series_name, version,
filename, stat.st_size, fp, mimetype)
|