How to Extend Spider Creator by Adding Custom Pipeline Stages

You can extend Spider Creator by creating a new Python module in the pipeline/ directory, exposing a callable that transforms data from the previous stage, and registering it in spidercreator.py to inject your logic into the execution flow.

Spider Creator, developed in the carlosplanchon/spidercreator repository, is built as a modular pipeline that converts recorded website interactions into fully functional Scrapy/Playwright spiders. Each logical step—from ROI classification to verification—lives in its own module under the pipeline/ package. Because the core orchestrator in spidercreator.py only calls functions that return data structures, you can drop in custom stages without modifying existing logic.

Understanding the Pipeline Architecture

The pipeline follows a linear data flow where each stage receives the output of the previous one. Key modules include:

• pipeline/xpath_builder_planning.py – Generates the structured planning model.
• pipeline/roiclf_spcandmkr.py – Classifies ROI HTML and creates candidate spider snippets.
• ctxexec/cand_sp_exec.py – Executes candidate spiders via CandSpiderExecutor.
• pipeline/verification_pipeline.py – Verifies extracted content against criteria.
• pipeline/sp_combination.py – Merges runnable fragments into the final script.

To extend Spider Creator by adding custom pipeline stages, you will insert a new module between execution and verification (or any other junction) to transform the data.

Step 1 – Create the Custom Stage Module

Create a new file at pipeline/postprocess_extracted.py. This example implements a post-processing stage that normalizes extracted data—such as trimming whitespace from prices and converting dates to ISO-8601 format—before verification occurs.

Define the Processing Function

# pipeline/postprocess_extracted.py

#!/usr/bin/env python3
"""
Custom stage that receives the raw spider output (a JSON string) and
applies user-defined normalisation rules.
"""

from typing import Any, Dict
import json
import re
from datetime import datetime

def normalise_price(text: str) -> str:
    """Strip currency symbols and whitespace."""
    return re.sub(r"[^\d.]", "", text).strip()

def normalise_date(text: str) -> str:
    """Try common date formats and output ISO-8601."""
    for fmt in ("%d/%m/%Y", "%Y-%m-%d", "%b %d, %Y"):
        try:
            return datetime.strptime(text.strip(), fmt).date().isoformat()
        except ValueError:
            continue
    return text  # fallback – return unchanged

def postprocess_extracted(
    spider_output: str,
    mapping: Dict[str, str] | None = None,
) -> str:
    """
    ``spider_output`` – JSON string generated by the candidate spider.
    ``mapping`` – optional dict where the key is the field name and the
    value indicates the normalisation function to apply
    (e.g. ``{"price": "normalise_price"}``).

    Returns a JSON string with the normalised payload.
    """
    # Load the raw output (assume a list of dicts)

    data = json.loads(spider_output)

    # Default mapping if the caller didn't provide one

    if mapping is None:
        mapping = {
            "price": "normalise_price",
            "date_listed": "normalise_date",
        }

    # Apply the transformations

    for item in data:
        for field, func_name in mapping.items():
            if field in item and isinstance(item[field], str):
                func = globals()[func_name]   # resolve by name

                item[field] = func(item[field])

    # Return a JSON string ready for the next pipeline stage

    return json.dumps(data, ensure_ascii=False, indent=2)

The postprocess_extracted function accepts the raw JSON string produced by CandSpiderExecutor (located in ctxexec/cand_sp_exec.py) and returns a transformed JSON string, maintaining the contract that pipeline stages pass serializable data.

Step 2 – Register the Stage in spidercreator.py

To integrate your custom stage, you must import the module and insert the call into the execution flow between candidate spider execution and verification.

Import the New Module

Open spidercreator.py and add the import near the other pipeline imports (around line 28):

# spidercreator.py

from pipeline.roiclf_spcandmkr import classify_roi_html_create_cand_spider
from pipeline.postprocess_extracted import postprocess_extracted   # ← NEW

Insert the Stage into the Execution Flow

Locate the block that executes candidate spiders (around line 84) and invoke your new stage before verification:

# After executing candidate spiders – insert before verification

CAND_SPIDER_EXEC_RESULTS = execute_cand_spiders(
    CAND_SPIDER_CREATION_RESULTS=CAND_SPIDER_CREATION_RESULTS,
    recordings_data=recordings_data
)

# -------------------------------------------------

# 5️⃣  NEW: Post-process the raw spider output

# -------------------------------------------------

POSTPROCESSED_RESULTS: dict[int, str] = {}
for key, exec_obj in CAND_SPIDER_EXEC_RESULTS.items():
    raw_output = exec_obj.spider_code_output_with_local_addresses
    # Apply custom normalisation (you can pass a custom mapping if needed)

    cleaned = postprocess_extracted(spider_output=raw_output)
    POSTPROCESSED_RESULTS[key] = cleaned
    print(f"\n--- POST-PROCESSED RESULT {key} ---")
    print(cleaned)

Finally, update the verification call to use the post-processed data instead of the raw output:

# Use the cleaned output for verification

xpath_verification_result = verify_spider_exec_result(
    spider_code_runnable=spider_code_runnable,
    verification_criteria=verification_criteria,
    extracted_content_on_rec=extracted_content_on_rec,
    spider_output=POSTPROCESSED_RESULTS[key]   # ← NEW

)

Step 3 – Configure via CLI (Optional)

To make your stage configurable without editing code, extend the argument parser in spidercreator.py to accept a JSON mapping file:

parser.add_argument(
    "--postprocess_map",
    help="Path to a JSON file that maps fields to normalisation functions",
    default=None,
)

Load the configuration before the pipeline runs:

if args.postprocess_map:
    with open(args.postprocess_map, "r", encoding="utf-8") as f:
        postprocess_map = json.load(f)
else:
    postprocess_map = None

Pass the mapping to your function:

cleaned = postprocess_extracted(
    spider_output=raw_output,
    mapping=postprocess_map
)

Step 4 – Test Your Extension

Run the tool on a small recording set to verify that your custom stage executes correctly:

python -m spidercreator --task_id sample01 --max_exec_instances 5

Confirm that:

1. The post-processed JSON appears in the console output under the --- POST-PROCESSED RESULT --- headers.
2. The verification step still succeeds (the verification logic in pipeline/verification_pipeline.py only evaluates content presence, not formatting).

You should see normalized output such as:

[
  {
    "title": "Cozy Studio",
    "price": "1250",
    "date_listed": "2024-03-12"
  }
]

Summary

• Spider Creator uses a modular pipeline where each stage in pipeline/ receives data from the previous step and returns enriched data for the next.
• To extend Spider Creator by adding custom pipeline stages, create a new module (e.g., pipeline/postprocess_extracted.py) that exposes a callable accepting and returning serializable data.
• Register the stage in spidercreator.py by importing the module and inserting the call between existing stages, such as after execute_cand_spiders and before verify_spider_exec_result.
• Optionally expose configuration via CLI arguments to allow runtime customization without code changes.
• Test your extension by running the full pipeline and verifying that downstream stages receive the transformed data correctly.

Frequently Asked Questions

What is the expected function signature for a custom pipeline stage?

A custom stage should expose a callable—typically a function—that accepts the output from the previous stage as its primary argument and returns the enriched data for the next stage. For example, postprocess_extracted(spider_output: str, mapping: dict | None = None) -> str receives a JSON string and returns a JSON string, maintaining the contract that pipeline stages pass serializable data structures.

Can I add multiple custom stages to the Spider Creator pipeline?

Yes, you can insert any number of bespoke stages by creating additional modules in the pipeline/ directory and chaining them in spidercreator.py. Each stage should follow the same pattern: import the module, call the function between existing stages, and pass the returned data to the next step. This allows you to build complex workflows such as data enrichment, caching, or logging without modifying the core logic.

How do I pass configuration options to my custom stage?

You can expose configuration via command-line arguments by extending the argument parser in spidercreator.py using parser.add_argument(). Load the configuration file (e.g., a JSON mapping) before the pipeline runs, then pass the loaded dictionary as a parameter to your custom stage function. This approach keeps the pipeline modular while allowing runtime customization without editing the source code.

Where should I place my custom stage module?

Custom stages should reside in the pipeline/ package alongside existing modules such as verification_pipeline.py and sp_combination.py. Create a new file (e.g., pipeline/my_custom_stage.py) and expose a callable function that follows the input/output contract of the pipeline. Import this module in spidercreator.py to register the stage in the execution flow.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}