deprecate_java_packages.py

import dataclasses
import logging
import csv
import requests
from typing import List
from urllib.request import urlopen


CSV_URL = 'https://raw.githubusercontent.com/Azure/azure-sdk/master/_data/releases/latest/java-packages.csv'
TRACK1_PACKAGE_PREFIX = 'azure-mgmt-'
TRACK2_PACKAGE_PREFIX = 'azure-resourcemanager-'

MIGRATION_LINK = 'https://github.com/Azure/azure-sdk-for-java/blob/main/sdk/resourcemanager/docs/MIGRATION_GUIDE.md'

CSV_FILENAME = 'deprecate_java_packages.csv'

SERVICE_NAME_MAP = {
    'cosmosdb': 'cosmos',
    'datalake-analytics': 'datalakeanalytics',
    'datalake-store': 'datalakestore',
    'datalake-store-uploader': 'datalakestore',
    'devices': 'deviceprovisioningservices',
    'devtestlab': 'devtestlabs',
    'documentdb': 'cosmos',
    'eventhub': 'eventhubs',
    'features': 'resources',
    'graph-rbac': 'authorization',
    'insights': 'monitor',
    'locks': 'resources',
    'machinelearning': 'machinelearningservices',
    'media': 'mediaservices',
    'dbformariadb': 'mariadb',
    'policy': 'resources',
    'powerbi': 'powerbidedicated',
    'samples': 'resources',
    'scheduler': 'resources',
    'subscriptions': 'resources',
    'traffic-manager': 'trafficmanager',
    'utility': 'resources',
    'website': 'appservice',
    'websites': 'appservice',
}


@dataclasses.dataclass
class Sdk:
    service: str
    package: str
    artifact: str


@dataclasses.dataclass
class SdkInfo:
    service: str
    t1_package: str
    t1_artifact: str
    t2_artifact: str

    def to_row(self) -> List[str]:
        return [self.service, 'Java', self.t1_maven_url(), self.t2_maven_url(), MIGRATION_LINK]

    def t1_maven_url(self):
        return f'https://mvnrepository.com/artifact/{self.t1_package}/{self.t1_artifact}'

    def t2_maven_url(self):
        return f'https://mvnrepository.com/artifact/com.azure.resourcemanager/{self.t2_artifact}'


def process_java_packages_csv() -> List[SdkInfo]:
    logging.info(f'query csv: {CSV_URL}')
    with urlopen(CSV_URL) as csv_response:
        csv_data = csv_response.read()
        csv_str = csv_data.decode('utf-8')
        csv_reader = csv.DictReader(csv_str.split('\n'), delimiter=',', quotechar='"')

    t1_sdks = []
    t2_sdks = {}
    for row in csv_reader:
        package = row['Package']
        namespace = row['GroupId']
        if package.startswith(TRACK1_PACKAGE_PREFIX):
            service = package[len(TRACK1_PACKAGE_PREFIX):]
            if service in SERVICE_NAME_MAP:
                service = SERVICE_NAME_MAP[service]
            sdk = Sdk(service, namespace, package)
            t1_sdks.append(sdk)
        elif package == 'azure':
            service = ''
            sdk = Sdk(service, namespace, package)
            t1_sdks.append(sdk)

        if package.startswith(TRACK2_PACKAGE_PREFIX) and package != TRACK2_PACKAGE_PREFIX + 'parent':
            service = package[len(TRACK2_PACKAGE_PREFIX):]
            sdk = Sdk(service, namespace, package)
            t2_sdks[service] = sdk
        elif package == 'azure-resourcemanager':
            service = ''
            sdk = Sdk(service, namespace, package)
            t2_sdks[service] = sdk

    t1_sdks.sort(key=lambda r: r.service)

    sdk_info_list = []
    for sdk in t1_sdks:
        t2_sdk = None
        if sdk.service in t2_sdks:
            t2_sdk = t2_sdks[sdk.service]
        if not t2_sdk:
            print(sdk.service)

        sdk_info = SdkInfo(sdk.service, sdk.package, sdk.artifact, t2_sdk.artifact if t2_sdk else '')
        sdk_info_list.append(sdk_info)

    # sdk_info_list.append(SdkInfo('', 'com.microsoft.azure', 'azure', 'azure-resourcemanager'))

    return sdk_info_list


def run():
    logging.basicConfig(level=logging.INFO)
    sdk_info_list = process_java_packages_csv()
    write_csv(sdk_info_list)

    validate_maven_packages(sdk_info_list)


def write_csv(sdk_info_list: List[SdkInfo]):
    logging.info(f'write csv: {CSV_FILENAME}')
    with open(CSV_FILENAME, 'w', newline='') as f:
        writer = csv.writer(f, delimiter=',', quotechar='"', quoting=csv.QUOTE_MINIMAL)

        writer.writerow(['Service', 'Language',
                         'Legacy package (package being replaced)',
                         'New Packages that replace the legacy package on the left',
                         'Migration Guide link'])
        for item in sdk_info_list:
            writer.writerow(item.to_row())


def validate_maven_packages(sdk_info_list: List[SdkInfo]):
    for sdk_info in sdk_info_list:
        check_maven_url(sdk_info.t1_maven_url())
        check_maven_url(sdk_info.t2_maven_url())


def check_maven_url(url: str):
    url = url\
              .replace('mvnrepository.com', 'mvnrepository#com')\
              .replace('.', '/')\
              .replace('mvnrepository#com', 'mvnrepository.com')\
              .replace('https://mvnrepository.com/artifact/', 'https://repo1.maven.org/maven2/') + '/maven-metadata.xml'
    r = requests.get(url)
    logging.info(f'check url {url}, status code {r.status_code}')
    r.raise_for_status()


if __name__ == "__main__":
    run()