main.py

#!/usr/bin/env python3
"""
Example Scanopy polling integration.

This template fetches hosts from Scanopy and syncs them to a target system.
Customize sync_to_target() to implement your integration logic.
"""

import os
import time
import logging

import requests
import schedule
from dotenv import load_dotenv

load_dotenv()

# Configuration
SCANOPY_URL = os.environ["SCANOPY_URL"].rstrip("/")
SCANOPY_API_KEY = os.environ["SCANOPY_API_KEY"]
TARGET_API_URL = os.environ["TARGET_API_URL"].rstrip("/")
TARGET_API_KEY = os.environ["TARGET_API_KEY"]
POLL_INTERVAL = int(os.environ.get("POLL_INTERVAL", 300))

# Setup logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger(__name__)


def fetch_scanopy_hosts() -> list[dict]:
    """Fetch all hosts from Scanopy API."""
    headers = {"Authorization": f"Bearer {SCANOPY_API_KEY}"}
    response = requests.get(
        f"{SCANOPY_URL}/api/v1/hosts",
        headers=headers,
        timeout=30,
    )
    response.raise_for_status()
    return response.json().get("data", [])


def fetch_scanopy_services(host_id: str) -> list[dict]:
    """Fetch services for a specific host."""
    headers = {"Authorization": f"Bearer {SCANOPY_API_KEY}"}
    response = requests.get(
        f"{SCANOPY_URL}/api/v1/hosts/{host_id}/services",
        headers=headers,
        timeout=30,
    )
    response.raise_for_status()
    return response.json().get("data", [])


def sync_to_target(hosts: list[dict]) -> None:
    """
    Sync hosts to the target system.

    TODO: Implement your sync logic here.

    Example:
        headers = {"Authorization": f"Bearer {TARGET_API_KEY}"}
        for host in hosts:
            requests.post(
                f"{TARGET_API_URL}/assets",
                headers=headers,
                json={"name": host["hostname"], "ip": host["ip_address"]},
            )
    """
    logger.info(f"Would sync {len(hosts)} hosts to {TARGET_API_URL}")
    # Placeholder - implement your logic here
    for host in hosts:
        logger.debug(f"  - {host.get('hostname', host.get('id'))}")


def sync() -> None:
    """Main sync function - runs on each poll interval."""
    try:
        logger.info("Starting sync...")
        hosts = fetch_scanopy_hosts()
        logger.info(f"Fetched {len(hosts)} hosts from Scanopy")
        sync_to_target(hosts)
        logger.info("Sync completed successfully")
    except requests.RequestException as e:
        logger.error(f"Sync failed: {e}")
    except Exception as e:
        logger.exception(f"Unexpected error during sync: {e}")


def main() -> None:
    """Entry point - runs sync immediately then on schedule."""
    logger.info(f"Starting integration (poll interval: {POLL_INTERVAL}s)")
    logger.info(f"Scanopy URL: {SCANOPY_URL}")
    logger.info(f"Target URL: {TARGET_API_URL}")

    # Run immediately on startup
    sync()

    # Schedule recurring syncs
    schedule.every(POLL_INTERVAL).seconds.do(sync)

    while True:
        schedule.run_pending()
        time.sleep(1)


if __name__ == "__main__":
    main()