diff --git a/_wt-mskcol/tradein-mvp/scripts/local-avito-msk/collect.py b/_wt-mskcol/tradein-mvp/scripts/local-avito-msk/collect.py index f172b10b..5b0f2461 100644 --- a/_wt-mskcol/tradein-mvp/scripts/local-avito-msk/collect.py +++ b/_wt-mskcol/tradein-mvp/scripts/local-avito-msk/collect.py @@ -434,18 +434,35 @@ def build_finalize_sql(batch_id: str, query: str, notes: str) -> str: ) -def run_psql(sql: str, ssh_host: str, container: str, db_user: str, db_name: str) -> None: +def run_psql(sql: str, ssh_host: str, container: str, db_user: str, db_name: str, + attempts: int = 4) -> None: + """Заливка батча через ssh с ретраем на обрыв транспорта. + + Прогон длится часами, и ssh рвётся: живьём поймано «Connection reset by peer» + (ssh возвращает 255) прямо посреди заливки — весь прогон умирал, а несброшенный + батч терялся. Ретраить безопасно: SQL идемпотентен (batch через ON CONFLICT DO + NOTHING, карточки через ON CONFLICT (source_id,batch_id,kind) DO NOTHING). + + Ретраится ТОЛЬКО транспорт (ssh 255). Ошибка самого psql (ON_ERROR_STOP, любой + другой код) — это дефект данных или SQL, её повтор не лечит: поднимаем сразу. + """ cmd = [ "ssh", ssh_host, f"docker exec -i {container} psql -U {db_user} -d {db_name} " "-v ON_ERROR_STOP=1 -f -", ] - proc = subprocess.run(cmd, input=sql.encode("utf-8"), capture_output=True) - out = (proc.stdout + proc.stderr).decode("utf-8", "replace").strip() - if proc.returncode != 0: - raise RuntimeError(f"psql через ssh вернул {proc.returncode}:\n{out}") - if out: - print(f" psql: {out}", flush=True) + for i in range(attempts): + proc = subprocess.run(cmd, input=sql.encode("utf-8"), capture_output=True) + out = (proc.stdout + proc.stderr).decode("utf-8", "replace").strip() + if proc.returncode == 0: + if out: + print(f" psql: {out}", flush=True) + return + if proc.returncode != 255 or i == attempts - 1: + raise RuntimeError(f"psql через ssh вернул {proc.returncode}:\n{out}") + tail = out.splitlines()[-1] if out else "без вывода" + print(f" ssh оборвался ({tail}), повтор заливки {i + 2}/{attempts}", flush=True) + time.sleep(15.0 * (i + 1)) # --- накопитель карточек ---------------------------------------------------