Source code for scrapy.logformatter
from __future__ import annotations
import logging
import os
from typing import TYPE_CHECKING, Any, Dict, Optional, Union
from twisted.python.failure import Failure
from scrapy import Request, Spider
from scrapy.http import Response
from scrapy.utils.request import referer_str
if TYPE_CHECKING:
# typing.Self requires Python 3.11
from typing_extensions import Self
from scrapy.crawler import Crawler
SCRAPEDMSG = "Scraped from %(src)s" + os.linesep + "%(item)s"
DROPPEDMSG = "Dropped: %(exception)s" + os.linesep + "%(item)s"
CRAWLEDMSG = "Crawled (%(status)s) %(request)s%(request_flags)s (referer: %(referer)s)%(response_flags)s"
ITEMERRORMSG = "Error processing %(item)s"
SPIDERERRORMSG = "Spider error processing %(request)s (referer: %(referer)s)"
DOWNLOADERRORMSG_SHORT = "Error downloading %(request)s"
DOWNLOADERRORMSG_LONG = "Error downloading %(request)s: %(errmsg)s"
[docs]class LogFormatter:
"""Class for generating log messages for different actions.
All methods must return a dictionary listing the parameters ``level``, ``msg``
and ``args`` which are going to be used for constructing the log message when
calling ``logging.log``.
Dictionary keys for the method outputs:
* ``level`` is the log level for that action, you can use those from the
`python logging library <https://docs.python.org/3/library/logging.html>`_ :
``logging.DEBUG``, ``logging.INFO``, ``logging.WARNING``, ``logging.ERROR``
and ``logging.CRITICAL``.
* ``msg`` should be a string that can contain different formatting placeholders.
This string, formatted with the provided ``args``, is going to be the long message
for that action.
* ``args`` should be a tuple or dict with the formatting placeholders for ``msg``.
The final log message is computed as ``msg % args``.
Users can define their own ``LogFormatter`` class if they want to customize how
each action is logged or if they want to omit it entirely. In order to omit
logging an action the method must return ``None``.
Here is an example on how to create a custom log formatter to lower the severity level of
the log message when an item is dropped from the pipeline::
class PoliteLogFormatter(logformatter.LogFormatter):
def dropped(self, item, exception, response, spider):
return {
'level': logging.INFO, # lowering the level from logging.WARNING
'msg': "Dropped: %(exception)s" + os.linesep + "%(item)s",
'args': {
'exception': exception,
'item': item,
}
}
"""
[docs] def crawled(self, request: Request, response: Response, spider: Spider) -> dict:
"""Logs a message when the crawler finds a webpage."""
request_flags = f" {str(request.flags)}" if request.flags else ""
response_flags = f" {str(response.flags)}" if response.flags else ""
return {
"level": logging.DEBUG,
"msg": CRAWLEDMSG,
"args": {
"status": response.status,
"request": request,
"request_flags": request_flags,
"referer": referer_str(request),
"response_flags": response_flags,
# backward compatibility with Scrapy logformatter below 1.4 version
"flags": response_flags,
},
}
[docs] def scraped(
self, item: Any, response: Union[Response, Failure], spider: Spider
) -> dict:
"""Logs a message when an item is scraped by a spider."""
src: Any
if isinstance(response, Failure):
src = response.getErrorMessage()
else:
src = response
return {
"level": logging.DEBUG,
"msg": SCRAPEDMSG,
"args": {
"src": src,
"item": item,
},
}
[docs] def dropped(
self, item: Any, exception: BaseException, response: Response, spider: Spider
) -> dict:
"""Logs a message when an item is dropped while it is passing through the item pipeline."""
return {
"level": logging.WARNING,
"msg": DROPPEDMSG,
"args": {
"exception": exception,
"item": item,
},
}
[docs] def item_error(
self, item: Any, exception: BaseException, response: Response, spider: Spider
) -> dict:
"""Logs a message when an item causes an error while it is passing
through the item pipeline.
.. versionadded:: 2.0
"""
return {
"level": logging.ERROR,
"msg": ITEMERRORMSG,
"args": {
"item": item,
},
}
[docs] def spider_error(
self,
failure: Failure,
request: Request,
response: Union[Response, Failure],
spider: Spider,
) -> dict:
"""Logs an error message from a spider.
.. versionadded:: 2.0
"""
return {
"level": logging.ERROR,
"msg": SPIDERERRORMSG,
"args": {
"request": request,
"referer": referer_str(request),
},
}
[docs] def download_error(
self,
failure: Failure,
request: Request,
spider: Spider,
errmsg: Optional[str] = None,
) -> dict:
"""Logs a download error message from a spider (typically coming from
the engine).
.. versionadded:: 2.0
"""
args: Dict[str, Any] = {"request": request}
if errmsg:
msg = DOWNLOADERRORMSG_LONG
args["errmsg"] = errmsg
else:
msg = DOWNLOADERRORMSG_SHORT
return {
"level": logging.ERROR,
"msg": msg,
"args": args,
}
@classmethod
def from_crawler(cls, crawler: Crawler) -> Self:
return cls()