Google Workspace Cli

{
  "_comment": "Google Workspace CLI automation config template. Copy and customize for your environment.",
  "auth": {
    "method": "oauth",
    "client_id": "",
    "client_secret": "",
    "token_path": "~/.config/gws/token.json",
    "service_account_key": "",
    "delegated_user": ""
  },
  "defaults": {
    "output_format": "json",
    "pagination_limit": 100,
    "timeout_ms": 30000,
    "log_level": "warn"
  },
  "persona": "developer",
  "scopes": [
    "gmail.modify",
    "gmail.send",
    "drive.file",
    "drive.metadata.readonly",
    "spreadsheets",
    "calendar",
    "calendar.events",
    "tasks"
  ],
  "scheduled_tasks": [
    {
      "name": "morning-briefing",
      "recipe": "morning-briefing",
      "schedule": "0 8 * * 1-5",
      "output": "~/workspace-reports/morning-{date}.json"
    },
    {
      "name": "eod-wrap",
      "recipe": "eod-wrap",
      "schedule": "0 17 * * 1-5",
      "output": "~/workspace-reports/eod-{date}.json"
    },
    {
      "name": "weekly-summary",
      "recipe": "weekly-summary",
      "schedule": "0 9 * * 5",
      "output": "~/workspace-reports/weekly-{date}.json"
    },
    {
      "name": "security-audit",
      "command": "python3 scripts/workspace_audit.py --json",
      "schedule": "0 10 * * 1",
      "output": "~/workspace-reports/audit-{date}.json"
    }
  ],
  "aliases": {
    "inbox": "gws gmail users.messages list me --query 'is:inbox' --limit 20 --json",
    "unread": "gws gmail users.messages list me --query 'is:unread' --limit 20 --json",
    "files": "gws drive files list --limit 20 --json",
    "events": "gws calendar events list primary --timeMin $(date -u +%Y-%m-%dT%H:%M:%SZ) --maxResults 10 --json",
    "tasks": "gws tasks tasks list @default --json"
  }
}

#!/usr/bin/env python3
"""
Google Workspace CLI Auth Setup Guide — Guided authentication configuration.

Prints step-by-step instructions for OAuth and service account setup,
generates .env templates, lists required scopes, and validates auth.

Usage:
    python3 auth_setup_guide.py --guide oauth
    python3 auth_setup_guide.py --guide service-account
    python3 auth_setup_guide.py --scopes gmail,drive,calendar
    python3 auth_setup_guide.py --generate-env
    python3 auth_setup_guide.py --validate [--json]
    python3 auth_setup_guide.py --check [--json]
"""

import argparse
import json
import shutil
import subprocess
import sys
from dataclasses import dataclass, field, asdict
from typing import List, Dict


SERVICE_SCOPES: Dict[str, List[str]] = {
    "gmail": [
        "https://www.googleapis.com/auth/gmail.modify",
        "https://www.googleapis.com/auth/gmail.send",
        "https://www.googleapis.com/auth/gmail.labels",
        "https://www.googleapis.com/auth/gmail.settings.basic",
    ],
    "drive": [
        "https://www.googleapis.com/auth/drive",
        "https://www.googleapis.com/auth/drive.file",
        "https://www.googleapis.com/auth/drive.metadata.readonly",
    ],
    "sheets": [
        "https://www.googleapis.com/auth/spreadsheets",
    ],
    "calendar": [
        "https://www.googleapis.com/auth/calendar",
        "https://www.googleapis.com/auth/calendar.events",
    ],
    "tasks": [
        "https://www.googleapis.com/auth/tasks",
    ],
    "chat": [
        "https://www.googleapis.com/auth/chat.spaces.readonly",
        "https://www.googleapis.com/auth/chat.messages",
    ],
    "docs": [
        "https://www.googleapis.com/auth/documents",
    ],
    "admin": [
        "https://www.googleapis.com/auth/admin.directory.user.readonly",
        "https://www.googleapis.com/auth/admin.directory.group",
        "https://www.googleapis.com/auth/admin.directory.orgunit.readonly",
    ],
    "meet": [
        "https://www.googleapis.com/auth/meetings.space.created",
    ],
}

OAUTH_GUIDE = """
=== Google Workspace CLI: OAuth Setup Guide ===

Step 1: Create a Google Cloud Project
  1. Go to https://console.cloud.google.com/
  2. Click "Select a project" -> "New Project"
  3. Name it (e.g., "gws-cli-access") and click Create
  4. Note the Project ID

Step 2: Enable Required APIs
  1. Go to APIs & Services -> Library
  2. Search and enable each API you need:
     - Gmail API
     - Google Drive API
     - Google Sheets API
     - Google Calendar API
     - Tasks API
     - Admin SDK API (for admin operations)

Step 3: Configure OAuth Consent Screen
  1. Go to APIs & Services -> OAuth consent screen
  2. Select "Internal" (for Workspace) or "External" (for personal)
  3. Fill in app name, support email
  4. Add scopes for the services you need
  5. Save and continue

Step 4: Create OAuth Credentials
  1. Go to APIs & Services -> Credentials
  2. Click "Create Credentials" -> "OAuth client ID"
  3. Application type: "Desktop app"
  4. Name it "gws-cli"
  5. Download the JSON file

Step 5: Configure gws CLI
  1. Set environment variables:
     export GOOGLE_WORKSPACE_CLI_CLIENT_ID=<your-client-id>
     export GOOGLE_WORKSPACE_CLI_CLIENT_SECRET=<your-client-secret>

  2. Or point the CLI at a credentials JSON file:
     export GOOGLE_WORKSPACE_CLI_CREDENTIALS_FILE=/path/to/credentials.json

Step 6: Authenticate
  gws auth setup    # interactive setup (uses gcloud if available)
  gws auth login -s gmail,drive,calendar   # request only needed scopes

Step 7: Verify (check exact syntax with: gws gmail --help)
  gws gmail users getProfile --params '{"userId": "me"}'
"""

SERVICE_ACCOUNT_GUIDE = """
=== Google Workspace CLI: Service Account Setup Guide ===

Step 1: Create a Google Cloud Project
  (Same as OAuth Step 1)

Step 2: Create a Service Account
  1. Go to IAM & Admin -> Service Accounts
  2. Click "Create Service Account"
  3. Name: "gws-cli-service"
  4. Grant roles as needed (no role needed for Workspace API access)
  5. Click "Done"

Step 3: Create Key
  1. Click on the service account
  2. Go to "Keys" tab
  3. Add Key -> Create new key -> JSON
  4. Download and store securely

Step 4: Enable Domain-Wide Delegation
  1. On the service account page, click "Edit"
  2. Check "Enable Google Workspace domain-wide delegation"
  3. Save
  4. Note the Client ID (numeric)

Step 5: Authorize in Google Admin
  1. Go to admin.google.com
  2. Security -> API Controls -> Domain-wide Delegation
  3. Add new:
     - Client ID: <numeric client ID from Step 4>
     - Scopes: (paste required scopes)
  4. Authorize

Step 6: Configure gws CLI for headless use
  NOTE: The documented headless flow for gws is to export credentials from an
  interactive machine and reuse them (see: gws auth --help):
    gws auth export --unmasked > credentials.json
    export GOOGLE_WORKSPACE_CLI_CREDENTIALS_FILE=/path/to/credentials.json
  Or supply a pre-obtained OAuth token:
    export GOOGLE_WORKSPACE_CLI_TOKEN=<token>
  Verify service-account/domain-wide-delegation support against your installed
  version (gws auth --help) before relying on it.

Step 7: Verify (check exact syntax with: gws gmail --help)
  gws gmail users getProfile --params '{"userId": "me"}'
"""

ENV_TEMPLATE = """# Google Workspace CLI Configuration
# Copy to .env and fill in values

# OAuth Credentials (for interactive auth)
GOOGLE_WORKSPACE_CLI_CLIENT_ID=
GOOGLE_WORKSPACE_CLI_CLIENT_SECRET=

# Headless/CI auth (export from an interactive machine: gws auth export --unmasked)
# GOOGLE_WORKSPACE_CLI_CREDENTIALS_FILE=/path/to/credentials.json
# GOOGLE_WORKSPACE_CLI_TOKEN=

# Optional
# GOOGLE_WORKSPACE_CLI_CONFIG_DIR=~/.config/gws
# GOOGLE_WORKSPACE_CLI_LOG=debug
"""


@dataclass
class ValidationResult:
    service: str
    status: str  # PASS, FAIL
    message: str


@dataclass
class ValidationReport:
    auth_method: str = ""
    user: str = ""
    results: List[dict] = field(default_factory=list)
    summary: str = ""
    demo_mode: bool = False


DEMO_VALIDATION = ValidationReport(
    auth_method="oauth",
    user="admin@company.com",
    results=[
        {"service": "gmail", "status": "PASS", "message": "Gmail API accessible"},
        {"service": "drive", "status": "PASS", "message": "Drive API accessible"},
        {"service": "calendar", "status": "PASS", "message": "Calendar API accessible"},
        {"service": "sheets", "status": "PASS", "message": "Sheets API accessible"},
        {"service": "tasks", "status": "FAIL", "message": "Scope not authorized"},
    ],
    summary="4/5 services validated (demo mode)",
    demo_mode=True,
)


def check_auth_status() -> dict:
    """Check current gws auth status."""
    try:
        result = subprocess.run(
            ["gws", "auth", "status", "--json"],
            capture_output=True, text=True, timeout=15
        )
        if result.returncode == 0:
            try:
                return json.loads(result.stdout)
            except json.JSONDecodeError:
                return {"status": "authenticated", "raw": result.stdout.strip()}
        return {"status": "unknown",
                "note": "could not determine auth status ('gws auth status' may not exist "
                        "in your version; check 'gws auth --help')",
                "error": result.stderr.strip()[:200]}
    except (FileNotFoundError, OSError):
        return {"status": "gws_not_found"}


def validate_services(services: List[str]) -> ValidationReport:
    """Validate auth by testing each service."""
    report = ValidationReport()

    auth = check_auth_status()
    if auth.get("status") == "gws_not_found":
        report.summary = "gws CLI not installed"
        return report
    if auth.get("status") == "not_authenticated":
        report.auth_method = "none"
        report.summary = "Not authenticated"
        return report

    report.auth_method = auth.get("method", "oauth")
    report.user = auth.get("user", auth.get("email", "unknown"))

    service_cmds = {
        "gmail": ["gws", "gmail", "users", "getProfile", "--params", '{"userId": "me"}'],
        "drive": ["gws", "drive", "files", "list", "--params", '{"pageSize": 1}'],
        "calendar": ["gws", "calendar", "calendarList", "list", "--params", '{"maxResults": 1}'],
        "sheets": ["gws", "schema", "sheets.spreadsheets.get"],
        "tasks": ["gws", "tasks", "tasklists", "list", "--params", '{"maxResults": 1}'],
    }

    for svc in services:
        cmd = service_cmds.get(svc)
        if not cmd:
            report.results.append(asdict(
                ValidationResult(svc, "WARN", f"No test available for {svc}")
            ))
            continue
        try:
            result = subprocess.run(cmd, capture_output=True, text=True, timeout=15)
            if result.returncode == 0:
                report.results.append(asdict(
                    ValidationResult(svc, "PASS", f"{svc.title()} API accessible")
                ))
            else:
                report.results.append(asdict(
                    ValidationResult(svc, "FAIL", result.stderr.strip()[:100])
                ))
        except (subprocess.TimeoutExpired, OSError) as e:
            report.results.append(asdict(
                ValidationResult(svc, "FAIL", str(e)[:100])
            ))

    passed = sum(1 for r in report.results if r["status"] == "PASS")
    total = len(report.results)
    report.summary = f"{passed}/{total} services validated"
    return report


def main():
    parser = argparse.ArgumentParser(
        description="Guided authentication setup for Google Workspace CLI (gws)",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  %(prog)s --guide oauth               # OAuth setup instructions
  %(prog)s --guide service-account      # Service account setup
  %(prog)s --scopes gmail,drive         # Show required scopes
  %(prog)s --generate-env               # Generate .env template
  %(prog)s --check                      # Check current auth status
  %(prog)s --validate --json            # Validate all services (JSON)
        """,
    )
    parser.add_argument("--guide", choices=["oauth", "service-account"],
                        help="Print setup guide")
    parser.add_argument("--scopes", help="Comma-separated services to show scopes for")
    parser.add_argument("--generate-env", action="store_true",
                        help="Generate .env template")
    parser.add_argument("--check", action="store_true",
                        help="Check current auth status")
    parser.add_argument("--validate", action="store_true",
                        help="Validate auth by testing services")
    parser.add_argument("--services", default="gmail,drive,calendar,sheets,tasks",
                        help="Services to validate (default: gmail,drive,calendar,sheets,tasks)")
    parser.add_argument("--json", action="store_true", help="Output JSON")
    args = parser.parse_args()

    if not any([args.guide, args.scopes, args.generate_env, args.check, args.validate]):
        parser.print_help()
        return

    if args.guide:
        if args.guide == "oauth":
            print(OAUTH_GUIDE)
        else:
            print(SERVICE_ACCOUNT_GUIDE)
        return

    if args.scopes:
        services = [s.strip() for s in args.scopes.split(",") if s.strip()]
        if args.json:
            output = {}
            for svc in services:
                output[svc] = SERVICE_SCOPES.get(svc, [])
            print(json.dumps(output, indent=2))
        else:
            print(f"\n{'='*60}")
            print(f"  REQUIRED OAUTH SCOPES")
            print(f"{'='*60}\n")
            for svc in services:
                scopes = SERVICE_SCOPES.get(svc, [])
                print(f"  {svc.upper()}:")
                if scopes:
                    for scope in scopes:
                        print(f"    - {scope}")
                else:
                    print(f"    (no scopes defined for '{svc}')")
                print()
            # Print combined for easy copy-paste
            all_scopes = []
            for svc in services:
                all_scopes.extend(SERVICE_SCOPES.get(svc, []))
            if all_scopes:
                print(f"  COMBINED (for consent screen):")
                print(f"  {','.join(all_scopes)}")
            print(f"\n{'='*60}\n")
        return

    if args.generate_env:
        print(ENV_TEMPLATE)
        return

    if args.check:
        if shutil.which("gws"):
            status = check_auth_status()
        else:
            status = {"status": "gws_not_found",
                      "note": "Install gws first: npm install -g @googleworkspace/cli  OR  https://github.com/googleworkspace/cli/releases"}
        if args.json:
            print(json.dumps(status, indent=2))
        else:
            print(f"\nAuth Status: {status.get('status', 'unknown')}")
            for k, v in status.items():
                if k != "status":
                    print(f"  {k}: {v}")
            print()
        return

    if args.validate:
        services = [s.strip() for s in args.services.split(",") if s.strip()]
        if not shutil.which("gws"):
            report = DEMO_VALIDATION
        else:
            report = validate_services(services)

        if args.json:
            print(json.dumps(asdict(report), indent=2))
        else:
            print(f"\n{'='*60}")
            print(f"  AUTH VALIDATION REPORT")
            if report.demo_mode:
                print(f"  (DEMO MODE)")
            print(f"{'='*60}\n")
            if report.user:
                print(f"  User: {report.user}")
                print(f"  Method: {report.auth_method}\n")
            for r in report.results:
                icon = "PASS" if r["status"] == "PASS" else "FAIL"
                print(f"  [{icon}] {r['service']}: {r['message']}")
            print(f"\n  {report.summary}")
            print(f"\n{'='*60}\n")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Google Workspace CLI Doctor — Pre-flight diagnostics for gws CLI.

Checks installation, version, authentication status, and service
connectivity. Runs in demo mode with embedded sample data when gws
is not installed.

Usage:
    python3 gws_doctor.py
    python3 gws_doctor.py --json
    python3 gws_doctor.py --services gmail,drive,calendar
"""

import argparse
import json
import shutil
import subprocess
import sys
from dataclasses import dataclass, field, asdict
from typing import List, Optional


@dataclass
class Check:
    name: str
    status: str  # PASS, WARN, FAIL
    message: str
    fix: str = ""


@dataclass
class DiagnosticReport:
    gws_installed: bool = False
    gws_version: str = ""
    auth_status: str = ""
    checks: List[dict] = field(default_factory=list)
    summary: str = ""
    demo_mode: bool = False


DEMO_CHECKS = [
    Check("gws-installed", "PASS", "gws v0.9.2 found at /usr/local/bin/gws"),
    Check("gws-version", "PASS", "Version 0.9.2 (latest)"),
    Check("auth-status", "PASS", "Authenticated as admin@company.com"),
    Check("token-expiry", "WARN", "Token expires in 23 minutes",
          "Run 'gws auth refresh' to extend token lifetime"),
    Check("gmail-access", "PASS", "Gmail API accessible — user profile retrieved"),
    Check("drive-access", "PASS", "Drive API accessible — root folder listed"),
    Check("calendar-access", "PASS", "Calendar API accessible — primary calendar found"),
    Check("sheets-access", "PASS", "Sheets API accessible"),
    Check("tasks-access", "FAIL", "Tasks API not authorized",
          "Run 'gws auth setup' and add 'tasks' scope"),
]

SERVICE_TEST_COMMANDS = {
    "gmail": ["gws", "gmail", "users", "getProfile", "--params", '{"userId": "me"}'],
    "drive": ["gws", "drive", "files", "list", "--params", '{"pageSize": 1}'],
    "calendar": ["gws", "calendar", "calendarList", "list", "--params", '{"maxResults": 1}'],
    "sheets": ["gws", "schema", "sheets.spreadsheets.get"],
    "tasks": ["gws", "tasks", "tasklists", "list", "--params", '{"maxResults": 1}'],
    "chat": ["gws", "chat", "spaces", "list", "--params", '{"pageSize": 1}'],
    "docs": ["gws", "schema", "docs.documents.get"],
}


def check_installation() -> Check:
    """Check if gws is installed and on PATH."""
    path = shutil.which("gws")
    if path:
        return Check("gws-installed", "PASS", f"gws found at {path}")
    return Check("gws-installed", "FAIL", "gws not found on PATH",
                 "Install via: npm install -g @googleworkspace/cli  OR  download from https://github.com/googleworkspace/cli/releases")


def check_version() -> Check:
    """Get gws version."""
    try:
        result = subprocess.run(
            ["gws", "--version"], capture_output=True, text=True, timeout=10
        )
        version = result.stdout.strip()
        if version:
            return Check("gws-version", "PASS", f"Version: {version}")
        return Check("gws-version", "WARN", "Could not parse version output")
    except (subprocess.TimeoutExpired, FileNotFoundError, OSError) as e:
        return Check("gws-version", "FAIL", f"Version check failed: {e}")


def check_auth() -> Check:
    """Check authentication status."""
    try:
        result = subprocess.run(
            ["gws", "auth", "status", "--json"],
            capture_output=True, text=True, timeout=15
        )
        if result.returncode == 0:
            try:
                data = json.loads(result.stdout)
                user = data.get("user", data.get("email", "unknown"))
                return Check("auth-status", "PASS", f"Authenticated as {user}")
            except json.JSONDecodeError:
                return Check("auth-status", "PASS", "Authenticated (could not parse details)")
        return Check("auth-status", "WARN",
                     "Could not confirm authentication ('gws auth status' may not exist "
                     "in your version; check 'gws auth --help')",
                     "Run 'gws auth setup' then 'gws auth login' to configure authentication")
    except (subprocess.TimeoutExpired, FileNotFoundError, OSError) as e:
        return Check("auth-status", "FAIL", f"Auth check failed: {e}",
                     "Run 'gws auth setup' then 'gws auth login' to configure authentication")


def check_service(service: str) -> Check:
    """Test connectivity to a specific service."""
    cmd = SERVICE_TEST_COMMANDS.get(service)
    if not cmd:
        return Check(f"{service}-access", "WARN", f"No test command for {service}")
    try:
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=15)
        if result.returncode == 0:
            return Check(f"{service}-access", "PASS", f"{service.title()} API accessible")
        stderr = result.stderr.strip()[:100]
        if "403" in stderr or "permission" in stderr.lower():
            return Check(f"{service}-access", "FAIL",
                         f"{service.title()} API permission denied",
                         f"Add '{service}' scope: gws auth setup --scopes {service}")
        return Check(f"{service}-access", "FAIL",
                     f"{service.title()} API error: {stderr}",
                     f"Check scope and permissions for {service}")
    except (subprocess.TimeoutExpired, FileNotFoundError, OSError) as e:
        return Check(f"{service}-access", "FAIL", f"{service.title()} test failed: {e}")


def run_diagnostics(services: List[str]) -> DiagnosticReport:
    """Run all diagnostic checks."""
    report = DiagnosticReport()
    checks = []

    # Installation check
    install_check = check_installation()
    checks.append(install_check)
    report.gws_installed = install_check.status == "PASS"

    if not report.gws_installed:
        report.checks = [asdict(c) for c in checks]
        report.summary = "FAIL: gws is not installed"
        return report

    # Version check
    version_check = check_version()
    checks.append(version_check)
    if version_check.status == "PASS":
        report.gws_version = version_check.message.replace("Version: ", "")

    # Auth check
    auth_check = check_auth()
    checks.append(auth_check)
    report.auth_status = auth_check.status

    if auth_check.status != "PASS":
        report.checks = [asdict(c) for c in checks]
        report.summary = "FAIL: Authentication not configured"
        return report

    # Service checks
    for svc in services:
        checks.append(check_service(svc))

    report.checks = [asdict(c) for c in checks]

    # Summary
    fails = sum(1 for c in checks if c.status == "FAIL")
    warns = sum(1 for c in checks if c.status == "WARN")
    passes = sum(1 for c in checks if c.status == "PASS")
    if fails > 0:
        report.summary = f"ISSUES FOUND: {passes} passed, {warns} warnings, {fails} failures"
    elif warns > 0:
        report.summary = f"MOSTLY OK: {passes} passed, {warns} warnings"
    else:
        report.summary = f"ALL CLEAR: {passes}/{passes} checks passed"

    return report


def run_demo() -> DiagnosticReport:
    """Return demo report with embedded sample data."""
    report = DiagnosticReport(
        gws_installed=True,
        gws_version="0.9.2",
        auth_status="PASS",
        checks=[asdict(c) for c in DEMO_CHECKS],
        summary="MOSTLY OK: 7 passed, 1 warning, 1 failure (demo mode)",
        demo_mode=True,
    )
    return report


def main():
    parser = argparse.ArgumentParser(
        description="Pre-flight diagnostics for Google Workspace CLI (gws)",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  %(prog)s                          # Run all checks
  %(prog)s --json                   # JSON output
  %(prog)s --services gmail,drive   # Check specific services only
  %(prog)s --demo                   # Demo mode (no gws required)
        """,
    )
    parser.add_argument("--json", action="store_true", help="Output JSON")
    parser.add_argument(
        "--services", default="gmail,drive,calendar,sheets,tasks",
        help="Comma-separated services to check (default: gmail,drive,calendar,sheets,tasks)"
    )
    parser.add_argument("--demo", action="store_true", help="Run with demo data")
    args = parser.parse_args()

    services = [s.strip() for s in args.services.split(",") if s.strip()]

    # Use demo mode if requested or gws not installed
    if args.demo or not shutil.which("gws"):
        report = run_demo()
    else:
        report = run_diagnostics(services)

    if args.json:
        print(json.dumps(asdict(report), indent=2))
    else:
        print(f"\n{'='*60}")
        print(f"  GWS CLI DIAGNOSTIC REPORT")
        if report.demo_mode:
            print(f"  (DEMO MODE — sample data)")
        print(f"{'='*60}\n")

        for c in report.checks:
            icon = {"PASS": "PASS", "WARN": "WARN", "FAIL": "FAIL"}.get(c["status"], "????")
            print(f"  [{icon}] {c['name']}: {c['message']}")
            if c.get("fix") and c["status"] != "PASS":
                print(f"         -> {c['fix']}")

        print(f"\n  {'-'*56}")
        print(f"  {report.summary}")
        print(f"\n{'='*60}\n")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Google Workspace CLI Recipe Runner — Catalog, search, and execute gws command templates.

Browse 43 recipe command templates (a LOCAL catalog shipped with this skill —
NOT built into the gws CLI), filter by persona, search by keyword, and run
with dry-run support.

IMPORTANT: The gws CLI (github.com/googleworkspace/cli) generates its command
surface dynamically from Google's Discovery Service. Command strings in this
catalog are templates — verify each against `gws --help`, `gws <service> --help`,
or `gws schema <service>.<resource>.<method>` before relying on it.

Usage:
    python3 gws_recipe_runner.py --list
    python3 gws_recipe_runner.py --search "email"
    python3 gws_recipe_runner.py --describe standup-report
    python3 gws_recipe_runner.py --run standup-report --dry-run
    python3 gws_recipe_runner.py --persona pm --list
    python3 gws_recipe_runner.py --list --json
"""

import argparse
import json
import subprocess
import sys
from dataclasses import dataclass, field, asdict
from typing import List, Dict, Optional


TEMPLATE_NOTE = ("NOTE: Recipe commands are templates from this local catalog (not shipped by "
                 "the gws CLI). Verify each against 'gws --help' / 'gws schema' before use.")


@dataclass
class Recipe:
    name: str
    description: str
    category: str
    services: List[str]
    commands: List[str]
    prerequisites: str = ""


RECIPES: Dict[str, Recipe] = {
    # Email (8)
    "send-email": Recipe("send-email", "Send an email with optional attachments", "email",
                         ["gmail"], ["gws gmail users.messages send me --to {to} --subject {subject} --body {body}"]),
    "reply-to-thread": Recipe("reply-to-thread", "Reply to an existing email thread", "email",
                              ["gmail"], ["gws gmail users.messages reply me --thread-id {thread_id} --body {body}"]),
    "forward-email": Recipe("forward-email", "Forward an email to another recipient", "email",
                            ["gmail"], ["gws gmail users.messages forward me --message-id {msg_id} --to {to}"]),
    "search-emails": Recipe("search-emails", "Search emails with Gmail query syntax", "email",
                            ["gmail"], ["gws gmail users.messages list me --query {query} --json"]),
    "archive-old": Recipe("archive-old", "Archive read emails older than N days", "email",
                          ["gmail"], [
                              "gws gmail users.messages list me --query 'is:read older_than:{days}d' --json",
                              "# Pipe IDs to batch modify to remove INBOX label",
                          ]),
    "label-manager": Recipe("label-manager", "Create, list, and organize Gmail labels", "email",
                            ["gmail"], ["gws gmail users.labels list me --json", "gws gmail users.labels create me --name {name}"]),
    "filter-setup": Recipe("filter-setup", "Create email filters for auto-labeling", "email",
                           ["gmail"], ["gws gmail users.settings.filters create me --criteria {criteria} --action {action}"]),
    "unread-digest": Recipe("unread-digest", "Get digest of unread emails", "email",
                            ["gmail"], ["gws gmail users.messages list me --query 'is:unread' --limit 20 --json"]),

    # Files (7)
    "upload-file": Recipe("upload-file", "Upload a file to Google Drive", "files",
                          ["drive"], ["gws drive files create --name {name} --upload {path} --parents {folder_id}"]),
    "create-sheet": Recipe("create-sheet", "Create a new Google Spreadsheet", "files",
                           ["sheets"], ["gws sheets spreadsheets create --title {title} --json"]),
    "share-file": Recipe("share-file", "Share a Drive file with a user or domain", "files",
                          ["drive"], ["gws drive permissions create {file_id} --type user --role writer --emailAddress {email}"]),
    "export-file": Recipe("export-file", "Export a Google Doc/Sheet as PDF", "files",
                          ["drive"], ["gws drive files export {file_id} --mime application/pdf --output {output}"]),
    "list-files": Recipe("list-files", "List files in a Drive folder", "files",
                         ["drive"], ["gws drive files list --parents {folder_id} --json"]),
    "find-large-files": Recipe("find-large-files", "Find largest files in Drive", "files",
                               ["drive"], ["gws drive files list --orderBy 'quotaBytesUsed desc' --limit 20 --json"]),
    "cleanup-trash": Recipe("cleanup-trash", "Empty Drive trash", "files",
                            ["drive"], ["gws drive files emptyTrash"]),

    # Calendar (6)
    "create-event": Recipe("create-event", "Create a calendar event with attendees", "calendar",
                           ["calendar"], [
                               "gws calendar events insert primary --summary {title} "
                               "--start {start} --end {end} --attendees {attendees}"
                           ]),
    "quick-event": Recipe("quick-event", "Create an event via the calendar helper", "calendar",
                          ["calendar"], ["gws calendar +insert {details}  # see: gws calendar +insert --help"]),
    "find-time": Recipe("find-time", "Find available time slots via free/busy", "calendar",
                        ["calendar"], ["gws calendar freebusy query --json '<freebusy-request>'  # verify: gws schema calendar.freebusy.query"]),
    "today-schedule": Recipe("today-schedule", "Show today's calendar events", "calendar",
                             ["calendar"], ["gws calendar events list primary --timeMin {today_start} --timeMax {today_end} --json"]),
    "meeting-prep": Recipe("meeting-prep", "Prepare for an upcoming meeting (agenda + attendees)", "calendar",
                           ["calendar"], ["gws workflow +meeting-prep"]),
    "reschedule": Recipe("reschedule", "Move an event to a new time", "calendar",
                         ["calendar"], ["gws calendar events patch primary {event_id} --start {new_start} --end {new_end}"]),

    # Reporting (5)
    "standup-report": Recipe("standup-report", "Generate daily standup from calendar and tasks", "reporting",
                             ["calendar", "tasks"], ["gws workflow +standup-report"]),
    "weekly-summary": Recipe("weekly-summary", "Summarize week's emails, events, and tasks", "reporting",
                             ["gmail", "calendar", "tasks"], ["gws workflow +weekly-digest"]),
    "drive-activity": Recipe("drive-activity", "Report on Drive file activity", "reporting",
                             ["drive"], ["gws drive activities list --json"]),
    "email-stats": Recipe("email-stats", "Email volume statistics", "reporting",
                          ["gmail"], [
                              "gws gmail users.messages list me --query 'newer_than:7d' --json",
                              "# Pipe through output_analyzer.py --count",
                          ]),
    "task-progress": Recipe("task-progress", "Report on task completion", "reporting",
                            ["tasks"], ["gws tasks tasks list {tasklist_id} --json"]),

    # Collaboration (5)
    "share-folder": Recipe("share-folder", "Share a Drive folder with a team", "collaboration",
                           ["drive"], ["gws drive permissions create {folder_id} --type group --role writer --emailAddress {group}"]),
    "create-doc": Recipe("create-doc", "Create a Google Doc with initial content", "collaboration",
                         ["docs"], ["gws docs documents create --title {title} --json"]),
    "chat-message": Recipe("chat-message", "Send a message to a Google Chat space", "collaboration",
                           ["chat"], ["gws chat spaces.messages create {space} --text {message}"]),
    "list-spaces": Recipe("list-spaces", "List Google Chat spaces", "collaboration",
                          ["chat"], ["gws chat spaces list --json"]),
    "task-create": Recipe("task-create", "Create a task in Google Tasks", "collaboration",
                          ["tasks"], ["gws tasks tasks insert {tasklist_id} --title {title} --due {due_date}"]),

    # Data (4)
    "sheet-read": Recipe("sheet-read", "Read data from a spreadsheet range", "data",
                         ["sheets"], ["gws sheets spreadsheets.values get {sheet_id} --range {range} --json"]),
    "sheet-write": Recipe("sheet-write", "Write data to a spreadsheet", "data",
                          ["sheets"], ["gws sheets spreadsheets.values update {sheet_id} --range {range} --values {data}"]),
    "sheet-append": Recipe("sheet-append", "Append rows to a spreadsheet", "data",
                           ["sheets"], ["gws sheets spreadsheets.values append {sheet_id} --range {range} --values {data}"]),
    "export-contacts": Recipe("export-contacts", "Export contacts list", "data",
                              ["people"], ["gws people people.connections list me --personFields names,emailAddresses --json"]),

    # Admin (4)
    "list-users": Recipe("list-users", "List all users in the Workspace domain", "admin",
                         ["admin"], ["gws admin users list --domain {domain} --json"],
                         "Requires Admin SDK API and admin.directory.user.readonly scope"),
    "list-groups": Recipe("list-groups", "List all groups in the domain", "admin",
                          ["admin"], ["gws admin groups list --domain {domain} --json"]),
    "user-info": Recipe("user-info", "Get detailed user information", "admin",
                        ["admin"], ["gws admin users get {email} --json"]),
    "audit-logins": Recipe("audit-logins", "Audit recent login activity", "admin",
                           ["admin"], ["gws admin activities list login --json"]),

    # Cross-Service (4)
    "morning-briefing": Recipe("morning-briefing", "Today's events + unread emails + pending tasks", "cross-service",
                               ["gmail", "calendar", "tasks"], [
                                   "gws calendar events list primary --timeMin {today} --maxResults 10 --json",
                                   "gws gmail users.messages list me --query 'is:unread' --limit 10 --json",
                                   "gws tasks tasks list {default_tasklist} --json",
                               ]),
    "eod-wrap": Recipe("eod-wrap", "End-of-day wrap up: summarize completed, pending, tomorrow", "cross-service",
                       ["calendar", "tasks"], [
                           "gws calendar events list primary --timeMin {today_start} --timeMax {today_end} --json",
                           "gws tasks tasks list {default_tasklist} --json",
                       ]),
    "project-status": Recipe("project-status", "Aggregate project status from Drive, Sheets, Tasks", "cross-service",
                             ["drive", "sheets", "tasks"], [
                                 "gws drive files list --query 'name contains {project}' --json",
                                 "gws tasks tasks list {tasklist_id} --json",
                             ]),
    "inbox-zero": Recipe("inbox-zero", "Process inbox to zero: label, archive, reply, task", "cross-service",
                         ["gmail", "tasks"], [
                             "gws gmail users.messages list me --query 'is:inbox' --json",
                             "# Process each: label, archive, or create task",
                         ]),
}

PERSONAS: Dict[str, Dict] = {
    "executive-assistant": {
        "description": "Executive assistant managing schedules, emails, and communications",
        "recipes": ["morning-briefing", "today-schedule", "find-time", "send-email", "reply-to-thread",
                     "standup-report", "meeting-prep", "eod-wrap", "quick-event", "inbox-zero"],
    },
    "pm": {
        "description": "Project manager tracking tasks, meetings, and deliverables",
        "recipes": ["standup-report", "create-event", "find-time", "task-create", "task-progress",
                     "project-status", "weekly-summary", "share-folder", "sheet-read", "morning-briefing"],
    },
    "hr": {
        "description": "HR managing people, onboarding, and communications",
        "recipes": ["list-users", "user-info", "send-email", "create-event", "create-doc",
                     "share-folder", "chat-message", "list-groups", "export-contacts", "today-schedule"],
    },
    "sales": {
        "description": "Sales rep managing client communications and proposals",
        "recipes": ["send-email", "search-emails", "create-event", "find-time", "create-doc",
                     "share-file", "sheet-read", "sheet-write", "export-file", "morning-briefing"],
    },
    "it-admin": {
        "description": "IT administrator managing Workspace configuration and security",
        "recipes": ["list-users", "list-groups", "user-info", "audit-logins", "drive-activity",
                     "find-large-files", "cleanup-trash", "label-manager", "filter-setup", "share-folder"],
    },
    "developer": {
        "description": "Developer using Workspace APIs for automation",
        "recipes": ["sheet-read", "sheet-write", "sheet-append", "upload-file", "create-doc",
                     "chat-message", "task-create", "list-files", "export-file", "send-email"],
    },
    "marketing": {
        "description": "Marketing team member managing campaigns and content",
        "recipes": ["send-email", "create-doc", "share-file", "upload-file", "create-sheet",
                     "sheet-write", "chat-message", "create-event", "email-stats", "weekly-summary"],
    },
    "finance": {
        "description": "Finance team managing spreadsheets and reports",
        "recipes": ["sheet-read", "sheet-write", "sheet-append", "create-sheet", "export-file",
                     "share-file", "send-email", "find-large-files", "drive-activity", "weekly-summary"],
    },
    "legal": {
        "description": "Legal team managing documents and compliance",
        "recipes": ["create-doc", "share-file", "export-file", "search-emails", "send-email",
                     "upload-file", "list-files", "drive-activity", "audit-logins", "find-large-files"],
    },
    "support": {
        "description": "Customer support managing tickets and communications",
        "recipes": ["search-emails", "send-email", "reply-to-thread", "label-manager", "filter-setup",
                     "task-create", "chat-message", "unread-digest", "inbox-zero", "morning-briefing"],
    },
}


def list_recipes(persona: Optional[str], output_json: bool):
    """List all recipes, optionally filtered by persona."""
    if persona:
        if persona not in PERSONAS:
            print(f"Unknown persona: {persona}. Available: {', '.join(PERSONAS.keys())}")
            sys.exit(1)
        recipe_names = PERSONAS[persona]["recipes"]
        recipes = {k: v for k, v in RECIPES.items() if k in recipe_names}
        title = f"Recipes for {persona.upper()}: {PERSONAS[persona]['description']}"
    else:
        recipes = RECIPES
        title = "All 43 Google Workspace CLI Recipes"

    if output_json:
        output = []
        for name, r in recipes.items():
            output.append(asdict(r))
        print(json.dumps(output, indent=2))
        return

    print(f"\n{'='*60}")
    print(f"  {title}")
    print(f"{'='*60}\n")

    by_category: Dict[str, list] = {}
    for name, r in recipes.items():
        by_category.setdefault(r.category, []).append(r)

    for cat, cat_recipes in sorted(by_category.items()):
        print(f"  {cat.upper()} ({len(cat_recipes)})")
        for r in cat_recipes:
            svcs = ",".join(r.services)
            print(f"    {r.name:<24} {r.description:<40} [{svcs}]")
        print()

    print(f"  Total: {len(recipes)} recipes")
    print(f"\n{'='*60}\n")


def search_recipes(keyword: str, output_json: bool):
    """Search recipes by keyword."""
    keyword_lower = keyword.lower()
    matches = {k: v for k, v in RECIPES.items()
               if keyword_lower in k.lower()
               or keyword_lower in v.description.lower()
               or keyword_lower in v.category.lower()
               or any(keyword_lower in s for s in v.services)}

    if output_json:
        print(json.dumps([asdict(r) for r in matches.values()], indent=2))
        return

    print(f"\n  Search results for '{keyword}': {len(matches)} matches\n")
    for name, r in matches.items():
        print(f"    {r.name:<24} {r.description}")
    print()


def describe_recipe(name: str, output_json: bool):
    """Show full details for a recipe."""
    recipe = RECIPES.get(name)
    if not recipe:
        print(f"Unknown recipe: {name}")
        print(f"Use --list to see available recipes")
        sys.exit(1)

    if output_json:
        print(json.dumps(asdict(recipe), indent=2))
        return

    print(f"\n{'='*60}")
    print(f"  Recipe: {recipe.name}")
    print(f"{'='*60}\n")
    print(f"  Description: {recipe.description}")
    print(f"  Category:    {recipe.category}")
    print(f"  Services:    {', '.join(recipe.services)}")
    if recipe.prerequisites:
        print(f"  Prerequisites: {recipe.prerequisites}")
    print(f"\n  Commands:")
    for i, cmd in enumerate(recipe.commands, 1):
        print(f"    {i}. {cmd}")
    print(f"\n  {TEMPLATE_NOTE}")
    print(f"\n{'='*60}\n")


def run_recipe(name: str, dry_run: bool):
    """Execute a recipe (or print commands in dry-run mode)."""
    recipe = RECIPES.get(name)
    if not recipe:
        print(f"Unknown recipe: {name}")
        sys.exit(1)

    if dry_run:
        print(f"\n  [DRY RUN] Recipe: {recipe.name}\n")
        for i, cmd in enumerate(recipe.commands, 1):
            print(f"  {i}. {cmd}")
        print(f"\n  {TEMPLATE_NOTE}")
        print(f"\n  (No commands executed)")
        return

    print(f"\n  Executing recipe: {recipe.name}")
    print(f"  {TEMPLATE_NOTE}\n")
    for cmd in recipe.commands:
        if cmd.startswith("#"):
            print(f"  {cmd}")
            continue
        print(f"  $ {cmd}")
        try:
            result = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=30)
            if result.stdout:
                print(result.stdout)
            if result.returncode != 0 and result.stderr:
                print(f"  Error: {result.stderr.strip()[:200]}")
        except subprocess.TimeoutExpired:
            print(f"  Timeout after 30s")
        except OSError as e:
            print(f"  Execution error: {e}")


def list_personas(output_json: bool):
    """List all available personas."""
    if output_json:
        print(json.dumps(PERSONAS, indent=2))
        return

    print(f"\n{'='*60}")
    print(f"  10 PERSONA BUNDLES")
    print(f"{'='*60}\n")
    for name, p in PERSONAS.items():
        print(f"  {name:<24} {p['description']}")
        print(f"  {'':24} Recipes: {', '.join(p['recipes'][:5])}...")
        print()
    print(f"{'='*60}\n")


def main():
    parser = argparse.ArgumentParser(
        description="Catalog, search, and execute Google Workspace CLI recipes",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  %(prog)s --list                           # List all 43 recipes
  %(prog)s --list --persona pm              # Recipes for project managers
  %(prog)s --search "email"                 # Search by keyword
  %(prog)s --describe standup-report        # Full recipe details
  %(prog)s --run standup-report --dry-run   # Preview recipe commands
  %(prog)s --personas                       # List all 10 personas
  %(prog)s --list --json                    # JSON output
        """,
    )
    parser.add_argument("--list", action="store_true", help="List all recipes")
    parser.add_argument("--search", help="Search recipes by keyword")
    parser.add_argument("--describe", help="Show full details for a recipe")
    parser.add_argument("--run", help="Execute a recipe")
    parser.add_argument("--dry-run", action="store_true", help="Print commands without executing")
    parser.add_argument("--persona", help="Filter recipes by persona")
    parser.add_argument("--personas", action="store_true", help="List all personas")
    parser.add_argument("--json", action="store_true", help="Output JSON")
    args = parser.parse_args()

    if not any([args.list, args.search, args.describe, args.run, args.personas]):
        parser.print_help()
        return

    if args.personas:
        list_personas(args.json)
        return

    if args.list:
        list_recipes(args.persona, args.json)
        return

    if args.search:
        search_recipes(args.search, args.json)
        return

    if args.describe:
        describe_recipe(args.describe, args.json)
        return

    if args.run:
        run_recipe(args.run, args.dry_run)
        return


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Google Workspace CLI Output Analyzer — Parse, filter, and aggregate JSON/NDJSON output.

Reads JSON arrays or NDJSON streams from stdin or file, applies filters,
projections, sorting, grouping, and outputs in table/csv/json format.

Usage:
    gws drive files list | python3 output_analyzer.py --count
    gws drive files list | python3 output_analyzer.py --filter "mimeType=application/pdf"
    gws drive files list | python3 output_analyzer.py --select "name,size" --format table
    python3 output_analyzer.py --input results.json --group-by "mimeType"
    python3 output_analyzer.py --demo --select "name,mimeType,size" --format table
"""

import argparse
import csv
import io
import json
import sys
from dataclasses import dataclass
from typing import List, Dict, Any, Optional


DEMO_DATA = [
    {"id": "1", "name": "Q1 Report.pdf", "mimeType": "application/pdf", "size": "245760",
     "modifiedTime": "2026-03-10T14:30:00Z", "shared": True, "owners": [{"displayName": "Alice"}]},
    {"id": "2", "name": "Budget 2026.xlsx", "mimeType": "application/vnd.google-apps.spreadsheet",
     "size": "0", "modifiedTime": "2026-03-09T09:15:00Z", "shared": True,
     "owners": [{"displayName": "Bob"}]},
    {"id": "3", "name": "Meeting Notes.docx", "mimeType": "application/vnd.google-apps.document",
     "size": "0", "modifiedTime": "2026-03-08T16:00:00Z", "shared": False,
     "owners": [{"displayName": "Alice"}]},
    {"id": "4", "name": "Logo.png", "mimeType": "image/png", "size": "102400",
     "modifiedTime": "2026-03-07T11:00:00Z", "shared": False,
     "owners": [{"displayName": "Charlie"}]},
    {"id": "5", "name": "Presentation.pptx", "mimeType": "application/vnd.google-apps.presentation",
     "size": "0", "modifiedTime": "2026-03-06T10:00:00Z", "shared": True,
     "owners": [{"displayName": "Alice"}]},
    {"id": "6", "name": "Invoice-001.pdf", "mimeType": "application/pdf", "size": "89000",
     "modifiedTime": "2026-03-05T08:30:00Z", "shared": False,
     "owners": [{"displayName": "Bob"}]},
    {"id": "7", "name": "Project Plan.xlsx", "mimeType": "application/vnd.google-apps.spreadsheet",
     "size": "0", "modifiedTime": "2026-03-04T13:45:00Z", "shared": True,
     "owners": [{"displayName": "Charlie"}]},
    {"id": "8", "name": "Contract Draft.docx", "mimeType": "application/vnd.google-apps.document",
     "size": "0", "modifiedTime": "2026-03-03T09:00:00Z", "shared": False,
     "owners": [{"displayName": "Alice"}]},
]


def read_input(input_file: Optional[str]) -> List[Dict[str, Any]]:
    """Read JSON array or NDJSON from file or stdin."""
    if input_file:
        with open(input_file, "r") as f:
            text = f.read().strip()
    else:
        if sys.stdin.isatty():
            return []
        text = sys.stdin.read().strip()

    if not text:
        return []

    # Try JSON array first
    try:
        data = json.loads(text)
        if isinstance(data, list):
            return data
        if isinstance(data, dict):
            # Some gws commands wrap results in a key
            for key in ("files", "messages", "events", "items", "results",
                        "spreadsheets", "spaces", "tasks", "users", "groups"):
                if key in data and isinstance(data[key], list):
                    return data[key]
            return [data]
    except json.JSONDecodeError:
        pass

    # Try NDJSON
    records = []
    for line in text.split("\n"):
        line = line.strip()
        if line:
            try:
                records.append(json.loads(line))
            except json.JSONDecodeError:
                continue
    return records


def get_nested(obj: Dict, path: str) -> Any:
    """Get a nested value by dot-separated path."""
    parts = path.split(".")
    current = obj
    for part in parts:
        if isinstance(current, dict):
            current = current.get(part)
        elif isinstance(current, list) and part.isdigit():
            idx = int(part)
            current = current[idx] if idx < len(current) else None
        else:
            return None
        if current is None:
            return None
    return current


def apply_filter(records: List[Dict], filter_expr: str) -> List[Dict]:
    """Filter records by field=value expression."""
    if "=" not in filter_expr:
        return records
    field_path, value = filter_expr.split("=", 1)
    result = []
    for rec in records:
        rec_val = get_nested(rec, field_path)
        if rec_val is None:
            continue
        rec_str = str(rec_val).lower()
        if rec_str == value.lower() or value.lower() in rec_str:
            result.append(rec)
    return result


def apply_select(records: List[Dict], fields: str) -> List[Dict]:
    """Project specific fields from records."""
    field_list = [f.strip() for f in fields.split(",")]
    result = []
    for rec in records:
        projected = {}
        for f in field_list:
            projected[f] = get_nested(rec, f)
        result.append(projected)
    return result


def apply_sort(records: List[Dict], sort_field: str, reverse: bool = False) -> List[Dict]:
    """Sort records by a field."""
    def sort_key(rec):
        val = get_nested(rec, sort_field)
        if val is None:
            return ""
        if isinstance(val, (int, float)):
            return val
        try:
            return float(val)
        except (ValueError, TypeError):
            return str(val).lower()
    return sorted(records, key=sort_key, reverse=reverse)


def apply_group_by(records: List[Dict], field: str) -> Dict[str, int]:
    """Group records by a field and count."""
    groups: Dict[str, int] = {}
    for rec in records:
        val = get_nested(rec, field)
        key = str(val) if val is not None else "(null)"
        groups[key] = groups.get(key, 0) + 1
    return dict(sorted(groups.items(), key=lambda x: x[1], reverse=True))


def compute_stats(records: List[Dict], field: str) -> Dict[str, Any]:
    """Compute min/max/avg/sum for a numeric field."""
    values = []
    for rec in records:
        val = get_nested(rec, field)
        if val is not None:
            try:
                values.append(float(val))
            except (ValueError, TypeError):
                continue
    if not values:
        return {"field": field, "count": 0, "error": "No numeric values found"}
    return {
        "field": field,
        "count": len(values),
        "min": min(values),
        "max": max(values),
        "sum": sum(values),
        "avg": sum(values) / len(values),
    }


def format_table(records: List[Dict]) -> str:
    """Format records as an aligned text table."""
    if not records:
        return "(no records)"

    headers = list(records[0].keys())
    # Calculate column widths
    widths = {h: len(h) for h in headers}
    for rec in records:
        for h in headers:
            val = str(rec.get(h, ""))
            if len(val) > 60:
                val = val[:57] + "..."
            widths[h] = max(widths[h], len(val))

    # Header
    header_line = "  ".join(h.ljust(widths[h]) for h in headers)
    sep_line = "  ".join("-" * widths[h] for h in headers)
    lines = [header_line, sep_line]

    # Rows
    for rec in records:
        row = []
        for h in headers:
            val = str(rec.get(h, ""))
            if len(val) > 60:
                val = val[:57] + "..."
            row.append(val.ljust(widths[h]))
        lines.append("  ".join(row))

    return "\n".join(lines)


def format_csv_output(records: List[Dict]) -> str:
    """Format records as CSV."""
    if not records:
        return ""
    output = io.StringIO()
    writer = csv.DictWriter(output, fieldnames=records[0].keys())
    writer.writeheader()
    writer.writerows(records)
    return output.getvalue()


def main():
    parser = argparse.ArgumentParser(
        description="Parse, filter, and aggregate JSON/NDJSON from gws CLI output",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  gws drive files list | %(prog)s --count
  gws drive files list | %(prog)s --filter "mimeType=pdf" --select "name,size"
  gws drive files list | %(prog)s --group-by "mimeType" --format table
  gws drive files list | %(prog)s --sort "size" --reverse --format table
  gws drive files list | %(prog)s --stats "size"
  %(prog)s --input results.json --select "name,mimeType" --format csv
  %(prog)s --demo --select "name,mimeType,size" --format table
        """,
    )
    parser.add_argument("--input", help="Input file (default: stdin)")
    parser.add_argument("--demo", action="store_true", help="Use demo data")
    parser.add_argument("--count", action="store_true", help="Count records")
    parser.add_argument("--filter", help="Filter by field=value")
    parser.add_argument("--select", help="Comma-separated fields to project")
    parser.add_argument("--sort", help="Sort by field")
    parser.add_argument("--reverse", action="store_true", help="Reverse sort order")
    parser.add_argument("--group-by", help="Group by field and count")
    parser.add_argument("--stats", help="Compute stats for a numeric field")
    parser.add_argument("--format", choices=["json", "table", "csv"], default="json",
                        help="Output format (default: json)")
    parser.add_argument("--json", action="store_true",
                        help="Shorthand for --format json")
    args = parser.parse_args()

    if args.json:
        args.format = "json"

    # Read input
    if args.demo:
        records = DEMO_DATA[:]
    else:
        records = read_input(args.input)

    if not records and not args.demo:
        # If no pipe input and no file, use demo
        records = DEMO_DATA[:]
        print("(No input detected, using demo data)\n", file=sys.stderr)

    # Apply operations in order
    if args.filter:
        records = apply_filter(records, args.filter)

    if args.sort:
        records = apply_sort(records, args.sort, args.reverse)

    # Count
    if args.count:
        if args.format == "json":
            print(json.dumps({"count": len(records)}))
        else:
            print(f"Count: {len(records)}")
        return

    # Group by
    if args.group_by:
        groups = apply_group_by(records, args.group_by)
        if args.format == "json":
            print(json.dumps(groups, indent=2))
        elif args.format == "csv":
            print(f"{args.group_by},count")
            for k, v in groups.items():
                print(f"{k},{v}")
        else:
            print(f"\n  Group by: {args.group_by}\n")
            for k, v in groups.items():
                print(f"  {k:<50} {v}")
            print(f"\n  Total groups: {len(groups)}")
        return

    # Stats
    if args.stats:
        stats = compute_stats(records, args.stats)
        if args.format == "json":
            print(json.dumps(stats, indent=2))
        else:
            print(f"\n  Stats for '{args.stats}':")
            for k, v in stats.items():
                if isinstance(v, float):
                    print(f"    {k}: {v:,.2f}")
                else:
                    print(f"    {k}: {v}")
        return

    # Select fields
    if args.select:
        records = apply_select(records, args.select)

    # Output
    if args.format == "json":
        print(json.dumps(records, indent=2))
    elif args.format == "csv":
        print(format_csv_output(records))
    else:
        print(f"\n{format_table(records)}\n")
        print(f"  ({len(records)} records)\n")


if __name__ == "__main__":
    main()