Skip to content

deletion_risk_analysis

Deletion risk analysis for model references.

Provides functions to analyze models for deletion risk. Identifies issues like missing downloads, non-preferred hosts, low usage, etc.

UsageTrend

Bases: BaseModel

Usage trend metrics comparing different time periods.

Ratios help identify model momentum and activity patterns.

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
class UsageTrend(BaseModel):
    """Usage trend metrics comparing different time periods.

    Ratios help identify model momentum and activity patterns.
    """

    model_config = ConfigDict(use_attribute_docstrings=True)

    day_to_month_ratio: float | None = None
    """Ratio of day usage to month usage (day/month). None if month usage is zero."""
    month_to_total_ratio: float | None = None
    """Ratio of month usage to total usage (month/total). None if total usage is zero."""

model_config class-attribute instance-attribute

model_config = ConfigDict(use_attribute_docstrings=True)

day_to_month_ratio class-attribute instance-attribute

day_to_month_ratio: float | None = None

Ratio of day usage to month usage (day/month). None if month usage is zero.

month_to_total_ratio class-attribute instance-attribute

month_to_total_ratio: float | None = None

Ratio of month usage to total usage (month/total). None if total usage is zero.

BackendVariationStats

Bases: BaseModel

Per-backend statistics for a text generation model in deletion risk context.

Provides breakdown of workers and usage by backend (aphrodite, koboldcpp, canonical). Used in ungrouped deletion risk view to show backend-specific details for each model.

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
class BackendVariationStats(BaseModel):
    """Per-backend statistics for a text generation model in deletion risk context.

    Provides breakdown of workers and usage by backend (aphrodite, koboldcpp, canonical).
    Used in ungrouped deletion risk view to show backend-specific details for each model.
    """

    model_config = ConfigDict(use_attribute_docstrings=True)

    backend: str
    """Backend name (e.g., 'aphrodite', 'koboldcpp', 'canonical')."""
    variant_name: str
    """Full model name as reported by Horde API (may include backend prefix)."""
    worker_count: int = Field(ge=0, default=0)
    """Number of workers serving this backend variant."""
    performance: float | None = None
    """Performance metric for this backend variant."""
    usage_day: int = Field(ge=0, default=0)
    """Usage count for the past day from this backend."""
    usage_month: int = Field(ge=0, default=0)
    """Usage count for the past month from this backend."""
    usage_total: int = Field(ge=0, default=0)
    """Total usage count from this backend."""

model_config class-attribute instance-attribute

model_config = ConfigDict(use_attribute_docstrings=True)

backend instance-attribute

backend: str

Backend name (e.g., 'aphrodite', 'koboldcpp', 'canonical').

variant_name instance-attribute

variant_name: str

Full model name as reported by Horde API (may include backend prefix).

worker_count class-attribute instance-attribute

worker_count: int = Field(ge=0, default=0)

Number of workers serving this backend variant.

performance class-attribute instance-attribute

performance: float | None = None

Performance metric for this backend variant.

usage_day class-attribute instance-attribute

usage_day: int = Field(ge=0, default=0)

Usage count for the past day from this backend.

usage_month class-attribute instance-attribute

usage_month: int = Field(ge=0, default=0)

Usage count for the past month from this backend.

usage_total class-attribute instance-attribute

usage_total: int = Field(ge=0, default=0)

Total usage count from this backend.

DeletionRiskFlags

Bases: BaseModel

Flags indicating potential reasons for model deletion.

Each flag represents a specific issue that might warrant model removal. Aligned with frontend expectations for consistent presentation.

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
class DeletionRiskFlags(BaseModel):
    """Flags indicating potential reasons for model deletion.

    Each flag represents a specific issue that might warrant model removal.
    Aligned with frontend expectations for consistent presentation.
    """

    model_config = ConfigDict(use_attribute_docstrings=True)

    zero_usage_day: bool = False
    """Model has no usage in the past day."""
    zero_usage_month: bool = False
    """Model has no usage in the past month."""
    zero_usage_total: bool = False
    """Model has no usage in total (all time)."""
    no_active_workers: bool = False
    """Model has zero active workers (from Horde API data)."""
    has_multiple_hosts: bool = False
    """Model downloads are distributed across multiple file hosts."""
    has_non_preferred_host: bool = False
    """Model is hosted on non-preferred file hosts (non-HuggingFace)."""
    has_unknown_host: bool = False
    """Model has download URLs with unknown or unparseable hosts."""
    no_download_urls: bool = False
    """Model has no download URLs, empty download list, or invalid URLs."""
    missing_description: bool = False
    """Model lacks a description field."""
    missing_baseline: bool = False
    """Model lacks baseline information (for applicable categories)."""
    low_usage: bool = False
    """Model has very low usage (threshold defined by LOW_USAGE_THRESHOLD constant)."""

    def any_flags(self) -> bool:
        """Check if any deletion risk flags are set.

        Returns:
            True if at least one flag is set, False otherwise.

        """
        return any(
            [
                self.zero_usage_day,
                self.zero_usage_month,
                self.zero_usage_total,
                self.no_active_workers,
                self.has_multiple_hosts,
                self.has_non_preferred_host,
                self.has_unknown_host,
                self.no_download_urls,
                self.missing_description,
                self.missing_baseline,
                self.low_usage,
            ]
        )

    def flag_count(self) -> int:
        """Count the number of flags set.

        Returns:
            Number of deletion risk flags that are True.

        """
        return sum(
            [
                self.zero_usage_day,
                self.zero_usage_month,
                self.zero_usage_total,
                self.no_active_workers,
                self.has_multiple_hosts,
                self.has_non_preferred_host,
                self.has_unknown_host,
                self.no_download_urls,
                self.missing_description,
                self.missing_baseline,
                self.low_usage,
            ]
        )

model_config class-attribute instance-attribute

model_config = ConfigDict(use_attribute_docstrings=True)

zero_usage_day class-attribute instance-attribute

zero_usage_day: bool = False

Model has no usage in the past day.

zero_usage_month class-attribute instance-attribute

zero_usage_month: bool = False

Model has no usage in the past month.

zero_usage_total class-attribute instance-attribute

zero_usage_total: bool = False

Model has no usage in total (all time).

no_active_workers class-attribute instance-attribute

no_active_workers: bool = False

Model has zero active workers (from Horde API data).

has_multiple_hosts class-attribute instance-attribute

has_multiple_hosts: bool = False

Model downloads are distributed across multiple file hosts.

has_non_preferred_host class-attribute instance-attribute

has_non_preferred_host: bool = False

Model is hosted on non-preferred file hosts (non-HuggingFace).

has_unknown_host class-attribute instance-attribute

has_unknown_host: bool = False

Model has download URLs with unknown or unparseable hosts.

no_download_urls class-attribute instance-attribute

no_download_urls: bool = False

Model has no download URLs, empty download list, or invalid URLs.

missing_description class-attribute instance-attribute

missing_description: bool = False

Model lacks a description field.

missing_baseline class-attribute instance-attribute

missing_baseline: bool = False

Model lacks baseline information (for applicable categories).

low_usage class-attribute instance-attribute

low_usage: bool = False

Model has very low usage (threshold defined by LOW_USAGE_THRESHOLD constant).

any_flags

any_flags() -> bool

Check if any deletion risk flags are set.

Returns:

  • bool

    True if at least one flag is set, False otherwise.

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
def any_flags(self) -> bool:
    """Check if any deletion risk flags are set.

    Returns:
        True if at least one flag is set, False otherwise.

    """
    return any(
        [
            self.zero_usage_day,
            self.zero_usage_month,
            self.zero_usage_total,
            self.no_active_workers,
            self.has_multiple_hosts,
            self.has_non_preferred_host,
            self.has_unknown_host,
            self.no_download_urls,
            self.missing_description,
            self.missing_baseline,
            self.low_usage,
        ]
    )

flag_count

flag_count() -> int

Count the number of flags set.

Returns:

  • int

    Number of deletion risk flags that are True.

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
def flag_count(self) -> int:
    """Count the number of flags set.

    Returns:
        Number of deletion risk flags that are True.

    """
    return sum(
        [
            self.zero_usage_day,
            self.zero_usage_month,
            self.zero_usage_total,
            self.no_active_workers,
            self.has_multiple_hosts,
            self.has_non_preferred_host,
            self.has_unknown_host,
            self.no_download_urls,
            self.missing_description,
            self.missing_baseline,
            self.low_usage,
        ]
    )

FlagValidatorService

Service providing reusable flag validation methods.

All methods are static and stateless for efficient reuse.

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
class FlagValidatorService:
    """Service providing reusable flag validation methods.

    All methods are static and stateless for efficient reuse.
    """

    @staticmethod
    def validate_downloads(
        downloads: list[Any] | None,
        preferred_hosts: list[str] | None = None,
        category: MODEL_REFERENCE_CATEGORY | None = None,
    ) -> tuple[bool, bool, bool, bool]:
        """Validate download URLs and return related flags.

        Args:
            downloads: List of download records to validate.
            preferred_hosts: List of preferred file hosts. If None, uses settings.
            category: Model category - if text_generation and ignore setting is True, returns all False.

        Returns:
            Tuple of (no_download_urls, has_multiple_hosts, has_non_preferred_host, has_unknown_host)

        """
        # Skip download validation for text_generation if configured
        if (
            category == MODEL_REFERENCE_CATEGORY.text_generation
            and horde_model_reference_settings.text_gen_ignore_download_hosts
        ):
            return (False, False, False, False)

        if preferred_hosts is None:
            preferred_hosts = horde_model_reference_settings.preferred_file_hosts

        unique_hosts: set[str] = set()
        has_valid_url = False
        has_preferred_host = False
        has_unknown_host = False

        if not downloads or len(downloads) == 0:
            return (True, False, False, False)

        for download in downloads:
            url = download.file_url
            if url:
                try:
                    parsed = urlparse(url)
                    if parsed.scheme and parsed.netloc:
                        has_valid_url = True
                        unique_hosts.add(parsed.netloc)

                        # Check if this host is preferred
                        if preferred_hosts and any(host in parsed.netloc for host in preferred_hosts):
                            has_preferred_host = True
                except Exception:
                    # Failed to parse URL - unknown host
                    has_unknown_host = True

        no_download_urls = not has_valid_url
        has_multiple_hosts = len(unique_hosts) > 1
        has_non_preferred_host = has_valid_url and not has_preferred_host

        return (no_download_urls, has_multiple_hosts, has_non_preferred_host, has_unknown_host)

    @staticmethod
    def validate_description(description: str | None) -> bool:
        """Validate model description.

        Args:
            description: The model description to validate.

        Returns:
            True if description is missing or empty.

        """
        return not description or len(description.strip()) == 0

    @staticmethod
    def validate_baseline(baseline: str | None) -> bool:
        """Validate model baseline.

        Args:
            baseline: The model baseline to validate.

        Returns:
            True if baseline is missing.

        """
        return not baseline

    @staticmethod
    def validate_statistics(
        statistics: CombinedModelStatistics | None,
        category_total_usage: int,
        low_usage_threshold: float | None = None,
        category: MODEL_REFERENCE_CATEGORY | None = None,
    ) -> tuple[bool, bool, bool, bool, bool]:
        """Validate Horde API statistics and usage data.

        Args:
            statistics: Optional Horde API statistics.
            category_total_usage: Total monthly usage for the category.
            low_usage_threshold: Percentage threshold for low usage. If None, uses settings default.
            category: Model category for category-specific thresholds.

        Returns:
            Tuple of (zero_usage_day, zero_usage_month, zero_usage_total, no_active_workers, low_usage)

        """
        if not statistics:
            return (False, False, False, False, False)

        # Use category-specific threshold for text_generation
        if low_usage_threshold is None:
            if category == MODEL_REFERENCE_CATEGORY.text_generation:
                low_usage_threshold = horde_model_reference_settings.text_gen_low_usage_threshold_percentage
            else:
                low_usage_threshold = horde_model_reference_settings.low_usage_threshold_percentage

        # Check for zero workers
        no_active_workers = statistics.worker_count == 0

        zero_usage_day = False
        zero_usage_month = False
        zero_usage_total = False
        low_usage = False

        # Check usage statistics
        if statistics.usage_stats:
            usage_day = statistics.usage_stats.day
            usage_month = statistics.usage_stats.month
            usage_total = statistics.usage_stats.total

            zero_usage_day = usage_day == 0
            zero_usage_month = usage_month == 0
            zero_usage_total = usage_total == 0

            # Low usage (configurable threshold via settings)
            if category_total_usage > 0:
                usage_percentage = (usage_month / category_total_usage) * 100.0
                low_usage = usage_percentage < low_usage_threshold

        return (zero_usage_day, zero_usage_month, zero_usage_total, no_active_workers, low_usage)

validate_downloads staticmethod

validate_downloads(
    downloads: list[Any] | None,
    preferred_hosts: list[str] | None = None,
    category: MODEL_REFERENCE_CATEGORY | None = None,
) -> tuple[bool, bool, bool, bool]

Validate download URLs and return related flags.

Parameters:

  • downloads (list[Any] | None) –

    List of download records to validate.

  • preferred_hosts (list[str] | None, default: None ) –

    List of preferred file hosts. If None, uses settings.

  • category (MODEL_REFERENCE_CATEGORY | None, default: None ) –

    Model category - if text_generation and ignore setting is True, returns all False.

Returns:

  • tuple[bool, bool, bool, bool]

    Tuple of (no_download_urls, has_multiple_hosts, has_non_preferred_host, has_unknown_host)

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
@staticmethod
def validate_downloads(
    downloads: list[Any] | None,
    preferred_hosts: list[str] | None = None,
    category: MODEL_REFERENCE_CATEGORY | None = None,
) -> tuple[bool, bool, bool, bool]:
    """Validate download URLs and return related flags.

    Args:
        downloads: List of download records to validate.
        preferred_hosts: List of preferred file hosts. If None, uses settings.
        category: Model category - if text_generation and ignore setting is True, returns all False.

    Returns:
        Tuple of (no_download_urls, has_multiple_hosts, has_non_preferred_host, has_unknown_host)

    """
    # Skip download validation for text_generation if configured
    if (
        category == MODEL_REFERENCE_CATEGORY.text_generation
        and horde_model_reference_settings.text_gen_ignore_download_hosts
    ):
        return (False, False, False, False)

    if preferred_hosts is None:
        preferred_hosts = horde_model_reference_settings.preferred_file_hosts

    unique_hosts: set[str] = set()
    has_valid_url = False
    has_preferred_host = False
    has_unknown_host = False

    if not downloads or len(downloads) == 0:
        return (True, False, False, False)

    for download in downloads:
        url = download.file_url
        if url:
            try:
                parsed = urlparse(url)
                if parsed.scheme and parsed.netloc:
                    has_valid_url = True
                    unique_hosts.add(parsed.netloc)

                    # Check if this host is preferred
                    if preferred_hosts and any(host in parsed.netloc for host in preferred_hosts):
                        has_preferred_host = True
            except Exception:
                # Failed to parse URL - unknown host
                has_unknown_host = True

    no_download_urls = not has_valid_url
    has_multiple_hosts = len(unique_hosts) > 1
    has_non_preferred_host = has_valid_url and not has_preferred_host

    return (no_download_urls, has_multiple_hosts, has_non_preferred_host, has_unknown_host)

validate_description staticmethod

validate_description(description: str | None) -> bool

Validate model description.

Parameters:

  • description (str | None) –

    The model description to validate.

Returns:

  • bool

    True if description is missing or empty.

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
@staticmethod
def validate_description(description: str | None) -> bool:
    """Validate model description.

    Args:
        description: The model description to validate.

    Returns:
        True if description is missing or empty.

    """
    return not description or len(description.strip()) == 0

validate_baseline staticmethod

validate_baseline(baseline: str | None) -> bool

Validate model baseline.

Parameters:

  • baseline (str | None) –

    The model baseline to validate.

Returns:

  • bool

    True if baseline is missing.

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
@staticmethod
def validate_baseline(baseline: str | None) -> bool:
    """Validate model baseline.

    Args:
        baseline: The model baseline to validate.

    Returns:
        True if baseline is missing.

    """
    return not baseline

validate_statistics staticmethod

validate_statistics(
    statistics: CombinedModelStatistics | None,
    category_total_usage: int,
    low_usage_threshold: float | None = None,
    category: MODEL_REFERENCE_CATEGORY | None = None,
) -> tuple[bool, bool, bool, bool, bool]

Validate Horde API statistics and usage data.

Parameters:

  • statistics (CombinedModelStatistics | None) –

    Optional Horde API statistics.

  • category_total_usage (int) –

    Total monthly usage for the category.

  • low_usage_threshold (float | None, default: None ) –

    Percentage threshold for low usage. If None, uses settings default.

  • category (MODEL_REFERENCE_CATEGORY | None, default: None ) –

    Model category for category-specific thresholds.

Returns:

  • tuple[bool, bool, bool, bool, bool]

    Tuple of (zero_usage_day, zero_usage_month, zero_usage_total, no_active_workers, low_usage)

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
@staticmethod
def validate_statistics(
    statistics: CombinedModelStatistics | None,
    category_total_usage: int,
    low_usage_threshold: float | None = None,
    category: MODEL_REFERENCE_CATEGORY | None = None,
) -> tuple[bool, bool, bool, bool, bool]:
    """Validate Horde API statistics and usage data.

    Args:
        statistics: Optional Horde API statistics.
        category_total_usage: Total monthly usage for the category.
        low_usage_threshold: Percentage threshold for low usage. If None, uses settings default.
        category: Model category for category-specific thresholds.

    Returns:
        Tuple of (zero_usage_day, zero_usage_month, zero_usage_total, no_active_workers, low_usage)

    """
    if not statistics:
        return (False, False, False, False, False)

    # Use category-specific threshold for text_generation
    if low_usage_threshold is None:
        if category == MODEL_REFERENCE_CATEGORY.text_generation:
            low_usage_threshold = horde_model_reference_settings.text_gen_low_usage_threshold_percentage
        else:
            low_usage_threshold = horde_model_reference_settings.low_usage_threshold_percentage

    # Check for zero workers
    no_active_workers = statistics.worker_count == 0

    zero_usage_day = False
    zero_usage_month = False
    zero_usage_total = False
    low_usage = False

    # Check usage statistics
    if statistics.usage_stats:
        usage_day = statistics.usage_stats.day
        usage_month = statistics.usage_stats.month
        usage_total = statistics.usage_stats.total

        zero_usage_day = usage_day == 0
        zero_usage_month = usage_month == 0
        zero_usage_total = usage_total == 0

        # Low usage (configurable threshold via settings)
        if category_total_usage > 0:
            usage_percentage = (usage_month / category_total_usage) * 100.0
            low_usage = usage_percentage < low_usage_threshold

    return (zero_usage_day, zero_usage_month, zero_usage_total, no_active_workers, low_usage)

DeletionRiskFlagsBuilder

Builder for composing DeletionRiskFlags with type-safe methods.

Provides a fluent interface for setting flags without magic strings.

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
class DeletionRiskFlagsBuilder:
    """Builder for composing DeletionRiskFlags with type-safe methods.

    Provides a fluent interface for setting flags without magic strings.
    """

    def __init__(self) -> None:
        """Initialize the builder with default flag values."""
        self.zero_usage_day: bool = False
        self.zero_usage_month: bool = False
        self.zero_usage_total: bool = False
        self.no_active_workers: bool = False
        self.has_multiple_hosts: bool = False
        self.has_non_preferred_host: bool = False
        self.has_unknown_host: bool = False
        self.no_download_urls: bool = False
        self.missing_description: bool = False
        self.missing_baseline: bool = False
        self.low_usage: bool = False

    def with_download_flags(
        self,
        no_download_urls: bool,
        has_multiple_hosts: bool,
        has_non_preferred_host: bool,
        has_unknown_host: bool,
    ) -> DeletionRiskFlagsBuilder:
        """Set download-related flags.

        Args:
            no_download_urls: Whether model has no download URLs.
            has_multiple_hosts: Whether downloads span multiple hosts.
            has_non_preferred_host: Whether downloads use non-preferred hosts.
            has_unknown_host: Whether any download URLs couldn't be parsed.

        Returns:
            Self for method chaining.

        """
        self.no_download_urls = no_download_urls
        self.has_multiple_hosts = has_multiple_hosts
        self.has_non_preferred_host = has_non_preferred_host
        self.has_unknown_host = has_unknown_host
        return self

    def with_missing_description(self, missing: bool) -> DeletionRiskFlagsBuilder:
        """Set missing_description flag.

        Args:
            missing: Whether description is missing.

        Returns:
            Self for method chaining.

        """
        self.missing_description = missing
        return self

    def with_missing_baseline(self, missing: bool) -> DeletionRiskFlagsBuilder:
        """Set missing_baseline flag.

        Args:
            missing: Whether baseline is missing.

        Returns:
            Self for method chaining.

        """
        self.missing_baseline = missing
        return self

    def with_statistics_flags(
        self,
        zero_usage_day: bool,
        zero_usage_month: bool,
        zero_usage_total: bool,
        no_active_workers: bool,
        low_usage: bool,
    ) -> DeletionRiskFlagsBuilder:
        """Set statistics-related flags.

        Args:
            zero_usage_day: Whether day usage is zero.
            zero_usage_month: Whether month usage is zero.
            zero_usage_total: Whether total usage is zero.
            no_active_workers: Whether worker count is zero.
            low_usage: Whether usage is below threshold.

        Returns:
            Self for method chaining.

        """
        self.zero_usage_day = zero_usage_day
        self.zero_usage_month = zero_usage_month
        self.zero_usage_total = zero_usage_total
        self.no_active_workers = no_active_workers
        self.low_usage = low_usage
        return self

    def build(self) -> DeletionRiskFlags:
        """Build the final DeletionRiskFlags object.

        Returns:
            DeletionRiskFlags with all accumulated flag values.

        """
        return DeletionRiskFlags(
            zero_usage_day=self.zero_usage_day,
            zero_usage_month=self.zero_usage_month,
            zero_usage_total=self.zero_usage_total,
            no_active_workers=self.no_active_workers,
            has_multiple_hosts=self.has_multiple_hosts,
            has_non_preferred_host=self.has_non_preferred_host,
            has_unknown_host=self.has_unknown_host,
            no_download_urls=self.no_download_urls,
            missing_description=self.missing_description,
            missing_baseline=self.missing_baseline,
            low_usage=self.low_usage,
        )

zero_usage_day instance-attribute

zero_usage_day: bool = False

zero_usage_month instance-attribute

zero_usage_month: bool = False

zero_usage_total instance-attribute

zero_usage_total: bool = False

no_active_workers instance-attribute

no_active_workers: bool = False

has_multiple_hosts instance-attribute

has_multiple_hosts: bool = False

has_non_preferred_host instance-attribute

has_non_preferred_host: bool = False

has_unknown_host instance-attribute

has_unknown_host: bool = False

no_download_urls instance-attribute

no_download_urls: bool = False

missing_description instance-attribute

missing_description: bool = False

missing_baseline instance-attribute

missing_baseline: bool = False

low_usage instance-attribute

low_usage: bool = False

__init__

__init__() -> None

Initialize the builder with default flag values.

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
def __init__(self) -> None:
    """Initialize the builder with default flag values."""
    self.zero_usage_day: bool = False
    self.zero_usage_month: bool = False
    self.zero_usage_total: bool = False
    self.no_active_workers: bool = False
    self.has_multiple_hosts: bool = False
    self.has_non_preferred_host: bool = False
    self.has_unknown_host: bool = False
    self.no_download_urls: bool = False
    self.missing_description: bool = False
    self.missing_baseline: bool = False
    self.low_usage: bool = False

with_download_flags

with_download_flags(
    no_download_urls: bool,
    has_multiple_hosts: bool,
    has_non_preferred_host: bool,
    has_unknown_host: bool,
) -> DeletionRiskFlagsBuilder

Set download-related flags.

Parameters:

  • no_download_urls (bool) –

    Whether model has no download URLs.

  • has_multiple_hosts (bool) –

    Whether downloads span multiple hosts.

  • has_non_preferred_host (bool) –

    Whether downloads use non-preferred hosts.

  • has_unknown_host (bool) –

    Whether any download URLs couldn't be parsed.

Returns:

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
def with_download_flags(
    self,
    no_download_urls: bool,
    has_multiple_hosts: bool,
    has_non_preferred_host: bool,
    has_unknown_host: bool,
) -> DeletionRiskFlagsBuilder:
    """Set download-related flags.

    Args:
        no_download_urls: Whether model has no download URLs.
        has_multiple_hosts: Whether downloads span multiple hosts.
        has_non_preferred_host: Whether downloads use non-preferred hosts.
        has_unknown_host: Whether any download URLs couldn't be parsed.

    Returns:
        Self for method chaining.

    """
    self.no_download_urls = no_download_urls
    self.has_multiple_hosts = has_multiple_hosts
    self.has_non_preferred_host = has_non_preferred_host
    self.has_unknown_host = has_unknown_host
    return self

with_missing_description

with_missing_description(
    missing: bool,
) -> DeletionRiskFlagsBuilder

Set missing_description flag.

Parameters:

  • missing (bool) –

    Whether description is missing.

Returns:

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
def with_missing_description(self, missing: bool) -> DeletionRiskFlagsBuilder:
    """Set missing_description flag.

    Args:
        missing: Whether description is missing.

    Returns:
        Self for method chaining.

    """
    self.missing_description = missing
    return self

with_missing_baseline

with_missing_baseline(
    missing: bool,
) -> DeletionRiskFlagsBuilder

Set missing_baseline flag.

Parameters:

  • missing (bool) –

    Whether baseline is missing.

Returns:

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
def with_missing_baseline(self, missing: bool) -> DeletionRiskFlagsBuilder:
    """Set missing_baseline flag.

    Args:
        missing: Whether baseline is missing.

    Returns:
        Self for method chaining.

    """
    self.missing_baseline = missing
    return self

with_statistics_flags

with_statistics_flags(
    zero_usage_day: bool,
    zero_usage_month: bool,
    zero_usage_total: bool,
    no_active_workers: bool,
    low_usage: bool,
) -> DeletionRiskFlagsBuilder

Set statistics-related flags.

Parameters:

  • zero_usage_day (bool) –

    Whether day usage is zero.

  • zero_usage_month (bool) –

    Whether month usage is zero.

  • zero_usage_total (bool) –

    Whether total usage is zero.

  • no_active_workers (bool) –

    Whether worker count is zero.

  • low_usage (bool) –

    Whether usage is below threshold.

Returns:

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
def with_statistics_flags(
    self,
    zero_usage_day: bool,
    zero_usage_month: bool,
    zero_usage_total: bool,
    no_active_workers: bool,
    low_usage: bool,
) -> DeletionRiskFlagsBuilder:
    """Set statistics-related flags.

    Args:
        zero_usage_day: Whether day usage is zero.
        zero_usage_month: Whether month usage is zero.
        zero_usage_total: Whether total usage is zero.
        no_active_workers: Whether worker count is zero.
        low_usage: Whether usage is below threshold.

    Returns:
        Self for method chaining.

    """
    self.zero_usage_day = zero_usage_day
    self.zero_usage_month = zero_usage_month
    self.zero_usage_total = zero_usage_total
    self.no_active_workers = no_active_workers
    self.low_usage = low_usage
    return self

build

build() -> DeletionRiskFlags

Build the final DeletionRiskFlags object.

Returns:

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
def build(self) -> DeletionRiskFlags:
    """Build the final DeletionRiskFlags object.

    Returns:
        DeletionRiskFlags with all accumulated flag values.

    """
    return DeletionRiskFlags(
        zero_usage_day=self.zero_usage_day,
        zero_usage_month=self.zero_usage_month,
        zero_usage_total=self.zero_usage_total,
        no_active_workers=self.no_active_workers,
        has_multiple_hosts=self.has_multiple_hosts,
        has_non_preferred_host=self.has_non_preferred_host,
        has_unknown_host=self.has_unknown_host,
        no_download_urls=self.no_download_urls,
        missing_description=self.missing_description,
        missing_baseline=self.missing_baseline,
        low_usage=self.low_usage,
    )

ModelDeletionRiskInfo

Bases: BaseModel

Deletion risk information for a single model.

Contains model metadata along with deletion risk assessment and usage statistics.

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
class ModelDeletionRiskInfo(BaseModel):
    """Deletion risk information for a single model.

    Contains model metadata along with deletion risk assessment and usage statistics.
    """

    model_config = ConfigDict(use_attribute_docstrings=True)

    name: str
    """The model name."""
    category: MODEL_REFERENCE_CATEGORY
    """The category this model belongs to."""

    deletion_risk_flags: DeletionRiskFlags
    """Flags indicating potential deletion risks."""
    at_risk: bool
    """True if model has any deletion risk flags."""
    risk_score: int = Field(ge=0)
    """Total number of deletion risk flags (0 = no risk)."""

    worker_count: int = Field(ge=0, default=0)
    """Number of active workers serving this model."""
    usage_day: int = Field(ge=0, default=0)
    """Usage count for the past day."""
    usage_month: int = Field(ge=0, default=0)
    """Usage count for the past month."""
    usage_total: int = Field(ge=0, default=0)
    """Total usage count (all time)."""
    usage_hour: int | None = None
    """Usage count for the past hour (will be populated when alternative Horde API sources become available)."""
    usage_minute: int | None = None
    """Usage count for the past minute (will be populated when alternative Horde API sources become available)."""
    usage_percentage_of_category: float = Field(ge=0.0, le=100.0, default=0.0)
    """Percentage of category's total monthly usage."""
    usage_trend: UsageTrend
    """Usage trend comparing different time periods."""

    cost_benefit_score: float | None = None
    """Cost-benefit metric: usage per GB (usage_month / size_gb). Only for models with size info."""
    size_gb: float | None = None
    """Model size in gigabytes (if available)."""

    baseline: str | None = None
    """Model baseline (if applicable)."""
    nsfw: bool | None = None
    """Whether model is NSFW (if applicable)."""
    has_description: bool = False
    """Whether model has a description."""
    download_count: int = Field(ge=0, default=0)
    """Number of download entries."""
    download_hosts: list[str] = Field(default_factory=list)
    """List of download host domains."""
    backend_variations: list[BackendVariationStats] | None = Field(default=None)
    """Per-backend statistics for text generation models (ungrouped view)."""

    @property
    def flag_count(self) -> int:
        """Count of deletion risk flags set.

        Returns:
            Number of flags that are True.

        """
        return self.deletion_risk_flags.flag_count()

    @computed_field  # type: ignore[prop-decorator]
    @property
    def is_critical(self) -> bool:
        """Determine if model is in critical state.

        For text_generation: usage_month < threshold AND worker_count < threshold
        For other models: zero month usage AND no active workers (original logic)

        Returns:
            True if model meets critical criteria.

        """
        if self.category == MODEL_REFERENCE_CATEGORY.text_generation:
            usage_threshold = horde_model_reference_settings.text_gen_critical_usage_threshold
            worker_threshold = horde_model_reference_settings.text_gen_critical_worker_threshold
            return self.usage_month < usage_threshold and self.worker_count < worker_threshold
        return self.deletion_risk_flags.zero_usage_month and self.deletion_risk_flags.no_active_workers

    @computed_field  # type: ignore[prop-decorator]
    @property
    def has_warning(self) -> bool:
        """Determine if model has warning-level issues.

        Warnings include host-related issues or download problems.

        Returns:
            True if model has warning-level flags.

        """
        return (
            self.deletion_risk_flags.has_multiple_hosts
            or self.deletion_risk_flags.has_non_preferred_host
            or self.deletion_risk_flags.has_unknown_host
            or self.deletion_risk_flags.no_download_urls
        )

model_config class-attribute instance-attribute

model_config = ConfigDict(use_attribute_docstrings=True)

name instance-attribute

name: str

The model name.

category instance-attribute

category: MODEL_REFERENCE_CATEGORY

The category this model belongs to.

deletion_risk_flags instance-attribute

deletion_risk_flags: DeletionRiskFlags

Flags indicating potential deletion risks.

at_risk instance-attribute

at_risk: bool

True if model has any deletion risk flags.

risk_score class-attribute instance-attribute

risk_score: int = Field(ge=0)

Total number of deletion risk flags (0 = no risk).

worker_count class-attribute instance-attribute

worker_count: int = Field(ge=0, default=0)

Number of active workers serving this model.

usage_day class-attribute instance-attribute

usage_day: int = Field(ge=0, default=0)

Usage count for the past day.

usage_month class-attribute instance-attribute

usage_month: int = Field(ge=0, default=0)

Usage count for the past month.

usage_total class-attribute instance-attribute

usage_total: int = Field(ge=0, default=0)

Total usage count (all time).

usage_hour class-attribute instance-attribute

usage_hour: int | None = None

Usage count for the past hour (will be populated when alternative Horde API sources become available).

usage_minute class-attribute instance-attribute

usage_minute: int | None = None

Usage count for the past minute (will be populated when alternative Horde API sources become available).

usage_percentage_of_category class-attribute instance-attribute

usage_percentage_of_category: float = Field(
    ge=0.0, le=100.0, default=0.0
)

Percentage of category's total monthly usage.

usage_trend instance-attribute

usage_trend: UsageTrend

Usage trend comparing different time periods.

cost_benefit_score class-attribute instance-attribute

cost_benefit_score: float | None = None

Cost-benefit metric: usage per GB (usage_month / size_gb). Only for models with size info.

size_gb class-attribute instance-attribute

size_gb: float | None = None

Model size in gigabytes (if available).

baseline class-attribute instance-attribute

baseline: str | None = None

Model baseline (if applicable).

nsfw class-attribute instance-attribute

nsfw: bool | None = None

Whether model is NSFW (if applicable).

has_description class-attribute instance-attribute

has_description: bool = False

Whether model has a description.

download_count class-attribute instance-attribute

download_count: int = Field(ge=0, default=0)

Number of download entries.

download_hosts class-attribute instance-attribute

download_hosts: list[str] = Field(default_factory=list)

List of download host domains.

backend_variations class-attribute instance-attribute

backend_variations: list[BackendVariationStats] | None = (
    Field(default=None)
)

Per-backend statistics for text generation models (ungrouped view).

flag_count property

flag_count: int

Count of deletion risk flags set.

Returns:

  • int

    Number of flags that are True.

is_critical property

is_critical: bool

Determine if model is in critical state.

For text_generation: usage_month < threshold AND worker_count < threshold For other models: zero month usage AND no active workers (original logic)

Returns:

  • bool

    True if model meets critical criteria.

has_warning property

has_warning: bool

Determine if model has warning-level issues.

Warnings include host-related issues or download problems.

Returns:

  • bool

    True if model has warning-level flags.

CategoryDeletionRiskSummary

Bases: BaseModel

Summary statistics for a category deletion risk analysis.

Aggregates deletion risk information across all models in a category.

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
class CategoryDeletionRiskSummary(BaseModel):
    """Summary statistics for a category deletion risk analysis.

    Aggregates deletion risk information across all models in a category.
    """

    model_config = ConfigDict(use_attribute_docstrings=True)

    total_models: int = Field(ge=0)
    """Total number of models in the category."""
    models_at_risk: int = Field(ge=0)
    """Number of models with at least one deletion risk flag."""
    models_critical: int = Field(ge=0)
    """Number of models in critical state (zero month usage AND no workers)."""
    models_with_warnings: int = Field(ge=0)
    """Number of models with warning-level issues (host/download problems)."""

    models_with_zero_day_usage: int = Field(ge=0, default=0)
    """Number of models with zero usage in the past day."""
    models_with_zero_month_usage: int = Field(ge=0, default=0)
    """Number of models with zero usage in the past month."""
    models_with_zero_total_usage: int = Field(ge=0, default=0)
    """Number of models with zero usage all-time."""
    models_with_no_active_workers: int = Field(ge=0, default=0)
    """Number of models with no active workers."""
    models_with_no_downloads: int = Field(ge=0, default=0)
    """Number of models without download URLs."""
    models_with_non_preferred_hosts: int = Field(ge=0, default=0)
    """Number of models on non-preferred hosts."""
    models_with_multiple_hosts: int = Field(ge=0, default=0)
    """Number of models with downloads across multiple hosts."""
    models_with_low_usage: int = Field(ge=0, default=0)
    """Number of models with very low usage (< 0.1% of category)."""

    average_risk_score: float = Field(ge=0.0)
    """Average risk score across all models."""
    category_total_month_usage: int = Field(ge=0, default=0)
    """Total monthly usage for the entire category."""

    @classmethod
    def from_risk_models(cls, risk_models: list[ModelDeletionRiskInfo]) -> CategoryDeletionRiskSummary:
        """Calculate summary statistics from risk models.

        Args:
            risk_models: List of ModelDeletionRiskInfo objects.

        Returns:
            CategoryDeletionRiskSummary with aggregate statistics.

        """
        total_models = len(risk_models)
        models_at_risk = sum(1 for m in risk_models if m.at_risk)
        models_critical = sum(1 for m in risk_models if m.is_critical)
        models_with_warnings = sum(1 for m in risk_models if m.has_warning)

        models_with_zero_day_usage = sum(1 for m in risk_models if m.deletion_risk_flags.zero_usage_day)
        models_with_zero_month_usage = sum(1 for m in risk_models if m.deletion_risk_flags.zero_usage_month)
        models_with_zero_total_usage = sum(1 for m in risk_models if m.deletion_risk_flags.zero_usage_total)
        models_with_no_active_workers = sum(1 for m in risk_models if m.deletion_risk_flags.no_active_workers)
        models_with_no_downloads = sum(1 for m in risk_models if m.deletion_risk_flags.no_download_urls)
        models_with_non_preferred_hosts = sum(1 for m in risk_models if m.deletion_risk_flags.has_non_preferred_host)
        models_with_multiple_hosts = sum(1 for m in risk_models if m.deletion_risk_flags.has_multiple_hosts)
        models_with_low_usage = sum(1 for m in risk_models if m.deletion_risk_flags.low_usage)

        category_total_month_usage = sum(m.usage_month for m in risk_models)

        total_risk_score = sum(m.risk_score for m in risk_models)
        average_risk_score = total_risk_score / total_models if total_models > 0 else 0.0

        return cls(
            total_models=total_models,
            models_at_risk=models_at_risk,
            models_critical=models_critical,
            models_with_warnings=models_with_warnings,
            models_with_zero_day_usage=models_with_zero_day_usage,
            models_with_zero_month_usage=models_with_zero_month_usage,
            models_with_zero_total_usage=models_with_zero_total_usage,
            models_with_no_active_workers=models_with_no_active_workers,
            models_with_no_downloads=models_with_no_downloads,
            models_with_non_preferred_hosts=models_with_non_preferred_hosts,
            models_with_multiple_hosts=models_with_multiple_hosts,
            models_with_low_usage=models_with_low_usage,
            average_risk_score=round(average_risk_score, 2),
            category_total_month_usage=category_total_month_usage,
        )

model_config class-attribute instance-attribute

model_config = ConfigDict(use_attribute_docstrings=True)

total_models class-attribute instance-attribute

total_models: int = Field(ge=0)

Total number of models in the category.

models_at_risk class-attribute instance-attribute

models_at_risk: int = Field(ge=0)

Number of models with at least one deletion risk flag.

models_critical class-attribute instance-attribute

models_critical: int = Field(ge=0)

Number of models in critical state (zero month usage AND no workers).

models_with_warnings class-attribute instance-attribute

models_with_warnings: int = Field(ge=0)

Number of models with warning-level issues (host/download problems).

models_with_zero_day_usage class-attribute instance-attribute

models_with_zero_day_usage: int = Field(ge=0, default=0)

Number of models with zero usage in the past day.

models_with_zero_month_usage class-attribute instance-attribute

models_with_zero_month_usage: int = Field(ge=0, default=0)

Number of models with zero usage in the past month.

models_with_zero_total_usage class-attribute instance-attribute

models_with_zero_total_usage: int = Field(ge=0, default=0)

Number of models with zero usage all-time.

models_with_no_active_workers class-attribute instance-attribute

models_with_no_active_workers: int = Field(ge=0, default=0)

Number of models with no active workers.

models_with_no_downloads class-attribute instance-attribute

models_with_no_downloads: int = Field(ge=0, default=0)

Number of models without download URLs.

models_with_non_preferred_hosts class-attribute instance-attribute

models_with_non_preferred_hosts: int = Field(
    ge=0, default=0
)

Number of models on non-preferred hosts.

models_with_multiple_hosts class-attribute instance-attribute

models_with_multiple_hosts: int = Field(ge=0, default=0)

Number of models with downloads across multiple hosts.

models_with_low_usage class-attribute instance-attribute

models_with_low_usage: int = Field(ge=0, default=0)

Number of models with very low usage (< 0.1% of category).

average_risk_score class-attribute instance-attribute

average_risk_score: float = Field(ge=0.0)

Average risk score across all models.

category_total_month_usage class-attribute instance-attribute

category_total_month_usage: int = Field(ge=0, default=0)

Total monthly usage for the entire category.

from_risk_models classmethod

from_risk_models(
    risk_models: list[ModelDeletionRiskInfo],
) -> CategoryDeletionRiskSummary

Calculate summary statistics from risk models.

Parameters:

Returns:

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
@classmethod
def from_risk_models(cls, risk_models: list[ModelDeletionRiskInfo]) -> CategoryDeletionRiskSummary:
    """Calculate summary statistics from risk models.

    Args:
        risk_models: List of ModelDeletionRiskInfo objects.

    Returns:
        CategoryDeletionRiskSummary with aggregate statistics.

    """
    total_models = len(risk_models)
    models_at_risk = sum(1 for m in risk_models if m.at_risk)
    models_critical = sum(1 for m in risk_models if m.is_critical)
    models_with_warnings = sum(1 for m in risk_models if m.has_warning)

    models_with_zero_day_usage = sum(1 for m in risk_models if m.deletion_risk_flags.zero_usage_day)
    models_with_zero_month_usage = sum(1 for m in risk_models if m.deletion_risk_flags.zero_usage_month)
    models_with_zero_total_usage = sum(1 for m in risk_models if m.deletion_risk_flags.zero_usage_total)
    models_with_no_active_workers = sum(1 for m in risk_models if m.deletion_risk_flags.no_active_workers)
    models_with_no_downloads = sum(1 for m in risk_models if m.deletion_risk_flags.no_download_urls)
    models_with_non_preferred_hosts = sum(1 for m in risk_models if m.deletion_risk_flags.has_non_preferred_host)
    models_with_multiple_hosts = sum(1 for m in risk_models if m.deletion_risk_flags.has_multiple_hosts)
    models_with_low_usage = sum(1 for m in risk_models if m.deletion_risk_flags.low_usage)

    category_total_month_usage = sum(m.usage_month for m in risk_models)

    total_risk_score = sum(m.risk_score for m in risk_models)
    average_risk_score = total_risk_score / total_models if total_models > 0 else 0.0

    return cls(
        total_models=total_models,
        models_at_risk=models_at_risk,
        models_critical=models_critical,
        models_with_warnings=models_with_warnings,
        models_with_zero_day_usage=models_with_zero_day_usage,
        models_with_zero_month_usage=models_with_zero_month_usage,
        models_with_zero_total_usage=models_with_zero_total_usage,
        models_with_no_active_workers=models_with_no_active_workers,
        models_with_no_downloads=models_with_no_downloads,
        models_with_non_preferred_hosts=models_with_non_preferred_hosts,
        models_with_multiple_hosts=models_with_multiple_hosts,
        models_with_low_usage=models_with_low_usage,
        average_risk_score=round(average_risk_score, 2),
        category_total_month_usage=category_total_month_usage,
    )

CategoryDeletionRiskResponse

Bases: BaseModel

Complete deletion risk response for a category.

Contains both per-model deletion risk information and aggregate summary.

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
class CategoryDeletionRiskResponse(BaseModel):
    """Complete deletion risk response for a category.

    Contains both per-model deletion risk information and aggregate summary.
    """

    model_config = ConfigDict(use_attribute_docstrings=True)

    category: MODEL_REFERENCE_CATEGORY
    """The category being analyzed."""
    category_total_month_usage: int = Field(ge=0)
    """Total monthly usage for the entire category."""

    total_count: int = Field(ge=0, default=0)
    """Total number of models in the category (before pagination/filtering)."""
    returned_count: int = Field(ge=0, default=0)
    """Number of models returned in this response (after pagination/filtering)."""
    offset: int = Field(ge=0, default=0)
    """Starting index for pagination (0 if not paginated)."""
    limit: int | None = None
    """Maximum number of models per page (None if not paginated)."""

    models: list[ModelDeletionRiskInfo]
    """List of deletion risk information for each model."""
    summary: CategoryDeletionRiskSummary
    """Aggregate summary statistics."""

model_config class-attribute instance-attribute

model_config = ConfigDict(use_attribute_docstrings=True)

category instance-attribute

category: MODEL_REFERENCE_CATEGORY

The category being analyzed.

category_total_month_usage class-attribute instance-attribute

category_total_month_usage: int = Field(ge=0)

Total monthly usage for the entire category.

total_count class-attribute instance-attribute

total_count: int = Field(ge=0, default=0)

Total number of models in the category (before pagination/filtering).

returned_count class-attribute instance-attribute

returned_count: int = Field(ge=0, default=0)

Number of models returned in this response (after pagination/filtering).

offset class-attribute instance-attribute

offset: int = Field(ge=0, default=0)

Starting index for pagination (0 if not paginated).

limit class-attribute instance-attribute

limit: int | None = None

Maximum number of models per page (None if not paginated).

models instance-attribute

models: list[ModelDeletionRiskInfo]

List of deletion risk information for each model.

summary instance-attribute

summary: CategoryDeletionRiskSummary

Aggregate summary statistics.

DeletionRiskFlagsHandler

Abstract handler for creating DeletionRiskFlags from specific model record types.

Subclasses should implement type-specific flag generation logic.

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
class DeletionRiskFlagsHandler:
    """Abstract handler for creating DeletionRiskFlags from specific model record types.

    Subclasses should implement type-specific flag generation logic.
    """

    def can_handle(self, model_record: GenericModelRecord) -> bool:
        """Check if this handler can process the given model record.

        Args:
            model_record: The model record to check.

        Returns:
            True if this handler can process the model record type.

        """
        raise NotImplementedError("Subclasses must implement can_handle")

    def create_flags(
        self,
        *,
        model_record: GenericModelRecord,
        statistics: CombinedModelStatistics | None,
        category_total_usage: int,
    ) -> DeletionRiskFlags:
        """Create DeletionRiskFlags for a model record.

        Args:
            model_record: The model record.
            statistics: Optional Horde API statistics.
            category_total_usage: Total monthly usage for the category.

        Returns:
            DeletionRiskFlags object.

        """
        raise NotImplementedError("Subclasses must implement create_flags")

can_handle

can_handle(model_record: GenericModelRecord) -> bool

Check if this handler can process the given model record.

Parameters:

Returns:

  • bool

    True if this handler can process the model record type.

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
def can_handle(self, model_record: GenericModelRecord) -> bool:
    """Check if this handler can process the given model record.

    Args:
        model_record: The model record to check.

    Returns:
        True if this handler can process the model record type.

    """
    raise NotImplementedError("Subclasses must implement can_handle")

create_flags

create_flags(
    *,
    model_record: GenericModelRecord,
    statistics: CombinedModelStatistics | None,
    category_total_usage: int,
) -> DeletionRiskFlags

Create DeletionRiskFlags for a model record.

Parameters:

Returns:

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
def create_flags(
    self,
    *,
    model_record: GenericModelRecord,
    statistics: CombinedModelStatistics | None,
    category_total_usage: int,
) -> DeletionRiskFlags:
    """Create DeletionRiskFlags for a model record.

    Args:
        model_record: The model record.
        statistics: Optional Horde API statistics.
        category_total_usage: Total monthly usage for the category.

    Returns:
        DeletionRiskFlags object.

    """
    raise NotImplementedError("Subclasses must implement create_flags")

ImageGenerationDeletionRiskFlagsHandler

Bases: DeletionRiskFlagsHandler

Handler for image generation model deletion risk flags creation.

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
class ImageGenerationDeletionRiskFlagsHandler(DeletionRiskFlagsHandler):
    """Handler for image generation model deletion risk flags creation."""

    def can_handle(self, model_record: GenericModelRecord) -> bool:
        """Check if this handler can process the given model record.

        Args:
            model_record: The model record to check.

        Returns:
            True if the model record is an ImageGenerationModelRecord.

        """
        return isinstance(model_record, ImageGenerationModelRecord)

    def _create_flags_impl(
        self,
        model_record: ImageGenerationModelRecord,
        statistics: CombinedModelStatistics | None,
        category_total_usage: int,
    ) -> DeletionRiskFlags:
        """Analyze an image generation model and determine deletion risk flags.

        Args:
            model_record: Typed image generation model record.
            statistics: Optional Horde API statistics (worker_count, usage_stats).
            category_total_usage: Total monthly usage for the category (for percentage calculations).

        Returns:
            DeletionRiskFlags with appropriate flags set.

        """
        downloads = model_record.config.download if model_record.config else []

        return (
            DeletionRiskFlagsBuilder()
            .with_download_flags(
                *FlagValidatorService.validate_downloads(downloads, category=MODEL_REFERENCE_CATEGORY.image_generation)
            )
            .with_missing_description(FlagValidatorService.validate_description(model_record.description))
            .with_missing_baseline(FlagValidatorService.validate_baseline(model_record.baseline))
            .with_statistics_flags(
                *FlagValidatorService.validate_statistics(
                    statistics, category_total_usage, category=MODEL_REFERENCE_CATEGORY.image_generation
                )
            )
            .build()
        )

    def create_flags(
        self,
        *,
        model_record: GenericModelRecord,
        statistics: CombinedModelStatistics | None,
        category_total_usage: int,
    ) -> DeletionRiskFlags:
        """Create DeletionRiskFlags for an image generation model.

        Args:
            model_record: The image generation model record.
            statistics: Optional Horde API statistics.
            category_total_usage: Total monthly usage for the category.

        Returns:
            DeletionRiskFlags object.

        """
        if not isinstance(model_record, ImageGenerationModelRecord):
            error_message = f"Expected ImageGenerationModelRecord, got {type(model_record).__name__}"
            raise TypeError(error_message)

        return self._create_flags_impl(model_record, statistics, category_total_usage)

can_handle

can_handle(model_record: GenericModelRecord) -> bool

Check if this handler can process the given model record.

Parameters:

Returns:

  • bool

    True if the model record is an ImageGenerationModelRecord.

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
def can_handle(self, model_record: GenericModelRecord) -> bool:
    """Check if this handler can process the given model record.

    Args:
        model_record: The model record to check.

    Returns:
        True if the model record is an ImageGenerationModelRecord.

    """
    return isinstance(model_record, ImageGenerationModelRecord)

_create_flags_impl

_create_flags_impl(
    model_record: ImageGenerationModelRecord,
    statistics: CombinedModelStatistics | None,
    category_total_usage: int,
) -> DeletionRiskFlags

Analyze an image generation model and determine deletion risk flags.

Parameters:

  • model_record (ImageGenerationModelRecord) –

    Typed image generation model record.

  • statistics (CombinedModelStatistics | None) –

    Optional Horde API statistics (worker_count, usage_stats).

  • category_total_usage (int) –

    Total monthly usage for the category (for percentage calculations).

Returns:

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
def _create_flags_impl(
    self,
    model_record: ImageGenerationModelRecord,
    statistics: CombinedModelStatistics | None,
    category_total_usage: int,
) -> DeletionRiskFlags:
    """Analyze an image generation model and determine deletion risk flags.

    Args:
        model_record: Typed image generation model record.
        statistics: Optional Horde API statistics (worker_count, usage_stats).
        category_total_usage: Total monthly usage for the category (for percentage calculations).

    Returns:
        DeletionRiskFlags with appropriate flags set.

    """
    downloads = model_record.config.download if model_record.config else []

    return (
        DeletionRiskFlagsBuilder()
        .with_download_flags(
            *FlagValidatorService.validate_downloads(downloads, category=MODEL_REFERENCE_CATEGORY.image_generation)
        )
        .with_missing_description(FlagValidatorService.validate_description(model_record.description))
        .with_missing_baseline(FlagValidatorService.validate_baseline(model_record.baseline))
        .with_statistics_flags(
            *FlagValidatorService.validate_statistics(
                statistics, category_total_usage, category=MODEL_REFERENCE_CATEGORY.image_generation
            )
        )
        .build()
    )

create_flags

create_flags(
    *,
    model_record: GenericModelRecord,
    statistics: CombinedModelStatistics | None,
    category_total_usage: int,
) -> DeletionRiskFlags

Create DeletionRiskFlags for an image generation model.

Parameters:

  • model_record (GenericModelRecord) –

    The image generation model record.

  • statistics (CombinedModelStatistics | None) –

    Optional Horde API statistics.

  • category_total_usage (int) –

    Total monthly usage for the category.

Returns:

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
def create_flags(
    self,
    *,
    model_record: GenericModelRecord,
    statistics: CombinedModelStatistics | None,
    category_total_usage: int,
) -> DeletionRiskFlags:
    """Create DeletionRiskFlags for an image generation model.

    Args:
        model_record: The image generation model record.
        statistics: Optional Horde API statistics.
        category_total_usage: Total monthly usage for the category.

    Returns:
        DeletionRiskFlags object.

    """
    if not isinstance(model_record, ImageGenerationModelRecord):
        error_message = f"Expected ImageGenerationModelRecord, got {type(model_record).__name__}"
        raise TypeError(error_message)

    return self._create_flags_impl(model_record, statistics, category_total_usage)

TextGenerationDeletionRiskFlagsHandler

Bases: DeletionRiskFlagsHandler

Handler for text generation model deletion risk flags creation.

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
class TextGenerationDeletionRiskFlagsHandler(DeletionRiskFlagsHandler):
    """Handler for text generation model deletion risk flags creation."""

    def can_handle(self, model_record: GenericModelRecord) -> bool:
        """Check if this handler can process the given model record.

        Args:
            model_record: The model record to check.

        Returns:
            True if the model record is a TextGenerationModelRecord.

        """
        return isinstance(model_record, TextGenerationModelRecord)

    def _create_flags_impl(
        self,
        model_record: TextGenerationModelRecord,
        statistics: CombinedModelStatistics | None,
        category_total_usage: int,
    ) -> DeletionRiskFlags:
        """Analyze a text generation model and determine deletion risk flags.

        Args:
            model_record: Typed text generation model record.
            statistics: Optional Horde API statistics (worker_count, usage_stats).
            category_total_usage: Total monthly usage for the category (for percentage calculations).

        Returns:
            DeletionRiskFlags with appropriate flags set.

        """
        downloads = model_record.config.download if model_record.config else []

        return (
            DeletionRiskFlagsBuilder()
            .with_download_flags(
                *FlagValidatorService.validate_downloads(downloads, category=MODEL_REFERENCE_CATEGORY.text_generation)
            )
            .with_missing_description(FlagValidatorService.validate_description(model_record.description))
            .with_missing_baseline(FlagValidatorService.validate_baseline(model_record.baseline))
            .with_statistics_flags(
                *FlagValidatorService.validate_statistics(
                    statistics, category_total_usage, category=MODEL_REFERENCE_CATEGORY.text_generation
                )
            )
            .build()
        )

    def create_flags(
        self,
        *,
        model_record: GenericModelRecord,
        statistics: CombinedModelStatistics | None,
        category_total_usage: int,
    ) -> DeletionRiskFlags:
        """Create DeletionRiskFlags for a text generation model.

        Args:
            model_record: The text generation model record.
            statistics: Optional Horde API statistics.
            category_total_usage: Total monthly usage for the category.

        Returns:
            DeletionRiskFlags object.

        """
        if not isinstance(model_record, TextGenerationModelRecord):
            error_message = f"Expected TextGenerationModelRecord, got {type(model_record).__name__}"
            raise TypeError(error_message)

        return self._create_flags_impl(model_record, statistics, category_total_usage)

can_handle

can_handle(model_record: GenericModelRecord) -> bool

Check if this handler can process the given model record.

Parameters:

Returns:

  • bool

    True if the model record is a TextGenerationModelRecord.

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
def can_handle(self, model_record: GenericModelRecord) -> bool:
    """Check if this handler can process the given model record.

    Args:
        model_record: The model record to check.

    Returns:
        True if the model record is a TextGenerationModelRecord.

    """
    return isinstance(model_record, TextGenerationModelRecord)

_create_flags_impl

_create_flags_impl(
    model_record: TextGenerationModelRecord,
    statistics: CombinedModelStatistics | None,
    category_total_usage: int,
) -> DeletionRiskFlags

Analyze a text generation model and determine deletion risk flags.

Parameters:

  • model_record (TextGenerationModelRecord) –

    Typed text generation model record.

  • statistics (CombinedModelStatistics | None) –

    Optional Horde API statistics (worker_count, usage_stats).

  • category_total_usage (int) –

    Total monthly usage for the category (for percentage calculations).

Returns:

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
def _create_flags_impl(
    self,
    model_record: TextGenerationModelRecord,
    statistics: CombinedModelStatistics | None,
    category_total_usage: int,
) -> DeletionRiskFlags:
    """Analyze a text generation model and determine deletion risk flags.

    Args:
        model_record: Typed text generation model record.
        statistics: Optional Horde API statistics (worker_count, usage_stats).
        category_total_usage: Total monthly usage for the category (for percentage calculations).

    Returns:
        DeletionRiskFlags with appropriate flags set.

    """
    downloads = model_record.config.download if model_record.config else []

    return (
        DeletionRiskFlagsBuilder()
        .with_download_flags(
            *FlagValidatorService.validate_downloads(downloads, category=MODEL_REFERENCE_CATEGORY.text_generation)
        )
        .with_missing_description(FlagValidatorService.validate_description(model_record.description))
        .with_missing_baseline(FlagValidatorService.validate_baseline(model_record.baseline))
        .with_statistics_flags(
            *FlagValidatorService.validate_statistics(
                statistics, category_total_usage, category=MODEL_REFERENCE_CATEGORY.text_generation
            )
        )
        .build()
    )

create_flags

create_flags(
    *,
    model_record: GenericModelRecord,
    statistics: CombinedModelStatistics | None,
    category_total_usage: int,
) -> DeletionRiskFlags

Create DeletionRiskFlags for a text generation model.

Parameters:

  • model_record (GenericModelRecord) –

    The text generation model record.

  • statistics (CombinedModelStatistics | None) –

    Optional Horde API statistics.

  • category_total_usage (int) –

    Total monthly usage for the category.

Returns:

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
def create_flags(
    self,
    *,
    model_record: GenericModelRecord,
    statistics: CombinedModelStatistics | None,
    category_total_usage: int,
) -> DeletionRiskFlags:
    """Create DeletionRiskFlags for a text generation model.

    Args:
        model_record: The text generation model record.
        statistics: Optional Horde API statistics.
        category_total_usage: Total monthly usage for the category.

    Returns:
        DeletionRiskFlags object.

    """
    if not isinstance(model_record, TextGenerationModelRecord):
        error_message = f"Expected TextGenerationModelRecord, got {type(model_record).__name__}"
        raise TypeError(error_message)

    return self._create_flags_impl(model_record, statistics, category_total_usage)

GenericDeletionRiskFlagsHandler

Bases: DeletionRiskFlagsHandler

Fallback handler for unsupported model record types.

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
class GenericDeletionRiskFlagsHandler(DeletionRiskFlagsHandler):
    """Fallback handler for unsupported model record types."""

    def can_handle(self, model_record: GenericModelRecord) -> bool:
        """Check if this handler can process the given model record.

        This handler accepts all model records as a fallback.

        Args:
            model_record: The model record to check.

        Returns:
            True (accepts all model records).

        """
        return True

    def create_flags(
        self,
        *,
        model_record: GenericModelRecord,
        statistics: CombinedModelStatistics | None,
        category_total_usage: int,
    ) -> DeletionRiskFlags:
        """Create DeletionRiskFlags for a generic/unsupported model type.

        Args:
            model_record: The generic model record.
            statistics: Optional Horde API statistics.
            category_total_usage: Total monthly usage for the category.

        Returns:
            DeletionRiskFlags object.

        """
        logger.warning(f"Using fallback handler for unsupported model type: {type(model_record).__name__}")

        downloads = model_record.config.download if model_record.config else []

        return (
            DeletionRiskFlagsBuilder()
            .with_download_flags(*FlagValidatorService.validate_downloads(downloads))
            .with_missing_description(FlagValidatorService.validate_description(model_record.description))
            .with_statistics_flags(*FlagValidatorService.validate_statistics(statistics, category_total_usage))
            .build()
        )

can_handle

can_handle(model_record: GenericModelRecord) -> bool

Check if this handler can process the given model record.

This handler accepts all model records as a fallback.

Parameters:

Returns:

  • bool

    True (accepts all model records).

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
def can_handle(self, model_record: GenericModelRecord) -> bool:
    """Check if this handler can process the given model record.

    This handler accepts all model records as a fallback.

    Args:
        model_record: The model record to check.

    Returns:
        True (accepts all model records).

    """
    return True

create_flags

create_flags(
    *,
    model_record: GenericModelRecord,
    statistics: CombinedModelStatistics | None,
    category_total_usage: int,
) -> DeletionRiskFlags

Create DeletionRiskFlags for a generic/unsupported model type.

Parameters:

Returns:

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
def create_flags(
    self,
    *,
    model_record: GenericModelRecord,
    statistics: CombinedModelStatistics | None,
    category_total_usage: int,
) -> DeletionRiskFlags:
    """Create DeletionRiskFlags for a generic/unsupported model type.

    Args:
        model_record: The generic model record.
        statistics: Optional Horde API statistics.
        category_total_usage: Total monthly usage for the category.

    Returns:
        DeletionRiskFlags object.

    """
    logger.warning(f"Using fallback handler for unsupported model type: {type(model_record).__name__}")

    downloads = model_record.config.download if model_record.config else []

    return (
        DeletionRiskFlagsBuilder()
        .with_download_flags(*FlagValidatorService.validate_downloads(downloads))
        .with_missing_description(FlagValidatorService.validate_description(model_record.description))
        .with_statistics_flags(*FlagValidatorService.validate_statistics(statistics, category_total_usage))
        .build()
    )

DeletionRiskFlagsFactory

Factory for creating DeletionRiskFlags objects with extensible handler support.

Handlers are registered and checked in order. The first handler that can process a model record type will be used to create the deletion risk flags.

Examples:

# Using default handlers
factory = DeletionRiskFlagsFactory.create_default()
flags = factory.create_flags(
    model_record=image_model_record,
    statistics=stats,
    category_total_usage=10000,
)

# Adding custom handler
factory.register_handler(CustomDeletionRiskFlagsHandler())
Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
class DeletionRiskFlagsFactory:
    """Factory for creating DeletionRiskFlags objects with extensible handler support.

    Handlers are registered and checked in order. The first handler that can process
    a model record type will be used to create the deletion risk flags.

    Examples:
        ```python
        # Using default handlers
        factory = DeletionRiskFlagsFactory.create_default()
        flags = factory.create_flags(
            model_record=image_model_record,
            statistics=stats,
            category_total_usage=10000,
        )

        # Adding custom handler
        factory.register_handler(CustomDeletionRiskFlagsHandler())
        ```

    """

    def __init__(self, handlers: list[DeletionRiskFlagsHandler] | None = None) -> None:
        """Initialize the factory with optional handlers.

        Args:
            handlers: List of handlers to use. If None, no handlers are registered.

        """
        self._handlers: list[DeletionRiskFlagsHandler] = handlers or []

    @classmethod
    def create_default(cls) -> DeletionRiskFlagsFactory:
        """Create a factory with default handlers for standard model types.

        Returns:
            DeletionRiskFlagsFactory with default handlers registered.

        """
        return cls(
            handlers=[
                ImageGenerationDeletionRiskFlagsHandler(),
                TextGenerationDeletionRiskFlagsHandler(),
                GenericDeletionRiskFlagsHandler(),  # Fallback handler (must be last)
            ]
        )

    def register_handler(self, handler: DeletionRiskFlagsHandler) -> None:
        """Register a new handler.

        Handlers are checked in registration order. Register more specific handlers
        before generic ones.

        Args:
            handler: The handler to register.

        """
        self._handlers.append(handler)

    def create_flags(
        self,
        *,
        model_record: GenericModelRecord,
        statistics: CombinedModelStatistics | None,
        category_total_usage: int,
    ) -> DeletionRiskFlags:
        """Create DeletionRiskFlags for a model record using the appropriate handler.

        Args:
            model_record: The model record.
            statistics: Optional Horde API statistics.
            category_total_usage: Total monthly usage for the category.

        Returns:
            DeletionRiskFlags object.

        Raises:
            ValueError: If no handler can process the model record type.

        """
        for handler in self._handlers:
            if handler.can_handle(model_record):
                return handler.create_flags(
                    model_record=model_record,
                    statistics=statistics,
                    category_total_usage=category_total_usage,
                )

        error_message = f"No handler found for model record type: {type(model_record).__name__}"
        raise ValueError(error_message)

_handlers instance-attribute

_handlers: list[DeletionRiskFlagsHandler] = handlers or []

__init__

__init__(
    handlers: list[DeletionRiskFlagsHandler] | None = None,
) -> None

Initialize the factory with optional handlers.

Parameters:

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
def __init__(self, handlers: list[DeletionRiskFlagsHandler] | None = None) -> None:
    """Initialize the factory with optional handlers.

    Args:
        handlers: List of handlers to use. If None, no handlers are registered.

    """
    self._handlers: list[DeletionRiskFlagsHandler] = handlers or []

create_default classmethod

create_default() -> DeletionRiskFlagsFactory

Create a factory with default handlers for standard model types.

Returns:

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
@classmethod
def create_default(cls) -> DeletionRiskFlagsFactory:
    """Create a factory with default handlers for standard model types.

    Returns:
        DeletionRiskFlagsFactory with default handlers registered.

    """
    return cls(
        handlers=[
            ImageGenerationDeletionRiskFlagsHandler(),
            TextGenerationDeletionRiskFlagsHandler(),
            GenericDeletionRiskFlagsHandler(),  # Fallback handler (must be last)
        ]
    )

register_handler

register_handler(handler: DeletionRiskFlagsHandler) -> None

Register a new handler.

Handlers are checked in registration order. Register more specific handlers before generic ones.

Parameters:

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
def register_handler(self, handler: DeletionRiskFlagsHandler) -> None:
    """Register a new handler.

    Handlers are checked in registration order. Register more specific handlers
    before generic ones.

    Args:
        handler: The handler to register.

    """
    self._handlers.append(handler)

create_flags

create_flags(
    *,
    model_record: GenericModelRecord,
    statistics: CombinedModelStatistics | None,
    category_total_usage: int,
) -> DeletionRiskFlags

Create DeletionRiskFlags for a model record using the appropriate handler.

Parameters:

Returns:

Raises:

  • ValueError

    If no handler can process the model record type.

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
def create_flags(
    self,
    *,
    model_record: GenericModelRecord,
    statistics: CombinedModelStatistics | None,
    category_total_usage: int,
) -> DeletionRiskFlags:
    """Create DeletionRiskFlags for a model record using the appropriate handler.

    Args:
        model_record: The model record.
        statistics: Optional Horde API statistics.
        category_total_usage: Total monthly usage for the category.

    Returns:
        DeletionRiskFlags object.

    Raises:
        ValueError: If no handler can process the model record type.

    """
    for handler in self._handlers:
        if handler.can_handle(model_record):
            return handler.create_flags(
                model_record=model_record,
                statistics=statistics,
                category_total_usage=category_total_usage,
            )

    error_message = f"No handler found for model record type: {type(model_record).__name__}"
    raise ValueError(error_message)

ModelDeletionRiskInfoHandler

Abstract handler for creating ModelDeletionRiskInfo from specific model record types.

Subclasses should implement type-specific extraction and flag generation logic.

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
class ModelDeletionRiskInfoHandler:
    """Abstract handler for creating ModelDeletionRiskInfo from specific model record types.

    Subclasses should implement type-specific extraction and flag generation logic.
    """

    def can_handle(self, model_record: GenericModelRecord) -> bool:
        """Check if this handler can process the given model record.

        Args:
            model_record: The model record to check.

        Returns:
            True if this handler can process the model record type.

        """
        raise NotImplementedError("Subclasses must implement can_handle")

    def create_risk_info(
        self,
        *,
        model_name: str,
        model_record: GenericModelRecord,
        statistics: CombinedModelStatistics | None,
        category_total_usage: int,
        category: MODEL_REFERENCE_CATEGORY,
        include_backend_variations: bool = False,
    ) -> ModelDeletionRiskInfo:
        """Create ModelDeletionRiskInfo for a model record.

        Args:
            model_name: The model name.
            model_record: The model record.
            statistics: Optional Horde API statistics.
            category_total_usage: Total monthly usage for the category.
            category: The model reference category.
            include_backend_variations: Whether to include per-backend breakdown (text models only).

        Returns:
            ModelDeletionRiskInfo object.

        """
        raise NotImplementedError("Subclasses must implement create_risk_info")

    @staticmethod
    def _build_risk_info(
        *,
        model_name: str,
        model_record: GenericModelRecord,
        statistics: CombinedModelStatistics | None,
        category_total_usage: int,
        category: MODEL_REFERENCE_CATEGORY,
        flags: DeletionRiskFlags,
        baseline: str | None,
        nsfw: bool | None,
        size_bytes: int | None,
        include_backend_variations: bool = False,
    ) -> ModelDeletionRiskInfo:
        """Build ModelDeletionRiskInfo from common components.

        Args:
            model_name: The model name.
            model_record: Any model record type.
            statistics: Optional Horde API statistics.
            category_total_usage: Total monthly usage for the category.
            category: The model reference category.
            flags: Deletion risk flags.
            baseline: Model baseline (if applicable).
            nsfw: Whether model is NSFW (if applicable).
            size_bytes: Model size in bytes (if available).
            include_backend_variations: Whether to include per-backend breakdown.

        Returns:
            ModelDeletionRiskInfo object.

        """
        # Extract Horde API data from statistics
        worker_count = statistics.worker_count if statistics else 0
        usage_day = statistics.usage_stats.day if statistics and statistics.usage_stats else 0
        usage_month = statistics.usage_stats.month if statistics and statistics.usage_stats else 0
        usage_total = statistics.usage_stats.total if statistics and statistics.usage_stats else 0
        usage_hour = None
        usage_minute = None

        # Calculate usage percentage
        usage_percentage = 0.0
        if category_total_usage > 0:
            usage_percentage = (usage_month / category_total_usage) * 100.0

        # Calculate usage trend ratios
        day_to_month_ratio: float | None = None
        if usage_month > 0:
            day_to_month_ratio = usage_day / usage_month

        month_to_total_ratio: float | None = None
        if usage_total > 0:
            month_to_total_ratio = usage_month / usage_total

        usage_trend = UsageTrend(
            day_to_month_ratio=day_to_month_ratio,
            month_to_total_ratio=month_to_total_ratio,
        )

        # Check description
        description = model_record.description
        has_description = bool(description and len(description.strip()) > 0)

        # Calculate size in GB and cost-benefit score
        size_gb: float | None = None
        cost_benefit_score: float | None = None

        if size_bytes and size_bytes > 0:
            size_gb = size_bytes / (1024**3)
            if size_gb > 0:
                cost_benefit_score = usage_month / size_gb

        # Extract download information
        downloads = model_record.config.download if model_record.config else []
        download_count = len(downloads)

        download_hosts = []
        for download in downloads:
            url = download.file_url
            if url:
                try:
                    parsed = urlparse(url)
                    if parsed.netloc and parsed.netloc not in download_hosts:
                        download_hosts.append(parsed.netloc)
                except Exception:
                    pass

        # Build backend variations list if requested and available
        backend_variations_list: list[BackendVariationStats] | None = None
        if include_backend_variations and statistics and statistics.backend_variations:
            backend_variations_list = [
                BackendVariationStats(
                    backend=bv.backend,
                    variant_name=bv.variant_name,
                    worker_count=bv.worker_count,
                    performance=bv.performance,
                    usage_day=bv.usage_day,
                    usage_month=bv.usage_month,
                    usage_total=bv.usage_total,
                )
                for bv in statistics.backend_variations.values()
            ]

        # Create risk info
        return ModelDeletionRiskInfo(
            name=model_name,
            category=category,
            deletion_risk_flags=flags,
            at_risk=flags.any_flags(),
            risk_score=flags.flag_count(),
            worker_count=worker_count,
            usage_day=usage_day,
            usage_month=usage_month,
            usage_total=usage_total,
            usage_hour=usage_hour,
            usage_minute=usage_minute,
            usage_percentage_of_category=round(usage_percentage, 4),
            usage_trend=usage_trend,
            cost_benefit_score=round(cost_benefit_score, 2) if cost_benefit_score is not None else None,
            size_gb=round(size_gb, 2) if size_gb is not None else None,
            baseline=baseline,
            nsfw=nsfw,
            has_description=has_description,
            download_count=download_count,
            download_hosts=download_hosts,
            backend_variations=backend_variations_list,
        )

can_handle

can_handle(model_record: GenericModelRecord) -> bool

Check if this handler can process the given model record.

Parameters:

Returns:

  • bool

    True if this handler can process the model record type.

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
def can_handle(self, model_record: GenericModelRecord) -> bool:
    """Check if this handler can process the given model record.

    Args:
        model_record: The model record to check.

    Returns:
        True if this handler can process the model record type.

    """
    raise NotImplementedError("Subclasses must implement can_handle")

create_risk_info

create_risk_info(
    *,
    model_name: str,
    model_record: GenericModelRecord,
    statistics: CombinedModelStatistics | None,
    category_total_usage: int,
    category: MODEL_REFERENCE_CATEGORY,
    include_backend_variations: bool = False,
) -> ModelDeletionRiskInfo

Create ModelDeletionRiskInfo for a model record.

Parameters:

  • model_name (str) –

    The model name.

  • model_record (GenericModelRecord) –

    The model record.

  • statistics (CombinedModelStatistics | None) –

    Optional Horde API statistics.

  • category_total_usage (int) –

    Total monthly usage for the category.

  • category (MODEL_REFERENCE_CATEGORY) –

    The model reference category.

  • include_backend_variations (bool, default: False ) –

    Whether to include per-backend breakdown (text models only).

Returns:

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
def create_risk_info(
    self,
    *,
    model_name: str,
    model_record: GenericModelRecord,
    statistics: CombinedModelStatistics | None,
    category_total_usage: int,
    category: MODEL_REFERENCE_CATEGORY,
    include_backend_variations: bool = False,
) -> ModelDeletionRiskInfo:
    """Create ModelDeletionRiskInfo for a model record.

    Args:
        model_name: The model name.
        model_record: The model record.
        statistics: Optional Horde API statistics.
        category_total_usage: Total monthly usage for the category.
        category: The model reference category.
        include_backend_variations: Whether to include per-backend breakdown (text models only).

    Returns:
        ModelDeletionRiskInfo object.

    """
    raise NotImplementedError("Subclasses must implement create_risk_info")

_build_risk_info staticmethod

_build_risk_info(
    *,
    model_name: str,
    model_record: GenericModelRecord,
    statistics: CombinedModelStatistics | None,
    category_total_usage: int,
    category: MODEL_REFERENCE_CATEGORY,
    flags: DeletionRiskFlags,
    baseline: str | None,
    nsfw: bool | None,
    size_bytes: int | None,
    include_backend_variations: bool = False,
) -> ModelDeletionRiskInfo

Build ModelDeletionRiskInfo from common components.

Parameters:

  • model_name (str) –

    The model name.

  • model_record (GenericModelRecord) –

    Any model record type.

  • statistics (CombinedModelStatistics | None) –

    Optional Horde API statistics.

  • category_total_usage (int) –

    Total monthly usage for the category.

  • category (MODEL_REFERENCE_CATEGORY) –

    The model reference category.

  • flags (DeletionRiskFlags) –

    Deletion risk flags.

  • baseline (str | None) –

    Model baseline (if applicable).

  • nsfw (bool | None) –

    Whether model is NSFW (if applicable).

  • size_bytes (int | None) –

    Model size in bytes (if available).

  • include_backend_variations (bool, default: False ) –

    Whether to include per-backend breakdown.

Returns:

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
@staticmethod
def _build_risk_info(
    *,
    model_name: str,
    model_record: GenericModelRecord,
    statistics: CombinedModelStatistics | None,
    category_total_usage: int,
    category: MODEL_REFERENCE_CATEGORY,
    flags: DeletionRiskFlags,
    baseline: str | None,
    nsfw: bool | None,
    size_bytes: int | None,
    include_backend_variations: bool = False,
) -> ModelDeletionRiskInfo:
    """Build ModelDeletionRiskInfo from common components.

    Args:
        model_name: The model name.
        model_record: Any model record type.
        statistics: Optional Horde API statistics.
        category_total_usage: Total monthly usage for the category.
        category: The model reference category.
        flags: Deletion risk flags.
        baseline: Model baseline (if applicable).
        nsfw: Whether model is NSFW (if applicable).
        size_bytes: Model size in bytes (if available).
        include_backend_variations: Whether to include per-backend breakdown.

    Returns:
        ModelDeletionRiskInfo object.

    """
    # Extract Horde API data from statistics
    worker_count = statistics.worker_count if statistics else 0
    usage_day = statistics.usage_stats.day if statistics and statistics.usage_stats else 0
    usage_month = statistics.usage_stats.month if statistics and statistics.usage_stats else 0
    usage_total = statistics.usage_stats.total if statistics and statistics.usage_stats else 0
    usage_hour = None
    usage_minute = None

    # Calculate usage percentage
    usage_percentage = 0.0
    if category_total_usage > 0:
        usage_percentage = (usage_month / category_total_usage) * 100.0

    # Calculate usage trend ratios
    day_to_month_ratio: float | None = None
    if usage_month > 0:
        day_to_month_ratio = usage_day / usage_month

    month_to_total_ratio: float | None = None
    if usage_total > 0:
        month_to_total_ratio = usage_month / usage_total

    usage_trend = UsageTrend(
        day_to_month_ratio=day_to_month_ratio,
        month_to_total_ratio=month_to_total_ratio,
    )

    # Check description
    description = model_record.description
    has_description = bool(description and len(description.strip()) > 0)

    # Calculate size in GB and cost-benefit score
    size_gb: float | None = None
    cost_benefit_score: float | None = None

    if size_bytes and size_bytes > 0:
        size_gb = size_bytes / (1024**3)
        if size_gb > 0:
            cost_benefit_score = usage_month / size_gb

    # Extract download information
    downloads = model_record.config.download if model_record.config else []
    download_count = len(downloads)

    download_hosts = []
    for download in downloads:
        url = download.file_url
        if url:
            try:
                parsed = urlparse(url)
                if parsed.netloc and parsed.netloc not in download_hosts:
                    download_hosts.append(parsed.netloc)
            except Exception:
                pass

    # Build backend variations list if requested and available
    backend_variations_list: list[BackendVariationStats] | None = None
    if include_backend_variations and statistics and statistics.backend_variations:
        backend_variations_list = [
            BackendVariationStats(
                backend=bv.backend,
                variant_name=bv.variant_name,
                worker_count=bv.worker_count,
                performance=bv.performance,
                usage_day=bv.usage_day,
                usage_month=bv.usage_month,
                usage_total=bv.usage_total,
            )
            for bv in statistics.backend_variations.values()
        ]

    # Create risk info
    return ModelDeletionRiskInfo(
        name=model_name,
        category=category,
        deletion_risk_flags=flags,
        at_risk=flags.any_flags(),
        risk_score=flags.flag_count(),
        worker_count=worker_count,
        usage_day=usage_day,
        usage_month=usage_month,
        usage_total=usage_total,
        usage_hour=usage_hour,
        usage_minute=usage_minute,
        usage_percentage_of_category=round(usage_percentage, 4),
        usage_trend=usage_trend,
        cost_benefit_score=round(cost_benefit_score, 2) if cost_benefit_score is not None else None,
        size_gb=round(size_gb, 2) if size_gb is not None else None,
        baseline=baseline,
        nsfw=nsfw,
        has_description=has_description,
        download_count=download_count,
        download_hosts=download_hosts,
        backend_variations=backend_variations_list,
    )

ImageGenerationModelDeletionRiskHandler

Bases: ModelDeletionRiskInfoHandler

Handler for image generation model deletion risk info creation.

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
class ImageGenerationModelDeletionRiskHandler(ModelDeletionRiskInfoHandler):
    """Handler for image generation model deletion risk info creation."""

    def __init__(self, flags_factory: DeletionRiskFlagsFactory | None = None) -> None:
        """Initialize the handler with optional flags factory.

        Args:
            flags_factory: Optional factory for creating deletion risk flags.
                If None, uses default factory.

        """
        self._flags_factory = flags_factory or DeletionRiskFlagsFactory.create_default()

    def can_handle(self, model_record: GenericModelRecord) -> bool:
        """Check if this handler can process the given model record.

        Args:
            model_record: The model record to check.

        Returns:
            True if the model record is an ImageGenerationModelRecord.

        """
        return isinstance(model_record, ImageGenerationModelRecord)

    def create_risk_info(
        self,
        *,
        model_name: str,
        model_record: GenericModelRecord,
        statistics: CombinedModelStatistics | None,
        category_total_usage: int,
        category: MODEL_REFERENCE_CATEGORY,
        include_backend_variations: bool = False,
    ) -> ModelDeletionRiskInfo:
        """Create ModelDeletionRiskInfo for an image generation model.

        Args:
            model_name: The model name.
            model_record: The image generation model record.
            statistics: Optional Horde API statistics.
            category_total_usage: Total monthly usage for the category.
            category: The model reference category.
            include_backend_variations: Ignored for image models (always False).

        Returns:
            ModelDeletionRiskInfo object.

        """
        if not isinstance(model_record, ImageGenerationModelRecord):
            error_message = f"Expected ImageGenerationModelRecord, got {type(model_record).__name__}"
            raise TypeError(error_message)

        flags = self._flags_factory.create_flags(
            model_record=model_record,
            statistics=statistics,
            category_total_usage=category_total_usage,
        )
        baseline = str(model_record.baseline) if model_record.baseline else None
        nsfw = model_record.nsfw
        size_bytes = model_record.size_on_disk_bytes

        return ModelDeletionRiskInfoHandler._build_risk_info(
            model_name=model_name,
            model_record=model_record,
            statistics=statistics,
            category_total_usage=category_total_usage,
            category=category,
            flags=flags,
            baseline=baseline,
            nsfw=nsfw,
            size_bytes=size_bytes,
            include_backend_variations=False,  # Not applicable for image models
        )

_flags_factory instance-attribute

_flags_factory = flags_factory or create_default()

__init__

__init__(
    flags_factory: DeletionRiskFlagsFactory | None = None,
) -> None

Initialize the handler with optional flags factory.

Parameters:

  • flags_factory (DeletionRiskFlagsFactory | None, default: None ) –

    Optional factory for creating deletion risk flags. If None, uses default factory.

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
def __init__(self, flags_factory: DeletionRiskFlagsFactory | None = None) -> None:
    """Initialize the handler with optional flags factory.

    Args:
        flags_factory: Optional factory for creating deletion risk flags.
            If None, uses default factory.

    """
    self._flags_factory = flags_factory or DeletionRiskFlagsFactory.create_default()

can_handle

can_handle(model_record: GenericModelRecord) -> bool

Check if this handler can process the given model record.

Parameters:

Returns:

  • bool

    True if the model record is an ImageGenerationModelRecord.

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
def can_handle(self, model_record: GenericModelRecord) -> bool:
    """Check if this handler can process the given model record.

    Args:
        model_record: The model record to check.

    Returns:
        True if the model record is an ImageGenerationModelRecord.

    """
    return isinstance(model_record, ImageGenerationModelRecord)

create_risk_info

create_risk_info(
    *,
    model_name: str,
    model_record: GenericModelRecord,
    statistics: CombinedModelStatistics | None,
    category_total_usage: int,
    category: MODEL_REFERENCE_CATEGORY,
    include_backend_variations: bool = False,
) -> ModelDeletionRiskInfo

Create ModelDeletionRiskInfo for an image generation model.

Parameters:

  • model_name (str) –

    The model name.

  • model_record (GenericModelRecord) –

    The image generation model record.

  • statistics (CombinedModelStatistics | None) –

    Optional Horde API statistics.

  • category_total_usage (int) –

    Total monthly usage for the category.

  • category (MODEL_REFERENCE_CATEGORY) –

    The model reference category.

  • include_backend_variations (bool, default: False ) –

    Ignored for image models (always False).

Returns:

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
def create_risk_info(
    self,
    *,
    model_name: str,
    model_record: GenericModelRecord,
    statistics: CombinedModelStatistics | None,
    category_total_usage: int,
    category: MODEL_REFERENCE_CATEGORY,
    include_backend_variations: bool = False,
) -> ModelDeletionRiskInfo:
    """Create ModelDeletionRiskInfo for an image generation model.

    Args:
        model_name: The model name.
        model_record: The image generation model record.
        statistics: Optional Horde API statistics.
        category_total_usage: Total monthly usage for the category.
        category: The model reference category.
        include_backend_variations: Ignored for image models (always False).

    Returns:
        ModelDeletionRiskInfo object.

    """
    if not isinstance(model_record, ImageGenerationModelRecord):
        error_message = f"Expected ImageGenerationModelRecord, got {type(model_record).__name__}"
        raise TypeError(error_message)

    flags = self._flags_factory.create_flags(
        model_record=model_record,
        statistics=statistics,
        category_total_usage=category_total_usage,
    )
    baseline = str(model_record.baseline) if model_record.baseline else None
    nsfw = model_record.nsfw
    size_bytes = model_record.size_on_disk_bytes

    return ModelDeletionRiskInfoHandler._build_risk_info(
        model_name=model_name,
        model_record=model_record,
        statistics=statistics,
        category_total_usage=category_total_usage,
        category=category,
        flags=flags,
        baseline=baseline,
        nsfw=nsfw,
        size_bytes=size_bytes,
        include_backend_variations=False,  # Not applicable for image models
    )

_build_risk_info staticmethod

_build_risk_info(
    *,
    model_name: str,
    model_record: GenericModelRecord,
    statistics: CombinedModelStatistics | None,
    category_total_usage: int,
    category: MODEL_REFERENCE_CATEGORY,
    flags: DeletionRiskFlags,
    baseline: str | None,
    nsfw: bool | None,
    size_bytes: int | None,
    include_backend_variations: bool = False,
) -> ModelDeletionRiskInfo

Build ModelDeletionRiskInfo from common components.

Parameters:

  • model_name (str) –

    The model name.

  • model_record (GenericModelRecord) –

    Any model record type.

  • statistics (CombinedModelStatistics | None) –

    Optional Horde API statistics.

  • category_total_usage (int) –

    Total monthly usage for the category.

  • category (MODEL_REFERENCE_CATEGORY) –

    The model reference category.

  • flags (DeletionRiskFlags) –

    Deletion risk flags.

  • baseline (str | None) –

    Model baseline (if applicable).

  • nsfw (bool | None) –

    Whether model is NSFW (if applicable).

  • size_bytes (int | None) –

    Model size in bytes (if available).

  • include_backend_variations (bool, default: False ) –

    Whether to include per-backend breakdown.

Returns:

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
@staticmethod
def _build_risk_info(
    *,
    model_name: str,
    model_record: GenericModelRecord,
    statistics: CombinedModelStatistics | None,
    category_total_usage: int,
    category: MODEL_REFERENCE_CATEGORY,
    flags: DeletionRiskFlags,
    baseline: str | None,
    nsfw: bool | None,
    size_bytes: int | None,
    include_backend_variations: bool = False,
) -> ModelDeletionRiskInfo:
    """Build ModelDeletionRiskInfo from common components.

    Args:
        model_name: The model name.
        model_record: Any model record type.
        statistics: Optional Horde API statistics.
        category_total_usage: Total monthly usage for the category.
        category: The model reference category.
        flags: Deletion risk flags.
        baseline: Model baseline (if applicable).
        nsfw: Whether model is NSFW (if applicable).
        size_bytes: Model size in bytes (if available).
        include_backend_variations: Whether to include per-backend breakdown.

    Returns:
        ModelDeletionRiskInfo object.

    """
    # Extract Horde API data from statistics
    worker_count = statistics.worker_count if statistics else 0
    usage_day = statistics.usage_stats.day if statistics and statistics.usage_stats else 0
    usage_month = statistics.usage_stats.month if statistics and statistics.usage_stats else 0
    usage_total = statistics.usage_stats.total if statistics and statistics.usage_stats else 0
    usage_hour = None
    usage_minute = None

    # Calculate usage percentage
    usage_percentage = 0.0
    if category_total_usage > 0:
        usage_percentage = (usage_month / category_total_usage) * 100.0

    # Calculate usage trend ratios
    day_to_month_ratio: float | None = None
    if usage_month > 0:
        day_to_month_ratio = usage_day / usage_month

    month_to_total_ratio: float | None = None
    if usage_total > 0:
        month_to_total_ratio = usage_month / usage_total

    usage_trend = UsageTrend(
        day_to_month_ratio=day_to_month_ratio,
        month_to_total_ratio=month_to_total_ratio,
    )

    # Check description
    description = model_record.description
    has_description = bool(description and len(description.strip()) > 0)

    # Calculate size in GB and cost-benefit score
    size_gb: float | None = None
    cost_benefit_score: float | None = None

    if size_bytes and size_bytes > 0:
        size_gb = size_bytes / (1024**3)
        if size_gb > 0:
            cost_benefit_score = usage_month / size_gb

    # Extract download information
    downloads = model_record.config.download if model_record.config else []
    download_count = len(downloads)

    download_hosts = []
    for download in downloads:
        url = download.file_url
        if url:
            try:
                parsed = urlparse(url)
                if parsed.netloc and parsed.netloc not in download_hosts:
                    download_hosts.append(parsed.netloc)
            except Exception:
                pass

    # Build backend variations list if requested and available
    backend_variations_list: list[BackendVariationStats] | None = None
    if include_backend_variations and statistics and statistics.backend_variations:
        backend_variations_list = [
            BackendVariationStats(
                backend=bv.backend,
                variant_name=bv.variant_name,
                worker_count=bv.worker_count,
                performance=bv.performance,
                usage_day=bv.usage_day,
                usage_month=bv.usage_month,
                usage_total=bv.usage_total,
            )
            for bv in statistics.backend_variations.values()
        ]

    # Create risk info
    return ModelDeletionRiskInfo(
        name=model_name,
        category=category,
        deletion_risk_flags=flags,
        at_risk=flags.any_flags(),
        risk_score=flags.flag_count(),
        worker_count=worker_count,
        usage_day=usage_day,
        usage_month=usage_month,
        usage_total=usage_total,
        usage_hour=usage_hour,
        usage_minute=usage_minute,
        usage_percentage_of_category=round(usage_percentage, 4),
        usage_trend=usage_trend,
        cost_benefit_score=round(cost_benefit_score, 2) if cost_benefit_score is not None else None,
        size_gb=round(size_gb, 2) if size_gb is not None else None,
        baseline=baseline,
        nsfw=nsfw,
        has_description=has_description,
        download_count=download_count,
        download_hosts=download_hosts,
        backend_variations=backend_variations_list,
    )

TextGenerationModelDeletionRiskHandler

Bases: ModelDeletionRiskInfoHandler

Handler for text generation model deletion risk info creation.

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
class TextGenerationModelDeletionRiskHandler(ModelDeletionRiskInfoHandler):
    """Handler for text generation model deletion risk info creation."""

    def __init__(self, flags_factory: DeletionRiskFlagsFactory | None = None) -> None:
        """Initialize the handler with optional flags factory.

        Args:
            flags_factory: Optional factory for creating deletion risk flags.
                If None, uses default factory.

        """
        self._flags_factory = flags_factory or DeletionRiskFlagsFactory.create_default()

    def can_handle(self, model_record: GenericModelRecord) -> bool:
        """Check if this handler can process the given model record.

        Args:
            model_record: The model record to check.

        Returns:
            True if the model record is a TextGenerationModelRecord.

        """
        return isinstance(model_record, TextGenerationModelRecord)

    def create_risk_info(
        self,
        *,
        model_name: str,
        model_record: GenericModelRecord,
        statistics: CombinedModelStatistics | None,
        category_total_usage: int,
        category: MODEL_REFERENCE_CATEGORY,
        include_backend_variations: bool = False,
    ) -> ModelDeletionRiskInfo:
        """Create ModelDeletionRiskInfo for a text generation model.

        Args:
            model_name: The model name.
            model_record: The text generation model record.
            statistics: Optional Horde API statistics.
            category_total_usage: Total monthly usage for the category.
            category: The model reference category.
            include_backend_variations: Whether to include per-backend breakdown.

        Returns:
            ModelDeletionRiskInfo object.

        """
        if not isinstance(model_record, TextGenerationModelRecord):
            error_message = f"Expected TextGenerationModelRecord, got {type(model_record).__name__}"
            raise TypeError(error_message)

        flags = self._flags_factory.create_flags(
            model_record=model_record,
            statistics=statistics,
            category_total_usage=category_total_usage,
        )
        baseline = model_record.baseline
        nsfw = model_record.nsfw
        size_bytes = None  # Text generation models don't have size_on_disk_bytes

        return ModelDeletionRiskInfoHandler._build_risk_info(
            model_name=model_name,
            model_record=model_record,
            statistics=statistics,
            category_total_usage=category_total_usage,
            category=category,
            flags=flags,
            baseline=baseline,
            nsfw=nsfw,
            size_bytes=size_bytes,
            include_backend_variations=include_backend_variations,
        )

_flags_factory instance-attribute

_flags_factory = flags_factory or create_default()

__init__

__init__(
    flags_factory: DeletionRiskFlagsFactory | None = None,
) -> None

Initialize the handler with optional flags factory.

Parameters:

  • flags_factory (DeletionRiskFlagsFactory | None, default: None ) –

    Optional factory for creating deletion risk flags. If None, uses default factory.

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
def __init__(self, flags_factory: DeletionRiskFlagsFactory | None = None) -> None:
    """Initialize the handler with optional flags factory.

    Args:
        flags_factory: Optional factory for creating deletion risk flags.
            If None, uses default factory.

    """
    self._flags_factory = flags_factory or DeletionRiskFlagsFactory.create_default()

can_handle

can_handle(model_record: GenericModelRecord) -> bool

Check if this handler can process the given model record.

Parameters:

Returns:

  • bool

    True if the model record is a TextGenerationModelRecord.

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
def can_handle(self, model_record: GenericModelRecord) -> bool:
    """Check if this handler can process the given model record.

    Args:
        model_record: The model record to check.

    Returns:
        True if the model record is a TextGenerationModelRecord.

    """
    return isinstance(model_record, TextGenerationModelRecord)

create_risk_info

create_risk_info(
    *,
    model_name: str,
    model_record: GenericModelRecord,
    statistics: CombinedModelStatistics | None,
    category_total_usage: int,
    category: MODEL_REFERENCE_CATEGORY,
    include_backend_variations: bool = False,
) -> ModelDeletionRiskInfo

Create ModelDeletionRiskInfo for a text generation model.

Parameters:

  • model_name (str) –

    The model name.

  • model_record (GenericModelRecord) –

    The text generation model record.

  • statistics (CombinedModelStatistics | None) –

    Optional Horde API statistics.

  • category_total_usage (int) –

    Total monthly usage for the category.

  • category (MODEL_REFERENCE_CATEGORY) –

    The model reference category.

  • include_backend_variations (bool, default: False ) –

    Whether to include per-backend breakdown.

Returns:

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
def create_risk_info(
    self,
    *,
    model_name: str,
    model_record: GenericModelRecord,
    statistics: CombinedModelStatistics | None,
    category_total_usage: int,
    category: MODEL_REFERENCE_CATEGORY,
    include_backend_variations: bool = False,
) -> ModelDeletionRiskInfo:
    """Create ModelDeletionRiskInfo for a text generation model.

    Args:
        model_name: The model name.
        model_record: The text generation model record.
        statistics: Optional Horde API statistics.
        category_total_usage: Total monthly usage for the category.
        category: The model reference category.
        include_backend_variations: Whether to include per-backend breakdown.

    Returns:
        ModelDeletionRiskInfo object.

    """
    if not isinstance(model_record, TextGenerationModelRecord):
        error_message = f"Expected TextGenerationModelRecord, got {type(model_record).__name__}"
        raise TypeError(error_message)

    flags = self._flags_factory.create_flags(
        model_record=model_record,
        statistics=statistics,
        category_total_usage=category_total_usage,
    )
    baseline = model_record.baseline
    nsfw = model_record.nsfw
    size_bytes = None  # Text generation models don't have size_on_disk_bytes

    return ModelDeletionRiskInfoHandler._build_risk_info(
        model_name=model_name,
        model_record=model_record,
        statistics=statistics,
        category_total_usage=category_total_usage,
        category=category,
        flags=flags,
        baseline=baseline,
        nsfw=nsfw,
        size_bytes=size_bytes,
        include_backend_variations=include_backend_variations,
    )

_build_risk_info staticmethod

_build_risk_info(
    *,
    model_name: str,
    model_record: GenericModelRecord,
    statistics: CombinedModelStatistics | None,
    category_total_usage: int,
    category: MODEL_REFERENCE_CATEGORY,
    flags: DeletionRiskFlags,
    baseline: str | None,
    nsfw: bool | None,
    size_bytes: int | None,
    include_backend_variations: bool = False,
) -> ModelDeletionRiskInfo

Build ModelDeletionRiskInfo from common components.

Parameters:

  • model_name (str) –

    The model name.

  • model_record (GenericModelRecord) –

    Any model record type.

  • statistics (CombinedModelStatistics | None) –

    Optional Horde API statistics.

  • category_total_usage (int) –

    Total monthly usage for the category.

  • category (MODEL_REFERENCE_CATEGORY) –

    The model reference category.

  • flags (DeletionRiskFlags) –

    Deletion risk flags.

  • baseline (str | None) –

    Model baseline (if applicable).

  • nsfw (bool | None) –

    Whether model is NSFW (if applicable).

  • size_bytes (int | None) –

    Model size in bytes (if available).

  • include_backend_variations (bool, default: False ) –

    Whether to include per-backend breakdown.

Returns:

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
@staticmethod
def _build_risk_info(
    *,
    model_name: str,
    model_record: GenericModelRecord,
    statistics: CombinedModelStatistics | None,
    category_total_usage: int,
    category: MODEL_REFERENCE_CATEGORY,
    flags: DeletionRiskFlags,
    baseline: str | None,
    nsfw: bool | None,
    size_bytes: int | None,
    include_backend_variations: bool = False,
) -> ModelDeletionRiskInfo:
    """Build ModelDeletionRiskInfo from common components.

    Args:
        model_name: The model name.
        model_record: Any model record type.
        statistics: Optional Horde API statistics.
        category_total_usage: Total monthly usage for the category.
        category: The model reference category.
        flags: Deletion risk flags.
        baseline: Model baseline (if applicable).
        nsfw: Whether model is NSFW (if applicable).
        size_bytes: Model size in bytes (if available).
        include_backend_variations: Whether to include per-backend breakdown.

    Returns:
        ModelDeletionRiskInfo object.

    """
    # Extract Horde API data from statistics
    worker_count = statistics.worker_count if statistics else 0
    usage_day = statistics.usage_stats.day if statistics and statistics.usage_stats else 0
    usage_month = statistics.usage_stats.month if statistics and statistics.usage_stats else 0
    usage_total = statistics.usage_stats.total if statistics and statistics.usage_stats else 0
    usage_hour = None
    usage_minute = None

    # Calculate usage percentage
    usage_percentage = 0.0
    if category_total_usage > 0:
        usage_percentage = (usage_month / category_total_usage) * 100.0

    # Calculate usage trend ratios
    day_to_month_ratio: float | None = None
    if usage_month > 0:
        day_to_month_ratio = usage_day / usage_month

    month_to_total_ratio: float | None = None
    if usage_total > 0:
        month_to_total_ratio = usage_month / usage_total

    usage_trend = UsageTrend(
        day_to_month_ratio=day_to_month_ratio,
        month_to_total_ratio=month_to_total_ratio,
    )

    # Check description
    description = model_record.description
    has_description = bool(description and len(description.strip()) > 0)

    # Calculate size in GB and cost-benefit score
    size_gb: float | None = None
    cost_benefit_score: float | None = None

    if size_bytes and size_bytes > 0:
        size_gb = size_bytes / (1024**3)
        if size_gb > 0:
            cost_benefit_score = usage_month / size_gb

    # Extract download information
    downloads = model_record.config.download if model_record.config else []
    download_count = len(downloads)

    download_hosts = []
    for download in downloads:
        url = download.file_url
        if url:
            try:
                parsed = urlparse(url)
                if parsed.netloc and parsed.netloc not in download_hosts:
                    download_hosts.append(parsed.netloc)
            except Exception:
                pass

    # Build backend variations list if requested and available
    backend_variations_list: list[BackendVariationStats] | None = None
    if include_backend_variations and statistics and statistics.backend_variations:
        backend_variations_list = [
            BackendVariationStats(
                backend=bv.backend,
                variant_name=bv.variant_name,
                worker_count=bv.worker_count,
                performance=bv.performance,
                usage_day=bv.usage_day,
                usage_month=bv.usage_month,
                usage_total=bv.usage_total,
            )
            for bv in statistics.backend_variations.values()
        ]

    # Create risk info
    return ModelDeletionRiskInfo(
        name=model_name,
        category=category,
        deletion_risk_flags=flags,
        at_risk=flags.any_flags(),
        risk_score=flags.flag_count(),
        worker_count=worker_count,
        usage_day=usage_day,
        usage_month=usage_month,
        usage_total=usage_total,
        usage_hour=usage_hour,
        usage_minute=usage_minute,
        usage_percentage_of_category=round(usage_percentage, 4),
        usage_trend=usage_trend,
        cost_benefit_score=round(cost_benefit_score, 2) if cost_benefit_score is not None else None,
        size_gb=round(size_gb, 2) if size_gb is not None else None,
        baseline=baseline,
        nsfw=nsfw,
        has_description=has_description,
        download_count=download_count,
        download_hosts=download_hosts,
        backend_variations=backend_variations_list,
    )

GenericModelDeletionRiskHandler

Bases: ModelDeletionRiskInfoHandler

Fallback handler for unsupported model record types.

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
class GenericModelDeletionRiskHandler(ModelDeletionRiskInfoHandler):
    """Fallback handler for unsupported model record types."""

    def can_handle(self, model_record: GenericModelRecord) -> bool:
        """Check if this handler can process the given model record.

        This handler accepts all model records as a fallback.

        Args:
            model_record: The model record to check.

        Returns:
            True (accepts all model records).

        """
        return True

    def create_risk_info(
        self,
        *,
        model_name: str,
        model_record: GenericModelRecord,
        statistics: CombinedModelStatistics | None,
        category_total_usage: int,
        category: MODEL_REFERENCE_CATEGORY,
        include_backend_variations: bool = False,
    ) -> ModelDeletionRiskInfo:
        """Create ModelDeletionRiskInfo for a generic/unsupported model type.

        Args:
            model_name: The model name.
            model_record: The generic model record.
            statistics: Optional Horde API statistics.
            category_total_usage: Total monthly usage for the category.
            category: The model reference category.
            include_backend_variations: Ignored for generic models.

        Returns:
            ModelDeletionRiskInfo object.

        """
        logger.warning(f"Using fallback handler for unsupported model type: {type(model_record).__name__}")

        # Use DeletionRiskFlagsFactory for flag creation
        flags_factory = DeletionRiskFlagsFactory.create_default()
        flags = flags_factory.create_flags(
            model_record=model_record,
            statistics=statistics,
            category_total_usage=category_total_usage,
        )

        return ModelDeletionRiskInfoHandler._build_risk_info(
            model_name=model_name,
            model_record=model_record,
            statistics=statistics,
            category_total_usage=category_total_usage,
            category=category,
            flags=flags,
            baseline=None,
            nsfw=None,
            size_bytes=None,
            include_backend_variations=False,  # Not applicable for generic models
        )

can_handle

can_handle(model_record: GenericModelRecord) -> bool

Check if this handler can process the given model record.

This handler accepts all model records as a fallback.

Parameters:

Returns:

  • bool

    True (accepts all model records).

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
def can_handle(self, model_record: GenericModelRecord) -> bool:
    """Check if this handler can process the given model record.

    This handler accepts all model records as a fallback.

    Args:
        model_record: The model record to check.

    Returns:
        True (accepts all model records).

    """
    return True

create_risk_info

create_risk_info(
    *,
    model_name: str,
    model_record: GenericModelRecord,
    statistics: CombinedModelStatistics | None,
    category_total_usage: int,
    category: MODEL_REFERENCE_CATEGORY,
    include_backend_variations: bool = False,
) -> ModelDeletionRiskInfo

Create ModelDeletionRiskInfo for a generic/unsupported model type.

Parameters:

  • model_name (str) –

    The model name.

  • model_record (GenericModelRecord) –

    The generic model record.

  • statistics (CombinedModelStatistics | None) –

    Optional Horde API statistics.

  • category_total_usage (int) –

    Total monthly usage for the category.

  • category (MODEL_REFERENCE_CATEGORY) –

    The model reference category.

  • include_backend_variations (bool, default: False ) –

    Ignored for generic models.

Returns:

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
def create_risk_info(
    self,
    *,
    model_name: str,
    model_record: GenericModelRecord,
    statistics: CombinedModelStatistics | None,
    category_total_usage: int,
    category: MODEL_REFERENCE_CATEGORY,
    include_backend_variations: bool = False,
) -> ModelDeletionRiskInfo:
    """Create ModelDeletionRiskInfo for a generic/unsupported model type.

    Args:
        model_name: The model name.
        model_record: The generic model record.
        statistics: Optional Horde API statistics.
        category_total_usage: Total monthly usage for the category.
        category: The model reference category.
        include_backend_variations: Ignored for generic models.

    Returns:
        ModelDeletionRiskInfo object.

    """
    logger.warning(f"Using fallback handler for unsupported model type: {type(model_record).__name__}")

    # Use DeletionRiskFlagsFactory for flag creation
    flags_factory = DeletionRiskFlagsFactory.create_default()
    flags = flags_factory.create_flags(
        model_record=model_record,
        statistics=statistics,
        category_total_usage=category_total_usage,
    )

    return ModelDeletionRiskInfoHandler._build_risk_info(
        model_name=model_name,
        model_record=model_record,
        statistics=statistics,
        category_total_usage=category_total_usage,
        category=category,
        flags=flags,
        baseline=None,
        nsfw=None,
        size_bytes=None,
        include_backend_variations=False,  # Not applicable for generic models
    )

_build_risk_info staticmethod

_build_risk_info(
    *,
    model_name: str,
    model_record: GenericModelRecord,
    statistics: CombinedModelStatistics | None,
    category_total_usage: int,
    category: MODEL_REFERENCE_CATEGORY,
    flags: DeletionRiskFlags,
    baseline: str | None,
    nsfw: bool | None,
    size_bytes: int | None,
    include_backend_variations: bool = False,
) -> ModelDeletionRiskInfo

Build ModelDeletionRiskInfo from common components.

Parameters:

  • model_name (str) –

    The model name.

  • model_record (GenericModelRecord) –

    Any model record type.

  • statistics (CombinedModelStatistics | None) –

    Optional Horde API statistics.

  • category_total_usage (int) –

    Total monthly usage for the category.

  • category (MODEL_REFERENCE_CATEGORY) –

    The model reference category.

  • flags (DeletionRiskFlags) –

    Deletion risk flags.

  • baseline (str | None) –

    Model baseline (if applicable).

  • nsfw (bool | None) –

    Whether model is NSFW (if applicable).

  • size_bytes (int | None) –

    Model size in bytes (if available).

  • include_backend_variations (bool, default: False ) –

    Whether to include per-backend breakdown.

Returns:

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
@staticmethod
def _build_risk_info(
    *,
    model_name: str,
    model_record: GenericModelRecord,
    statistics: CombinedModelStatistics | None,
    category_total_usage: int,
    category: MODEL_REFERENCE_CATEGORY,
    flags: DeletionRiskFlags,
    baseline: str | None,
    nsfw: bool | None,
    size_bytes: int | None,
    include_backend_variations: bool = False,
) -> ModelDeletionRiskInfo:
    """Build ModelDeletionRiskInfo from common components.

    Args:
        model_name: The model name.
        model_record: Any model record type.
        statistics: Optional Horde API statistics.
        category_total_usage: Total monthly usage for the category.
        category: The model reference category.
        flags: Deletion risk flags.
        baseline: Model baseline (if applicable).
        nsfw: Whether model is NSFW (if applicable).
        size_bytes: Model size in bytes (if available).
        include_backend_variations: Whether to include per-backend breakdown.

    Returns:
        ModelDeletionRiskInfo object.

    """
    # Extract Horde API data from statistics
    worker_count = statistics.worker_count if statistics else 0
    usage_day = statistics.usage_stats.day if statistics and statistics.usage_stats else 0
    usage_month = statistics.usage_stats.month if statistics and statistics.usage_stats else 0
    usage_total = statistics.usage_stats.total if statistics and statistics.usage_stats else 0
    usage_hour = None
    usage_minute = None

    # Calculate usage percentage
    usage_percentage = 0.0
    if category_total_usage > 0:
        usage_percentage = (usage_month / category_total_usage) * 100.0

    # Calculate usage trend ratios
    day_to_month_ratio: float | None = None
    if usage_month > 0:
        day_to_month_ratio = usage_day / usage_month

    month_to_total_ratio: float | None = None
    if usage_total > 0:
        month_to_total_ratio = usage_month / usage_total

    usage_trend = UsageTrend(
        day_to_month_ratio=day_to_month_ratio,
        month_to_total_ratio=month_to_total_ratio,
    )

    # Check description
    description = model_record.description
    has_description = bool(description and len(description.strip()) > 0)

    # Calculate size in GB and cost-benefit score
    size_gb: float | None = None
    cost_benefit_score: float | None = None

    if size_bytes and size_bytes > 0:
        size_gb = size_bytes / (1024**3)
        if size_gb > 0:
            cost_benefit_score = usage_month / size_gb

    # Extract download information
    downloads = model_record.config.download if model_record.config else []
    download_count = len(downloads)

    download_hosts = []
    for download in downloads:
        url = download.file_url
        if url:
            try:
                parsed = urlparse(url)
                if parsed.netloc and parsed.netloc not in download_hosts:
                    download_hosts.append(parsed.netloc)
            except Exception:
                pass

    # Build backend variations list if requested and available
    backend_variations_list: list[BackendVariationStats] | None = None
    if include_backend_variations and statistics and statistics.backend_variations:
        backend_variations_list = [
            BackendVariationStats(
                backend=bv.backend,
                variant_name=bv.variant_name,
                worker_count=bv.worker_count,
                performance=bv.performance,
                usage_day=bv.usage_day,
                usage_month=bv.usage_month,
                usage_total=bv.usage_total,
            )
            for bv in statistics.backend_variations.values()
        ]

    # Create risk info
    return ModelDeletionRiskInfo(
        name=model_name,
        category=category,
        deletion_risk_flags=flags,
        at_risk=flags.any_flags(),
        risk_score=flags.flag_count(),
        worker_count=worker_count,
        usage_day=usage_day,
        usage_month=usage_month,
        usage_total=usage_total,
        usage_hour=usage_hour,
        usage_minute=usage_minute,
        usage_percentage_of_category=round(usage_percentage, 4),
        usage_trend=usage_trend,
        cost_benefit_score=round(cost_benefit_score, 2) if cost_benefit_score is not None else None,
        size_gb=round(size_gb, 2) if size_gb is not None else None,
        baseline=baseline,
        nsfw=nsfw,
        has_description=has_description,
        download_count=download_count,
        download_hosts=download_hosts,
        backend_variations=backend_variations_list,
    )

ModelDeletionRiskInfoFactory

Factory for creating ModelDeletionRiskInfo objects with extensible handler support.

Handlers are registered and checked in order. The first handler that can process a model record type will be used to create the risk info.

Examples:

# Using default handlers
factory = ModelDeletionRiskInfoFactory.create_default()
risk_info = factory.create_risk_info(
    model_name="my_model",
    model_record=image_model_record,
    statistics=stats,
    category_total_usage=10000,
    category=MODEL_REFERENCE_CATEGORY.image_generation,
)

# Adding custom handler
factory.register_handler(CustomModelRiskHandler())
Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
class ModelDeletionRiskInfoFactory:
    """Factory for creating ModelDeletionRiskInfo objects with extensible handler support.

    Handlers are registered and checked in order. The first handler that can process
    a model record type will be used to create the risk info.

    Examples:
        ```python
        # Using default handlers
        factory = ModelDeletionRiskInfoFactory.create_default()
        risk_info = factory.create_risk_info(
            model_name="my_model",
            model_record=image_model_record,
            statistics=stats,
            category_total_usage=10000,
            category=MODEL_REFERENCE_CATEGORY.image_generation,
        )

        # Adding custom handler
        factory.register_handler(CustomModelRiskHandler())
        ```

    """

    def __init__(self, handlers: list[ModelDeletionRiskInfoHandler] | None = None) -> None:
        """Initialize the factory with optional handlers.

        Args:
            handlers: List of handlers to use. If None, no handlers are registered.

        """
        self._handlers: list[ModelDeletionRiskInfoHandler] = handlers or []

    @classmethod
    def create_default(cls) -> ModelDeletionRiskInfoFactory:
        """Create a factory with default handlers for standard model types.

        Returns:
            ModelDeletionRiskInfoFactory with default handlers registered.

        """
        return cls(
            handlers=[
                ImageGenerationModelDeletionRiskHandler(),
                TextGenerationModelDeletionRiskHandler(),
                GenericModelDeletionRiskHandler(),  # Fallback handler (must be last)
            ]
        )

    def register_handler(self, handler: ModelDeletionRiskInfoHandler) -> None:
        """Register a new handler.

        Handlers are checked in registration order. Register more specific handlers
        before generic ones.

        Args:
            handler: The handler to register.

        """
        self._handlers.append(handler)

    def create_risk_info(
        self,
        *,
        model_name: str,
        model_record: GenericModelRecord,
        statistics: CombinedModelStatistics | None,
        category_total_usage: int,
        category: MODEL_REFERENCE_CATEGORY,
        include_backend_variations: bool = False,
    ) -> ModelDeletionRiskInfo:
        """Create ModelDeletionRiskInfo for a model record using the appropriate handler.

        Args:
            model_name: The model name.
            model_record: The model record.
            statistics: Optional Horde API statistics.
            category_total_usage: Total monthly usage for the category.
            category: The model reference category.
            include_backend_variations: Whether to include per-backend breakdown (text models only).

        Returns:
            ModelDeletionRiskInfo object.

        Raises:
            ValueError: If no handler can process the model record type.

        """
        for handler in self._handlers:
            if handler.can_handle(model_record):
                return handler.create_risk_info(
                    model_name=model_name,
                    model_record=model_record,
                    statistics=statistics,
                    category_total_usage=category_total_usage,
                    category=category,
                    include_backend_variations=include_backend_variations,
                )

        error_message = f"No handler found for model record type: {type(model_record).__name__}"
        raise ValueError(error_message)

    def analyze_models(
        self,
        model_records: (
            dict[str, GenericModelRecord]
            | dict[str, ImageGenerationModelRecord]
            | dict[str, TextGenerationModelRecord]
        ),
        model_statistics: dict[str, CombinedModelStatistics],
        category_total_usage: int,
        category: MODEL_REFERENCE_CATEGORY,
        include_backend_variations: bool = False,
    ) -> list[ModelDeletionRiskInfo]:
        """Analyze model records and statistics to create deletion risk information.

        Args:
            model_records: Dictionary of model names to typed model records.
            model_statistics: Dictionary of model names to Horde API statistics.
            category_total_usage: Total monthly usage for the entire category.
            category: The model reference category.
            include_backend_variations: Whether to include per-backend breakdown (text models only).

        Returns:
            List of ModelDeletionRiskInfo sorted by usage (descending).

        """
        risk_models: list[ModelDeletionRiskInfo] = []

        model_record: GenericModelRecord | ImageGenerationModelRecord | TextGenerationModelRecord

        for model_name, model_record in model_records.items():
            # Get statistics for this model (may be None if not in Horde data)
            statistics = model_statistics.get(model_name)

            # Use factory to create risk info
            risk_info = self.create_risk_info(
                model_name=model_name,
                model_record=model_record,
                statistics=statistics,
                category_total_usage=category_total_usage,
                category=category,
                include_backend_variations=include_backend_variations,
            )

            risk_models.append(risk_info)

        # Sort by usage (descending) for easier review
        risk_models.sort(key=lambda x: x.usage_month, reverse=True)

        logger.info(
            f"Analyzed {len(risk_models)} models for deletion risk: {sum(1 for m in risk_models if m.at_risk)} at risk"
        )

        return risk_models

    def create_deletion_risk_response(
        self,
        model_records: (
            dict[str, GenericModelRecord]
            | dict[str, ImageGenerationModelRecord]
            | dict[str, TextGenerationModelRecord]
        ),
        model_statistics: dict[str, CombinedModelStatistics],
        category_total_usage: int,
        category: MODEL_REFERENCE_CATEGORY,
        include_backend_variations: bool = False,
    ) -> CategoryDeletionRiskResponse:
        """Analyze models and create complete deletion risk response with summary.

        Args:
            model_records: Dictionary of model names to typed model records.
            model_statistics: Dictionary of model names to Horde API statistics.
            category_total_usage: Total monthly usage for the entire category.
            category: The model reference category.
            include_backend_variations: Whether to include per-backend breakdown (text models only).

        Returns:
            CategoryDeletionRiskResponse with models and summary.

        """
        # Analyze all models
        risk_models = self.analyze_models(
            model_records=model_records,
            model_statistics=model_statistics,
            category_total_usage=category_total_usage,
            category=category,
            include_backend_variations=include_backend_variations,
        )

        # Calculate summary
        summary = CategoryDeletionRiskSummary.from_risk_models(risk_models)

        # Create response
        return CategoryDeletionRiskResponse(
            category=category,
            category_total_month_usage=category_total_usage,
            total_count=len(risk_models),
            returned_count=len(risk_models),
            offset=0,
            limit=None,
            models=risk_models,
            summary=summary,
        )

_handlers instance-attribute

_handlers: list[ModelDeletionRiskInfoHandler] = (
    handlers or []
)

__init__

__init__(
    handlers: list[ModelDeletionRiskInfoHandler]
    | None = None,
) -> None

Initialize the factory with optional handlers.

Parameters:

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
def __init__(self, handlers: list[ModelDeletionRiskInfoHandler] | None = None) -> None:
    """Initialize the factory with optional handlers.

    Args:
        handlers: List of handlers to use. If None, no handlers are registered.

    """
    self._handlers: list[ModelDeletionRiskInfoHandler] = handlers or []

create_default classmethod

create_default() -> ModelDeletionRiskInfoFactory

Create a factory with default handlers for standard model types.

Returns:

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
@classmethod
def create_default(cls) -> ModelDeletionRiskInfoFactory:
    """Create a factory with default handlers for standard model types.

    Returns:
        ModelDeletionRiskInfoFactory with default handlers registered.

    """
    return cls(
        handlers=[
            ImageGenerationModelDeletionRiskHandler(),
            TextGenerationModelDeletionRiskHandler(),
            GenericModelDeletionRiskHandler(),  # Fallback handler (must be last)
        ]
    )

register_handler

register_handler(
    handler: ModelDeletionRiskInfoHandler,
) -> None

Register a new handler.

Handlers are checked in registration order. Register more specific handlers before generic ones.

Parameters:

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
def register_handler(self, handler: ModelDeletionRiskInfoHandler) -> None:
    """Register a new handler.

    Handlers are checked in registration order. Register more specific handlers
    before generic ones.

    Args:
        handler: The handler to register.

    """
    self._handlers.append(handler)

create_risk_info

create_risk_info(
    *,
    model_name: str,
    model_record: GenericModelRecord,
    statistics: CombinedModelStatistics | None,
    category_total_usage: int,
    category: MODEL_REFERENCE_CATEGORY,
    include_backend_variations: bool = False,
) -> ModelDeletionRiskInfo

Create ModelDeletionRiskInfo for a model record using the appropriate handler.

Parameters:

  • model_name (str) –

    The model name.

  • model_record (GenericModelRecord) –

    The model record.

  • statistics (CombinedModelStatistics | None) –

    Optional Horde API statistics.

  • category_total_usage (int) –

    Total monthly usage for the category.

  • category (MODEL_REFERENCE_CATEGORY) –

    The model reference category.

  • include_backend_variations (bool, default: False ) –

    Whether to include per-backend breakdown (text models only).

Returns:

Raises:

  • ValueError

    If no handler can process the model record type.

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
def create_risk_info(
    self,
    *,
    model_name: str,
    model_record: GenericModelRecord,
    statistics: CombinedModelStatistics | None,
    category_total_usage: int,
    category: MODEL_REFERENCE_CATEGORY,
    include_backend_variations: bool = False,
) -> ModelDeletionRiskInfo:
    """Create ModelDeletionRiskInfo for a model record using the appropriate handler.

    Args:
        model_name: The model name.
        model_record: The model record.
        statistics: Optional Horde API statistics.
        category_total_usage: Total monthly usage for the category.
        category: The model reference category.
        include_backend_variations: Whether to include per-backend breakdown (text models only).

    Returns:
        ModelDeletionRiskInfo object.

    Raises:
        ValueError: If no handler can process the model record type.

    """
    for handler in self._handlers:
        if handler.can_handle(model_record):
            return handler.create_risk_info(
                model_name=model_name,
                model_record=model_record,
                statistics=statistics,
                category_total_usage=category_total_usage,
                category=category,
                include_backend_variations=include_backend_variations,
            )

    error_message = f"No handler found for model record type: {type(model_record).__name__}"
    raise ValueError(error_message)

analyze_models

analyze_models(
    model_records: dict[str, GenericModelRecord]
    | dict[str, ImageGenerationModelRecord]
    | dict[str, TextGenerationModelRecord],
    model_statistics: dict[str, CombinedModelStatistics],
    category_total_usage: int,
    category: MODEL_REFERENCE_CATEGORY,
    include_backend_variations: bool = False,
) -> list[ModelDeletionRiskInfo]

Analyze model records and statistics to create deletion risk information.

Parameters:

Returns:

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
def analyze_models(
    self,
    model_records: (
        dict[str, GenericModelRecord]
        | dict[str, ImageGenerationModelRecord]
        | dict[str, TextGenerationModelRecord]
    ),
    model_statistics: dict[str, CombinedModelStatistics],
    category_total_usage: int,
    category: MODEL_REFERENCE_CATEGORY,
    include_backend_variations: bool = False,
) -> list[ModelDeletionRiskInfo]:
    """Analyze model records and statistics to create deletion risk information.

    Args:
        model_records: Dictionary of model names to typed model records.
        model_statistics: Dictionary of model names to Horde API statistics.
        category_total_usage: Total monthly usage for the entire category.
        category: The model reference category.
        include_backend_variations: Whether to include per-backend breakdown (text models only).

    Returns:
        List of ModelDeletionRiskInfo sorted by usage (descending).

    """
    risk_models: list[ModelDeletionRiskInfo] = []

    model_record: GenericModelRecord | ImageGenerationModelRecord | TextGenerationModelRecord

    for model_name, model_record in model_records.items():
        # Get statistics for this model (may be None if not in Horde data)
        statistics = model_statistics.get(model_name)

        # Use factory to create risk info
        risk_info = self.create_risk_info(
            model_name=model_name,
            model_record=model_record,
            statistics=statistics,
            category_total_usage=category_total_usage,
            category=category,
            include_backend_variations=include_backend_variations,
        )

        risk_models.append(risk_info)

    # Sort by usage (descending) for easier review
    risk_models.sort(key=lambda x: x.usage_month, reverse=True)

    logger.info(
        f"Analyzed {len(risk_models)} models for deletion risk: {sum(1 for m in risk_models if m.at_risk)} at risk"
    )

    return risk_models

create_deletion_risk_response

create_deletion_risk_response(
    model_records: dict[str, GenericModelRecord]
    | dict[str, ImageGenerationModelRecord]
    | dict[str, TextGenerationModelRecord],
    model_statistics: dict[str, CombinedModelStatistics],
    category_total_usage: int,
    category: MODEL_REFERENCE_CATEGORY,
    include_backend_variations: bool = False,
) -> CategoryDeletionRiskResponse

Analyze models and create complete deletion risk response with summary.

Parameters:

Returns:

Source code in src/horde_model_reference/analytics/deletion_risk_analysis.py
def create_deletion_risk_response(
    self,
    model_records: (
        dict[str, GenericModelRecord]
        | dict[str, ImageGenerationModelRecord]
        | dict[str, TextGenerationModelRecord]
    ),
    model_statistics: dict[str, CombinedModelStatistics],
    category_total_usage: int,
    category: MODEL_REFERENCE_CATEGORY,
    include_backend_variations: bool = False,
) -> CategoryDeletionRiskResponse:
    """Analyze models and create complete deletion risk response with summary.

    Args:
        model_records: Dictionary of model names to typed model records.
        model_statistics: Dictionary of model names to Horde API statistics.
        category_total_usage: Total monthly usage for the entire category.
        category: The model reference category.
        include_backend_variations: Whether to include per-backend breakdown (text models only).

    Returns:
        CategoryDeletionRiskResponse with models and summary.

    """
    # Analyze all models
    risk_models = self.analyze_models(
        model_records=model_records,
        model_statistics=model_statistics,
        category_total_usage=category_total_usage,
        category=category,
        include_backend_variations=include_backend_variations,
    )

    # Calculate summary
    summary = CategoryDeletionRiskSummary.from_risk_models(risk_models)

    # Create response
    return CategoryDeletionRiskResponse(
        category=category,
        category_total_month_usage=category_total_usage,
        total_count=len(risk_models),
        returned_count=len(risk_models),
        offset=0,
        limit=None,
        models=risk_models,
        summary=summary,
    )