Merge branch 'codex/docling-281' into development

Author: fffoivos

src/glossapi/scripts/openarchives_pdf_stage_pull.py

@@ -45,6 +45,7 @@ class TransferItem:
 """
 
 PDF_NAME_PATTERN = re.compile(r"([A-Za-z0-9._-]+\.pdf(?:\.[A-Za-z0-9_-]+)?)", re.IGNORECASE)
+FILENAME_KEYS = ("filename", "canonical_filename", "md_filename", "source_filename")
 
 
 def parse_args(argv: Optional[Sequence[str]] = None) -> argparse.Namespace:
@@ -56,6 +57,7 @@ def parse_args(argv: Optional[Sequence[str]] = None) -> argparse.Namespace:
     p.add_argument("--work-root", required=True, help="Root directory for downloads, partials, logs, and state.")
     p.add_argument("--remote-host", default="debian@83.212.80.170")
     p.add_argument("--password-env", default="GREECE_BOX_PASSWORD", help="Environment variable containing the remote SSH password.")
+    p.add_argument("--transport", choices=("sftp", "rsync"), default="sftp")
     p.add_argument("--max-attempts", type=int, default=20)
     p.add_argument("--connect-timeout", type=int, default=30)
     p.add_argument("--io-timeout", type=int, default=180)
@@ -323,6 +325,40 @@ def append_event(path: Path, payload: dict) -> None:
         handle.write(json.dumps(payload, ensure_ascii=False) + "\n")
 
 
+def sshpass_env(password_env: str) -> dict[str, str]:
+    env = os.environ.copy()
+    secret = env.get(password_env)
+    if not secret:
+        raise SystemExit(f"Password env var '{password_env}' is not set.")
+    env["SSHPASS"] = secret
+    return env
+
+
+def ssh_transport_options(connect_timeout: int) -> list[str]:
+    return [
+        "-o",
+        "BatchMode=no",
+        "-o",
+        "PreferredAuthentications=password",
+        "-o",
+        "PubkeyAuthentication=no",
+        "-o",
+        "KbdInteractiveAuthentication=yes",
+        "-o",
+        f"ConnectTimeout={int(connect_timeout)}",
+        "-o",
+        "ServerAliveInterval=15",
+        "-o",
+        "ServerAliveCountMax=3",
+        "-o",
+        "ConnectionAttempts=3",
+        "-o",
+        "StrictHostKeyChecking=no",
+        "-o",
+        "UserKnownHostsFile=/tmp/greece_box_known_hosts",
+    ]
+
+
 def canonicalize_pdf_name(raw: str) -> Optional[str]:
     text = os.path.basename(str(raw).strip())
     if not text:
@@ -337,6 +373,74 @@ def canonicalize_pdf_name(raw: str) -> Optional[str]:
     return None
 
 
+def _walk_json_strings(obj) -> Iterable[str]:
+    if isinstance(obj, dict):
+        for key, value in obj.items():
+            if isinstance(key, str):
+                yield key
+            yield from _walk_json_strings(value)
+    elif isinstance(obj, list):
+        for item in obj:
+            yield from _walk_json_strings(item)
+    elif isinstance(obj, str):
+        yield obj
+
+
+def _extract_priority_filenames_from_csv(path: Path) -> set[str]:
+    results: set[str] = set()
+    with path.open("r", encoding="utf-8", errors="ignore", newline="") as handle:
+        reader = csv.DictReader(handle)
+        fields = {field.strip() for field in (reader.fieldnames or []) if field}
+        keyed = any(key in fields for key in FILENAME_KEYS)
+        for row in reader:
+            if keyed:
+                for key in FILENAME_KEYS:
+                    value = row.get(key)
+                    if value:
+                        canonical = canonicalize_pdf_name(value)
+                        if canonical is not None:
+                            results.add(canonical)
+                            break
+            else:
+                for value in row.values():
+                    if not value:
+                        continue
+                    for match in PDF_NAME_PATTERN.findall(str(value)):
+                        canonical = canonicalize_pdf_name(match)
+                        if canonical is not None:
+                            results.add(canonical)
+    return results
+
+
+def _extract_priority_filenames_from_json(path: Path) -> set[str]:
+    results: set[str] = set()
+    data = json.loads(path.read_text(encoding="utf-8", errors="ignore"))
+    for text in _walk_json_strings(data):
+        canonical = canonicalize_pdf_name(text)
+        if canonical is not None:
+            results.add(canonical)
+            continue
+        for match in PDF_NAME_PATTERN.findall(text):
+            canonical = canonicalize_pdf_name(match)
+            if canonical is not None:
+                results.add(canonical)
+    return results
+
+
+def _extract_priority_filenames_from_text(path: Path) -> set[str]:
+    results: set[str] = set()
+    text = path.read_text(encoding="utf-8", errors="ignore")
+    for line in text.splitlines():
+        canonical = canonicalize_pdf_name(line)
+        if canonical is not None:
+            results.add(canonical)
+    for match in PDF_NAME_PATTERN.findall(text):
+        canonical = canonicalize_pdf_name(match)
+        if canonical is not None:
+            results.add(canonical)
+    return results
+
+
 def load_priority_filenames(priority_dir: Path) -> set[str]:
     results: set[str] = set()
     if not priority_dir.exists():
@@ -348,17 +452,60 @@ def load_priority_filenames(priority_dir: Path) -> set[str]:
         if direct is not None:
             results.add(direct)
             continue
+        suffix = path.suffix.lower()
         try:
-            text = path.read_text(encoding="utf-8", errors="ignore")
+            if suffix == ".csv":
+                results.update(_extract_priority_filenames_from_csv(path))
+            elif suffix == ".json":
+                results.update(_extract_priority_filenames_from_json(path))
+            elif suffix in {".txt", ".list", ".lst", ".log"}:
+                results.update(_extract_priority_filenames_from_text(path))
+            else:
+                continue
         except Exception:
             continue
-        for match in PDF_NAME_PATTERN.findall(text):
-            canonical = canonicalize_pdf_name(match)
-            if canonical is not None:
-                results.add(canonical)
     return results
 
 
+def rsync_one(
+    *,
+    remote_host: str,
+    remote_path: str,
+    temp_path: Path,
+    password_env: str,
+    connect_timeout: int,
+    io_timeout: int,
+) -> subprocess.CompletedProcess[str]:
+    ssh_cmd = (
+        "ssh "
+        "-o BatchMode=no "
+        "-o PreferredAuthentications=password "
+        "-o PubkeyAuthentication=no "
+        "-o KbdInteractiveAuthentication=yes "
+        f"-o ConnectTimeout={int(connect_timeout)} "
+        "-o ServerAliveInterval=15 "
+        "-o ServerAliveCountMax=3 "
+        "-o ConnectionAttempts=3 "
+        "-o StrictHostKeyChecking=no "
+        "-o UserKnownHostsFile=/tmp/greece_box_known_hosts"
+    )
+    cmd = [
+        "sshpass",
+        "-e",
+        "rsync",
+        "-av",
+        "--partial",
+        "--append-verify",
+        "--inplace",
+        f"--timeout={int(io_timeout)}",
+        "-e",
+        ssh_cmd,
+        f"{remote_host}:{remote_path}",
+        str(temp_path),
+    ]
+    return subprocess.run(cmd, capture_output=True, text=True, env=sshpass_env(password_env))
+
+
 def sftp_one(
     *,
     remote_host: str,
@@ -372,37 +519,13 @@ def sftp_one(
         "sshpass",
         "-e",
         "sftp",
-        "-o",
-        "BatchMode=no",
-        "-o",
-        "PreferredAuthentications=password",
-        "-o",
-        "PubkeyAuthentication=no",
-        "-o",
-        "KbdInteractiveAuthentication=yes",
-        "-o",
-        f"ConnectTimeout={int(connect_timeout)}",
-        "-o",
-        "ServerAliveInterval=15",
-        "-o",
-        "ServerAliveCountMax=3",
-        "-o",
-        "ConnectionAttempts=3",
-        "-o",
-        "StrictHostKeyChecking=no",
-        "-o",
-        "UserKnownHostsFile=/tmp/greece_box_known_hosts",
+        *ssh_transport_options(connect_timeout),
         "-b",
         "-",
         remote_host,
     ]
-    env = os.environ.copy()
-    secret = env.get(password_env)
-    if not secret:
-        raise SystemExit(f"Password env var '{password_env}' is not set.")
-    env["SSHPASS"] = secret
     batch = f'reget "{remote_path}" "{temp_path}"\n'
-    return subprocess.run(cmd, capture_output=True, text=True, env=env, input=batch)
+    return subprocess.run(cmd, capture_output=True, text=True, env=sshpass_env(password_env), input=batch)
 
 
 def run(argv: Optional[Sequence[str]] = None) -> int:
@@ -496,6 +619,7 @@ def refresh_priorities() -> dict[str, int]:
             current_path,
             {
                 "updated_at": utc_now(),
+                "transport": str(args.transport),
                 "canonical_filename": canonical,
                 "remote_path": remote_path,
                 "remote_size_bytes": remote_size,
@@ -509,6 +633,7 @@ def refresh_priorities() -> dict[str, int]:
             {
                 "ts": utc_now(),
                 "event": "start",
+                "transport": str(args.transport),
                 "canonical_filename": canonical,
                 "remote_path": remote_path,
                 "remote_size_bytes": remote_size,
@@ -517,14 +642,18 @@ def refresh_priorities() -> dict[str, int]:
             },
         )
 
-        result = sftp_one(
-            remote_host=str(args.remote_host),
-            remote_path=remote_path,
-            temp_path=temp_path,
-            password_env=str(args.password_env),
-            connect_timeout=int(args.connect_timeout),
-            io_timeout=int(args.io_timeout),
-        )
+        transfer_kwargs = {
+            "remote_host": str(args.remote_host),
+            "remote_path": remote_path,
+            "temp_path": temp_path,
+            "password_env": str(args.password_env),
+            "connect_timeout": int(args.connect_timeout),
+            "io_timeout": int(args.io_timeout),
+        }
+        if str(args.transport) == "rsync":
+            result = rsync_one(**transfer_kwargs)
+        else:
+            result = sftp_one(**transfer_kwargs)
 
         if result.returncode == 0 and temp_path.exists():
             actual_size = temp_path.stat().st_size
@@ -543,6 +672,7 @@ def refresh_priorities() -> dict[str, int]:
                     {
                         "ts": utc_now(),
                         "event": "completed",
+                        "transport": str(args.transport),
                         "canonical_filename": canonical,
                         "size_bytes": actual_size,
                     },
@@ -556,6 +686,7 @@ def refresh_priorities() -> dict[str, int]:
                 {
                     "ts": utc_now(),
                     "event": "failed",
+                    "transport": str(args.transport),
                     "canonical_filename": canonical,
                     "return_code": int(result.returncode),
                     "partial_size_bytes": actual_size,

tests/test_openarchives_pdf_stage_pull.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import subprocess
 from pathlib import Path
 
 from glossapi.scripts.openarchives_pdf_stage_pull import (
@@ -8,6 +9,7 @@
     canonicalize_pdf_name,
     load_priority_filenames,
     read_manifest,
+    run,
 )
 
 
@@ -131,3 +133,59 @@ def test_transfer_state_priorities_are_selected_first(tmp_path: Path) -> None:
     assert counts["priority_total"] == 1
     assert counts["priority_pending"] == 1
     state.close()
+
+
+def test_load_priority_filenames_ignores_parquet_and_reads_csv_columns(tmp_path: Path) -> None:
+    priority_dir = tmp_path / "priority"
+    priority_dir.mkdir()
+    (priority_dir / "unreachable_from_source_20260331.csv").write_text(
+        "filename,source_unreachable_reason\n"
+        "ZFV_051.pdf,connect_timeout\n"
+        "ZGA_056.pdf,connect_timeout\n",
+        encoding="utf-8",
+    )
+    (priority_dir / "unreachable_from_source_20260331.parquet").write_bytes(b"PAR1junkZXY_999.pdfjunk")
+
+    names = load_priority_filenames(priority_dir)
+
+    assert names == {"ZFV_051.pdf", "ZGA_056.pdf"}
+
+
+def test_run_uses_rsync_transport_when_requested(tmp_path: Path, monkeypatch) -> None:
+    manifest = tmp_path / "manifest.tsv"
+    _write_manifest(manifest)
+    work_root = tmp_path / "work"
+    seen: list[str] = []
+
+    def _fake_rsync_one(**kwargs):
+        seen.append("rsync")
+        Path(kwargs["temp_path"]).parent.mkdir(parents=True, exist_ok=True)
+        Path(kwargs["temp_path"]).write_bytes(b"x" * 10)
+        return subprocess.CompletedProcess(args=["rsync"], returncode=0, stdout="", stderr="")
+
+    def _fake_sftp_one(**kwargs):
+        seen.append("sftp")
+        return subprocess.CompletedProcess(args=["sftp"], returncode=1, stdout="", stderr="unexpected")
+
+    monkeypatch.setenv("GREECE_BOX_PASSWORD", "secret")
+    monkeypatch.setattr("glossapi.scripts.openarchives_pdf_stage_pull.rsync_one", _fake_rsync_one)
+    monkeypatch.setattr("glossapi.scripts.openarchives_pdf_stage_pull.sftp_one", _fake_sftp_one)
+
+    rc = run(
+        [
+            "--manifest",
+            str(manifest),
+            "--work-root",
+            str(work_root),
+            "--transport",
+            "rsync",
+            "--limit",
+            "1",
+            "--summary-interval-seconds",
+            "0",
+        ]
+    )
+
+    assert rc == 0
+    assert seen == ["rsync"]
+    assert (work_root / "downloads" / "AAA_456.pdf").exists()