home-assistant/utils/validation/validate_entities.py

"""Validate entity IDs against Home Assistant database."""

import os
import sys
from pathlib import Path

import sqlalchemy as sa
from sqlalchemy.orm import Session

# Add parent directory to path for imports
sys.path.insert(0, str(Path(__file__).parent.parent.parent))

from utils.database.get_db_session import get_db_session
from utils.database.tables.states_meta import StatesMeta
from utils.validation.parse_config import parse_config_files


def get_valid_entity_ids(session: Session) -> set[str]:
    """
    Query database for all valid entity IDs.
    
    Args:
        session: SQLAlchemy database session
    
    Returns:
        Set of entity_id strings from states_meta table
    """
    stmt = sa.select(StatesMeta.entity_id)
    result = session.execute(stmt)
    return {row[0] for row in result}


def validate_entities(config_dir: Path) -> tuple[set[str], set[str], set[str]]:
    """
    Validate entity IDs from config files against database.
    
    Args:
        config_dir: Path to Home Assistant config directory
    
    Returns:
        Tuple of (valid_entities, invalid_entities, all_config_entities)
    """
    # Get environment variables for database connection
    db_host = os.environ.get('POSTGRES_HOST', 'localhost')
    db_port = int(os.environ.get('POSTGRES_PORT', '5432'))
    db_name = os.environ.get('DB_NAME', 'home_assistant')
    db_user = os.environ.get('DB_READONLY_USER', os.environ.get('POSTGRES_USER'))
    db_password = os.environ.get('DB_READONLY_PASSWORD', os.environ.get('POSTGRES_PASSWORD'))
    
    if not all([db_user, db_password]):
        raise ValueError(
            "Database credentials not found. Set DB_READONLY_USER and DB_READONLY_PASSWORD "
            "environment variables."
        )
    
    # Parse config files to get entity IDs
    config_entities, _ = parse_config_files(config_dir)
    
    # Get database session
    session = get_db_session(
        host=db_host,
        port=db_port,
        user=db_user,
        password=db_password,
        database=db_name
    )
    
    # Query valid entity IDs from database
    db_entities = get_valid_entity_ids(session)
    
    # Find valid and invalid entities
    valid = config_entities & db_entities
    invalid = config_entities - db_entities
    
    session.close()
    
    return valid, invalid, config_entities


def main():
    """Main entry point for entity validation."""
    if len(sys.argv) < 2:
        print("Usage: python validate_entities.py <config_directory>")
        sys.exit(1)
    
    config_path = Path(sys.argv[1])
    
    if not config_path.exists():
        print(f"Error: Config directory not found: {config_path}")
        sys.exit(1)
    
    print("Validating entity IDs against database...")
    
    try:
        valid, invalid, all_entities = validate_entities(config_path)
        
        print(f"\n✓ Total entity IDs in config: {len(all_entities)}")
        print(f"✓ Valid entity IDs: {len(valid)}")
        
        if invalid:
            print(f"\n⚠ Warning: {len(invalid)} entity IDs not found in database:")
            for entity_id in sorted(invalid):
                print(f"  - {entity_id}")
            print("\nNote: These might be template entities defined in templates.yaml")
            print("or entities that haven't been created yet.")
            # Exit with warning code (not failure)
            sys.exit(0)
        else:
            print("\n✓ All entity IDs validated successfully!")
            sys.exit(0)
    
    except Exception as e:
        print(f"\n✗ Error during validation: {e}")
        sys.exit(1)


if __name__ == '__main__':
    main()