fix(scheduler): read inbound bytes from the cached local_path, not read_file(f.name)
In real (paramiko) mode, `_download_and_parse` called `client.read_file(f.name)` with a bare filename. paramiko's `sftp.open(f.name)` opens at the SFTP root, not at `paths.inbound` (FromHPE) — so the scheduler would fail to download any file in real mode even with the path swap from the previous commit. But this round-trip is also unnecessary: `_list_inbound_paramiko` already downloads each entry into the local cache (cache_path) and returns it as `InboundFile.local_path` as part of the listing pass. Reading from disk is faster than re-fetching and avoids the path bug. Stub mode was already reading from `f.local_path`. Now both modes do, which is the simpler invariant. Verified: 36 tests pass (test_scheduler, test_api_scheduler, test_sftp_stub, test_sftp_paramiko).
This commit is contained in:
@@ -650,20 +650,21 @@ class Scheduler:
|
|||||||
def _download_and_parse(
|
def _download_and_parse(
|
||||||
self, f: InboundFile, file_type: str,
|
self, f: InboundFile, file_type: str,
|
||||||
) -> tuple[Path, str, int]:
|
) -> tuple[Path, str, int]:
|
||||||
"""Download from MFT, run the right handler. Returns (path, parser, count).
|
"""Run the right handler on one inbound file. Returns (path, parser, count).
|
||||||
|
|
||||||
Stub mode: ``f.local_path`` already points at the staged file
|
Both stub and real modes read from ``f.local_path`` — the
|
||||||
(set by ``SftpClient._list_inbound_stub``). Real mode: the
|
inbound file is already on disk:
|
||||||
remote name is ``f.name`` and we round-trip through paramiko.
|
* Stub mode: ``_list_inbound_stub`` points ``local_path`` at
|
||||||
|
the operator-dropped staging file.
|
||||||
|
* Real mode: ``_list_inbound_paramiko`` downloads each
|
||||||
|
``listdir_attr`` entry into the local cache as part of the
|
||||||
|
listing pass. Re-reading from the MFT would require
|
||||||
|
``SftpClient.read_file`` with a full remote path, which the
|
||||||
|
scheduler was passing just ``f.name`` for (i.e. a bare
|
||||||
|
filename at the SFTP root, not the inbound dir). Use the
|
||||||
|
cached bytes instead.
|
||||||
"""
|
"""
|
||||||
if self._sftp_block.stub:
|
content = f.local_path.read_bytes()
|
||||||
# In stub mode the InboundFile already has a local_path;
|
|
||||||
# reading the staged bytes directly avoids the stub's
|
|
||||||
# remote-path semantics (which expect a full inbound path).
|
|
||||||
content = f.local_path.read_bytes()
|
|
||||||
else:
|
|
||||||
client = self._sftp_client_factory(self._sftp_block)
|
|
||||||
content = client.read_file(f.name)
|
|
||||||
text = content.decode("utf-8")
|
text = content.decode("utf-8")
|
||||||
handler = HANDLERS[file_type]
|
handler = HANDLERS[file_type]
|
||||||
parser_used, claim_count = handler(text, f.name)
|
parser_used, claim_count = handler(text, f.name)
|
||||||
|
|||||||
Reference in New Issue
Block a user