subscription.py 文件源码

def _handle_http_codes(self, parsed: feedparser.FeedParserDict) -> "UpdateResult":
        """
        Given feedparser parse result, determine if parse succeeded, and what to do about that.
        """
        # Feedparser gives no status if you feedparse a local file.
        if "status" not in parsed:
            LOG.debug("Saw status 200 - OK, all is well.")
            return UpdateResult.SUCCESS

        status = parsed.get("status", 200)
        result = UpdateResult.SUCCESS
        if status == requests.codes["NOT_FOUND"]:
            LOG.error(f"Saw status {status}, unable to retrieve feed text for "
                      f"{self.metadata['name']}."
                      f"\nStored URL {self.url} for {self.metadata['name']} will be preserved"
                      f"and checked again on next attempt."
                     )

            result = UpdateResult.FAILURE

        elif status in [requests.codes["UNAUTHORIZED"], requests.codes["GONE"]]:
            LOG.error(f"Saw status {status}, unable to retrieve feed text for "
                      f"{self.metadata['name']}."
                      f"\nClearing stored URL {self.url} for {self.metadata['name']}."
                      f"\nPlease provide new URL and authorization for subscription "
                      f"{self.metadata['name']}."
                     )

            self.url = None
            result = UpdateResult.FAILURE

        # Handle redirecting errors
        elif status in [requests.codes["MOVED_PERMANENTLY"], requests.codes["PERMANENT_REDIRECT"]]:
            LOG.warning(f"Saw status {status} indicating permanent URL change."
                        f"\nChanging stored URL {self.url} for {self.metadata['name']} to "
                        f"{parsed.get('href')} and attempting get with new URL."
                       )

            self.url = parsed.get("href")
            result = UpdateResult.ATTEMPT_AGAIN

        elif status in [requests.codes["FOUND"], requests.codes["SEE_OTHER"],
                        requests.codes["TEMPORARY_REDIRECT"]]:
            LOG.warning(f"Saw status {status} indicating temporary URL change."
                        f"\nAttempting with new URL {parsed.get('href')}."
                        f"\nStored URL {self.url} for {self.metadata['name']} will be unchanged.")

            self.temp_url = self.url
            self.url = parsed.get("href")
            result = UpdateResult.ATTEMPT_AGAIN

        elif status != 200:
            LOG.warning(f"Saw '{status}'. Retrying retrieve for {self.metadata['name']} " +
                        f"at {self.url}.")
            result = UpdateResult.ATTEMPT_AGAIN

        else:
            LOG.debug("Saw status 200. Success!")

        return result