Function evaluate_series_rules_impl

apps/channels/tasks.py:452–477

Synchronous implementation of series rule evaluation; returns details for debugging.

(tvg_id: str | None = None)

def evaluate_series_rules_impl(tvg_id: str | None = None):
    """Synchronous implementation of series rule evaluation; returns details for debugging."""
    result = {"scheduled": 0, "details": []}

    # Serialize all invocations to prevent concurrent evaluations from
    # racing to create duplicate recordings (e.g. multiple EPG sources
    # refreshing simultaneously each firing evaluate_series_rules.delay()).
    # If Redis is unavailable, proceed without lock — the primary and
    # secondary dedup guards still prevent duplicates.
    lock_acquired = False
    try:
        lock_acquired = acquire_task_lock('evaluate_series_rules', 'all')
        if not lock_acquired:
            result["details"].append({"status": "skipped", "reason": "concurrent evaluation in progress"})
            return result
    except (ConnectionError, OSError, AttributeError):
        logger.warning("Could not acquire series rule evaluation lock (Redis unavailable), proceeding without lock")

    try:
        return _evaluate_series_rules_locked(tvg_id, result)
    finally:
        if lock_acquired:
            try:
                release_task_lock('evaluate_series_rules', 'all')
            except (ConnectionError, OSError, AttributeError):
                logger.warning("Could not release series rule evaluation lock")


def _evaluate_series_rules_locked(tvg_id, result):