dss/scripts/figma-sync.py

#!/usr/bin/env python3
"""
DSS Figma Sync CLI.

This script is a lightweight CLI wrapper around the FigmaTokenSource from the
dss.ingest module. It fetches tokens and components from Figma and saves them
to the project's .dss directory.

The core extraction and processing logic resides in:
dss.ingest.sources.figma.FigmaTokenSource
"""

import argparse
import asyncio
import json
import os
import sys
from pathlib import Path
from typing import Dict

# Ensure the project root is in the Python path BEFORE importing dss modules
DSS_ROOT = Path(__file__).parent.parent
if str(DSS_ROOT) not in sys.path:
    sys.path.insert(0, str(DSS_ROOT))

from dss.ingest.base import TokenCollection
from dss.ingest.sources.figma import FigmaTokenSource


# =============================================================================
# CONFIGURATION
# =============================================================================

CACHE_DIR = DSS_ROOT / ".dss/cache"
TOKENS_DIR = DSS_ROOT / ".dss/data/_system/tokens"
COMPONENTS_DIR = DSS_ROOT / ".dss/components"

# =============================================================================
# OUTPUT WRITER
# =============================================================================


class OutputWriter:
    """Writes extraction results to the DSS file structure."""

    def __init__(self, verbose: bool = False):
        self.verbose = verbose

    def write_token_collection(self, collection: TokenCollection, output_dir: Path = TOKENS_DIR):
        """Write TokenCollection to a structured JSON file."""
        output_dir.mkdir(parents=True, exist_ok=True)
        tokens_file = output_dir / "figma-tokens.json"

        if self.verbose:
            print(f"  [OUT] Writing {len(collection)} tokens to {tokens_file}")

        with open(tokens_file, "w") as f:
            json.dump(json.loads(collection.to_json()), f, indent=2)
        print(f"  [OUT] Tokens: {tokens_file}")

    def write_components(self, components: Dict, output_dir: Path = COMPONENTS_DIR):
        """Write component registry."""
        output_dir.mkdir(parents=True, exist_ok=True)
        comp_file = output_dir / "figma-registry.json"

        if self.verbose:
            print(
                f"  [OUT] Writing {components.get('component_count', 0)} components to {comp_file}"
            )

        with open(comp_file, "w") as f:
            json.dump(components, f, indent=2)
        print(f"  [OUT] Components: {comp_file}")


# =============================================================================
# MAIN ORCHESTRATOR
# =============================================================================


async def main():
    """Main CLI orchestration function."""
    parser = argparse.ArgumentParser(description="DSS Intelligent Figma Sync")
    parser.add_argument("--file-key", help="Figma file key to sync")
    parser.add_argument("--force", action="store_true", help="Force sync, ignoring cache")
    parser.add_argument("--verbose", "-v", action="store_true", help="Enable verbose output")
    args = parser.parse_args()

    # --- Configuration Loading ---
    config = load_config()
    file_key = args.file_key or config.get("uikit_reference", {}).get("file_key")
    token = os.environ.get("FIGMA_TOKEN") or config.get("token")

    if not file_key:
        print("[ERROR] No Figma file key provided.", file=sys.stderr)
        print("  Provide it with --file-key or in .dss/config/figma.json", file=sys.stderr)
        sys.exit(1)

    if not token:
        print("[ERROR] No Figma token found.", file=sys.stderr)
        print("  Set FIGMA_TOKEN env var or add 'token' to .dss/config/figma.json", file=sys.stderr)
        sys.exit(1)

    print_header(file_key, token, args.force)

    # --- Extraction ---
    try:
        source = FigmaTokenSource(figma_token=token, verbose=args.verbose)
        token_collection, components = await source.extract(file_key)
    except Exception as e:
        print(f"\n[ERROR] An error occurred during extraction: {e}", file=sys.stderr)
        # In verbose mode, print more details
        if args.verbose:
            import traceback

            traceback.print_exc()
        sys.exit(1)

    # --- Writing Output ---
    print("\n[3/3] Writing output...")
    writer = OutputWriter(verbose=args.verbose)
    writer.write_token_collection(token_collection)

    # Build component registry dict from List[Component]
    component_registry = {
        "file_key": file_key,
        "component_count": len(components),
        "components": [
            {
                "name": c.name,
                "classification": c.classification.value if hasattr(c.classification, 'value') else str(c.classification),
                "variant_count": len(c.variants) if c.variants else 0,
            }
            for c in components
        ]
    }
    writer.write_components(component_registry)

    # --- Summary ---
    print_summary(
        file_name=file_key,
        token_count=len(token_collection),
        component_count=len(components),
    )

    print("\n[OK] Sync successful!")
    print("  Next: Run the translation and theming pipeline.")
    sys.exit(0)


def load_config() -> Dict:
    """Load Figma config from .dss/config/figma.json."""
    config_path = DSS_ROOT / ".dss/config/figma.json"
    if config_path.exists():
        try:
            with open(config_path) as f:
                return json.load(f)
        except (json.JSONDecodeError, IOError) as e:
            print(
                f"[WARN] Could not read or parse config file: {config_path}\n{e}", file=sys.stderr
            )
    return {}


def print_header(file_key: str, token: str, force: bool):
    """Prints the CLI header."""
    print("╔══════════════════════════════════════════════════════════════╗")
    print("║            DSS FIGMA SYNC (INGESTION CLI) v3.0               ║")
    print("╚══════════════════════════════════════════════════════════════╝")
    print(f"  File:  {file_key}")
    print(f"  Token: {token[:10]}...")
    print(f"  Force: {force}")
    print("\n[1/3] Initializing Figma Ingestion Source...")


def print_summary(file_name: str, token_count: int, component_count: int):
    """Prints the final summary."""
    print("\n" + "=" * 60)
    print("SYNC COMPLETE")
    print("=" * 60)
    print(f"  File:       {file_name}")
    print(f"  Tokens:     {token_count}")
    print(f"  Components: {component_count}")


if __name__ == "__main__":
    asyncio.run(main())