Repository: scrapy/scrapy
Pull request #5892 (2023-04-11 19:07:23)

implement request_to_curl function
Implement request_to_curl util function that generates a curl command string from a scrapy.Request object


Patch:
diff --git a/scrapy/linkextractors/lxmlhtml.py b/scrapy/linkextractors/lxmlhtml.py
index dd8dcdf7c..23cbd0116 100644
--- a/scrapy/linkextractors/lxmlhtml.py
+++ b/scrapy/linkextractors/lxmlhtml.py
@@ -1,6 +1,7 @@
 """
 Link extractor based on lxml.html
 """
+import logging
 import operator
 from functools import partial
 from urllib.parse import urljoin, urlparse
@@ -23,6 +24,8 @@ from scrapy.utils.python import unique as unique_list
 from scrapy.utils.response import get_base_url
 from scrapy.utils.url import url_has_any_extension, url_is_from_any_domain
 
+logger = logging.getLogger(__name__)
+
 # from lxml/src/lxml/html/__init__.py
 XHTML_NAMESPACE = "http://www.w3.org/1999/xhtml"
 
@@ -88,7 +91,12 @@ class LxmlParserLinkExtractor:
                 url = self.process_attr(attr_val)
                 if url is None:
                     continue
-            url = safe_url_string(url, encoding=response_encoding)
+            try:
+                url = safe_url_string(url, encoding=response_encoding)
+            except ValueError:
+                logger.debug(f"Skipping extraction of link with bad URL {url!r}")
+                continue
+
             # to fix relative links after process_value
             url = urljoin(response_url, url)
             link = Link(
diff --git a/scrapy/utils/python.py b/scrapy/utils/python.py
index fc50e0f12..818fa5d6b 100644
--- a/scrapy/utils/python.py
+++ b/scrapy/utils/python.py
@@ -174,33 +174,33 @@ def binary_is_text(data):
 
 
 def get_func_args(func, stripself=False):
-    """Return the argument name list of a callable"""
-    if inspect.isfunction(func):
-        spec = inspect.getfullargspec(func)
-        func_args = spec.args + spec.kwonlyargs
-    elif inspect.isclass(func):
-        return get_func_args(func.__init__, True)
-    elif inspect.ismethod(func):
-        return get_func_args(func.__func__, True)
-    elif inspect.ismethoddescriptor(func):
-        return []
-    elif isinstance(func, partial):
-        return [
-            x
-            for x in get_func_args(func.func)[len(func.args) :]
-            if not (func.keywords and x in func.keywords)
-        ]
-    elif hasattr(func, "__call__"):
-        if inspect.isroutine(func):
-            return []
-        if getattr(func, "__name__", None) == "__call__":
-            return []
-        return get_func_args(func.__call__, True)
+    """Return the argument name list of a callable object"""
+    if not callable(func):
+        raise TypeError(f"func must be callable, got '{type(func).__name__}'")
+
+    args = []
+    try:
+        sig = inspect.signature(func)
+    except ValueError:
+        return args
+
+    if isinstance(func, partial):
+        partial_args = func.args
+        partial_kw = func.keywords
+
+        for name, param in sig.parameters.items():
+            if param.name in partial_args:
+                continue
+            if partial_kw and param.name in partial_kw:
+                continue
+            args.append(name)
     else:
-        raise TypeError(f"{type(func)} is not callable")
-    if stripself:
-        func_args.pop(0)
-    return func_args
+        for name in sig.parameters.keys():
+            args.append(name)
+
+    if stripself and args and args[0] == "self":
+        args = args[1:]
+    return args
 
 
 def get_spec(func):
diff --git a/scrapy/utils/request.py b/scrapy/utils/request.py
index 409ca2e52..6d8be991d 100644
--- a/scrapy/utils/request.py
+++ b/scrapy/utils/request.py
@@ -327,3 +327,34 @@ def _get_method(obj, name):
         return getattr(obj, name)
     except AttributeError:
         raise ValueError(f"Method {name!r} not found in: {obj}")
+
+
+def request_to_curl(request: Request) -> str:
+    """
+    Converts a :class:`~scrapy.Request` object to a curl command.
+
+    :param :class:`~scrapy.Request`: Request object to be converted
+    :return: string containing the curl command
+    """
+    method = request.method
+
+    data = f"--data-raw '{request.body.decode('utf-8')}'" if request.body else ""
+
+    headers = " ".join(
+        f"-H '{k.decode()}: {v[0].decode()}'" for k, v in request.headers.items()
+    )
+
+    url = request.url
+    cookies = ""
+    if request.cookies:
+        if isinstance(request.cookies, dict):
+            cookie = "; ".join(f"{k}={v}" for k, v in request.cookies.items())
+            cookies = f"--cookie '{cookie}'"
+        elif isinstance(request.cookies, list):
+            cookie = "; ".join(
+                f"{list(c.keys())[0]}={list(c.values())[0]}" for c in request.cookies
+            )
+            cookies = f"--cookie '{cookie}'"
+
+    curl_cmd = f"curl -X {method} {url} {data} {headers} {cookies}".strip()
+    return " ".join(curl_cmd.split())
diff --git a/tox.ini b/tox.ini
index d96a278ea..06b52f3dc 100644
--- a/tox.ini
+++ b/tox.ini
@@ -101,6 +101,7 @@ install_command =
     python -I -m pip install {opts} {packages}
 
 [testenv:pinned]
+basepython = python3.7
 deps =
     {[pinned]deps}
     PyDispatcher==2.0.5


Test patch:
diff --git a/tests/test_linkextractors.py b/tests/test_linkextractors.py
index f663013ba..e1ec19601 100644
--- a/tests/test_linkextractors.py
+++ b/tests/test_linkextractors.py
@@ -2,6 +2,10 @@ import pickle
 import re
 import unittest
 
+from packaging.version import Version
+from pytest import mark
+from w3lib import __version__ as w3lib_version
+
 from scrapy.http import HtmlResponse, XmlResponse
 from scrapy.link import Link
 from scrapy.linkextractors.lxmlhtml import LxmlLinkExtractor
@@ -815,3 +819,34 @@ class LxmlLinkExtractorTestCase(Base.LinkExtractorTestCase):
 
     def test_restrict_xpaths_with_html_entities(self):
         super().test_restrict_xpaths_with_html_entities()
+
+    @mark.skipif(
+        Version(w3lib_version) < Version("2.0.0"),
+        reason=(
+            "Before w3lib 2.0.0, w3lib.url.safe_url_string would not complain "
+            "about an invalid port value."
+        ),
+    )
+    def test_skip_bad_links(self):
+        html = b"""
+        <a href="http://example.org:non-port">Why would you do this?</a>
+        <a href="http://example.org/item2.html">Good Link</a>
+        <a href="http://example.org/item3.html">Good Link 2</a>
+        """
+        response = HtmlResponse("http://example.org/index.html", body=html)
+        lx = self.extractor_cls()
+        self.assertEqual(
+            [link for link in lx.extract_links(response)],
+            [
+                Link(
+                    url="http://example.org/item2.html",
+                    text="Good Link",
+                    nofollow=False,
+                ),
+                Link(
+                    url="http://example.org/item3.html",
+                    text="Good Link 2",
+                    nofollow=False,
+                ),
+            ],
+        )
diff --git a/tests/test_utils_python.py b/tests/test_utils_python.py
index 57f40c2e5..80d2e8da1 100644
--- a/tests/test_utils_python.py
+++ b/tests/test_utils_python.py
@@ -235,20 +235,16 @@ class UtilsPythonTestCase(unittest.TestCase):
         self.assertEqual(get_func_args(partial_f3), ["c"])
         self.assertEqual(get_func_args(cal), ["a", "b", "c"])
         self.assertEqual(get_func_args(object), [])
+        self.assertEqual(get_func_args(str.split, stripself=True), ["sep", "maxsplit"])
+        self.assertEqual(get_func_args(" ".join, stripself=True), ["iterable"])
 
         if platform.python_implementation() == "CPython":
-            # TODO: how do we fix this to return the actual argument names?
-            self.assertEqual(get_func_args(str.split), [])
-            self.assertEqual(get_func_args(" ".join), [])
+            # doesn't work on CPython: https://bugs.python.org/issue42785
             self.assertEqual(get_func_args(operator.itemgetter(2)), [])
         elif platform.python_implementation() == "PyPy":
-            self.assertEqual(
-                get_func_args(str.split, stripself=True), ["sep", "maxsplit"]
-            )
             self.assertEqual(
                 get_func_args(operator.itemgetter(2), stripself=True), ["obj"]
             )
-            self.assertEqual(get_func_args(" ".join, stripself=True), ["iterable"])
 
     def test_without_none_values(self):
         self.assertEqual(without_none_values([1, None, 3, 4]), [1, 3, 4])
diff --git a/tests/test_utils_request.py b/tests/test_utils_request.py
index 6ca272de1..e6d1abe3f 100644
--- a/tests/test_utils_request.py
+++ b/tests/test_utils_request.py
@@ -1,3 +1,4 @@
+import json
 import unittest
 import warnings
 from hashlib import sha1
@@ -18,6 +19,7 @@ from scrapy.utils.request import (
     request_authenticate,
     request_fingerprint,
     request_httprepr,
+    request_to_curl,
 )
 from scrapy.utils.test import get_crawler
 
@@ -666,5 +668,67 @@ class CustomRequestFingerprinterTestCase(unittest.TestCase):
         self.assertEqual(fingerprint, settings["FINGERPRINT"])
 
 
+class RequestToCurlTest(unittest.TestCase):
+    def _test_request(self, request_object, expected_curl_command):
+        curl_command = request_to_curl(request_object)
+        self.assertEqual(curl_command, expected_curl_command)
+
+    def test_get(self):
+        request_object = Request("https://www.example.com")
+        expected_curl_command = "curl -X GET https://www.example.com"
+        self._test_request(request_object, expected_curl_command)
+
+    def test_post(self):
+        request_object = Request(
+            "https://www.httpbin.org/post",
+            method="POST",
+            body=json.dumps({"foo": "bar"}),
+        )
+        expected_curl_command = (
+            'curl -X POST https://www.httpbin.org/post --data-raw \'{"foo": "bar"}\''
+        )
+        self._test_request(request_object, expected_curl_command)
+
+    def test_headers(self):
+        request_object = Request(
+            "https://www.httpbin.org/post",
+            method="POST",
+            headers={"Content-Type": "application/json", "Accept": "application/json"},
+            body=json.dumps({"foo": "bar"}),
+        )
+        expected_curl_command = (
+            "curl -X POST https://www.httpbin.org/post"
+            ' --data-raw \'{"foo": "bar"}\''
+            " -H 'Content-Type: application/json' -H 'Accept: application/json'"
+        )
+        self._test_request(request_object, expected_curl_command)
+
+    def test_cookies_dict(self):
+        request_object = Request(
+            "https://www.httpbin.org/post",
+            method="POST",
+            cookies={"foo": "bar"},
+            body=json.dumps({"foo": "bar"}),
+        )
+        expected_curl_command = (
+            "curl -X POST https://www.httpbin.org/post"
+            " --data-raw '{\"foo\": \"bar\"}' --cookie 'foo=bar'"
+        )
+        self._test_request(request_object, expected_curl_command)
+
+    def test_cookies_list(self):
+        request_object = Request(
+            "https://www.httpbin.org/post",
+            method="POST",
+            cookies=[{"foo": "bar"}],
+            body=json.dumps({"foo": "bar"}),
+        )
+        expected_curl_command = (
+            "curl -X POST https://www.httpbin.org/post"
+            " --data-raw '{\"foo\": \"bar\"}' --cookie 'foo=bar'"
+        )
+        self._test_request(request_object, expected_curl_command)
+
+
 if __name__ == "__main__":
     unittest.main()