SkillForge

### Generated Skill: `DataSanitizer`

**Description**: Automatically sanitizes and validates CSV data files before ingestion into a data pipeline.

**Primary Function**: The skill processes raw CSV files, removes duplicates, validates required fields (e.g., `email`, `timestamp`), and flags records with anomalies (e.g., malformed emails, future dates). It integrates with AWS S3 for file storage and Slack for alerting.

**Key Features**:
- **Duplicate Removal**: Uses a rolling 7-day hash-based deduplication system to avoid false positives.
- **Validation Rules**: Enforces regex patterns for emails (`^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$`), checks for null values in required fields, and validates timestamps against the current date.
- **Anomaly Detection**: Flags records where the `timestamp` is more than 30 days in the future or where the `email` domain doesn’t match a predefined allowlist (e.g., `@company.com`).
- **Integration**: Automatically uploads sanitized files to an S3 bucket (`s3://data-pipeline-sanitized/`) and sends a Slack alert to the `#data-team` channel with a summary of processed records and any errors.
- **Unit Tests**: Includes tests for edge cases like empty files, files with no header row, and files with 100% invalid records.

**Sample Output**:
```python
# skill.py
import csv
import re
import hashlib
from datetime import datetime
import boto3
from slack_sdk import WebClient

class DataSanitizer:
    def __init__(self):
        self.s3 = boto3.client('s3')
        self.slack = WebClient(token='xoxb-your-token')
        self.allowlisted_domains = ['@company.com']

    def sanitize_csv(self, input_bucket, input_key, output_bucket):
        # Download file from S3
        response = self.s3.get_object(Bucket=input_bucket, Key=input_key)
        content = response['Body'].read().decode('utf-8').splitlines()
        reader = csv.DictReader(content)

        # Process records
        sanitized_records = []
        seen_hashes = set()
        errors = []

        for row in reader:
            # Check for duplicates
            row_hash = hashlib.md5(str(row).encode()).hexdigest()
            if row_hash in seen_hashes:
                continue
            seen_hashes.add(row_hash)

            # Validate email
            email = row.get('email')
            if not email or not re.match(r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$', email):
                errors.append(f"Invalid email: {email}")
                continue

            # Validate domain
            if not any(email.endswith(domain) for domain in self.allowlisted_domains):
                errors.append(f"Email domain not allowlisted: {email}")
                continue

            # Validate timestamp
            timestamp = row.get('timestamp')
            try:
                parsed_date = datetime.strptime(timestamp, '%Y-%m-%d %H:%M:%S')
                if parsed_date > datetime.now():
                    errors.append(f"Future timestamp: {timestamp}")
                    continue
            except ValueError:
                errors.append(f"Invalid timestamp format: {timestamp}")
                continue

            sanitized_records.append(row)

        # Upload to S3
        output_key = f"sanitized/{input_key}"
        self.s3.put_object(Bucket=output_bucket, Key=output_key, Body='\n'.join([str(record) for record in sanitized_records]))

        # Send Slack alert
        summary = f"Processed {len(sanitized_records)} records. Found {len(errors)} errors."
        self.slack.chat_postMessage(channel='#data-team', text=summary)
        return summary

# Unit tests
import unittest
from io import StringIO

class TestDataSanitizer(unittest.TestCase):
    def setUp(self):
        self.sanitizer = DataSanitizer()

    def test_empty_file(self):
        csv_content = ''
        reader = csv.DictReader(StringIO(csv_content))
        self.assertEqual(len(list(reader)), 0)

    def test_invalid_email(self):
        csv_content = 'email,timestamp\ninvalid-email,2023-01-01 00:00:00'
        reader = csv.DictReader(StringIO(csv_content))
        records = list(reader)
        self.assertEqual(len(records), 0)

    def test_future_timestamp(self):
        csv_content = 'email,timestamp\ntest@company.com,2099-01-01 00:00:00'
        reader = csv.DictReader(StringIO(csv_content))
        records = list(reader)
        self.assertEqual(len(records), 0)

if __name__ == '__main__':
    unittest.main()
```

**README.md**:
```markdown
# DataSanitizer Skill

A Claude Code skill for sanitizing and validating CSV data files.

## Setup
1. Install dependencies:
   ```bash
   pip install boto3 slack-sdk
   ```
2. Configure AWS credentials:
   ```bash
   aws configure
   ```
3. Set Slack token in environment:
   ```bash
   export SLACK_TOKEN=xoxb-your-token
   ```

## Usage
Run the skill with:
```bash
claude code skill.py --input-bucket your-input-bucket --input-key raw/data.csv --output-bucket your-output-bucket
```

## Testing
Run unit tests:
```bash
python -m unittest skill.py
```
```