Menü
>-
Standardmäßig ist der Prompt ausgewählt, der zuerst die Quelle prüft. Sie können zu einem direkten Befehl wechseln oder eine lokale Kopie herunterladen.
Lesen Sie SKILL.md und alle von SkillsMP angezeigten Begleitdateien, bevor Sie sich für eine Installation entscheiden.
Mit Codex oder Claude installieren Kopieren Sie diesen Prompt, fügen Sie ihn in Codex, Claude oder einen anderen Assistant ein und lassen Sie die Skill-Seite prüfen und installieren.
Ein direkter Befehl überspringt den Prüf-Prompt. Prüfen Sie die Quelle, bevor Sie ihn ausführen.
npx skills add https://github.com/CleanExpo/NodeJS-Starter-V1 --skill csv-processorDer Befehl bleibt in einer Zeile. Scrollen Sie horizontal, um ihn vor dem Kopieren vollständig zu prüfen.
Sie bevorzugen eine lokale Kopie? Laden Sie die Dateien herunter, die SkillsMP derzeit vorliegen.
SKILL.md wird angezeigt
Route complex requests to the right specialist agent or chain of agents. This skill acts as the central brain of an agent swarm — it analyses what the user needs, determines which specialist domain(s) are required, and coordinates parallel or sequential agent execution. Use this skill when a request spans multiple domains (e.g., "research competitors and create a pitch deck"), when you need to decide which specialist should handle an ambiguous request, or when a task requires a multi-step pipeline across different skills. Triggers on: multi-step requests, cross-domain tasks, "coordinate", "plan this out", "I need help with multiple things", or any complex request that touches more than one specialist area. Also triggers when the user seems unsure which tool or approach to use.
>-
Hybrid DAG execution primitive combining deterministic and agentic nodes with hard iteration caps
Basierend auf der SOC-Berufsklassifikation
| id | csv-processor |
| name | csv-processor |
| type | skill |
| version | 1.0.0 |
| created | 20/03/2026 |
| modified | 20/03/2026 |
| status | active |
| metadata | {"author":"NodeJS-Starter-V1","version":"1.0.0","locale":"en-AU"} |
| description | >- |
| context | fork |
Provides streaming CSV parse and generate patterns for NodeJS-Starter-V1, covering large file imports, exports, and transformations across Next.js frontend and FastAPI backend. Enforces row-by-row streaming, Zod and Pydantic row validation, and Australian locale formatting (DD/MM/YYYY dates, AUD currency).
data-validation instead)report-generator patterns)api-contract instead)| Library | Purpose | Install |
|---|---|---|
papaparse | Streaming CSV parse (browser + Node) | pnpm add papaparse |
@types/papaparse | TypeScript definitions | pnpm add -D @types/papaparse |
| Library | Purpose | Install |
|---|---|---|
python-multipart | File upload handling (already installed) | — |
Built-in csv module |
| Streaming read/write |
| — |
aiofiles | Async file I/O | uv add aiofiles |
No additional backend library needed — Python's built-in csv module supports streaming via csv.reader and csv.DictWriter.
import Papa from 'papaparse';
import * as z from 'zod';
// Define row schema — mirrors backend Pydantic model
const contractorRowSchema = z.object({
name: z.string().min(1, 'Name is required'),
email: z.string().email('Invalid email'),
phone: z.string().regex(/^04\d{8}$/, 'Australian mobile required'),
abn: z.string().regex(/^\d{11}$/, 'ABN must be 11 digits'),
state: z.enum(['QLD', 'NSW', 'VIC', 'SA', 'WA', 'TAS', 'NT', 'ACT']),
});
type ContractorRow = z.infer<typeof contractorRowSchema>;
interface ParseResult {
valid: ContractorRow[];
errors: Array<{ row: number; issues: z.ZodIssue[] }>;
}
function parseContractorCsv(file: File): Promise<ParseResult> {
return new Promise((resolve) => {
const valid: ContractorRow[] = [];
const errors: ParseResult['errors'] = [];
let rowIndex = 0;
Papa.parse(file, {
header: true,
skipEmptyLines: true,
step(results) {
rowIndex++;
const parsed = contractorRowSchema.safeParse(results.data);
if (parsed.success) {
valid.push(parsed.data);
} else {
errors.push({ row: rowIndex, issues: parsed.error.issues });
}
},
complete() {
resolve({ valid, errors });
},
});
});
}
import Papa from 'papaparse';
interface ExportOptions {
filename: string;
data: Record<string, unknown>[];
columns?: string[];
}
function exportToCsv({ filename, data, columns }: ExportOptions): void {
// Format Australian dates and currency before export
const formatted = data.map((row) => {
const out: Record<string, unknown> = {};
for (const [key, value] of Object.entries(row)) {
if (value instanceof Date) {
out[key] = value.toLocaleDateString('en-AU'); // DD/MM/YYYY
} else if (typeof value === 'number' && key.includes('amount')) {
out[key] = `$${value.toFixed(2)}`; // AUD
} else {
out[key] = value;
}
}
return out;
});
const csv = Papa.unparse(formatted, { columns });
const blob = new Blob([csv], { type: 'text/csv;charset=utf-8;' });
const url = URL.createObjectURL(blob);
const link = document.createElement('a');
link.href = url;
link.download = filename.endsWith('.csv') ? filename : `${filename}.csv`;
link.click();
URL.revokeObjectURL(url);
}
Wrap the parse function in a <input type="file" accept=".csv,text/csv"> component. Validate file extension (.csv), MIME type (text/csv), and size (default 10 MB cap) before parsing. Display row-level errors using ParseResult.errors.
import csv
import io
from typing import Any
from fastapi import APIRouter, File, HTTPException, UploadFile, status
from pydantic import BaseModel, Field, field_validator
router = APIRouter(prefix="/import", tags=["Import"])
class ContractorImportRow(BaseModel):
"""Schema for a single CSV row."""
name: str = Field(min_length=1)
email: str
phone: str
abn: str
state: str
@field_validator("phone")
@classmethod
def validate_phone(cls, v: str) -> str:
import re
cleaned = re.sub(r"[^\d]", "", v)
if not re.match(r"^04\d{8}$", cleaned):
raise ValueError("Australian mobile required (04XX XXX XXX)")
return cleaned
class ImportResult(BaseModel):
"""Result of CSV import."""
total_rows: int
valid_rows: int
error_rows: int
errors: list[dict[str, Any]] = Field(default_factory=list)
@router.post("/contractors", response_model=ImportResult)
async def import_contractors(file: UploadFile = File(...)) -> ImportResult:
"""Import contractors from CSV file."""
if not file.filename or not file.filename.endswith(".csv"):
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail="Only CSV files accepted",
)
content = await file.read()
reader = csv.DictReader(io.StringIO(content.decode("utf-8")))
valid: list[ContractorImportRow] = []
errors: list[dict[str, Any]] = []
for i, row in enumerate(reader, start=1):
try:
parsed = ContractorImportRow(**row)
valid.append(parsed)
except Exception as e:
errors.append({"row": i, "error": str(e)})
# Process valid rows (insert into database)
# await bulk_insert_contractors(valid)
return ImportResult(
total_rows=len(valid) + len(errors),
valid_rows=len(valid),
error_rows=len(errors),
errors=errors[:50], # Cap error list
)
import csv
import io
from datetime import datetime
from fastapi import APIRouter
from fastapi.responses import StreamingResponse
router = APIRouter(prefix="/export", tags=["Export"])
@router.get("/contractors")
async def export_contractors() -> StreamingResponse:
"""Export contractors as CSV download."""
# Fetch data
contractors = await get_all_contractors()
# Stream CSV output
output = io.StringIO()
writer = csv.DictWriter(
output,
fieldnames=["name", "email", "phone", "abn", "state", "created_at"],
)
writer.writeheader()
for c in contractors:
writer.writerow({
"name": c.name,
"email": c.email,
"phone": c.phone,
"abn": c.abn,
"state": c.state,
"created_at": _format_au_date(c.created_at),
})
output.seek(0)
timestamp = datetime.now().strftime("%d-%m-%Y")
return StreamingResponse(
iter([output.getvalue()]),
media_type="text/csv",
headers={
"Content-Disposition": f'attachment; filename="contractors-{timestamp}.csv"'
},
)
def _format_au_date(dt: datetime) -> str:
"""Format datetime as DD/MM/YYYY for CSV export."""
return dt.strftime("%d/%m/%Y")
For files over 10 MB, use async generators:
from collections.abc import AsyncGenerator
async def stream_csv_rows(
file: UploadFile,
chunk_size: int = 8192,
) -> AsyncGenerator[dict[str, str], None]:
"""Stream CSV rows without loading entire file into memory."""
buffer = ""
reader = None
while chunk := await file.read(chunk_size):
buffer += chunk.decode("utf-8")
lines = buffer.split("\n")
buffer = lines.pop() # Keep incomplete line in buffer
if reader is None and lines:
# First chunk — extract headers
header_line = lines.pop(0)
headers = next(csv.reader([header_line]))
reader = headers
for line in lines:
if line.strip() and reader:
values = next(csv.reader([line]))
yield dict(zip(reader, values))
# Process remaining buffer
if buffer.strip() and reader:
values = next(csv.reader([buffer]))
yield dict(zip(reader, values))
| Context | Format | Example |
|---|---|---|
| CSV export | DD/MM/YYYY | 23/01/2026 |
| CSV import (accept) | DD/MM/YYYY or ISO 8601 | 23/01/2026, 2026-01-23 |
| Database storage | ISO 8601 | 2026-01-23T00:00:00Z |
function parseAustralianDate(value: string): Date {
// Try DD/MM/YYYY first
const auMatch = value.match(/^(\d{2})\/(\d{2})\/(\d{4})$/);
if (auMatch) {
return new Date(`${auMatch[3]}-${auMatch[2]}-${auMatch[1]}`);
}
// Fall back to ISO 8601
const iso = new Date(value);
if (!isNaN(iso.getTime())) return iso;
throw new Error(`Invalid date: ${value}`);
}
from datetime import datetime
def parse_au_date(value: str) -> datetime:
"""Parse DD/MM/YYYY or ISO 8601 date string."""
for fmt in ("%d/%m/%Y", "%Y-%m-%d", "%Y-%m-%dT%H:%M:%S"):
try:
return datetime.strptime(value, fmt)
except ValueError:
continue
raise ValueError(f"Invalid date: {value}")
// Export: format as AUD
const amount = 1234.5;
const formatted = `$${amount.toFixed(2)}`; // "$1234.50"
// Import: strip $ and commas
const raw = '$1,234.50';
const parsed = parseFloat(raw.replace(/[$,]/g, '')); // 1234.5
| Anti-Pattern | Why It Fails | Correct Approach |
|---|---|---|
file.text() then split by \n | Breaks on quoted fields with newlines | Use PapaParse or csv module |
| Load entire file into array | Memory overflow on large files | Stream row-by-row |
| No row validation | Bad data corrupts database | Zod/Pydantic per row |
Hardcoded column indices (row[3]) | Breaks when columns reorder | Use header-based access (row.name) |
| MM/DD/YYYY date format | Australian users expect DD/MM/YYYY | Always DD/MM/YYYY in CSV |
| UTF-8 BOM not handled | Excel exports include BOM bytes | Strip BOM before parsing |
// PapaParse handles BOM automatically with `skipEmptyLines: true`
// Manual stripping if needed
function stripBom(text: string): string {
return text.charCodeAt(0) === 0xfeff ? text.slice(1) : text;
}
.csv extension and text/csv MIME)step or Python csv.reader)StreamingResponse (backend) or Blob download (frontend)$X,XXX.XX)contractors-23-01-2026.csv)Content-Disposition header set for download[AGENT_ACTIVATED]: CSV Processor
[PHASE]: {Design | Implementation | Review}
[STATUS]: {in_progress | complete}
{CSV processing analysis or implementation guidance}
[NEXT_ACTION]: {what to do next}
data-validation patternsUploadFile with response_model=ImportResultStreamingResponse with text/csv media typecsv_import_started, csv_import_completed, csv_import_failedtotal_rows, valid_rows, error_rows, duration_msDATA_VALIDATION_INVALID_FORMAT (422)ImportResult.errors (not HTTP errors)DATA_VALIDATION_FILE_TOO_LARGE (413)$X,XXX.XX format in CSV columns