diff --git a/scripts/sxyprn_reap_dead.py b/scripts/sxyprn_reap_dead.py index 05ebaa6..2dc6ad7 100644 --- a/scripts/sxyprn_reap_dead.py +++ b/scripts/sxyprn_reap_dead.py @@ -22,7 +22,13 @@ miesięcznie). Dlatego niska równoległość i odstęp między pobraniami, żeb rate-limitem ani banem IP. To celowo wolne. Wznawialne za darmo: bierzemy tylko żywe źródła, więc ponowny przebieg pomija już oznaczone. -Użycie (w kontenerze workera): +UWAGA NA RESTART: `docker compose restart worker` UBIJA to zadanie, jesli chodzi w +kontenerze workera (2026-08-27: stracilem tak przebieg po 151 tys. sprawdzonych, log +urwal sie 4 minuty przed restartem). Odpalaj przez `docker compose run -d --rm worker`, +bo to OSOBNY kontener, ktorego restart workera nie dotyka. Wznawianie jest darmowe: +bierzemy tylko zywe zrodla, wiec ponowne odpalenie samo pomija juz oznaczone. + +Użycie: python3 /srv/scripts/sxyprn_reap_dead.py --limit 300 --dry-run python3 /srv/scripts/sxyprn_reap_dead.py --limit 5000 """ @@ -100,25 +106,36 @@ def main(argv: list[str] | None = None) -> int: sid, page_url = row return sid, _probe(page_url) - with ThreadPoolExecutor(max_workers=_CONCURRENCY) as pool: - for i, (sid, verdict) in enumerate(pool.map(work, rows), 1): - cnt[verdict] += 1 - if verdict == _DEAD and not a.dry_run: - with session_scope() as s: - ubite += s.execute( - text( - "UPDATE playback_sources SET dead_at = now(), " - "dead_reason = 'sxyprn-reap: Post Not Found' " - "WHERE id = :id AND dead_at IS NULL" - ), - {"id": sid}, - ).rowcount - if i % 200 == 0: - log.info( - "[%d/%d] żywe=%d martwe=%d nieznane=%d, ubite=%d (%.0f min)", - i, len(rows), cnt[_ALIVE], cnt[_DEAD], cnt[_UNKNOWN], ubite, - (time.time() - t0) / 60, - ) + # Porcjami, ze SWIEZA pula na kazda porcje. `pool.map` na calej liscie kolejkowal + # 188 tys. zadan naraz, wiec postep byl widoczny dopiero na koncu, a przerwanie w + # polowie nie zostawialo zadnego punktu zaczepienia. + _PORCJA = 500 + for start in range(0, len(rows), _PORCJA): + porcja = rows[start : start + _PORCJA] + do_ubicia: list = [] + with ThreadPoolExecutor(max_workers=_CONCURRENCY) as pool: + for sid, verdict in pool.map(work, porcja): + cnt[verdict] += 1 + if verdict == _DEAD: + do_ubicia.append(sid) + if do_ubicia and not a.dry_run: + # Jeden UPDATE na porcje zamiast jednego na wiersz. + with session_scope() as s: + ubite += s.execute( + text( + "UPDATE playback_sources SET dead_at = now(), " + "dead_reason = 'sxyprn-reap: Post Not Found' " + "WHERE id = ANY(:ids) AND dead_at IS NULL" + ), + {"ids": do_ubicia}, + ).rowcount + log.info( + "[%d/%d] żywe=%d martwe=%d nieznane=%d, ubite=%d (%.0f min)", + min(start + _PORCJA, len(rows)), len(rows), + cnt[_ALIVE], cnt[_DEAD], cnt[_UNKNOWN], ubite, (time.time() - t0) / 60, + ) + for h in logging.getLogger().handlers: + h.flush() # bez tego log do pliku milczal godzinami (buforowanie) total = sum(cnt.values()) log.info(