<commit_before>#!/usr/bin/env python3
# -*- coding: utf-8 -*-
##
# Copyright (c) 2010-2022 WALLIX, SARL. All rights reserved.
# Licensed computer software. Property of WALLIX.
# Product name: Packager
# Author(s): <NAME>, <NAME>, <NAME>, <NAME>
# Module description: Create package and git tag
##

import sys
import os
import re
import shutil
import argparse
import datetime
import subprocess
import traceback
from collections.abc import Iterable

var_ident = '[A-Z][A-Z0-9_]*'


class PackagerError(Exception):
    pass


def replace_dict_all(text: str, dico: dict[str, str]) -> str:
    rgx_replace_var = re.compile(f'%{var_ident}%')
    text_parts = []
    pos = 0
    for m in rgx_replace_var.finditer(text):
        text_parts.append(text[pos:m.start()])
        var = m.group()
        text_parts.append(dico.get(var[1:-1], var))
        pos = m.end()
    text_parts.append(text[pos:])
    return ''.join(text_parts)


def read_and_update_config(filename: str, configs: dict[str, str] = None) -> dict[str, str]:
    """ Parse target Config files """
    parse_config_rgx = re.compile(rf'^({var_ident})\s*=(.*)')
    configs = {} if configs is None else configs

    with open(filename) as f:
        for line in f:
            if line.startswith('include '):
                directory = os.path.dirname(filename)
                included_path = os.path.join(directory, line[8:].strip())
                read_and_update_config(included_path, configs)
            elif m := re.search(parse_config_rgx, line):
                configs[m.group(1).strip()] = m.group(2).strip()

    if 'PKG_DISTRIBUTION' not in configs:
        if configs['DIST_NAME'].lower() == 'ubuntu':
            configs['PKG_DISTRIBUTION'] = 'unstable'
        else:
            configs['PKG_DISTRIBUTION'] = configs['DIST_ID']

    if 'TARGET_NAME' not in configs:
        if configs['DIST_NAME'].lower() == 'ubuntu':
            configs['TARGET_NAME'] = '+' + configs['DIST_ID']

    return configs


def update_config_variables(config: dict[str, str], variables: Iterable[str]) -> list[str]:
    """Return unparsed variable"""
    rgx_var = re.compile(rf'^({var_ident})(\+?=)(.*)')
    variable_errors = []
    for var in variables:
        m = rgx_var.match(var)
        if m is None:
            variable_errors.append(var)
            continue
        k = m.group(1)
        newvalue = m.group(3) if m.group(2) == '=' else config.get(k, '') + m.group(3)
        config[k] = newvalue
    return variable_errors


def print_configs(configs: dict[str, str]) -> None:
    variables = []
    for k, v in configs.items():
        variables.append(f'{k} = {v}')
    print('\n'.join(variables))


def readall(filename: str) -> str:
    with open(filename) as f:
        return f.read()


def writeall(filename: str, s: str) -> None:
    with open(filename, 'w+') as f:
        f.write(s)


def get_changelog_entry(project_name: str, version: str, maintainer: str, urgency: str, utc: str) -> str:
    editor = os.environ.get('EDITOR') or 'nano'

    changelog = []

    tmp_changelog = f'/tmp/{version}-changelog.tmp'
    cmd = f'{editor} {tmp_changelog}'
    if os.system(cmd):
        raise PackagerError(f'Error in `{cmd}`')

    with open(tmp_changelog) as f:
        for line in f:
            if line and line != '\n':
                changelog.append('  * ')
                changelog.append(line)

    os.remove(tmp_changelog)

    if not changelog:
        raise PackagerError('Change log is empty')

    now = datetime.datetime.today().strftime(f'%a, %d %b %Y %H:%M:%S +{utc}')
    return (f'{project_name or "%PROJECT_NAME%"} ({version}%TARGET_NAME%) %PKG_DISTRIBUTION%; '
            f'urgency={urgency}\n\n{"".join(changelog)}\n\n -- {maintainer}  {now}\n\n')


def update_changelog(changelog_path: str, changelog: str) -> None:
    changelog += readall(changelog_path)
    writeall(changelog_path, changelog)


def create_build_directory(package_template: str, output_build: str, configs: dict[str, str]) -> None:
    rgx_tempfile = re.compile('^#.*#$|~$')

    try:
        os.mkdir(output_build, 0o766)
    except FileExistsError:
        pass

    filenames = filter(lambda fname: rgx_tempfile.search(fname) is None,
                       os.listdir(package_template))

    for filename in filenames:
        out = readall(f'{package_template}/{filename}')
        out = replace_dict_all(out, configs)
        writeall(f'{output_build}/{filename}', out)


def less_version(lhs_version: str, rhs_version: str) -> bool:
    rgx_split_version = re.compile(
        r'[^\d]*(\d+)(?:\.(\d+))?(?:[-.](\d+))?(?:[-.](\d+))?(?:[-.](\d+))?(.*)')

    def to_tuple(m: re.Match): return (
        int(m.group(1)),
        int(m.group(2) or 0),
        int(m.group(3) or 0),
        int(m.group(4) or 0),
        int(m.group(5) or 0),
        m.group(6)
    )
    m1 = rgx_split_version.match(lhs_version)
    m2 = rgx_split_version.match(rhs_version)
    return to_tuple(m1) < to_tuple(m2)


def check_version(old_version: str, new_version: str) -> None:
    if not less_version(old_version, new_version):
        raise PackagerError(f'New version ({new_version}) is less than'
                            f' old version ({old_version})')


def check_pattern_version(old_version: str, new_version: str, pattern: str) -> None:
    if pattern:
        m = re.match(pattern, old_version)
        if m is None:
            raise PackagerError('Unable to retrieve the version number'
                                f'from {old_version}')
        if not new_version.startswith(m.group(0)):
            raise PackagerError(f'The last tag number ({old_version}) of '
                                f'the repository is not compatible with'
                                f'the new tag number {new_version}. '
                                f'Pattern is "{pattern}"')


def shell_cmd(cmd: list[str]) -> str:
    print('$', ' '.join(cmd))
    return subprocess.check_output(cmd, env={'GIT_PAGER': ''}, text=True)


def git_uncommited_changes() -> str:
    return shell_cmd(['git', 'diff', '--shortstat'])


def git_tag_exists(tag: str) -> tuple[bool, str]:<commit_msg>split less_version()<commit_after>#!/usr/bin/env python3
# -*- coding: utf-8 -*-
##
# Copyright (c) 2010-2022 WALLIX, SARL. All rights reserved.
# Licensed computer software. Property of WALLIX.
# Product name: Packager
# Author(s): <NAME>, <NAME>, <NAME>, <NAME>
# Module description: Create package and git tag
##

import sys
import os
import re
import shutil
import argparse
import datetime
import subprocess
import traceback
from collections.abc import Iterable

var_ident = '[A-Z][A-Z0-9_]*'


class PackagerError(Exception):
    pass


def replace_dict_all(text: str, dico: dict[str, str]) -> str:
    rgx_replace_var = re.compile(f'%{var_ident}%')
    text_parts = []
    pos = 0
    for m in rgx_replace_var.finditer(text):
        text_parts.append(text[pos:m.start()])
        var = m.group()
        text_parts.append(dico.get(var[1:-1], var))
        pos = m.end()
    text_parts.append(text[pos:])
    return ''.join(text_parts)


def read_and_update_config(filename: str, configs: dict[str, str] = None) -> dict[str, str]:
    """ Parse target Config files """
    parse_config_rgx = re.compile(rf'^({var_ident})\s*=(.*)')
    configs = {} if configs is None else configs

    with open(filename) as f:
        for line in f:
            if line.startswith('include '):
                directory = os.path.dirname(filename)
                included_path = os.path.join(directory, line[8:].strip())
                read_and_update_config(included_path, configs)
            elif m := re.search(parse_config_rgx, line):
                configs[m.group(1).strip()] = m.group(2).strip()

    if 'PKG_DISTRIBUTION' not in configs:
        if configs['DIST_NAME'].lower() == 'ubuntu':
            configs['PKG_DISTRIBUTION'] = 'unstable'
        else:
            configs['PKG_DISTRIBUTION'] = configs['DIST_ID']

    if 'TARGET_NAME' not in configs:
        if configs['DIST_NAME'].lower() == 'ubuntu':
            configs['TARGET_NAME'] = '+' + configs['DIST_ID']

    return configs


def update_config_variables(config: dict[str, str], variables: Iterable[str]) -> list[str]:
    """Return unparsed variable"""
    rgx_var = re.compile(rf'^({var_ident})(\+?=)(.*)')
    variable_errors = []
    for var in variables:
        m = rgx_var.match(var)
        if m is None:
            variable_errors.append(var)
            continue
        k = m.group(1)
        newvalue = m.group(3) if m.group(2) == '=' else config.get(k, '') + m.group(3)
        config[k] = newvalue
    return variable_errors


def print_configs(configs: dict[str, str]) -> None:
    variables = []
    for k, v in configs.items():
        variables.append(f'{k} = {v}')
    print('\n'.join(variables))


def readall(filename: str) -> str:
    with open(filename) as f:
        return f.read()


def writeall(filename: str, s: str) -> None:
    with open(filename, 'w+') as f:
        f.write(s)


def get_changelog_entry(project_name: str, version: str, maintainer: str, urgency: str, utc: str) -> str:
    editor = os.environ.get('EDITOR') or 'nano'

    changelog = []

    tmp_changelog = f'/tmp/{version}-changelog.tmp'
    cmd = f'{editor} {tmp_changelog}'
    if os.system(cmd):
        raise PackagerError(f'Error in `{cmd}`')

    with open(tmp_changelog) as f:
        for line in f:
            if line and line != '\n':
                changelog.append('  * ')
                changelog.append(line)

    os.remove(tmp_changelog)

    if not changelog:
        raise PackagerError('Change log is empty')

    now = datetime.datetime.today().strftime(f'%a, %d %b %Y %H:%M:%S +{utc}')
    return (f'{project_name or "%PROJECT_NAME%"} ({version}%TARGET_NAME%) %PKG_DISTRIBUTION%; '
            f'urgency={urgency}\n\n{"".join(changelog)}\n\n -- {maintainer}  {now}\n\n')


def update_changelog(changelog_path: str, changelog: str) -> None:
    changelog += readall(changelog_path)
    writeall(changelog_path, changelog)


def create_build_directory(package_template: str, output_build: str, configs: dict[str, str]) -> None:
    rgx_tempfile = re.compile('^#.*#$|~$')

    try:
        os.mkdir(output_build, 0o766)
    except FileExistsError:
        pass

    filenames = filter(lambda fname: rgx_tempfile.search(fname) is None,
                       os.listdir(package_template))

    for filename in filenames:
        out = readall(f'{package_template}/{filename}')
        out = replace_dict_all(out, configs)
        writeall(f'{output_build}/{filename}', out)


def get_version_extractor() -> re.Pattern:
    return re.compile(
        r'[^\d]*(\d+)(?:\.(\d+))?(?:[-.](\d+))?(?:[-.](\d+))?(?:[-.](\d+))?(.*)')


def re_match_version_to_tuple(m: re.Match) -> tuple[int, int, int, int, int, str]:
    return (
        int(m.group(1)),
        int(m.group(2) or 0),
        int(m.group(3) or 0),
        int(m.group(4) or 0),
        int(m.group(5) or 0),
        m.group(6)
    )


def less_version(lhs_version: str, rhs_version: str) -> bool:
    patt = get_version_extractor()
    m1 = patt.match(lhs_version)
    m2 = patt.match(rhs_version)
    return re_match_version_to_tuple(m1) < re_match_version_to_tuple(m2)


def check_version(old_version: str, new_version: str) -> None:
    if not less_version(old_version, new_version):
        raise PackagerError(f'New version ({new_version}) is less than'
                            f' old version ({old_version})')


def check_pattern_version(old_version: str, new_version: str, pattern: str) -> None:
    if pattern:
        m = re.match(pattern, old_version)
        if m is None:
            raise PackagerError('Unable to retrieve the version number'
                                f'from {old_version}')
        if not new_version.startswith(m.group(0)):
            raise PackagerError(f'The last tag number ({old_version}) of '
                                f'the repository is not compatible with'
                                f'the new tag number {new_version}. '
                                f'Pattern is "{pattern}"')


def shell_cmd(cmd: list[str]) -> str:
    print('$', ' '.join(cmd))
    return subprocess.check_output(cmd, env={'GIT_PAGER': ''}, text=True)


def git_uncommited_changes() -> str:
    return shell_cmd(['git', 'diff', '--shortstat'])


def git_tag_exists(tag: str) -> tuple[bool, str]:
{'fraction_non_alphanumeric': 0.15247684252919855, 'fraction_numerical': 0.006604913411196134, 'mean_word_length': 2.7544602358633203, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 43, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '45468', 'n_tokens_mistral': 4353, 'n_tokens_neox': 4186, 'n_words': 947}