Feature: Added DuckDB Connector (Resolves Issue #3)

src/agents/pii_anonymiser.py

@@ -0,0 +1,54 @@
+import re
+import pandas as pd
+from src.models import PIIAnonymiserResult
+
+# PII patterns
+PATTERNS = {
+    "email":       (r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}',
+                    lambda m: m.split('@')[0][0] + "***@***.com"),
+    "phone":       (r'(\+?\d[\d\s\-().]{7,}\d)',
+                    lambda m: re.sub(r'\d', '*', m[:-2]) + m[-2:]),
+    "postcode":    (r'\b[A-Z]{1,2}\d[A-Z\d]?\s*\d[A-Z]{2}\b',
+                    lambda m: m[:3] + "***"),
+    "card_number": (r'\b\d{4}[\s\-]?\d{4}[\s\-]?\d{4}[\s\-]?\d{4}\b',
+                    lambda m: "**** **** **** " + m.replace(" ","").replace("-","")[-4:]),
+}
+
+def anonymise_text(text: str) -> tuple[str, list]:
+    """Anonymise PII in a string, return cleaned text and list of findings."""
+    findings = []
+    for pii_type, (pattern, replacement) in PATTERNS.items():
+        matches = re.findall(pattern, str(text), re.IGNORECASE)
+        if matches:
+            findings.append(f"{pii_type}: {len(matches)} found")
+            text = re.sub(pattern,
+                         lambda m: replacement(m.group()),
+                         str(text),
+                         flags=re.IGNORECASE)
+    return text, findings
+
+def run(csv_preview: str, total_rows: int) -> "PIIAnonymiserResult":
+    print("[PII Anonymiser] Starting...")
+    lines = csv_preview.strip().split("\n")
+    cleaned_lines = []
+    all_findings = []
+    rows_affected = 0
+
+    for i, line in enumerate(lines):
+        cleaned_line, findings = anonymise_text(line)
+        cleaned_lines.append(cleaned_line)
+        if findings:
+            rows_affected += 1
+            all_findings.extend([f"Row {i}: {f}" for f in findings])
+
+    cleaned_preview = "\n".join(cleaned_lines)
+    pii_types = list({f.split(":")[0].strip() for f in all_findings})
+
+    print(f"[PII Anonymiser] Done — {rows_affected} rows had PII")
+
+    return PIIAnonymiserResult(
+        pii_found=all_findings,
+        rows_affected=rows_affected,
+        pii_types_detected=pii_types,
+        anonymised_preview=cleaned_preview
+    )

src/models.py

@@ -28,6 +28,15 @@ class SummariserResult(BaseModel):
     key_stats: dict
     recommendations: List[str]
 
+# ==========================================
+# 1. NEW PII MODEL ADDED HERE
+# ==========================================
+class PIIAnonymiserResult(BaseModel):
+    pii_found: List[str]
+    rows_affected: int
+    pii_types_detected: List[str]
+    anonymised_preview: str
+
 class PipelineResult(BaseModel):
     file_name: str
     total_rows: int
@@ -36,4 +45,8 @@ class PipelineResult(BaseModel):
     transformer: Optional[TransformerResult] = None
     anomaly: Optional[AnomalyResult] = None
     summariser: Optional[SummariserResult] = None
+    # ==========================================
+    # 2. INJECTED INTO FINAL PIPELINE PAYLOAD
+    # ==========================================
+    pii: Optional[PIIAnonymiserResult] = None 
     status: str = "complete"

src/pipeline.py

@@ -6,7 +6,7 @@
 from rich.table import Table
 from rich.panel import Panel
 from src.models import PipelineResult
-from src.agents import cleaner, validator, transformer, anomaly, summariser
+from src.agents import cleaner, validator, transformer, anomaly, summariser, pii_anonymiser
 
 console = Console()
 
@@ -33,42 +33,51 @@ def run_pipeline(file_path: str) -> PipelineResult:
         console=console
     ) as progress:
 
-        task1 = progress.add_task("[green]Agent 1/5 — Cleaner...", total=None)
+        task1 = progress.add_task("[green]Agent 1/6 — Cleaner...", total=None)
         cleaner_result = cleaner.run(preview, total_rows)
-        progress.update(task1, description="[green]Agent 1/5 — Cleaner ✓")
+        progress.update(task1, description="[green]Agent 1/6 — Cleaner ✓")
         progress.stop_task(task1)
 
-        task2 = progress.add_task("[blue]Agent 2/5 — Validator...", total=None)
-        validator_result = validator.run(preview, total_rows)
-        progress.update(task2, description="[blue]Agent 2/5 — Validator ✓")
+        task2 = progress.add_task("[cyan]Agent 2/6 — PII Anonymiser...", total=None)
+        pii_result = pii_anonymiser.run(preview, total_rows)
+
+        safe_preview = pii_result.anonymised_preview  
+
+        progress.update(task2, description="[cyan]Agent 2/6 — PII Anonymiser ✓")
         progress.stop_task(task2)
 
-        task3 = progress.add_task("[yellow]Agent 3/5 — Transformer...", total=None)
-        transformer_result = transformer.run(preview, total_rows)
-        progress.update(task3, description="[yellow]Agent 3/5 — Transformer ✓")
+        task3 = progress.add_task("[blue]Agent 3/6 — Validator...", total=None)
+        validator_result = validator.run(preview, total_rows)
+        progress.update(task3, description="[blue]Agent 3/6 — Validator ✓")
         progress.stop_task(task3)
 
-        task4 = progress.add_task("[red]Agent 4/5 — Anomaly Detector...", total=None)
-        anomaly_result = anomaly.run(preview, total_rows)
-        progress.update(task4, description="[red]Agent 4/5 — Anomaly Detector ✓")
+        task4 = progress.add_task("[yellow]Agent 4/6 — Transformer...", total=None)
+        transformer_result = transformer.run(preview, total_rows)
+        progress.update(task4, description="[yellow]Agent 4/6 — Transformer ✓")
         progress.stop_task(task4)
 
+        task5 = progress.add_task("[red]Agent 5/6 — Anomaly Detector...", total=None)
+        anomaly_result = anomaly.run(preview, total_rows)
+        progress.update(task5, description="[red]Agent 5/6 — Anomaly Detector ✓")
+        progress.stop_task(task5)
+
         context = f"""
         Cleaner found {len(cleaner_result.issues_fixed)} issues affecting {cleaner_result.rows_affected} rows.
         Validator score: {validator_result.completeness_score}% completeness, {len(validator_result.violations)} violations.
         Transformer applied {len(transformer_result.transformations_applied)} transformations.
         Anomaly detector found {anomaly_result.anomaly_count} anomalies with risk score {anomaly_result.anomaly_score}/10.
         """
 
-        task5 = progress.add_task("[magenta]Agent 5/5 — Summariser...", total=None)
+        task6 = progress.add_task("[magenta]Agent 6/6 — Summariser...", total=None)
         summariser_result = summariser.run(preview, total_rows, context)
-        progress.update(task5, description="[magenta]Agent 5/5 — Summariser ✓")
-        progress.stop_task(task5)
+        progress.update(task6, description="[magenta]Agent 6/6 — Summariser ✓")
+        progress.stop_task(task6)
 
     result = PipelineResult(
         file_name=file_name,
         total_rows=total_rows,
         cleaner=cleaner_result,
+        pii=pii_result,
         validator=validator_result,
         transformer=transformer_result,
         anomaly=anomaly_result,
@@ -89,6 +98,12 @@ def _print_summary(result: PipelineResult):
         "Cleaner",
         f"{result.cleaner.rows_affected} rows fixed, {len(result.cleaner.issues_fixed)} issues"
     )
+
+    table.add_row(
+        "PII Anonymiser",
+        f"{result.pii.rows_affected} rows masked, types: {', '.join(result.pii.pii_types_detected) or 'none detected'}"
+    )
+
     table.add_row(
         "Validator",
         f"{result.validator.completeness_score}% complete, {len(result.validator.violations)} violations"