def generate_web_osci_change_ranking()

in osci/publishing/transfer_monthly_change_ranking.py [0:0]

• 43 lines of code
• 2 McCabe index (conditional complexity)

def generate_web_osci_change_ranking(date: datetime) -> dict:
    log.info(f'Load OSCI change ranking {date:%Y-%m-%d}')
    ranking_df = OSCIChangeRanking(date=date).read()
    ranking_df[OSCIChangeRanking.schema.position] = range(1, len(ranking_df)+1)
    ranking_df = ranking_df.rename(columns=WebOSCIChangeRankingData.mapping).fillna(0)
    last_year = date - relativedelta(years=1)
    before_last_year = date - relativedelta(years=2)
    log.info(f'Load previous OSCI change ranking {last_year:%Y-%m-%d}')
    last_year_contributors = generate_company_to_active_contributors_map(OSCIRankingYTD(date=last_year).read())
    log.info(f'Load previous OSCI change ranking {before_last_year:%Y-%m-%d}')
    before_last_year_contributors = generate_company_to_active_contributors_map(
        OSCIRankingYTD(date=before_last_year).read()
    )
    log.info(f'Load OSCI Contributors ranking {date:%Y-%m-%d}')
    contributors_map = generate_company_to_contributors_map(OSCIContributorsRankingYTD(date=date).read().dropna())
    log.info(f'Load OSCI Languages ranking {date:%Y-%m-%d}')
    languages_map = generate_company_to_languages_map(OSCILanguagesYTD(date=date).read().dropna())
    log.info(f'Load OSCI Licenses ranking {date:%Y-%m-%d}')
    licenses_map = generate_company_to_licenses_map(OSCILicensesYTD(date=date).read().dropna())

    def _data_field_value_generator(rows: Iterable[dict]):
        for row in rows:
            yield {
                **row,
                WebOSCIChangeRankingData.Columns.year_of_year: [
                    dict(date=before_last_year,
                         active=before_last_year_contributors.get(row[WebOSCIChangeRankingData.Columns.company], [])),
                    dict(date=last_year,
                         active=last_year_contributors.get(row[WebOSCIChangeRankingData.Columns.company], []))
                ],
                WebOSCIChangeRankingData.Columns.contributors:
                    contributors_map.get(row[WebOSCIChangeRankingData.Columns.company]),
                WebOSCIChangeRankingData.Columns.languages:
                    languages_map.get(row[WebOSCIChangeRankingData.Columns.company]),
                WebOSCIChangeRankingData.Columns.licenses:
                    licenses_map.get(row[WebOSCIChangeRankingData.Columns.company]),
            }

    return {
        WebOSCIChangeRanking.Columns.date: date,
        WebOSCIChangeRanking.Columns.compared_date: get_compared_date(date),
        WebOSCIChangeRanking.Columns.data: list(
            _data_field_value_generator(ranking_df.to_dict(orient='records'))
        )
    }