New configuration setting LINKBACKS_IGNORED_URLS_PATTERN

Author: Lucas-C

.pylintrc

@@ -1,5 +1,5 @@
 [MESSAGES CONTROL]
-disable = broad-except, missing-docstring, multiple-imports, too-few-public-methods, too-many-arguments, too-many-locals, too-many-positional-arguments
+disable = broad-except, missing-docstring, multiple-imports, too-few-public-methods, too-many-arguments, too-many-branches, too-many-locals, too-many-positional-arguments, wrong-import-order
 
 [FORMAT]
 max-line-length = 180

CHANGELOG.md

@@ -11,6 +11,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 
 ### Added
 
+- new configuration setting `LINKBACKS_IGNORED_URLS_PATTERN` to define some URLs that should never be considered for linkbacks (_e.g._ `youtube.com`)
 - manual execution mode: `python linkbacks.py $pelican_generated_html_file`
 
 ### Changed

pelican/plugins/linkbacks/linkbacks.py

@@ -9,6 +9,7 @@
 import json
 import logging
 import os
+import re
 import sys
 from os import makedirs
 from os.path import basename, splitext
@@ -32,6 +33,8 @@
 DEFAULT_USER_AGENT = 'pelican-plugin-linkbacks'
 DEFAULT_CERT_VERIFY = True
 DEFAULT_TIMEOUT = 3
+DEFAULT_IGNORED_URLS_PATTERN = 'artstation.com|deviantart.com|github.com|github.io|itch.io|readthedocs.io|youtube.com|wikipedia.org'
+IMAGE_EXTENSIONS = ('.gif', '.jpg', '.pdf', '.png', '.svg')
 WEBMENTION_POSS_REL = ('webmention', 'http://webmention.org', 'http://webmention.org/', 'https://webmention.org', 'https://webmention.org/')
 
 LOGGER = logging.getLogger(__name__)
@@ -79,9 +82,12 @@ def process_all_links_of_an_article(config, cache, url, slug, content):
         if config.siteurl and link_url.startswith(config.siteurl):
             LOGGER.debug("Link url %s skipped because is starts with %s", link_url, config.siteurl)
             continue
-        if splitext(link_url)[1] in ('.gif', '.jpg', '.pdf', '.png', '.svg'):
+        if splitext(link_url)[1] in IMAGE_EXTENSIONS:
             LOGGER.debug("Link url %s skipped because it appears to be an image or PDF file", link_url)
             continue
+        if config.ignored_urls_pattern.search(link_url):
+            LOGGER.debug("Link url %s skipped because it matches the ignored URLs pattern", link_url)
+            continue
         cache_status = cache.get_status(slug, link_url)
         if cache_status:
             LOGGER.debug("Link url %s skipped because it is present in cache with status: %s", link_url, cache_status)
@@ -104,9 +110,9 @@ def process_all_links_of_an_article(config, cache, url, slug, content):
                     continue
                 response = notifier.send()
                 LOGGER.info("%s notification sent for URL %s, endpoint response: %s", notifier.kind, link_url, response)
-                cache.add_success(slug, link_url, notifier.kind, notifier.server_uri)
+                cache.add_success(slug, link_url, notifier.kind, notifier.server_uri, response)
                 successful_notifs_count += 1
-            except (ConnectionError, HTTPError, RequestException, SSLError, xmlrpc.client.ProtocolError) as error:
+            except (ConnectionError, HTTPError, NotifierError, RequestException, SSLError, xmlrpc.client.ProtocolError) as error:
                 LOGGER.error("Failed to send %s for link url %s: [%s] %s", notifier.kind, link_url, error.__class__.__name__, error)
                 cache.add_failure(slug, link_url, error, notifier.kind, notifier.server_uri)
             except Exception as error:  # unexpected exception => we display the stacktrace:
@@ -128,6 +134,9 @@ def __init__(self, settings=None):
         self.cert_verify = settings.get('LINKBACKS_CERT_VERIFY', DEFAULT_CERT_VERIFY)
         self.timeout = settings.get('LINKBACKS_REQUEST_TIMEOUT', DEFAULT_TIMEOUT)
         self.user_agent = settings.get('LINKBACKS_USERAGENT', DEFAULT_USER_AGENT)
+        self.ignored_urls_pattern = settings.get('LINKBACKS_IGNORED_URLS_PATTERN', DEFAULT_IGNORED_URLS_PATTERN)
+        if self.ignored_urls_pattern and isinstance(self.ignored_urls_pattern, str):
+            self.ignored_urls_pattern = re.compile(self.ignored_urls_pattern)
 
 class Cache:
     def __init__(self, config, data):
@@ -137,12 +146,14 @@ def __init__(self, config, data):
         #   $article_slug: {
         #     $link_url: {
         #       "pingback": {
+        #         "error": // string or null if successful
+        #         "response": // string or null if failed
         #         "server_uri": "http...", // optional string
-        #         "error": // string or null if successfull
         #       },
         #       "webmention": {
+        #         "error": // string or null if successful
+        #         "response": // string or null if failed
         #         "server_uri": "http...", // optional string
-        #         "error": // string or null if successfull
         #       }
         #     },
         #     ...
@@ -151,13 +162,14 @@ def __init__(self, config, data):
         # }
         self.data = defaultdict(dict)
         self.data.update(data)
-    def add_success(self, article_slug, link_url, kind, server_uri):
+    def add_success(self, article_slug, link_url, kind, server_uri, response):
         article_links = self.data[article_slug]
         link_status = article_links.get(link_url)
         if link_status is None:
             link_status = {}
             article_links[link_url] = link_status
         link_status[kind] = {
+            "response": response,
             "server_uri": server_uri
         }
     def add_failure(self, article_slug, link_url, error, notifier_kind=None, server_uri=None):
@@ -186,11 +198,9 @@ def get_status(self, article_slug, link_url):
             return None  # defensive, should never happen
         # For now we never retry sending pingbacks & webmentions if there is already an entry in the cache.
         # Later on, we could for example consider retrying on HTTP 5XX errors.
-        pingback_error = pingback_status.get("error")
-        webmention_error = webmention_status.get("error")
-        if pingback_error is None or webmention_error is None:
+        if pingback_status.get("response") or webmention_status.get("response"):
             return "ALREADY SUBMITTED"
-        return pingback_error or webmention_error
+        return pingback_status.get("error") or webmention_status.get("error")
     def links_count(self):
         return sum(len(url_statuses) for url_statuses in self.data.values())
     @classmethod
@@ -227,6 +237,9 @@ def discover_server_uri(self):
     def send(self):
         "Sends the actual notification."
 
+class NotifierError(RuntimeError):
+    pass
+
 class PingbackNotifier(Notifier):
     def __init__(self, source_url, target_url, config=LinkbackConfig()):
         self.kind = "pingback"
@@ -253,8 +266,8 @@ def send(self):
             return xml_rpc_client.pingback.ping(self.source_url, self.target_url)
         except xmlrpc.client.Fault as fault:
             if fault.faultCode == 48:  # pingback already registered
-                raise RuntimeError(f"Pingback already registered for URL {self.target_url}, XML-RPC response: code={fault.faultCode} - {fault.faultString}") from fault
-            raise RuntimeError(f"Pingback XML-RPC request failed for URL {self.target_url}: code={fault.faultCode} - {fault.faultString}") from fault
+                raise NotifierError(f"Pingback already registered for URL {self.target_url}, XML-RPC response: code={fault.faultCode} - {fault.faultString}") from fault
+            raise NotifierError(f"Pingback XML-RPC request failed for URL {self.target_url}: code={fault.faultCode} - {fault.faultString}") from fault
 
 class WebmentionNotifier(Notifier):
     def __init__(self, source_url, target_url, config=LinkbackConfig()):

pelican/plugins/linkbacks/test_linkbacks.py

@@ -1,18 +1,17 @@
 import json, logging, os
 
 import httpretty
+from pelican.generators import ArticlesGenerator
+from pelican.tests.support import get_settings
+import pytest
+
 from linkbacks import (
     process_all_articles_linkbacks,
     Cache,
     LinkbackConfig,
     CACHE_FILENAME,
     MAX_RESPONSE_LENGTH,
 )
-import pytest
-
-from pelican.generators import ArticlesGenerator
-from pelican.tests.support import get_settings
-
 
 CUR_DIR = os.path.dirname(__file__)
 TEST_CONTENT_DIR = os.path.join(CUR_DIR, 'test_content')