Page MenuHomeSoftware Heritage
Files F9339447

No OneTemporary
Actions

View Options

diff --git a/swh/loader/package/loader.py b/swh/loader/package/loader.py
index c1cd674..75d8d83 100644
--- a/swh/loader/package/loader.py
+++ b/swh/loader/package/loader.py
@@ -1,477 +1,477 @@
# Copyright (C) 2019 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information
import os
import shutil
import requests
try:
from _version import __version__
except ImportError:
__version__ = 'devel'
from tempfile import mkdtemp
from swh.core import tarball
from swh.loader.core.utils import clean_dangling_folders
from swh.loader.core.loader import BufferedLoader
from swh.model.identifiers import normalize_timestamp
from swh.model.hashutil import MultiHash, HASH_BLOCK_SIZE
from swh.model.from_disk import Directory
from swh.model.identifiers import (
identifier_to_bytes, revision_identifier, snapshot_identifier
)
DEBUG_MODE = '** DEBUG MODE **'
class GNULoader(BufferedLoader):
SWH_PERSON = {
'name': b'Software Heritage',
'fullname': b'Software Heritage',
'email': b'robot@softwareheritage.org'
}
REVISION_MESSAGE = b'swh-loader-package: synthetic revision message'
visit_type = 'gnu'
def __init__(self):
self.TEMPORARY_DIR_PREFIX_PATTERN = 'swh.loader.gnu.'
super().__init__(logging_class='swh.loader.package.GNULoader')
self.dir_path = None
temp_directory = self.config['temp_directory']
os.makedirs(temp_directory, exist_ok=True)
self.temp_directory = mkdtemp(
suffix='-%s' % os.getpid(),
prefix=self.TEMPORARY_DIR_PREFIX_PATTERN,
dir=temp_directory)
self.debug = self.config.get('debug', False)
self.session = requests.session()
self.params = {
'headers': {
'User-Agent': 'Software Heritage Loader (%s)' % (
__version__
)
}
}
def pre_cleanup(self):
"""To prevent disk explosion if some other workers exploded
in mid-air (OOM killed), we try and clean up dangling files.
"""
if self.debug:
self.log.warning('%s Will not pre-clean up temp dir %s' % (
DEBUG_MODE, self.temp_directory
))
return
clean_dangling_folders(self.temp_directory,
pattern_check=self.TEMPORARY_DIR_PREFIX_PATTERN,
log=self.log)
def prepare_origin_visit(self, name, origin_url, **kwargs):
"""Prepare package visit.
Args:
name (str): Package Name
origin_url (str): Package origin url
**kwargs: Arbitrary keyword arguments passed by the lister.
"""
# reset statuses
self._load_status = 'uneventful'
self._visit_status = 'full'
self.done = False
self.origin = {
'url': origin_url,
'type': self.visit_type,
}
self.visit_date = None # loader core will populate it
def prepare(self, name, origin_url, **kwargs):
"""Prepare effective loading of source tarballs for a package manager
package.
Args:
name (str): Package Name
origin_url (str): Package origin url
**kwargs: Arbitrary keyword arguments passed by the lister.
"""
self.package_contents = []
self.package_directories = []
self.package_revisions = []
self.all_version_data = []
self.latest_timestamp = 0
# Conceled the data into one dictionary to eleminate the need of
- # passing all the paramaters when required in some method
+ # passing all the parameters when required in some method
self.package_details = {
'name': name,
'origin_url': origin_url,
'tarballs': kwargs['tarballs'],
}
self.package_temp_dir = os.path.join(self.temp_directory,
self.package_details['name'])
self.new_versions = \
self.prepare_package_versions(self.package_details['tarballs'])
def prepare_package_versions(self, tarballs):
"""
Instantiate a generator that will process a specific package release
version at each iteration step. The following operations will be
performed:
1. Create a temporary directory to download and extract the
release tarball.
2. Download the tarball.
3. Uncompress the tarball.
4. Parse the file associated to the package version to extract
metadata (optional).
5. Delete unnecessary files (optional).
Args:
tarballs (list): a list of dicts containing information about the
respective tarball that is provided by lister.
known_versions (dict): may be provided by the loader, it enables
to filter out versions already ingested in the archive.
Yields:
Tuple[dict, str]: tuples containing the following
members:
* a dict holding package tarball information and metadata
* a string holding the path of the uncompressed package to
load into the archive
"""
for package_version_data in tarballs:
tarball_url = package_version_data['archive']
tarball_request = self._request(tarball_url,
throw_error=False)
if tarball_request.status_code == 404:
self.log.warning('Tarball url %s returns a 404 error.',
tarball_url)
self._visit_status = 'partial'
# FIX ME: Do we need to mark it `partial` here
continue
yield self._prepare_package_version(package_version_data,
tarball_request)
def _request(self, url, throw_error=True):
"""Request the remote tarball url.
Args:
url (str): Url (file or http*).
Raises:
ValueError in case of failing to query.
Returns:
Tuple of local (filepath, hashes of filepath).
"""
response = self.session.get(url, **self.params, stream=True)
if response.status_code != 200 and throw_error:
raise ValueError("Fail to query '%s'. Reason: %s" % (
url, response.status_code))
return response
def _prepare_package_version(self, package_version_data, tarball_request):
"""Process the package release version.
The following operations are performed:
1. Download the tarball
2. Uncompress the tarball
3. Delete unnecessary files (optional)
4. Parse the file associated to the package version to extract
metadata (optional)
Args:
package_version_data (dict): containing information
about the focused package version.
known_versions (dict): may be provided by the loader, it enables
to filter out versions already ingested in the archive.
Return:
Tuple[dict, str]: tuples containing the following
members:
* a dict holding package tarball information and metadata
* a string holding the path of the uncompressed package to
load into the archive
"""
url = package_version_data['archive']
tarball_path, hashes = self.download_generate_hash(tarball_request,
url)
uncompressed_path = os.path.join(self.package_temp_dir, 'uncompressed',
os.path.basename(url)) # SEE ME
self.uncompress_tarball(tarball_path, uncompressed_path)
# remove tarball
os.remove(tarball_path)
if self.tarball_invalid:
return None, None
return package_version_data, uncompressed_path
def download_generate_hash(self, response, url):
"""Store file in temp directory and computes hash of its filepath.
Args:
response (Response): Server response of the url
url (str): Url of the tarball
Returns:
Tuple of local (filepath, hashes of filepath)
"""
length = int(response.headers['content-length'])
os.makedirs(self.package_temp_dir, exist_ok=True)
# SEE ME
filepath = os.path.join(self.package_temp_dir, os.path.basename(url))
# Convert the server response to a file.
h = MultiHash(length=length)
with open(filepath, 'wb') as f:
for chunk in response.iter_content(chunk_size=HASH_BLOCK_SIZE):
h.update(chunk)
f.write(chunk)
# Check for the validity of the tarball downloaded.
actual_length = os.path.getsize(filepath)
if length != actual_length:
raise ValueError('Error when checking size: %s != %s' % (
length, actual_length))
hashes = {
'length': length,
**h.hexdigest()
}
return filepath, hashes
def uncompress_tarball(self, filepath, path):
"""Uncompress a tarball.
Args:
filepath (str): Path of tarball to uncompress
path (str): The destination folder where to uncompress the tarball
Returns:
The nature of the tarball, zip or tar.
"""
try:
self.tarball_invalid = False
tarball.uncompress(filepath, path)
except Exception:
self.tarball_invalid = True
self._visit_status = 'partial'
def fetch_data(self):
"""Called once per release artifact version (can be many for one
release).
This will for each call:
- retrieve a release artifact (associated to a release version)
- Computes the swh objects
Returns:
True as long as data to fetch exist
"""
data = None
if self.done:
return False
try:
data = next(self.new_versions)
self._load_status = 'eventful'
except StopIteration:
self.done = True
return False
package_version_data, dir_path = data
# package release tarball was corrupted
if self.tarball_invalid:
return not self.done
dir_path = dir_path.encode('utf-8')
directory = Directory.from_disk(path=dir_path, data=True)
objects = directory.collect()
if 'content' not in objects:
objects['content'] = {}
if 'directory' not in objects:
objects['directory'] = {}
self.package_contents = objects['content'].values()
self.package_directories = objects['directory'].values()
revision = self.build_revision(directory,
package_version_data)
revision['id'] = identifier_to_bytes(
revision_identifier(revision))
self.package_revisions.append(revision)
self.log.debug(revision)
package_version_data['id'] = revision['id']
self.all_version_data.append(package_version_data)
# To find the latest version
if self.latest_timestamp < int(package_version_data['date']):
self.latest_timestamp = int(package_version_data['date'])
self.log.debug('Removing unpacked package files at %s', dir_path)
shutil.rmtree(dir_path)
return not self.done
def build_revision(self, directory, package_version_data):
normalize_date = normalize_timestamp(int(package_version_data['date']))
return {
'metadata': {
'package': {
'date': package_version_data['date'],
'archive': package_version_data['archive'],
},
},
'date': normalize_date,
'committer_date': normalize_date,
'author': self.SWH_PERSON,
'committer': self.SWH_PERSON,
'type': 'tar',
'message': self.REVISION_MESSAGE,
'directory': directory.hash,
'synthetic': True,
'parents': [],
}
def store_data(self):
"""Store fetched data in the database.
"""
self.maybe_load_contents(self.package_contents)
self.maybe_load_directories(self.package_directories)
self.maybe_load_revisions(self.package_revisions)
if self.done:
self.generate_and_load_snapshot()
self.flush()
def generate_and_load_snapshot(self):
"""Generate and load snapshot for the package visit.
"""
branches = {}
for version_data in self.all_version_data:
branch_name = self.find_branch_name(version_data['archive'])
target = self.target_from_version(version_data['id'])
branches[branch_name] = target
branches = self.find_head(branches, branch_name,
version_data['date'])
if not target:
self._visit_status = 'partial'
snapshot = {
'branches': branches,
}
snapshot['id'] = identifier_to_bytes(snapshot_identifier(snapshot))
self.maybe_load_snapshot(snapshot)
def find_branch_name(self, url):
"""Extract branch name from tarball url
Args:
url (str): Tarball URL
Returns:
byte: Branch name
Example:
For url = https://ftp.gnu.org/gnu/8sync/8sync-0.2.0.tar.gz
>>> find_branch_name(url)
b'release/8sync-0.2.0'
"""
branch_name = ''
filename = os.path.basename(url)
filename_parts = filename.split(".")
if len(filename_parts) > 1 and filename_parts[-2] == 'tar':
for part in filename_parts[:-2]:
branch_name += '.' + part
elif len(filename_parts) > 1 and filename_parts[-1] == 'zip':
for part in filename_parts[:-1]:
branch_name += '.' + part
return (('release/%s') % branch_name[1:]).encode('ascii')
def find_head(self, branches, branch_name, timestamp):
"""Make branch head.
Checks if the current version is the latest version. Make it as head
if it is the latest version.
Args:
branches (dict): Branches for the focused package.
branch_name (str): Branch name
Returns:
dict: Branches for the focused package
"""
if self.latest_timestamp == int(timestamp):
branches[b'HEAD'] = {
'target_type': 'alias',
'target': branch_name,
}
return branches
def target_from_version(self, revision_id):
return {
'target': revision_id,
'target_type': 'revision',
} if revision_id else None
def load_status(self):
return {
'status': self._load_status,
}
def visit_status(self):
return self._visit_status
def cleanup(self):
"""Clean up temporary disk use after downloading and extracting
package tarballs.
"""
if self.debug:
self.log.warning('%s Will not clean up temp dir %s' % (
DEBUG_MODE, self.temp_directory
))
return
if os.path.exists(self.temp_directory):
self.log.debug('Clean up %s' % self.temp_directory)
shutil.rmtree(self.temp_directory)

File Metadata

Mime Type
text/x-diff
Expires
Jul 4 2025, 9:41 AM (5 w, 3 d ago)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
3237827

Event Timeline

About · Developer information · Legal