slpkg/slpkg/install_data.py
Dimitris Zlatanidis 3a5d7c4b49 Fixed for checksum
2023-03-29 18:20:41 +03:00

416 lines
16 KiB
Python

#!/usr/bin/python3
# -*- coding: utf-8 -*-
from pathlib import Path
from slpkg.configs import Configs
from slpkg.utilities import Utilities
from slpkg.repositories import Repositories
from slpkg.models.models import session as Session
from slpkg.models.models import SBoTable, PonceTable, BinariesTable
class InstallData(Configs):
def __init__(self):
super(Configs, self).__init__()
self.session = Session
self.utils = Utilities()
self.repos = Repositories()
self.color = self.colour()
self.bold: str = self.color['bold']
self.yellow: str = self.color['yellow']
self.byellow: str = f'{self.bold}{self.yellow}'
self.endc: str = self.color['endc']
def install_sbos_data(self) -> None:
""" Install the data for SBo repository. """
sbo_tags = [
'SLACKBUILD NAME:',
'SLACKBUILD LOCATION:',
'SLACKBUILD FILES:',
'SLACKBUILD VERSION:',
'SLACKBUILD DOWNLOAD:',
'SLACKBUILD DOWNLOAD_x86_64:',
'SLACKBUILD MD5SUM:',
'SLACKBUILD MD5SUM_x86_64:',
'SLACKBUILD REQUIRES:',
'SLACKBUILD SHORT DESCRIPTION:'
]
sbo_table = SBoTable
path = Path(self.repos.sbo_repo_path, self.repos.sbo_repo_slackbuilds)
if self.repos.ponce_repo:
sbo_table = PonceTable
path = Path(self.repos.ponce_repo_path, self.repos.ponce_repo_slackbuilds)
slackbuilds_txt: list = self.utils.read_file(path)
cache: list = [] # init cache
print('Creating the database... ', end='', flush=True)
for i, line in enumerate(slackbuilds_txt, 1):
for tag in sbo_tags:
if line.startswith(tag):
line = line.replace(tag, '').strip()
cache.append(line)
if (i % 11) == 0:
data: str = sbo_table(name=cache[0], location=cache[1].split('/')[1],
files=cache[2], version=cache[3],
download=cache[4], download64=cache[5],
md5sum=cache[6], md5sum64=cache[7],
requires=cache[8], short_description=cache[9])
self.session.add(data)
cache: list = [] # reset cache after 11 lines
print(f'{self.byellow}Done{self.endc}')
self.session.commit()
def install_alien_data(self) -> None:
""" Install the data for alien repository. """
checksums_dict: dict = {}
pkg_tag = [
'PACKAGE NAME:',
'PACKAGE LOCATION:',
'PACKAGE SIZE (compressed):',
'PACKAGE SIZE (uncompressed):',
'PACKAGE REQUIRED:',
'PACKAGE DESCRIPTION:'
]
path_packages: Path = Path(self.repos.alien_repo_path, self.repos.alien_repo_packages)
path_checksums: Path = Path(self.repos.alien_repo_path, self.repos.alien_repo_checksums)
packages_txt: list = self.utils.read_file(path_packages)
checksums_md5: list = self.utils.read_file(path_checksums)
for line in checksums_md5:
line = line.strip()
if line.endswith(('.txz', '.tgz')):
file: str = line.split('./')[1].split('/')[-1].strip()
checksum: str = line.split('./')[0].strip()
checksums_dict[file] = checksum
cache: list = [] # init cache
print('Creating the database... ', end='', flush=True)
for line in packages_txt:
if line.startswith(pkg_tag[0]):
package_name = line.replace(pkg_tag[0], '').strip()
split_package: list = self.utils.split_binary_pkg(package_name)
cache.append(split_package[0]) # package name
cache.append(split_package[1]) # package version
cache.append(package_name)
cache.append(self.repos.alien_repo_mirror)
try:
cache.append(checksums_dict[package_name])
except KeyError:
cache.append('error checksum')
if line.startswith(pkg_tag[1]):
package_location = line.replace(pkg_tag[1], '').strip()
cache.append(package_location[1:]) # Do not install (.) dot
if line.startswith(pkg_tag[2]):
package_size_comp = line.replace(pkg_tag[2], '').strip()
cache.append(f'{package_size_comp}B')
if line.startswith(pkg_tag[3]):
package_size_uncomp = line.replace(pkg_tag[3], '').strip()
cache.append(f'{package_size_uncomp}B')
if line.startswith(pkg_tag[4]):
required = line.replace(pkg_tag[4], '').strip()
package_required = required.replace(',', ' ').strip()
cache.append(package_required)
if line.startswith(pkg_tag[5]):
package_description = line.replace(pkg_tag[5], '').strip()
cache.append(package_description)
if len(cache) == 10:
data: str = BinariesTable(
repo=self.repos.alien_repo_name,
name=cache[0],
version=cache[1],
package=cache[2],
mirror=cache[3],
checksum=cache[4],
location=cache[5],
size_comp=cache[6],
size_uncomp=cache[7],
required=cache[8],
description=cache[9]
)
self.session.add(data)
cache: list = [] # reset cache
print(f'{self.byellow}Done{self.endc}')
self.session.commit()
def install_gnome_data(self) -> None:
""" Install the data for gnome repository. """
checksums_dict: dict = {}
pkg_tag = [
'PACKAGE NAME:',
'PACKAGE MIRROR:',
'PACKAGE LOCATION:',
'PACKAGE SIZE (compressed):',
'PACKAGE SIZE (uncompressed):',
'PACKAGE DESCRIPTION:'
]
path_packages: Path = Path(self.repos.gnome_repo_path, self.repos.gnome_repo_packages)
path_checksums: Path = Path(self.repos.gnome_repo_path, self.repos.gnome_repo_checksums)
packages_txt: list = self.utils.read_file(path_packages)
checksums_md5: list = self.utils.read_file(path_checksums)
for line in checksums_md5:
line = line.strip()
if line.endswith(('.txz', '.tgz')):
file: str = line.split('./')[1].split('/')[-1].strip()
checksum: str = line.split('./')[0].strip()
checksums_dict[file] = checksum
cache: list = [] # init cache
print('Creating the database... ', end='', flush=True)
for line in packages_txt:
if line.startswith(pkg_tag[0]):
package_name = line.replace(pkg_tag[0], '').strip()
split_package: list = self.utils.split_binary_pkg(package_name)
cache.append(split_package[0]) # package name
cache.append(split_package[1]) # package version
cache.append(package_name)
try:
cache.append(checksums_dict[package_name])
except KeyError:
cache.append('error checksum')
if line.startswith(pkg_tag[1]):
package_mirror = line.replace(pkg_tag[1], '').strip()
if not package_mirror.endswith('/'):
package_mirror: str = f'{package_mirror}/'
cache.append(package_mirror)
if line.startswith(pkg_tag[2]):
package_location = line.replace(pkg_tag[2], '').strip()
cache.append(package_location[1:]) # Do not install (.) dot
if line.startswith(pkg_tag[3]):
package_size_comp = line.replace(pkg_tag[3], '').strip()
cache.append(f'{package_size_comp}B')
if line.startswith(pkg_tag[4]):
package_size_uncomp = line.replace(pkg_tag[4], '').strip()
cache.append(f'{package_size_uncomp}B')
if line.startswith(pkg_tag[5]):
package_description = line.replace(pkg_tag[5], '').strip()
cache.append(package_description)
if len(cache) == 9:
data: str = BinariesTable(
repo=self.repos.gnome_repo_name,
name=cache[0],
version=cache[1],
package=cache[2],
checksum=cache[3],
mirror=cache[4],
location=cache[5],
size_comp=cache[6],
size_uncomp=cache[7],
description=cache[8]
)
self.session.add(data)
cache: list = [] # reset cache
print(f'{self.byellow}Done{self.endc}')
self.session.commit()
def install_conraid_data(self) -> None:
""" Install the data for conraid repository. """
checksums_dict: dict = {}
pkg_tag = [
'PACKAGE NAME:',
'PACKAGE MIRROR:',
'PACKAGE LOCATION:',
'PACKAGE SIZE (compressed):',
'PACKAGE SIZE (uncompressed):',
'PACKAGE DESCRIPTION:'
]
path_packages: Path = Path(self.repos.conraid_repo_path, self.repos.conraid_repo_packages)
path_checksums: Path = Path(self.repos.conraid_repo_path, self.repos.conraid_repo_checksums)
checksums_md5: list = self.utils.read_file(path_checksums)
packages_txt: list = self.utils.read_file(path_packages)
for line in checksums_md5:
line = line.strip()
if line.endswith(('.txz', '.tgz')):
file: str = line.split('./')[1].split('/')[-1].strip()
checksum: str = line.split('./')[0].strip()
checksums_dict[file] = checksum
cache: list = []
print('Creating the database... ', end='', flush=True)
for line in packages_txt:
if line.startswith(pkg_tag[0]):
package_name: str = line.replace(pkg_tag[0], '').strip()
split_package: list = self.utils.split_binary_pkg(package_name)
cache.append(split_package[0]) # package name
cache.append(split_package[1]) # package version
cache.append(package_name)
try:
cache.append(checksums_dict[package_name])
except KeyError:
cache.append('error checksum')
if line.startswith(pkg_tag[1]):
package_mirror: str = line.replace(pkg_tag[1], '').strip()
if not package_mirror.endswith('/'):
package_mirror: str = f'{package_mirror}/'
cache.append(package_mirror)
if line.startswith(pkg_tag[2]):
package_location: str = line.replace(pkg_tag[2], '').strip()
cache.append(package_location[2:]) # Do not install (./) dot
if line.startswith(pkg_tag[3]):
package_size_comp: str = line.replace(pkg_tag[3], '').strip()
cache.append(f'{package_size_comp}B')
if line.startswith(pkg_tag[4]):
package_size_uncomp: str = line.replace(pkg_tag[4], '').strip()
cache.append(f'{package_size_uncomp}B')
if line.startswith(pkg_tag[5]):
package_description: str = line.replace(pkg_tag[5], '').strip()
cache.append(package_description)
if len(cache) == 9:
data: str = BinariesTable(
repo=self.repos.conraid_repo_name,
name=cache[0],
version=cache[1],
package=cache[2],
checksum=cache[3],
mirror=cache[4],
location=cache[5],
size_comp=cache[6],
size_uncomp=cache[7],
description=cache[8]
)
self.session.add(data)
cache: list = [] # reset cache
print(f'{self.byellow}Done{self.endc}')
self.session.commit()
def install_slackonly_data(self) -> None:
""" Install the data for slackonly repository. """
checksums_dict: dict = {}
pkg_tag = [
'PACKAGE NAME:',
'PACKAGE LOCATION:',
'PACKAGE SIZE (compressed):',
'PACKAGE SIZE (uncompressed):',
'PACKAGE REQUIRED:',
'PACKAGE DESCRIPTION:'
]
path_packages: Path = Path(self.repos.slackonly_repo_path, self.repos.slackonly_repo_packages)
path_checksums: Path = Path(self.repos.slackonly_repo_path, self.repos.slackonly_repo_checksums)
packages_txt: list = self.utils.read_file(path_packages)
checksums_md5: list = self.utils.read_file(path_checksums)
for line in checksums_md5:
line = line.strip()
if line.endswith(('.txz', '.tgz')):
file: str = line.split('./')[1].split('/')[-1].strip()
checksum: str = line.split('./')[0].strip()
checksums_dict[file] = checksum
cache: list = [] # init cache
print('Creating the database... ', end='', flush=True)
for line in packages_txt:
if line.startswith(pkg_tag[0]):
package_name = line.replace(pkg_tag[0], '').strip()
split_package: list = self.utils.split_binary_pkg(package_name)
cache.append(split_package[0]) # package name
cache.append(split_package[1]) # package version
cache.append(package_name)
cache.append(self.repos.slackonly_repo_mirror)
try:
cache.append(checksums_dict[package_name])
except KeyError:
cache.append('error checksum')
if line.startswith(pkg_tag[1]):
package_location = line.replace(pkg_tag[1], '').strip()
cache.append(package_location[1:]) # Do not install (.) dot
if line.startswith(pkg_tag[2]):
package_size_comp = line.replace(pkg_tag[2], '').strip()
cache.append(f'{package_size_comp}B')
if line.startswith(pkg_tag[3]):
package_size_uncomp = line.replace(pkg_tag[3], '').strip()
cache.append(f'{package_size_uncomp}B')
if line.startswith(pkg_tag[4]):
required = line.replace(pkg_tag[4], '').strip()
package_required = required.replace(',', ' ').strip()
cache.append(package_required)
if line.startswith(pkg_tag[5]):
package_description = line.replace(pkg_tag[5], '').strip()
cache.append(package_description)
if len(cache) == 10:
data: str = BinariesTable(
repo=self.repos.slackonly_repo_name,
name=cache[0],
version=cache[1],
package=cache[2],
mirror=cache[3],
checksum=cache[4],
location=cache[5],
size_comp=cache[6],
size_uncomp=cache[7],
required=cache[8],
description=cache[9]
)
self.session.add(data)
cache: list = [] # reset cache
print(f'{self.byellow}Done{self.endc}')
self.session.commit()