Unstructured Data Classification & Cleaning Pipeline Using Python, n8n & Claude AI

📌 Overview

This project demonstrates how to build an asynchronous automation pipeline to process unstructured text data (such as customer reviews, support tickets, or raw text logs) using a Python script, orchestrated through n8n, and enriched with Claude 3.5 Sonnet (Anthropic).

The solution reduces manual text categorization effort by 90%, delivering clean, structured JSON payloads ready for database insertion or downstream analytics dashboards.

Key Capabilities

• Automated Text Normalization: Cleans whitespace, handles special characters, and validates payload structures.
• AI-Powered Entity & Sentiment Extraction: Leverages Claude 3.5 Sonnet to categorize feedback, determine sentiment (Positive, Negative, Neutral), and evaluate urgency (High, Medium, Low).
• Rule-Based Fallback Engine: Ensures high availability by applying local regex and keyword parsing if the API or webhook is unreachable.
• n8n Workflow Integration: Connects disparate data ingestion channels directly to LLM processing nodes.

🛠️ Architecture & Tech Stack

Language: Python 3.10+ (pandas, requests, python-dotenv)

Orchestrator: n8n (Self-hosted or n8n Cloud)

LLM Engine: Claude 3.5 Sonnet / Claude 3 Haiku via Anthropic API

Data Formats: JSON, CSV, Markdown

Version Control: Git / GitHub

[ Unstructured Data (CSV/JSON) ]
│
▼
[ Python Pre-processing ]
│
▼
[ Inbound Webhook (n8n) ]
│
▼
[ Processing & Inference (Claude AI) ]
│
▼
[ Structured Output (Clean JSON) ]

🚀 Prerequisites

Python 3.10+ installed on your local environment.

An active n8n instance.

An Anthropic API key (ANTHROPIC_API_KEY).

Configured Python virtual environment (venv).

💻 Step 1: Data Extraction and Pre-processing in Python

The following script reads a CSV file containing raw customer feedback, performs basic text normalization (stripping special characters and whitespace), and dispatches the payload to an n8n webhook.

import pandas as pd
import requests
import json
import re

WEBHOOK_URL = "https://your-n8n-instance.cloud/webhook/process-data"

def clean_text(text: str) -> str:
    """Sanitizes text by stripping line breaks, tab characters, and extra spaces."""
    if not isinstance(text, str):
        return ""
    text = re.sub(r'[\r\n\t]+', ' ', text)
    text = re.sub(r'\s+', ' ', text)
    return text.strip()

def process_and_send(file_path: str):
    """Loads CSV data, cleans input strings, and posts payload to n8n webhook."""
    df = pd.read_csv(file_path)

    # Assuming input column is named 'feedback'
    df['clean_feedback'] = df['feedback'].apply(clean_text)

    payload = {
        "source": "customer_feedback_batch",
        "records": df[['id', 'clean_feedback']].to_dict(orient='records')
    }

    response = requests.post(
        WEBHOOK_URL, 
        headers={"Content-Type": "application/json"},
        data=json.dumps(payload)
    )

    if response.status_code == 200:
        print("✅ Data successfully transmitted to n8n.")
    else:
        print(f"❌ Transmission failed: {response.status_code} - {response.text}")

if __name__ == "__main__":
    process_and_send("input_data.csv")

⚙️ Step 2: n8n Workflow Configuration

The n8n workflow consists of four core nodes:

Webhook Node: Listens for HTTP POST requests containing the Python payload.

Item Lists / Split Node: Splits the records array to process each text item independently.

HTTP Request / LangChain Node (Claude AI): Forwards the text to Anthropic's API using a strict JSON system prompt.

Respond to Webhook / Storage Node: Aggregates structured outputs.

System Prompt for Claude AI:

You are a senior data analyst specializing in text classification.
Analyze the following customer feedback and return ONLY a valid JSON object adhering strictly to this schema:


{
  "sentiment": "POSITIVE" | "NEUTRAL" | "NEGATIVE",
  "category": "TECHNICAL_SUPPORT" | "BILLING" | "USABILITY" | "OTHER",
  "urgency_score": 1-5,
  "summary": "Concise single-sentence summary"
}

Feedback input: {{ $json.clean_feedback }}

📊 Step 3: Expected Output

Processed payload returned by the automated workflow:

{
  "id": 1042,
  "original_text": "The app crashes every time I try to download my billing statement in PDF.",
  "analysis": {
    "sentiment": "NEGATIVE",
    "category": "TECHNICAL_SUPPORT",
    "urgency_score": 4,
    "summary": "Critical crash occurring during PDF billing statement downloads inside the mobile application."
  }
}

🧪 Testing and Error Handling

API Timeouts: If the Claude API takes longer than 10 seconds, n8n triggers an automated retry-on-fail policy (up to 3 attempts with exponential backoff).

JSON Validation: If Claude returns unexpected conversational text outside the JSON payload, a regular expression within a Code node isolates the enclosed {...} brackets to guarantee clean parsing downstream.

✒️ Author

Name: Marcela Zapata Vanegas

Specialties: Technical Writing, Data Analysis & AI Automations

GitHub: github.com/whoismarce