Batch processes multiple URLs simultaneously to extract clean text content AND discover all links from those pages
| 5 | from utils.url_validator import filter_valid_urls |
| 6 | |
| 7 | class CrawlAndExtract(BatchNode): |
| 8 | """Batch processes multiple URLs simultaneously to extract clean text content AND discover all links from those pages""" |
| 9 | |
| 10 | def prep(self, shared): |
| 11 | # The calling application is responsible for populating `urls_to_process`. |
| 12 | # This node just consumes the list. |
| 13 | urls_to_crawl = [] |
| 14 | for url_idx in shared.get("urls_to_process", []): |
| 15 | if url_idx < len(shared.get("all_discovered_urls", [])): |
| 16 | urls_to_crawl.append((url_idx, shared["all_discovered_urls"][url_idx])) |
| 17 | |
| 18 | return urls_to_crawl |
| 19 | |
| 20 | def exec(self, url_data): |
| 21 | """Process a single URL to extract content and links""" |
| 22 | url_idx, url = url_data |
| 23 | content, links = crawl_webpage(url) |
| 24 | return url_idx, content, links |
| 25 | |
| 26 | def exec_fallback(self, url_data, exc): |
| 27 | """Fallback when crawling fails. The 'None' for links signals a failure.""" |
| 28 | url_idx, url = url_data |
| 29 | print(f"Error crawling {url}: {exc}") |
| 30 | return url_idx, f"Error crawling page", None # Return None for links |
| 31 | |
| 32 | def post(self, shared, prep_res, exec_res_list): |
| 33 | """Store results and update URL tracking""" |
| 34 | new_urls = [] |
| 35 | content_max_chars = shared.get("content_max_chars", 10000) |
| 36 | max_links_per_page = shared.get("max_links_per_page", 300) |
| 37 | |
| 38 | successful_crawls = 0 |
| 39 | for url_idx, content, links in exec_res_list: |
| 40 | # This part only runs for successful crawls |
| 41 | successful_crawls += 1 |
| 42 | |
| 43 | # Truncate content to max chars |
| 44 | truncated_content = content[:content_max_chars] |
| 45 | if len(content) > content_max_chars: |
| 46 | truncated_content += f"\n... [Content truncated - original length: {len(content)} chars]" |
| 47 | |
| 48 | shared["url_content"][url_idx] = truncated_content |
| 49 | shared["visited_urls"].add(url_idx) |
| 50 | |
| 51 | valid_links = filter_valid_urls(links, shared["allowed_domains"]) |
| 52 | |
| 53 | if len(valid_links) > max_links_per_page: |
| 54 | valid_links = valid_links[:max_links_per_page] |
| 55 | |
| 56 | link_indices = [] |
| 57 | for link in valid_links: |
| 58 | if link not in shared["all_discovered_urls"]: |
| 59 | shared["all_discovered_urls"].append(link) |
| 60 | new_urls.append(len(shared["all_discovered_urls"]) - 1) |
| 61 | link_idx = shared["all_discovered_urls"].index(link) |
| 62 | link_indices.append(link_idx) |
| 63 | |
| 64 | shared["url_graph"][url_idx] = link_indices |
no outgoing calls
no test coverage detected