| 1 | """Parallel multi-entity fan-out for the --competitors flag. |
| 2 | |
| 3 | The orchestrator accepts a `main_runner()` for the topic and a |
| 4 | `competitor_runner(entity)` for each peer. It parallelizes their execution |
| 5 | via a `ThreadPoolExecutor` and collects per-entity Reports. Per-entity |
| 6 | failures are logged and dropped; the run survives as long as the main topic |
| 7 | plus at least one competitor succeed. |
| 8 | |
| 9 | This module owns no business logic about pipeline arguments — the caller |
| 10 | (scripts/last30days.py main) builds the closures with the appropriate |
| 11 | config, depth, and overrides for each entity. |
| 12 | """ |
| 13 | |
| 14 | from __future__ import annotations |
| 15 | |
| 16 | from concurrent.futures import ThreadPoolExecutor, as_completed |
| 17 | from typing import Callable |
| 18 | |
| 19 | from . import log, schema, youtube_yt |
| 20 | |
| 21 | # Sub-runs hit the same upstream APIs as the main topic. Cap parallelism so a |
| 22 | # 6-way fan-out does not stampede a single backend's rate limit. |
| 23 | MAX_PARALLEL_SUBRUNS = 6 |
| 24 | |
| 25 | |
| 26 | def _log(msg: str) -> None: |
| 27 | log.source_log("Fanout", msg, tty_only=False) |
| 28 | |
| 29 | |
| 30 | def run_competitor_fanout( |
| 31 | *, |
| 32 | main_topic: str, |
| 33 | main_runner: Callable[[], schema.Report], |
| 34 | competitors: list[str], |
| 35 | competitor_runner: Callable[[str], schema.Report], |
| 36 | ) -> list[tuple[str, schema.Report]]: |
| 37 | """Run main + competitor pipelines in parallel; return surviving reports. |
| 38 | |
| 39 | Args: |
| 40 | main_topic: Display label for the user's primary topic. |
| 41 | main_runner: Zero-arg callable returning the main topic's Report. |
| 42 | competitors: Ordered list of competitor entity names. |
| 43 | competitor_runner: Callable(entity_name) -> Report for each peer. |
| 44 | |
| 45 | Returns: |
| 46 | Ordered list of (entity_name, Report) tuples for runs that succeeded. |
| 47 | Empty list if every run raised; the caller decides how to surface |
| 48 | partial-failure modes. |
| 49 | |
| 50 | ``main_topic`` is NOT guaranteed to be present: a main run that raised |
| 51 | is dropped like any other. Since the render treats element 0 as the |
| 52 | comparison's subject, a caller must verify ``main_topic`` survived |
| 53 | before using the list, or it will silently head the report with a |
| 54 | competitor. |
| 55 | """ |
| 56 | if not competitors: |
| 57 | report = main_runner() |
| 58 | return [(main_topic, report)] |
| 59 | |
| 60 | # One clear for the whole comparison so entity sub-runs share the YouTube |
| 61 | # search cache without inheriting a prior run's results in this process. |
| 62 | youtube_yt.reset_search_cache() |
| 63 | |
| 64 | workers = min(len(competitors) + 1, MAX_PARALLEL_SUBRUNS) |
| 65 | |
| 66 | def _run_one(label: str, fn: Callable[[], schema.Report]) -> tuple[str, schema.Report | None, Exception | None]: |
| 67 | try: |
| 68 | return label, fn(), None |
| 69 | except Exception as exc: |
| 70 | return label, None, exc |
| 71 | |
| 72 | submissions: list[tuple[str, Callable[[], schema.Report]]] = [ |
| 73 | (main_topic, main_runner), |
| 74 | ] |
| 75 | for entity in competitors: |
| 76 | submissions.append((entity, lambda e=entity: competitor_runner(e))) |
| 77 | |
| 78 | with ThreadPoolExecutor(max_workers=workers) as executor: |
| 79 | futures = { |
| 80 | executor.submit(_run_one, label, fn): label |
| 81 | for label, fn in submissions |
| 82 | } |
| 83 | results: dict[str, schema.Report] = {} |
| 84 | for future in as_completed(futures): |
| 85 | label, report, exc = future.result() |
| 86 | if exc is not None: |
| 87 | _log(f"Sub-run failed for {label!r}: {type(exc).__name__}: {exc}") |
| 88 | continue |
| 89 | assert report is not None |
| 90 | results[label] = report |
| 91 | |
| 92 | # Preserve the original submission order rather than completion order so |
| 93 | # the comparison render is deterministic across runs. |
| 94 | return [(label, results[label]) for label, _ in submissions if label in results] |
| 95 |