"""ARCHIVED — v1 run_distribuicao (per-company × per-tribunal).

Superseded on 2026-04-19 by the tribunal-sweep version in jobs/scheduler.py.
Kept as a code-only reference (not importable). See _archive/distribuicao_v1_per_company.py
for the matching scraper code.
"""
async def run_distribuicao(
    date_from: date | None = None,
    date_to: date | None = None,
    types: tuple[str, ...] = ("internal", "competitor"),
    company_id: str | None = None,
    isolated: bool | None = False,
) -> str:
    """Scrape Distribuição for a date range.

    When both `date_from` and `date_to` are None (scheduled run), looks up the
    last successful scrape's `date_to` and scrapes from (last + 1) through
    yesterday. Capped at 30 days back to avoid runaway ranges. If no history,
    defaults to yesterday only.

    For manual runs, `date_from` is hard-capped to `date_to - 180 days`: the
    CITIUS web filter only returns rows from roughly the last 6 months no
    matter what range we send, so older windows just burn HTTP round-trips.

    Commits per company — if the process is killed mid-run, completed companies
    stay persisted and the next catch-up picks up where we left off.
    """
    today = date.today()
    yesterday = today - timedelta(days=1)
    auto_mode = date_from is None and date_to is None

    if auto_mode and company_id:
        # On-create backfill: always the full 180-day window, regardless of the
        # global run history (which covers OTHER companies, not this one).
        date_from = yesterday - timedelta(days=DISTRIBUICAO_MAX_LOOKBACK_DAYS)
        date_to = yesterday
    elif auto_mode:
        async with AsyncSessionLocal() as session:
            # Isolated and main runs keep separate catch-up histories. If no
            # isolated history exists yet, fall back to the full 180d window so
            # the first isolated run actually backfills.
            last = await _last_successful_distribuicao_date(
                session, isolated=bool(isolated)
            )
        if last and last < yesterday:
            date_from = max(last + timedelta(days=1), yesterday - timedelta(days=30))
            date_to = yesterday
        elif isolated:
            date_from = yesterday - timedelta(days=DISTRIBUICAO_MAX_LOOKBACK_DAYS)
            date_to = yesterday
        else:
            date_from = date_to = yesterday
    elif date_from is None:
        date_from = date_to
    elif date_to is None:
        date_to = date_from
    assert date_from is not None and date_to is not None
    if date_from > date_to:
        date_from, date_to = date_to, date_from

    # Hard cap manual ranges. CITIUS only serves ~6 months — anything older just
    # adds latency for empty responses.
    capped_from = date_to - timedelta(days=DISTRIBUICAO_MAX_LOOKBACK_DAYS)
    if date_from < capped_from:
        logger.info(
            "distribuicao: capping date_from %s -> %s (CITIUS 180d limit)",
            date_from, capped_from,
        )
        date_from = capped_from

    async with AsyncSessionLocal() as session:
        params_log: dict[str, Any] = {
            "date_from": date_from.isoformat(),
            "date_to": date_to.isoformat(),
            "types": list(types),
            "auto": auto_mode,
            "isolated": bool(isolated) if isolated is not None else False,
        }
        if company_id:
            params_log["company_id"] = company_id
            params_log["backfill_new"] = True
        log_id = await _log_start(session, "distribuicao", params_log)
        companies = await _monitored_companies(
            session, types=types, company_id=company_id, isolated=isolated,
        )

    try:
        if not companies:
            async with AsyncSessionLocal() as session:
                await _log_finish(session, log_id, status="ok", rows_seen=0, rows_new=0)
            return log_id

        tribunals = await discover_tribunal_codes()
        total_seen = 0
        total_new = 0
        error_count = 0

        # Seed the progress denominator so the UI can compute % immediately.
        async with AsyncSessionLocal() as session:
            await session.execute(
                text("UPDATE scraping_logs SET progress_total = :t WHERE id = :id"),
                {"t": len(companies), "id": log_id},
            )
            await session.commit()

        from app.config import settings as _settings
        sem = asyncio.Semaphore(_settings.SCRAPE_CONCURRENCY)

        async def worker(
            company: dict[str, Any], t: dict[str, str], search: str
        ) -> list[dict[str, Any]]:
            nonlocal error_count
            async with sem:
                try:
                    return await scrape_distribuicao_range(
                        t["value"], t["label"], date_from, date_to,
                        [company], party_filter=search,
                    )
                except Exception as e:
                    error_count += 1
                    logger.warning(
                        "distribuicao company=%s tribunal=%s failed: %r",
                        company["legal_name"], t["label"], e,
                    )
                    return []

        # Process companies sequentially; within each, tribunals run concurrently
        # (bounded by the shared semaphore). Commit after EACH company, so if the
        # backend is restarted or crashes, finished companies stay persisted.
        for idx, company in enumerate(companies, 1):
            search = short_search_term(company["legal_name"])
            if not search:
                logger.info(
                    "distribuicao [%d/%d] %s: SKIPPED (ambiguous/short name)",
                    idx, len(companies), company["legal_name"][:40],
                )
                continue
            results = await asyncio.gather(
                *(worker(company, t, search) for t in tribunals),
                return_exceptions=False,
            )
            matches: list[dict[str, Any]] = [row for r in results for row in r]
            if matches: