Following on from the first example, this adds one more step to the workflow.
The crawl is still triggered from self-hosted n8n, but once Screaming Frog has finished, a second Python script now preprocesses the export ready for whatever comes next downstream.
The additional step looks for the latest internal_all.csv, keeps only the columns I actually want, renames them into cleaner field names, and outputs a new:
So instead of pushing the full Screaming Frog Internal export into the next stage, I can strip it down first.
The fields are controlled directly in the script. If I do not want one included, I simply comment that mapping out.
For example:
The script also includes:
Set that to True if you want to keep the original internal_all.csv as well.
The workflow is now:
That cleaned file is then ready for the next step, whether that is reporting, comparisons, Google Sheets, a database, an LLM or something more agent-led.
As with the other guides, this is geared towards self-hosted n8n, but the same setup can be adapted to other automation platforms and cloud workflows.
Python preprocessing script:
The crawl is still triggered from self-hosted n8n, but once Screaming Frog has finished, a second Python script now preprocesses the export ready for whatever comes next downstream.
The additional step looks for the latest internal_all.csv, keeps only the columns I actually want, renames them into cleaner field names, and outputs a new:
Code:
internal_all_preprocessed.csvSo instead of pushing the full Screaming Frog Internal export into the next stage, I can strip it down first.
The fields are controlled directly in the script. If I do not want one included, I simply comment that mapping out.
For example:
Code:
"Address": "address",
"Status Code": "status_code",
"Indexability": "indexability",
"Title 1": "title",
"H1-1": "h1",
"Canonical Link Element 1": "canonical_url",The script also includes:
Code:
KEEP_ORIGINAL_FILE = FalseSet that to True if you want to keep the original internal_all.csv as well.
The workflow is now:
Code:
Scheduler
↓
Python SF CLI Trigger
↓
Python PreprocessThat cleaned file is then ready for the next step, whether that is reporting, comparisons, Google Sheets, a database, an LLM or something more agent-led.
As with the other guides, this is geared towards self-hosted n8n, but the same setup can be adapted to other automation platforms and cloud workflows.
Python preprocessing script:
Code:
"""Create a downstream-ready version of the latest Screaming Frog Internal:All export."""
import csv
from pathlib import Path
OUTPUT_ROOT = Path(r"C:\Users\Administrator\Desktop\workflows\output")
OUTPUT_FILENAME = "internal_all_preprocessed.csv"
# True keeps internal_all.csv after preprocessing; False removes it after a successful run.
KEEP_ORIGINAL_FILE = False
# Comment out any mapping you don't want included in the downstream CSV.
COLUMN_MAP = {
"Address": "address",
"Content Type": "content_type",
"Status Code": "status_code",
"Status": "status",
"Indexability": "indexability",
# "Indexability Status": "indexability_status",
"Title 1": "title",
"Title 1 Length": "title_length",
# "Title 1 Pixel Width": "title_pixel_width",
"Meta Description 1": "meta_description",
"Meta Description 1 Length": "meta_description_length",
# "Meta Description 1 Pixel Width": "meta_description_pixel_width",
# "Meta Keywords 1": "meta_keywords",
# "Meta Keywords 1 Length": "meta_keywords_length",
"H1-1": "h1",
# "H1-1 Length": "h1_length",
# "H2-1": "h2_1",
# "H2-1 Length": "h2_1_length",
# "H2-2": "h2_2",
# "H2-2 Length": "h2_2_length",
# "Meta Robots 1": "meta_robots",
# "X-Robots-Tag 1": "x_robots_tag",
# "Meta Refresh 1": "meta_refresh",
"Canonical Link Element 1": "canonical_url",
# 'rel="next" 1': "rel_next",
# 'rel="prev" 1': "rel_prev",
# 'HTTP rel="next" 1': "http_rel_next",
# 'HTTP rel="prev" 1': "http_rel_prev",
# "amphtml Link Element": "amphtml_url",
"Size (Bytes)": "size_bytes",
# "Transferred (Bytes)": "transferred_bytes",
# "Total Transferred (Bytes)": "total_transferred_bytes",
# "CO2 (mg)": "co2_mg",
# "Carbon Rating": "carbon_rating",
"Word Count": "word_count",
# "Sentence Count": "sentence_count",
# "Average Words Per Sentence": "average_words_per_sentence",
# "Flesch Reading Ease Score": "flesch_reading_ease_score",
# "Readability": "readability",
# "Text Ratio": "text_ratio",
# "Crawl Depth": "crawl_depth",
"Folder Depth": "folder_depth",
# "Link Score": "link_score",
"Inlinks": "inlinks",
# "Unique Inlinks": "unique_inlinks",
# "Unique JS Inlinks": "unique_js_inlinks",
# "% of Total": "percent_of_total",
"Outlinks": "outlinks",
# "Unique Outlinks": "unique_outlinks",
# "Unique JS Outlinks": "unique_js_outlinks",
# "External Outlinks": "external_outlinks",
# "Unique External Outlinks": "unique_external_outlinks",
# "Unique External JS Outlinks": "unique_external_js_outlinks",
# "Closest Near Duplicate Match": "closest_near_duplicate_match",
# "No. Near Duplicates": "near_duplicate_count",
# "Spelling Errors": "spelling_errors",
# "Grammar Errors": "grammar_errors",
# "Hash": "content_hash",
"Response Time": "response_time",
# "Last Modified": "last_modified",
"Redirect URL": "redirect_url",
"Redirect Type": "redirect_type",
# "Cookies": "cookies",
# "Language": "language",
"HTTP Version": "http_version",
# "Mobile Alternate Link": "mobile_alternate_url",
# "Closest Semantically Similar Address": "closest_semantically_similar_address",
# "Semantic Similarity Score": "semantic_similarity_score",
# "No. Semantically Similar": "semantically_similar_count",
# "Semantic Relevance Score": "semantic_relevance_score",
# "URL Encoded Address": "url_encoded_address",
# "Crawl Timestamp": "crawl_timestamp",
}
def latest_export() -> Path:
exports = list(OUTPUT_ROOT.rglob("internal_all.csv"))
if not exports:
raise FileNotFoundError(f"No internal_all.csv found under {OUTPUT_ROOT}")
return max(exports, key=lambda path: path.stat().st_mtime)
def preprocess(source: Path) -> Path:
destination = source.with_name(OUTPUT_FILENAME)
with source.open("r", encoding="utf-8-sig", newline="") as input_file:
reader = csv.DictReader(input_file)
missing_headers = set(COLUMN_MAP).difference(reader.fieldnames or [])
if missing_headers:
raise ValueError(f"Expected headers not found: {sorted(missing_headers)}")
with destination.open("w", encoding="utf-8", newline="") as output_file:
writer = csv.DictWriter(output_file, fieldnames=COLUMN_MAP.values())
writer.writeheader()
for row in reader:
writer.writerow({output: row[source] for source, output in COLUMN_MAP.items()})
return destination
if __name__ == "__main__":
source_file = latest_export()
output_file = preprocess(source_file)
if not KEEP_ORIGINAL_FILE:
source_file.unlink()
print(f"Removed original export: {source_file}")
print(f"Preprocessed {source_file.name} to {output_file}")