historical_daily_importer.py 文件源码

def import_history(self, start_date, end_date):
        dataframe_columns = RESCUETIME_EFFICIENCY_HEADERS + [PRODUCTIVITY_PULSE]
        historical_df = pd.DataFrame(columns=dataframe_columns)

        query_dates = pd.date_range(start=start_date, end=end_date).date

        for query_date in query_dates:
            response = self._get_rescuetime_efficiency_for_date(query_date)

            if response.status_code != 200:
                continue

            efficiency_timeseries = self.get_efficiency_timeseries_from_response(response)
            pulse = calculate_rescue_time_pulse_from_dataframe(efficiency_timeseries)
            efficiency_timeseries[PRODUCTIVITY_PULSE] = pulse

            # Update the dataframe with history
            historical_df.loc[query_date] = efficiency_timeseries

        # when done, update into the results
        self.results = historical_df