Analytics Tracking

#!/usr/bin/env python3
"""Event Schema Checker - Validate analytics event names and parameters against a taxonomy.

Checks event names follow naming conventions (snake_case, noun_verb),
validates required parameters per event type, and detects PII leaks.

Usage:
    python event_schema_checker.py events.json
    python event_schema_checker.py events.json --json
    python event_schema_checker.py --generate-schema > schema.json
"""

import argparse
import json
import re
import sys


# Default SaaS event schema
DEFAULT_SCHEMA = {
    "naming_rules": {
        "case": "snake_case",
        "pattern": "noun_verb",
        "allowed_verbs": [
            "started", "completed", "failed", "viewed", "clicked",
            "submitted", "selected", "created", "updated", "deleted",
            "cancelled", "renewed", "opened", "closed", "sent",
            "received", "activated", "deactivated", "requested",
            "downloaded", "uploaded", "connected", "disconnected",
        ],
    },
    "events": {
        "signup_started": {"required_params": [], "optional_params": ["method", "source"]},
        "signup_completed": {"required_params": ["method"], "optional_params": ["user_id", "plan_name"]},
        "trial_started": {"required_params": ["plan_name"], "optional_params": ["user_id"]},
        "onboarding_step_completed": {"required_params": ["step_name", "step_number"], "optional_params": ["user_id"]},
        "feature_activated": {"required_params": ["feature_name"], "optional_params": ["user_id"]},
        "plan_selected": {"required_params": ["plan_name", "billing_period"], "optional_params": ["value", "currency"]},
        "checkout_started": {"required_params": ["value", "currency", "plan_name"], "optional_params": []},
        "checkout_completed": {"required_params": ["value", "currency", "transaction_id"], "optional_params": ["plan_name"]},
        "subscription_renewed": {"required_params": ["value", "plan_name"], "optional_params": ["currency"]},
        "subscription_cancelled": {"required_params": ["cancel_reason", "plan_name"], "optional_params": []},
        "pricing_viewed": {"required_params": [], "optional_params": ["source"]},
        "demo_requested": {"required_params": [], "optional_params": ["source"]},
        "form_submitted": {"required_params": ["form_name"], "optional_params": ["form_location"]},
        "content_downloaded": {"required_params": ["content_name"], "optional_params": ["content_type"]},
        "video_started": {"required_params": ["video_title"], "optional_params": []},
        "video_completed": {"required_params": ["video_title"], "optional_params": ["percent_watched"]},
    },
    "global_params": {
        "required_when_authenticated": ["user_id"],
        "required_with_value": ["currency"],
    },
}

# PII detection patterns
PII_PATTERNS = {
    "email": re.compile(r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}"),
    "phone": re.compile(r"(\+?\d{1,3}[-.\s]?)?\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4}"),
    "ssn": re.compile(r"\b\d{3}-\d{2}-\d{4}\b"),
    "credit_card": re.compile(r"\b\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}\b"),
    "ip_address": re.compile(r"\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b"),
}

PII_PARAM_NAMES = {
    "email", "mail", "e_mail", "user_email", "customer_email",
    "phone", "phone_number", "mobile", "telephone",
    "name", "first_name", "last_name", "full_name", "customer_name",
    "address", "street", "city", "zip", "postal_code",
    "ssn", "social_security", "tax_id",
    "credit_card", "card_number", "cvv", "expiry",
    "password", "secret", "token",
}


def check_naming_convention(event_name):
    """Check if event name follows snake_case noun_verb convention."""
    issues = []

    # Check snake_case
    if event_name != event_name.lower():
        issues.append({
            "type": "naming",
            "severity": "error",
            "message": f"Event '{event_name}' is not lowercase snake_case",
            "suggestion": event_name.lower(),
        })

    if "-" in event_name:
        issues.append({
            "type": "naming",
            "severity": "error",
            "message": f"Event '{event_name}' uses hyphens instead of underscores",
            "suggestion": event_name.replace("-", "_"),
        })

    if " " in event_name:
        issues.append({
            "type": "naming",
            "severity": "error",
            "message": f"Event '{event_name}' contains spaces",
            "suggestion": event_name.replace(" ", "_").lower(),
        })

    # Check camelCase
    if re.match(r"^[a-z]+[A-Z]", event_name):
        issues.append({
            "type": "naming",
            "severity": "error",
            "message": f"Event '{event_name}' appears to be camelCase",
            "suggestion": re.sub(r"([A-Z])", r"_\1", event_name).lower(),
        })

    # Check noun_verb pattern (last segment should be a verb)
    parts = event_name.lower().split("_")
    if len(parts) < 2:
        issues.append({
            "type": "naming",
            "severity": "warning",
            "message": f"Event '{event_name}' should follow noun_verb pattern (e.g., 'form_submitted')",
        })

    return issues


def check_pii(params):
    """Check event parameters for potential PII."""
    issues = []

    for key, value in params.items():
        # Check param name
        if key.lower() in PII_PARAM_NAMES:
            issues.append({
                "type": "pii",
                "severity": "error",
                "message": f"Parameter name '{key}' likely contains PII. Remove or hash before sending.",
                "param": key,
            })

        # Check param value for PII patterns
        if isinstance(value, str):
            for pii_type, pattern in PII_PATTERNS.items():
                if pattern.search(value):
                    issues.append({
                        "type": "pii",
                        "severity": "error",
                        "message": f"Parameter '{key}' value appears to contain {pii_type}: '{value[:20]}...'",
                        "param": key,
                        "pii_type": pii_type,
                    })

    return issues


def check_schema_compliance(event_name, params, schema):
    """Check if event parameters match the expected schema."""
    issues = []
    events = schema.get("events", {})

    if event_name not in events:
        issues.append({
            "type": "schema",
            "severity": "info",
            "message": f"Event '{event_name}' is not in the defined schema. This may be intentional.",
        })
        return issues

    event_def = events[event_name]

    # Check required params
    for req_param in event_def.get("required_params", []):
        if req_param not in params:
            issues.append({
                "type": "schema",
                "severity": "error",
                "message": f"Event '{event_name}' missing required parameter: '{req_param}'",
                "param": req_param,
            })

    # Check for unknown params
    known_params = set(event_def.get("required_params", []) + event_def.get("optional_params", []))
    known_params.update(schema.get("global_params", {}).get("required_when_authenticated", []))
    known_params.update(["currency", "value"])  # Common global params

    for key in params:
        if key not in known_params:
            issues.append({
                "type": "schema",
                "severity": "info",
                "message": f"Event '{event_name}' has undocumented parameter: '{key}'",
                "param": key,
            })

    # Check value/currency pairing
    global_params = schema.get("global_params", {})
    if "value" in params and "currency" not in params:
        if "currency" in global_params.get("required_with_value", []):
            issues.append({
                "type": "schema",
                "severity": "error",
                "message": f"Event '{event_name}' has 'value' without 'currency'",
            })

    return issues


def validate_events(events_data, schema=None):
    """Validate a list of events against naming conventions and schema."""
    if schema is None:
        schema = DEFAULT_SCHEMA

    all_issues = []
    event_counts = {}

    for i, event in enumerate(events_data):
        event_name = event.get("event", event.get("name", ""))
        params = event.get("params", event.get("parameters", {}))

        if not event_name:
            all_issues.append({
                "type": "structure",
                "severity": "error",
                "message": f"Event at index {i} has no event name",
                "event_index": i,
            })
            continue

        # Track event counts for duplicate detection
        event_counts[event_name] = event_counts.get(event_name, 0) + 1

        # Naming convention checks
        naming_issues = check_naming_convention(event_name)
        for issue in naming_issues:
            issue["event"] = event_name
            issue["event_index"] = i
        all_issues.extend(naming_issues)

        # PII checks
        pii_issues = check_pii(params)
        for issue in pii_issues:
            issue["event"] = event_name
            issue["event_index"] = i
        all_issues.extend(pii_issues)

        # Schema compliance checks
        schema_issues = check_schema_compliance(event_name, params, schema)
        for issue in schema_issues:
            issue["event"] = event_name
            issue["event_index"] = i
        all_issues.extend(schema_issues)

    return all_issues, event_counts


def format_report(all_issues, event_counts, total_events):
    """Format human-readable report."""
    errors = [i for i in all_issues if i["severity"] == "error"]
    warnings = [i for i in all_issues if i["severity"] == "warning"]
    infos = [i for i in all_issues if i["severity"] == "info"]

    lines = []
    lines.append("=" * 60)
    lines.append("EVENT SCHEMA VALIDATION REPORT")
    lines.append("=" * 60)
    lines.append(f"Events checked:  {total_events}")
    lines.append(f"Unique events:   {len(event_counts)}")
    lines.append(f"Errors:          {len(errors)}")
    lines.append(f"Warnings:        {len(warnings)}")
    lines.append(f"Info:            {len(infos)}")
    lines.append("")

    if errors:
        lines.append("--- ERRORS ---")
        for issue in errors:
            event_label = issue.get("event", "unknown")
            lines.append(f"  [{event_label}] {issue['message']}")
            if "suggestion" in issue:
                lines.append(f"    Suggestion: {issue['suggestion']}")
        lines.append("")

    if warnings:
        lines.append("--- WARNINGS ---")
        for issue in warnings:
            event_label = issue.get("event", "unknown")
            lines.append(f"  [{event_label}] {issue['message']}")
        lines.append("")

    if infos:
        lines.append("--- INFO ---")
        for issue in infos:
            event_label = issue.get("event", "unknown")
            lines.append(f"  [{event_label}] {issue['message']}")
        lines.append("")

    pii_count = len([i for i in all_issues if i.get("type") == "pii"])
    if pii_count > 0:
        lines.append(f"PII ALERT: {pii_count} potential PII violations detected!")
        lines.append("")

    score = max(0, 100 - (len(errors) * 10) - (len(warnings) * 3) - (len(infos) * 1))
    lines.append(f"Schema Compliance Score: {score}/100")

    return "\n".join(lines)


def main():
    parser = argparse.ArgumentParser(
        description="Validate analytics event names and parameters against a taxonomy"
    )
    parser.add_argument(
        "input",
        nargs="?",
        help="JSON file containing events array",
    )
    parser.add_argument(
        "--schema",
        help="Custom schema JSON file (default: built-in SaaS schema)",
    )
    parser.add_argument(
        "--json",
        action="store_true",
        dest="json_output",
        help="Output results in JSON format",
    )
    parser.add_argument(
        "--generate-schema",
        action="store_true",
        help="Print the default schema as JSON for customization",
    )
    args = parser.parse_args()

    if args.generate_schema:
        print(json.dumps(DEFAULT_SCHEMA, indent=2))
        sys.exit(0)

    if not args.input:
        parser.print_help()
        sys.exit(1)

    # Load events
    try:
        with open(args.input, "r", encoding="utf-8") as f:
            data = json.load(f)
    except FileNotFoundError:
        print(f"Error: File not found: {args.input}", file=sys.stderr)
        sys.exit(1)
    except json.JSONDecodeError as e:
        print(f"Error: Invalid JSON: {e}", file=sys.stderr)
        sys.exit(1)

    events = data if isinstance(data, list) else data.get("events", [])

    # Load custom schema
    schema = DEFAULT_SCHEMA
    if args.schema:
        try:
            with open(args.schema, "r", encoding="utf-8") as f:
                schema = json.load(f)
        except Exception as e:
            print(f"Error loading schema: {e}", file=sys.stderr)
            sys.exit(1)

    all_issues, event_counts = validate_events(events, schema)

    if args.json_output:
        result = {
            "total_events": len(events),
            "unique_events": len(event_counts),
            "event_counts": event_counts,
            "total_issues": len(all_issues),
            "errors": len([i for i in all_issues if i["severity"] == "error"]),
            "warnings": len([i for i in all_issues if i["severity"] == "warning"]),
            "info": len([i for i in all_issues if i["severity"] == "info"]),
            "pii_violations": len([i for i in all_issues if i.get("type") == "pii"]),
            "score": max(
                0,
                100
                - len([i for i in all_issues if i["severity"] == "error"]) * 10
                - len([i for i in all_issues if i["severity"] == "warning"]) * 3
                - len([i for i in all_issues if i["severity"] == "info"]) * 1,
            ),
            "issues": all_issues,
        }
        print(json.dumps(result, indent=2))
    else:
        print(format_report(all_issues, event_counts, len(events)))

    sys.exit(1 if any(i["severity"] == "error" for i in all_issues) else 0)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Funnel Drop-Off Analyzer - Analyze conversion funnels and identify drop-off points.

Reads funnel stage data (stage name + count), calculates stage-to-stage conversion
rates, identifies the biggest drop-off points, and suggests investigation priorities.

Usage:
    python funnel_drop_off_analyzer.py funnel_data.json
    python funnel_drop_off_analyzer.py funnel_data.json --json
    python funnel_drop_off_analyzer.py --stages "Visitors:10000,Signups:1200,Activated:480,Paid:120"
"""

import argparse
import json
import sys
import math


def parse_stages_string(stages_str):
    """Parse stages from a comma-separated string like 'Name:Count,Name:Count'."""
    stages = []
    for pair in stages_str.split(","):
        pair = pair.strip()
        if ":" not in pair:
            raise ValueError(f"Invalid stage format: '{pair}'. Use 'Name:Count'")
        name, count = pair.rsplit(":", 1)
        stages.append({"name": name.strip(), "count": int(count.strip())})
    return stages


def analyze_funnel(stages):
    """Analyze funnel stages and return detailed metrics."""
    if len(stages) < 2:
        return {"error": "Need at least 2 stages for funnel analysis"}

    results = []
    total_entry = stages[0]["count"]

    for i, stage in enumerate(stages):
        entry = {
            "stage": stage["name"],
            "count": stage["count"],
            "cumulative_conversion": (stage["count"] / total_entry * 100) if total_entry > 0 else 0,
        }

        if i > 0:
            prev_count = stages[i - 1]["count"]
            drop_off = prev_count - stage["count"]
            conversion_rate = (stage["count"] / prev_count * 100) if prev_count > 0 else 0
            drop_off_rate = 100 - conversion_rate

            entry["from_stage"] = stages[i - 1]["name"]
            entry["drop_off_count"] = drop_off
            entry["stage_conversion_rate"] = round(conversion_rate, 2)
            entry["drop_off_rate"] = round(drop_off_rate, 2)
            entry["drop_off_pct_of_total"] = round(
                (drop_off / total_entry * 100) if total_entry > 0 else 0, 2
            )

        results.append(entry)

    # Find biggest drop-off points
    drop_offs = [r for r in results if "drop_off_rate" in r]
    if drop_offs:
        # By absolute count
        biggest_absolute = max(drop_offs, key=lambda x: x["drop_off_count"])
        # By rate
        biggest_rate = max(drop_offs, key=lambda x: x["drop_off_rate"])
    else:
        biggest_absolute = None
        biggest_rate = None

    # Calculate overall metrics
    overall_conversion = (stages[-1]["count"] / total_entry * 100) if total_entry > 0 else 0

    # Benchmark comparison
    benchmarks = {
        "overall": {"good": 5.0, "average": 2.5, "poor": 1.0},
        "stage": {"good": 70.0, "average": 50.0, "poor": 30.0},
    }

    # Generate recommendations
    recommendations = []
    for r in drop_offs:
        if r["drop_off_rate"] > 70:
            recommendations.append({
                "priority": "critical",
                "stage": f"{r['from_stage']} -> {r['stage']}",
                "drop_off_rate": r["drop_off_rate"],
                "recommendation": f"Critical drop-off ({r['drop_off_rate']:.1f}%). "
                f"Investigate {r['stage'].lower()} experience immediately. "
                f"Check for UX friction, unclear value proposition, or technical issues.",
            })
        elif r["drop_off_rate"] > 50:
            recommendations.append({
                "priority": "high",
                "stage": f"{r['from_stage']} -> {r['stage']}",
                "drop_off_rate": r["drop_off_rate"],
                "recommendation": f"High drop-off ({r['drop_off_rate']:.1f}%). "
                f"Review {r['stage'].lower()} step for friction. "
                f"Consider A/B testing simplified flow or adding social proof.",
            })
        elif r["drop_off_rate"] > 30:
            recommendations.append({
                "priority": "medium",
                "stage": f"{r['from_stage']} -> {r['stage']}",
                "drop_off_rate": r["drop_off_rate"],
                "recommendation": f"Moderate drop-off ({r['drop_off_rate']:.1f}%). "
                f"Optimize {r['stage'].lower()} with better copy, clearer CTAs, or incentives.",
            })

    # Revenue impact estimation
    revenue_impact = []
    for r in drop_offs:
        if r["drop_off_rate"] > 30:
            # If we improve this stage by 10%, how many more reach the end?
            improvement_pct = 10
            additional_passed = int(r["drop_off_count"] * improvement_pct / 100)
            # Estimate downstream conversion
            downstream_rate = (stages[-1]["count"] / r["count"]) if r["count"] > 0 else 0
            additional_final = int(additional_passed * downstream_rate)
            revenue_impact.append({
                "stage": f"{r['from_stage']} -> {r['stage']}",
                "improvement_scenario": f"10% improvement",
                "additional_users_passed": additional_passed,
                "estimated_additional_conversions": additional_final,
            })

    return {
        "stages": results,
        "summary": {
            "total_stages": len(stages),
            "entry_count": total_entry,
            "exit_count": stages[-1]["count"],
            "overall_conversion_rate": round(overall_conversion, 2),
            "biggest_drop_off_absolute": {
                "stage": f"{biggest_absolute['from_stage']} -> {biggest_absolute['stage']}",
                "count": biggest_absolute["drop_off_count"],
                "rate": biggest_absolute["drop_off_rate"],
            } if biggest_absolute else None,
            "biggest_drop_off_rate": {
                "stage": f"{biggest_rate['from_stage']} -> {biggest_rate['stage']}",
                "count": biggest_rate["drop_off_count"],
                "rate": biggest_rate["drop_off_rate"],
            } if biggest_rate else None,
        },
        "recommendations": sorted(recommendations, key=lambda x: x["drop_off_rate"], reverse=True),
        "revenue_impact": revenue_impact,
    }


def format_report(analysis):
    """Format human-readable funnel report."""
    lines = []
    lines.append("=" * 65)
    lines.append("FUNNEL DROP-OFF ANALYSIS")
    lines.append("=" * 65)

    summary = analysis["summary"]
    lines.append(f"Stages:              {summary['total_stages']}")
    lines.append(f"Entry count:         {summary['entry_count']:,}")
    lines.append(f"Final count:         {summary['exit_count']:,}")
    lines.append(f"Overall conversion:  {summary['overall_conversion_rate']:.2f}%")
    lines.append("")

    # Stage breakdown
    lines.append("--- STAGE BREAKDOWN ---")
    lines.append(f"{'Stage':<25} {'Count':>10} {'Conv %':>8} {'Drop-off':>10} {'Drop %':>8}")
    lines.append("-" * 65)

    for stage in analysis["stages"]:
        conv = f"{stage.get('stage_conversion_rate', 100):.1f}%" if "stage_conversion_rate" in stage else "entry"
        drop = f"{stage.get('drop_off_count', 0):,}" if "drop_off_count" in stage else "-"
        drop_pct = f"{stage.get('drop_off_rate', 0):.1f}%" if "drop_off_rate" in stage else "-"
        lines.append(f"{stage['stage']:<25} {stage['count']:>10,} {conv:>8} {drop:>10} {drop_pct:>8}")

    lines.append("")

    # Biggest drop-offs
    if summary.get("biggest_drop_off_absolute"):
        lines.append("--- BIGGEST DROP-OFFS ---")
        ba = summary["biggest_drop_off_absolute"]
        lines.append(f"  By count: {ba['stage']} ({ba['count']:,} users lost, {ba['rate']:.1f}%)")
        br = summary["biggest_drop_off_rate"]
        if br["stage"] != ba["stage"]:
            lines.append(f"  By rate:  {br['stage']} ({br['count']:,} users lost, {br['rate']:.1f}%)")
        lines.append("")

    # Recommendations
    if analysis["recommendations"]:
        lines.append("--- RECOMMENDATIONS ---")
        for rec in analysis["recommendations"]:
            priority_marker = {"critical": "!!!", "high": "!!", "medium": "!"}.get(rec["priority"], "")
            lines.append(f"  [{rec['priority'].upper()}] {priority_marker} {rec['recommendation']}")
        lines.append("")

    # Revenue impact
    if analysis["revenue_impact"]:
        lines.append("--- REVENUE IMPACT SCENARIOS ---")
        for impact in analysis["revenue_impact"]:
            lines.append(
                f"  {impact['stage']}: {impact['improvement_scenario']} -> "
                f"+{impact['additional_users_passed']:,} pass through, "
                f"+{impact['estimated_additional_conversions']:,} final conversions"
            )
        lines.append("")

    return "\n".join(lines)


def main():
    parser = argparse.ArgumentParser(
        description="Analyze conversion funnels and identify drop-off points"
    )
    parser.add_argument(
        "input",
        nargs="?",
        help="JSON file with funnel stages [{name, count}, ...]",
    )
    parser.add_argument(
        "--stages",
        help='Inline stages: "Visitors:10000,Signups:1200,Paid:120"',
    )
    parser.add_argument(
        "--json",
        action="store_true",
        dest="json_output",
        help="Output results in JSON format",
    )
    args = parser.parse_args()

    if not args.input and not args.stages:
        parser.print_help()
        sys.exit(1)

    if args.stages:
        try:
            stages = parse_stages_string(args.stages)
        except ValueError as e:
            print(f"Error: {e}", file=sys.stderr)
            sys.exit(1)
    else:
        try:
            with open(args.input, "r", encoding="utf-8") as f:
                data = json.load(f)
            stages = data if isinstance(data, list) else data.get("stages", data.get("funnel", []))
        except FileNotFoundError:
            print(f"Error: File not found: {args.input}", file=sys.stderr)
            sys.exit(1)
        except json.JSONDecodeError as e:
            print(f"Error: Invalid JSON: {e}", file=sys.stderr)
            sys.exit(1)

    analysis = analyze_funnel(stages)

    if "error" in analysis:
        print(f"Error: {analysis['error']}", file=sys.stderr)
        sys.exit(1)

    if args.json_output:
        print(json.dumps(analysis, indent=2))
    else:
        print(format_report(analysis))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""UTM Parameter Validator - Validate UTM parameters for consistency and best practices.

Checks URLs or UTM parameter sets against naming conventions, detects common
mistakes (mixed case, spaces, missing required params), and reports violations.

Usage:
    python utm_validator.py urls.csv
    python utm_validator.py urls.csv --json
    python utm_validator.py --url "https://example.com?utm_source=Google&utm_medium=CPC"
"""

import argparse
import csv
import json
import re
import sys
from urllib.parse import urlparse, parse_qs


VALID_MEDIUMS = {
    "cpc", "ppc", "email", "social", "organic", "referral", "display",
    "affiliate", "video", "podcast", "sms", "push", "qr", "print",
    "partner", "retargeting", "native", "cpm", "cpa", "cpl",
}

VALID_SOURCES = {
    "google", "facebook", "meta", "linkedin", "twitter", "x", "instagram",
    "youtube", "tiktok", "bing", "reddit", "pinterest", "newsletter",
    "email", "partner", "direct", "referral", "quora", "snapchat",
}

REQUIRED_PARAMS = {"utm_source", "utm_medium", "utm_campaign"}
ALL_PARAMS = {"utm_source", "utm_medium", "utm_campaign", "utm_content", "utm_term"}


def extract_utms_from_url(url):
    """Extract UTM parameters from a URL string."""
    try:
        parsed = urlparse(url if "://" in url else f"https://{url}")
        params = parse_qs(parsed.query)
        utms = {}
        for key in ALL_PARAMS:
            if key in params:
                utms[key] = params[key][0]
        return utms
    except Exception:
        return {}


def validate_utms(utms, row_id=None):
    """Validate a set of UTM parameters and return issues found."""
    issues = []
    label = f"Row {row_id}" if row_id else "URL"

    # Check required params
    for param in REQUIRED_PARAMS:
        if param not in utms or not utms[param].strip():
            issues.append({
                "severity": "error",
                "param": param,
                "issue": f"Missing required parameter: {param}",
                "label": label,
            })

    for param, value in utms.items():
        # Check for uppercase characters
        if value != value.lower():
            issues.append({
                "severity": "error",
                "param": param,
                "value": value,
                "issue": f"Contains uppercase characters (should be lowercase): '{value}'",
                "suggestion": value.lower(),
                "label": label,
            })

        # Check for spaces
        if " " in value:
            issues.append({
                "severity": "error",
                "param": param,
                "value": value,
                "issue": f"Contains spaces: '{value}'",
                "suggestion": value.replace(" ", "-"),
                "label": label,
            })

        # Check for special characters (allow hyphens, underscores, dots)
        if re.search(r"[^a-zA-Z0-9\-_.]", value.replace(" ", "")):
            issues.append({
                "severity": "warning",
                "param": param,
                "value": value,
                "issue": f"Contains special characters: '{value}'",
                "label": label,
            })

        # Check for common encoding issues
        if "%" in value and not re.match(r".*%[0-9a-fA-F]{2}.*", value):
            issues.append({
                "severity": "warning",
                "param": param,
                "value": value,
                "issue": f"Contains percent sign that may not be URL-encoded: '{value}'",
                "label": label,
            })

    # Validate utm_medium against known values
    if "utm_medium" in utms:
        medium = utms["utm_medium"].lower().strip()
        if medium and medium not in VALID_MEDIUMS:
            issues.append({
                "severity": "warning",
                "param": "utm_medium",
                "value": utms["utm_medium"],
                "issue": f"Non-standard medium: '{medium}'. Consider using one of: {', '.join(sorted(VALID_MEDIUMS))}",
                "label": label,
            })

    # Validate utm_source against known values
    if "utm_source" in utms:
        source = utms["utm_source"].lower().strip()
        if source and source not in VALID_SOURCES:
            issues.append({
                "severity": "info",
                "param": "utm_source",
                "value": utms["utm_source"],
                "issue": f"Custom source: '{source}'. Ensure this is intentional and documented.",
                "label": label,
            })

    # Check utm_campaign naming pattern
    if "utm_campaign" in utms:
        campaign = utms["utm_campaign"]
        if campaign and not re.match(r"^[a-z0-9][a-z0-9\-_.]*$", campaign.lower()):
            issues.append({
                "severity": "warning",
                "param": "utm_campaign",
                "value": campaign,
                "issue": f"Campaign name may not follow naming convention: '{campaign}'",
                "label": label,
            })

    # Check for utm_term on non-search mediums
    if "utm_term" in utms and "utm_medium" in utms:
        medium = utms["utm_medium"].lower()
        if medium not in ("cpc", "ppc", "paid-search", "search"):
            issues.append({
                "severity": "info",
                "param": "utm_term",
                "value": utms["utm_term"],
                "issue": f"utm_term is typically used for paid search only, but medium is '{medium}'",
                "label": label,
            })

    return issues


def parse_csv(filepath):
    """Parse a CSV file containing URLs or UTM parameters."""
    entries = []
    with open(filepath, "r", newline="", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        headers = [h.lower().strip() for h in (reader.fieldnames or [])]

        for i, row in enumerate(reader, start=2):
            row_lower = {k.lower().strip(): v for k, v in row.items()}

            if "url" in row_lower:
                utms = extract_utms_from_url(row_lower["url"])
                entries.append({"row": i, "source": row_lower["url"], "utms": utms})
            else:
                utms = {}
                for param in ALL_PARAMS:
                    if param in row_lower and row_lower[param]:
                        utms[param] = row_lower[param]
                    # Also check without utm_ prefix
                    short = param.replace("utm_", "")
                    if short in row_lower and row_lower[short]:
                        utms[param] = row_lower[short]
                entries.append({"row": i, "source": "csv_row", "utms": utms})

    return entries


def format_report(all_issues, total_checked):
    """Format a human-readable report."""
    errors = [i for i in all_issues if i["severity"] == "error"]
    warnings = [i for i in all_issues if i["severity"] == "warning"]
    infos = [i for i in all_issues if i["severity"] == "info"]

    lines = []
    lines.append("=" * 60)
    lines.append("UTM VALIDATION REPORT")
    lines.append("=" * 60)
    lines.append(f"URLs/rows checked: {total_checked}")
    lines.append(f"Errors:   {len(errors)}")
    lines.append(f"Warnings: {len(warnings)}")
    lines.append(f"Info:     {len(infos)}")
    lines.append("")

    if errors:
        lines.append("--- ERRORS (must fix) ---")
        for issue in errors:
            lines.append(f"  [{issue['label']}] {issue['issue']}")
            if "suggestion" in issue:
                lines.append(f"    Suggestion: {issue['suggestion']}")
        lines.append("")

    if warnings:
        lines.append("--- WARNINGS (should fix) ---")
        for issue in warnings:
            lines.append(f"  [{issue['label']}] {issue['issue']}")
            if "suggestion" in issue:
                lines.append(f"    Suggestion: {issue['suggestion']}")
        lines.append("")

    if infos:
        lines.append("--- INFO (review) ---")
        for issue in infos:
            lines.append(f"  [{issue['label']}] {issue['issue']}")
        lines.append("")

    if not all_issues:
        lines.append("All UTM parameters pass validation checks.")

    # Summary
    score = max(0, 100 - (len(errors) * 15) - (len(warnings) * 5) - (len(infos) * 1))
    lines.append(f"UTM Quality Score: {score}/100")

    return "\n".join(lines)


def main():
    parser = argparse.ArgumentParser(
        description="Validate UTM parameters for consistency and best practices"
    )
    parser.add_argument(
        "input",
        nargs="?",
        help="CSV file with URLs or UTM parameters",
    )
    parser.add_argument(
        "--url",
        help="Single URL to validate",
    )
    parser.add_argument(
        "--json",
        action="store_true",
        dest="json_output",
        help="Output results in JSON format",
    )
    args = parser.parse_args()

    if not args.input and not args.url:
        parser.print_help()
        sys.exit(1)

    all_issues = []
    total_checked = 0

    if args.url:
        utms = extract_utms_from_url(args.url)
        issues = validate_utms(utms, row_id="single")
        all_issues.extend(issues)
        total_checked = 1
    elif args.input:
        try:
            entries = parse_csv(args.input)
        except FileNotFoundError:
            print(f"Error: File not found: {args.input}", file=sys.stderr)
            sys.exit(1)
        except Exception as e:
            print(f"Error reading file: {e}", file=sys.stderr)
            sys.exit(1)

        for entry in entries:
            issues = validate_utms(entry["utms"], row_id=entry["row"])
            all_issues.extend(issues)
            total_checked += 1

    if args.json_output:
        result = {
            "total_checked": total_checked,
            "total_issues": len(all_issues),
            "errors": len([i for i in all_issues if i["severity"] == "error"]),
            "warnings": len([i for i in all_issues if i["severity"] == "warning"]),
            "info": len([i for i in all_issues if i["severity"] == "info"]),
            "score": max(
                0,
                100
                - len([i for i in all_issues if i["severity"] == "error"]) * 15
                - len([i for i in all_issues if i["severity"] == "warning"]) * 5
                - len([i for i in all_issues if i["severity"] == "info"]) * 1,
            ),
            "issues": all_issues,
        }
        print(json.dumps(result, indent=2))
    else:
        print(format_report(all_issues, total_checked))

    sys.exit(1 if any(i["severity"] == "error" for i in all_issues) else 0)


if __name__ == "__main__":
    main()