Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/package/debian/loader.py
# Copyright (C) 2017-2019 The Software Heritage developers | # Copyright (C) 2017-2019 The Software Heritage developers | ||||
# See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
# License: GNU General Public License version 3, or any later version | # License: GNU General Public License version 3, or any later version | ||||
# See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
import email.utils | import email.utils | ||||
import iso8601 | |||||
import logging | import logging | ||||
from os import path | |||||
import re | import re | ||||
import subprocess | import subprocess | ||||
from dateutil.parser import parse as parse_date | from dateutil.parser import parse as parse_date | ||||
from debian.changelog import Changelog | from debian.changelog import Changelog | ||||
from debian.deb822 import Dsc | from debian.deb822 import Dsc | ||||
from os import path | |||||
from typing import ( | from typing import ( | ||||
Any, Dict, Generator, List, Mapping, Optional, Sequence, Tuple | Any, Generator, List, Mapping, Optional, Sequence, Tuple | ||||
) | ) | ||||
from swh.loader.package.loader import PackageLoader | from swh.loader.package.loader import PackageLoader | ||||
from swh.loader.package.utils import download, release_name | from swh.loader.package.utils import download, release_name | ||||
from swh.model.model import ( | |||||
Sha1Git, Person, Revision, RevisionType, TimestampWithTimezone | |||||
) | |||||
logger = logging.getLogger(__name__) | logger = logging.getLogger(__name__) | ||||
UPLOADERS_SPLIT = re.compile(r'(?<=\>)\s*,\s*') | UPLOADERS_SPLIT = re.compile(r'(?<=\>)\s*,\s*') | ||||
class DebianLoader(PackageLoader): | class DebianLoader(PackageLoader): | ||||
"""Load debian origins into swh archive. | """Load debian origins into swh archive. | ||||
▲ Show 20 Lines • Show All 84 Lines • ▼ Show 20 Lines | def download_package(self, p_info: Mapping[str, Any], | ||||
logger.debug('res: %s', res) | logger.debug('res: %s', res) | ||||
return res | return res | ||||
def uncompress(self, dl_artifacts: List[Tuple[str, Mapping[str, Any]]], | def uncompress(self, dl_artifacts: List[Tuple[str, Mapping[str, Any]]], | ||||
dest: str) -> str: | dest: str) -> str: | ||||
logger.debug('dl_artifacts: %s', dl_artifacts) | logger.debug('dl_artifacts: %s', dl_artifacts) | ||||
return extract_package(dl_artifacts, dest=dest) | return extract_package(dl_artifacts, dest=dest) | ||||
def build_revision(self, a_metadata: Mapping[str, Any], | def build_revision( | ||||
uncompressed_path: str) -> Dict: | self, a_metadata: Mapping[str, Any], uncompressed_path: str, | ||||
directory: Sha1Git) -> Optional[Revision]: | |||||
dsc_url, dsc_name = dsc_information(a_metadata) | dsc_url, dsc_name = dsc_information(a_metadata) | ||||
if not dsc_name: | if not dsc_name: | ||||
raise ValueError( | raise ValueError( | ||||
'dsc name for url %s should not be None' % dsc_url) | 'dsc name for url %s should not be None' % dsc_url) | ||||
dsc_path = path.join(path.dirname(uncompressed_path), dsc_name) | dsc_path = path.join(path.dirname(uncompressed_path), dsc_name) | ||||
i_metadata = get_package_metadata( | i_metadata = get_package_metadata( | ||||
a_metadata, dsc_path, uncompressed_path) | a_metadata, dsc_path, uncompressed_path) | ||||
logger.debug('i_metadata: %s', i_metadata) | logger.debug('i_metadata: %s', i_metadata) | ||||
logger.debug('a_metadata: %s', a_metadata) | logger.debug('a_metadata: %s', a_metadata) | ||||
msg = 'Synthetic revision for Debian source package %s version %s' % ( | msg = 'Synthetic revision for Debian source package %s version %s' % ( | ||||
a_metadata['name'], a_metadata['version']) | a_metadata['name'], a_metadata['version']) | ||||
date = iso8601.parse_date(i_metadata['changelog']['date']) | date = TimestampWithTimezone.from_iso8601( | ||||
i_metadata['changelog']['date']) | |||||
author = prepare_person(i_metadata['changelog']['person']) | author = prepare_person(i_metadata['changelog']['person']) | ||||
# inspired from swh.loader.debian.converters.package_metadata_to_revision # noqa | # inspired from swh.loader.debian.converters.package_metadata_to_revision # noqa | ||||
return { | return Revision( | ||||
'type': 'dsc', | type=RevisionType.DSC, | ||||
'message': msg.encode('utf-8'), | message=msg.encode('utf-8'), | ||||
'author': author, | author=author, | ||||
'date': date, | date=date, | ||||
'committer': author, | committer=author, | ||||
'committer_date': date, | committer_date=date, | ||||
'parents': [], | parents=[], | ||||
'metadata': { | directory=directory, | ||||
synthetic=True, | |||||
metadata={ | |||||
'intrinsic': { | 'intrinsic': { | ||||
'tool': 'dsc', | 'tool': 'dsc', | ||||
'raw': i_metadata, | 'raw': i_metadata, | ||||
}, | }, | ||||
'extrinsic': { | 'extrinsic': { | ||||
'provider': dsc_url, | 'provider': dsc_url, | ||||
'when': self.visit_date.isoformat(), | 'when': self.visit_date.isoformat(), | ||||
'raw': a_metadata, | 'raw': a_metadata, | ||||
}, | }, | ||||
} | }, | ||||
} | ) | ||||
def resolve_revision_from(known_package_artifacts: Mapping, | def resolve_revision_from(known_package_artifacts: Mapping, | ||||
artifact_metadata: Mapping) -> Optional[bytes]: | artifact_metadata: Mapping) -> Optional[bytes]: | ||||
"""Given known package artifacts (resolved from the snapshot of previous | """Given known package artifacts (resolved from the snapshot of previous | ||||
visit) and the new artifact to fetch, try to solve the corresponding | visit) and the new artifact to fetch, try to solve the corresponding | ||||
revision. | revision. | ||||
▲ Show 20 Lines • Show All 48 Lines • ▼ Show 20 Lines | def uid_to_person(uid: str) -> Mapping[str, str]: | ||||
if name and email: | if name and email: | ||||
ret['name'] = name | ret['name'] = name | ||||
ret['email'] = mail | ret['email'] = mail | ||||
else: | else: | ||||
ret['name'] = uid | ret['name'] = uid | ||||
return ret | return ret | ||||
def prepare_person(person: Mapping[str, str]) -> Mapping[str, bytes]: | def prepare_person(person: Mapping[str, str]) -> Person: | ||||
"""Prepare person for swh serialization... | """Prepare person for swh serialization... | ||||
Args: | Args: | ||||
A person dict | A person dict | ||||
Returns: | Returns: | ||||
A person dict ready for storage | A person ready for storage | ||||
""" | """ | ||||
ret = {} | return Person.from_dict({ | ||||
for key, value in person.items(): | key: value.encode('utf-8') | ||||
ret[key] = value.encode('utf-8') | for (key, value) in person.items() | ||||
return ret | }) | ||||
def download_package( | def download_package( | ||||
package: Mapping[str, Any], tmpdir: Any) -> Mapping[str, Any]: | package: Mapping[str, Any], tmpdir: Any) -> Mapping[str, Any]: | ||||
"""Fetch a source package in a temporary directory and check the checksums | """Fetch a source package in a temporary directory and check the checksums | ||||
for all files. | for all files. | ||||
Args: | Args: | ||||
▲ Show 20 Lines • Show All 157 Lines • Show Last 20 Lines |