_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))
 
 
 # --- накопитель карточек ---------------------------------------------------