(ctx, sync)
Validate third-party license list.
Validate third-party license list.
Validate third - party license list .
def licenses(ctx, sync):
    """Validate third-party license list."""
    if ctx.obj['repo_choice'] != 'core':
        return

    agent_requirements_file = get_agent_requirements()
    if not file_exists(agent_requirements_file):
        abort('Out of sync, run again with the --sync flag')

    packages = defaultdict(set)
    for i, line in enumerate(read_file_lines(agent_requirements_file)):
        try:
            requirement = Requirement(line.strip())
            packages[requirement.name].add(str(requirement.specifier)[2:])
        except Exception as e:
            rel_file = os.path.basename(agent_requirements_file)
            line = i + 1
            annotate_error(agent_requirements_file, str(e).split(":")[1], line=line)
            echo_failure(f"Detected error in {rel_file}:{line} {e}")

    api_urls = []
    for package, versions in packages.items():
        for version in versions:
            api_urls.append(f'https://pypi.org/pypi/{package}/{version}/json')

    package_data = asyncio.run(scrape_license_data(api_urls))
    known_spdx_licenses = {license_id.lower(): license_id for license_id in get_known_spdx_licenses()}

    package_license_errors = defaultdict(list)

    header_line = "{}\n".format(','.join(HEADERS))

    lines = [header_line]
    for package_name, data in sorted(package_data.items()):
        if package_name in EXPLICIT_LICENSES:
            for license_id in sorted(EXPLICIT_LICENSES[package_name]):
                lines.append(format_attribution_line(package_name, license_id, data['copyright']))

            continue

        license_ids = set()
        for package_license in data['licenses']:
            package_license = package_license.strip('"')

            expanded_licenses = []
            for separator in ('/', ' OR ', ' or '):
                if separator in package_license:
                    expanded_licenses.extend(package_license.split(separator))
                    break
            else:
                expanded_licenses.append(package_license)

            for expanded_license in expanded_licenses:
                normalized_license = expanded_license.lower()
                if normalized_license in IGNORED_LICENSES:
                    continue
                elif normalized_license in KNOWN_LICENSES:
                    license_ids.add(KNOWN_LICENSES[normalized_license])
                elif normalized_license in known_spdx_licenses:
                    license_ids.add(known_spdx_licenses[normalized_license])
                else:
                    license_ids.add(expanded_license)
                    package_license_errors[package_name].append(f'unknown license: {expanded_license}')

        for classifier in data['classifiers']:
            if classifier in KNOWN_CLASSIFIERS:
                license_ids.add(KNOWN_CLASSIFIERS[classifier])
            elif classifier in CLASSIFIER_TO_HIGHEST_SPDX:
                license_ids.add(CLASSIFIER_TO_HIGHEST_SPDX[classifier])
            else:
                package_license_errors[package_name].append(f'unknown classifier: {classifier}')

        if license_ids:
            for license_id in sorted(license_ids):
                lines.append(format_attribution_line(package_name, license_id, data['copyright']))
        else:
            package_license_errors[package_name].append('no license information')

    if package_license_errors:
        for package_name, errors in package_license_errors.items():
            echo_info(package_name)
            for error in errors:
                echo_failure(error, indent=True)

        abort()

    extra_licenses_lines, any_errors = validate_extra_licenses()
    lines.extend(extra_licenses_lines)
    lines.sort()
    license_attribution_file = get_license_attribution_file()
    if sync:
        write_file_lines(license_attribution_file, lines)
        if any_errors:
            abort('Failed to write all extra licenses. Please fix any reported errors')
        else:
            echo_success('Success!')
    elif read_file_lines(license_attribution_file) != lines:
        abort('Out of sync, run again with the --sync flag')
    elif any_errors:
        abort()
def licenses ( ctx , sync ) : if ctx . obj [ 'repo_choice' ] != 'core' : return agent_requirements_file = get_agent_requirements ( ) if not file_exists ( agent_requirements_file ) : abort ( 'Out of sync, run again with the --sync flag' ) packages = defaultdict ( set ) for i , line in enumerate ( read_file_lines ( agent_requirements_file ) ) : try : requirement = Requirement ( line . strip ( ) ) packages [ requirement . name ] . add ( str ( requirement . specifier ) [ 2 : ] ) except Exception as e : rel_file = os . path . basename ( agent_requirements_file ) line = i + 1 annotate_error ( agent_requirements_file , str ( e ) . split ( ":" ) [ 1 ] , line = line ) echo_failure ( f"Detected error in {rel_file}:{line} {e}" ) api_urls = [ ] for package , versions in packages . items ( ) : for version in versions : api_urls . append ( f'https://pypi.org/pypi/{package}/{version}/json' ) package_data = asyncio . run ( scrape_license_data ( api_urls ) ) known_spdx_licenses = { license_id . lower ( ) : license_id for license_id in get_known_spdx_licenses ( ) } package_license_errors = defaultdict ( list ) header_line = "{}\n" . format ( ',' . join ( HEADERS ) ) lines = [ header_line ] for package_name , data in sorted ( package_data . items ( ) ) : if package_name in EXPLICIT_LICENSES : for license_id in sorted ( EXPLICIT_LICENSES [ package_name ] ) : lines . append ( format_attribution_line ( package_name , license_id , data [ 'copyright' ] ) ) continue license_ids = set ( ) for package_license in data [ 'licenses' ] : package_license = package_license . strip ( '"' ) expanded_licenses = [ ] for separator in ( '/' , ' OR ' , ' or ' ) : if separator in package_license : expanded_licenses . extend ( package_license . split ( separator ) ) break else : expanded_licenses . append ( package_license ) for expanded_license in expanded_licenses : normalized_license = expanded_license . lower ( ) if normalized_license in IGNORED_LICENSES : continue elif normalized_license in KNOWN_LICENSES : license_ids . add ( KNOWN_LICENSES [ normalized_license ] ) elif normalized_license in known_spdx_licenses : license_ids . add ( known_spdx_licenses [ normalized_license ] ) else : license_ids . add ( expanded_license ) package_license_errors [ package_name ] . append ( f'unknown license: {expanded_license}' ) for classifier in data [ 'classifiers' ] : if classifier in KNOWN_CLASSIFIERS : license_ids . add ( KNOWN_CLASSIFIERS [ classifier ] ) elif classifier in CLASSIFIER_TO_HIGHEST_SPDX : license_ids . add ( CLASSIFIER_TO_HIGHEST_SPDX [ classifier ] ) else : package_license_errors [ package_name ] . append ( f'unknown classifier: {classifier}' ) if license_ids : for license_id in sorted ( license_ids ) : lines . append ( format_attribution_line ( package_name , license_id , data [ 'copyright' ] ) ) else : package_license_errors [ package_name ] . append ( 'no license information' ) if package_license_errors : for package_name , errors in package_license_errors . items ( ) : echo_info ( package_name ) for error in errors : echo_failure ( error , indent = True ) abort ( ) extra_licenses_lines , any_errors = validate_extra_licenses ( ) lines . extend ( extra_licenses_lines ) lines . sort ( ) license_attribution_file = get_license_attribution_file ( ) if sync : write_file_lines ( license_attribution_file , lines ) if any_errors : abort ( 'Failed to write all extra licenses. Please fix any reported errors' ) else : echo_success ( 'Success!' ) elif read_file_lines ( license_attribution_file ) != lines : abort ( 'Out of sync, run again with the --sync flag' ) elif any_errors : abort ( )