(self, args: list[str], opts: Namespace)
| 67 | sys.stdout.buffer.write(bytes_ + b"\n") |
| 68 | |
| 69 | def run(self, args: list[str], opts: Namespace) -> None: |
| 70 | if len(args) != 1 or not is_url(args[0]): |
| 71 | raise UsageError |
| 72 | request = Request( |
| 73 | args[0], |
| 74 | callback=self._print_response, |
| 75 | cb_kwargs={"opts": opts}, |
| 76 | dont_filter=True, |
| 77 | ) |
| 78 | # by default, let the framework handle redirects, |
| 79 | # i.e. command handles all codes expect 3xx |
| 80 | if not opts.no_redirect: |
| 81 | request.meta["handle_httpstatus_list"] = SequenceExclude(range(300, 400)) |
| 82 | else: |
| 83 | request.meta["handle_httpstatus_all"] = True |
| 84 | |
| 85 | spidercls: type[Spider] = DefaultSpider |
| 86 | assert self.crawler_process |
| 87 | spider_loader = self.crawler_process.spider_loader |
| 88 | if opts.spider: |
| 89 | spidercls = spider_loader.load(opts.spider) |
| 90 | else: |
| 91 | spidercls = spidercls_for_request(spider_loader, request, spidercls) |
| 92 | |
| 93 | async def start(self: Spider) -> AsyncIterator[Any]: |
| 94 | yield request |
| 95 | |
| 96 | spidercls.start = start # type: ignore[method-assign] |
| 97 | |
| 98 | self.crawler_process.crawl(spidercls) |
| 99 | self.crawler_process.start() |
nothing calls this directly
no test coverage detected