fix(sftp): case-insensitive inbound regex, skip _warn.txt, add targeted pull
Three changes that unblock the daily inbound pull from Gainwell's FromHPE MFT path: 1. INBOUND_RE now accepts both 'TP' and 'tp' prefixes via inline (?i:TP) scoping — case-folding the whole pattern would let lowercase '837p' / tracking IDs through, which is invalid HCPF. The rest of the pattern is still case-sensitive. 2. _list_inbound_paramiko skips *_warn.txt entries. Gainwell's MFT drops ~583 advisory text-format notes in the same inbound dir; they come first alphabetically and were padding every poll with ~80 min of pointless downloads. 3. New SftpClient.list_inbound_names() + download_inbound() pair gives a metadata-only listing and on-demand fetch. The scheduler's existing full-listing path still works (now without the warn padding); the new path is what the new /api/admin/scheduler/pull-inbound endpoint and the 'cyclone pull-inbound' CLI use to fast-target a date range without paying the cost of a full ~6000-file download. Scheduler.process_inbound_files() runs the same per-file pipeline as a regular tick on the pre-fetched list, so dedup via processed_inbound_files still applies. Tests added in test_filenames.py (lowercase + mixed-case cases), test_sftp_paramiko.py (warn skip + no-download listing), and test_scheduler.py (process_inbound_files idempotency). With this, the daily 385-file pull for 20260624 completes in seconds via 'docker exec cyclone-backend-1 python -m cyclone pull-inbound --date 20260624 --block dzinesco' (or the equivalent POST to /api/admin/scheduler/pull-inbound?date=20260624).
This commit is contained in:
@@ -18,6 +18,7 @@ Additional origins (LAN IPs, staging hosts) can be appended via the
|
|||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import asyncio
|
||||||
import csv
|
import csv
|
||||||
import io
|
import io
|
||||||
import json
|
import json
|
||||||
@@ -36,6 +37,7 @@ from pydantic import ValidationError
|
|||||||
|
|
||||||
from cyclone import __version__, db
|
from cyclone import __version__, db
|
||||||
from cyclone.auth.deps import matrix_gate
|
from cyclone.auth.deps import matrix_gate
|
||||||
|
from cyclone.clearhouse import InboundFile
|
||||||
from cyclone.db import Batch, Claim, ClaimState, Remittance
|
from cyclone.db import Batch, Claim, ClaimState, Remittance
|
||||||
from sqlalchemy import desc, or_
|
from sqlalchemy import desc, or_
|
||||||
from sqlalchemy.exc import IntegrityError
|
from sqlalchemy.exc import IntegrityError
|
||||||
@@ -3419,6 +3421,128 @@ async def scheduler_tick() -> Any:
|
|||||||
return {"ok": True, "tick": result.as_dict()}
|
return {"ok": True, "tick": result.as_dict()}
|
||||||
|
|
||||||
|
|
||||||
|
@app.post("/api/admin/scheduler/pull-inbound", dependencies=[Depends(matrix_gate)])
|
||||||
|
async def scheduler_pull_inbound(
|
||||||
|
date: str = Query(
|
||||||
|
..., pattern=r"^\d{8}$",
|
||||||
|
description="Date filter as YYYYMMDD; only filenames whose 8-digit "
|
||||||
|
"timestamp (the 9th positional group in the inbound "
|
||||||
|
"filename) matches are downloaded and processed.",
|
||||||
|
),
|
||||||
|
file_types: str | None = Query(
|
||||||
|
default=None,
|
||||||
|
description="Optional comma-separated whitelist of file_types "
|
||||||
|
"(999, TA1, 277, 277CA, 835). Defaults to 999+TA1.",
|
||||||
|
),
|
||||||
|
limit: int = Query(default=2000, ge=1, le=10000),
|
||||||
|
) -> Any:
|
||||||
|
"""Targeted pull: list, filter to a date, download, and process.
|
||||||
|
|
||||||
|
Bypasses the alphabetical full-listing pass. Workflow:
|
||||||
|
1. ``SftpClient.list_inbound_names()`` — sub-second metadata-only
|
||||||
|
listing of the inbound MFT dir (skips ``*_warn.txt``).
|
||||||
|
2. Client-side filter: keep files whose 8-digit timestamp
|
||||||
|
substring equals ``date`` and whose ``file_type`` is in the
|
||||||
|
allowlist.
|
||||||
|
3. ``SftpClient.download_inbound(f)`` for each — fetches bytes
|
||||||
|
into the local cache.
|
||||||
|
4. ``Scheduler.process_inbound_files(files)`` — runs the same
|
||||||
|
per-file pipeline as a regular tick (already-processed files
|
||||||
|
are deduped via ``processed_inbound_files``).
|
||||||
|
|
||||||
|
Use this for the daily "process today's 999s" workflow without
|
||||||
|
paying the cost of downloading the full inbound set.
|
||||||
|
|
||||||
|
Returns ``{"ok": True, "summary": {...}}`` with
|
||||||
|
``listed / matched / downloaded / processed / skipped / errored``
|
||||||
|
counters and the date / file_type filters applied.
|
||||||
|
"""
|
||||||
|
import time
|
||||||
|
|
||||||
|
from cyclone.clearhouse import SftpClient
|
||||||
|
from cyclone.edi.filenames import (
|
||||||
|
ALLOWED_FILE_TYPES,
|
||||||
|
parse_inbound_filename,
|
||||||
|
)
|
||||||
|
from cyclone.providers import SftpBlock
|
||||||
|
|
||||||
|
sched = _scheduler_or_503()
|
||||||
|
block: SftpBlock = sched._sftp_block # noqa: SLF001 — internal but stable
|
||||||
|
client = SftpClient(block)
|
||||||
|
|
||||||
|
if file_types:
|
||||||
|
wanted = {t.strip().upper() for t in file_types.split(",") if t.strip()}
|
||||||
|
unknown = wanted - ALLOWED_FILE_TYPES
|
||||||
|
if unknown:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=400,
|
||||||
|
detail=f"file_types {sorted(unknown)!r} not in "
|
||||||
|
f"{sorted(ALLOWED_FILE_TYPES)}",
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
wanted = {"999", "TA1"} # daily default — what the operator needs
|
||||||
|
|
||||||
|
started = time.monotonic()
|
||||||
|
try:
|
||||||
|
# Single SFTP listdir — fast, no download.
|
||||||
|
all_files = await asyncio.to_thread(client.list_inbound_names)
|
||||||
|
except Exception as exc:
|
||||||
|
log.exception("SFTP list_inbound_names failed")
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=502,
|
||||||
|
detail=f"SFTP list failed: {type(exc).__name__}: {exc}",
|
||||||
|
) from exc
|
||||||
|
|
||||||
|
listed = len(all_files)
|
||||||
|
matched: list[InboundFile] = []
|
||||||
|
for f in all_files:
|
||||||
|
if f.name.find(date) == -1:
|
||||||
|
continue
|
||||||
|
try:
|
||||||
|
parsed = parse_inbound_filename(f.name)
|
||||||
|
except ValueError:
|
||||||
|
continue
|
||||||
|
if parsed.file_type not in wanted:
|
||||||
|
continue
|
||||||
|
matched.append(f)
|
||||||
|
if len(matched) >= limit:
|
||||||
|
break
|
||||||
|
|
||||||
|
# Download in parallel-ish via to_thread (SftpClient serializes per
|
||||||
|
# connection; the overhead is dominated by the SFTP round trip).
|
||||||
|
downloaded = 0
|
||||||
|
download_errors: list[str] = []
|
||||||
|
for f in matched:
|
||||||
|
try:
|
||||||
|
await asyncio.to_thread(client.download_inbound, f)
|
||||||
|
downloaded += 1
|
||||||
|
except Exception as exc: # noqa: BLE001
|
||||||
|
log.warning("Failed to download %s: %s", f.name, exc)
|
||||||
|
download_errors.append(f"{f.name}: {type(exc).__name__}: {exc}")
|
||||||
|
|
||||||
|
# Hand off to the scheduler pipeline (idempotent; dedupes via
|
||||||
|
# processed_inbound_files).
|
||||||
|
tick = await sched.process_inbound_files(matched)
|
||||||
|
duration = round(time.monotonic() - started, 3)
|
||||||
|
return {
|
||||||
|
"ok": True,
|
||||||
|
"summary": {
|
||||||
|
"date": date,
|
||||||
|
"file_types": sorted(wanted),
|
||||||
|
"limit": limit,
|
||||||
|
"listed": listed,
|
||||||
|
"matched": len(matched),
|
||||||
|
"downloaded": downloaded,
|
||||||
|
"download_errors": download_errors,
|
||||||
|
"processed": tick.files_processed,
|
||||||
|
"skipped": tick.files_skipped,
|
||||||
|
"errored": tick.files_errored,
|
||||||
|
"duration_s": duration,
|
||||||
|
},
|
||||||
|
"tick": tick.as_dict(),
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
@app.get("/api/admin/scheduler/status", dependencies=[Depends(matrix_gate)])
|
@app.get("/api/admin/scheduler/status", dependencies=[Depends(matrix_gate)])
|
||||||
def scheduler_status() -> Any:
|
def scheduler_status() -> Any:
|
||||||
"""Return the scheduler's runtime snapshot (running, counters, last tick)."""
|
"""Return the scheduler's runtime snapshot (running, counters, last tick)."""
|
||||||
|
|||||||
@@ -87,11 +87,70 @@ class SftpClient:
|
|||||||
dir and returns :class:`InboundFile` records pointing at the
|
dir and returns :class:`InboundFile` records pointing at the
|
||||||
cache copy. The remote file is *not* deleted — the operator
|
cache copy. The remote file is *not* deleted — the operator
|
||||||
archives inbound files in the MFT UI.
|
archives inbound files in the MFT UI.
|
||||||
|
|
||||||
|
Gainwell's MFT puts advisory ``*_warn.txt`` files in the same
|
||||||
|
inbound path. They're text-format side-channel notes (not X12
|
||||||
|
envelopes) and are skipped at list time. Use
|
||||||
|
:meth:`list_inbound_names` if you need the raw list without
|
||||||
|
the download.
|
||||||
"""
|
"""
|
||||||
if self._stub:
|
if self._stub:
|
||||||
return self._list_inbound_stub()
|
return self._list_inbound_stub()
|
||||||
return self._list_inbound_paramiko()
|
return self._list_inbound_paramiko()
|
||||||
|
|
||||||
|
def list_inbound_names(self) -> list[InboundFile]:
|
||||||
|
"""Lightweight listing: returns metadata only, no file download.
|
||||||
|
|
||||||
|
Use this when you want to filter the inbound set (e.g. by date)
|
||||||
|
before paying the download cost. Pair with
|
||||||
|
:meth:`download_inbound` to fetch a filtered subset on demand.
|
||||||
|
|
||||||
|
Real mode is implemented as a single SFTP ``listdir_attr`` call
|
||||||
|
— sub-second on Gainwell's MFT — versus the full
|
||||||
|
:meth:`list_inbound` which downloads every file. The
|
||||||
|
``*_warn.txt`` advisory files are filtered out the same way.
|
||||||
|
|
||||||
|
Stub mode returns the same as :meth:`list_inbound` (the stub
|
||||||
|
only knows about local files; no download cost).
|
||||||
|
"""
|
||||||
|
if self._stub:
|
||||||
|
return self._list_inbound_stub()
|
||||||
|
return self._list_inbound_names_paramiko()
|
||||||
|
|
||||||
|
def download_inbound(self, f: InboundFile) -> Path:
|
||||||
|
"""Download a single inbound file to its ``local_path``.
|
||||||
|
|
||||||
|
Idempotent: if ``f.local_path`` already exists and is
|
||||||
|
non-empty, the download is skipped. Callers should use the
|
||||||
|
``InboundFile`` returned by :meth:`list_inbound_names` and pass
|
||||||
|
it back here — ``local_path`` is the planned cache location
|
||||||
|
and matches the path the scheduler will read from.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
The on-disk path (same as ``f.local_path``).
|
||||||
|
|
||||||
|
Raises:
|
||||||
|
FileNotFoundError: if the file is missing locally in stub
|
||||||
|
mode, or if the remote file disappears between list
|
||||||
|
and download in real mode.
|
||||||
|
"""
|
||||||
|
if f.local_path.exists() and f.local_path.stat().st_size > 0:
|
||||||
|
log.debug(
|
||||||
|
"SFTP: %s already cached at %s, skipping download",
|
||||||
|
f.name, f.local_path,
|
||||||
|
)
|
||||||
|
return f.local_path
|
||||||
|
if self._stub:
|
||||||
|
# Stub mode: no remote — the file is supposed to already be
|
||||||
|
# at f.local_path (operator-dropped). If it isn't there, the
|
||||||
|
# operator hasn't seeded the stub; raise loudly.
|
||||||
|
if not f.local_path.is_file():
|
||||||
|
raise FileNotFoundError(
|
||||||
|
f"inbound stub file not found: {f.local_path}"
|
||||||
|
)
|
||||||
|
return f.local_path
|
||||||
|
return self._download_inbound_paramiko(f)
|
||||||
|
|
||||||
def read_file(self, remote_path: str) -> bytes:
|
def read_file(self, remote_path: str) -> bytes:
|
||||||
"""Read bytes from a remote path.
|
"""Read bytes from a remote path.
|
||||||
|
|
||||||
@@ -284,6 +343,12 @@ class SftpClient:
|
|||||||
if attr.st_mode and (attr.st_mode & 0o170000) == 0o040000:
|
if attr.st_mode and (attr.st_mode & 0o170000) == 0o040000:
|
||||||
# Directory entry — skip.
|
# Directory entry — skip.
|
||||||
continue
|
continue
|
||||||
|
if attr.filename.endswith("_warn.txt"):
|
||||||
|
# Gainwell's MFT drops text-format advisory notes in
|
||||||
|
# the same inbound path. They're side-channel noise,
|
||||||
|
# not X12 envelopes — skip at list time so we don't
|
||||||
|
# download ~600 advisory files per poll.
|
||||||
|
continue
|
||||||
remote = f"{inbound_dir.rstrip('/')}/{attr.filename}"
|
remote = f"{inbound_dir.rstrip('/')}/{attr.filename}"
|
||||||
cache_path = cache_dir / attr.filename
|
cache_path = cache_dir / attr.filename
|
||||||
# Download into cache. We use ``prefetch`` to keep memory
|
# Download into cache. We use ``prefetch`` to keep memory
|
||||||
@@ -299,6 +364,56 @@ class SftpClient:
|
|||||||
))
|
))
|
||||||
return files
|
return files
|
||||||
|
|
||||||
|
def _list_inbound_names_paramiko(self) -> list[InboundFile]:
|
||||||
|
"""List inbound names via paramiko; do NOT download (lightweight).
|
||||||
|
|
||||||
|
Same ``listdir_attr`` iteration as
|
||||||
|
:meth:`_list_inbound_paramiko`, but the returned
|
||||||
|
:class:`InboundFile` records have ``local_path`` set to the
|
||||||
|
planned cache location without actually fetching the file.
|
||||||
|
Pair with :meth:`_download_inbound_paramiko` to fetch on
|
||||||
|
demand. Skips ``*_warn.txt`` advisory files the same way.
|
||||||
|
"""
|
||||||
|
with self._connect() as (ssh, sftp):
|
||||||
|
inbound_dir = self._block.paths.get("inbound", "/")
|
||||||
|
staging = Path(self._block.staging_dir).resolve()
|
||||||
|
inbound_rel = inbound_dir.lstrip("/")
|
||||||
|
cache_dir = staging / inbound_rel
|
||||||
|
cache_dir.mkdir(parents=True, exist_ok=True)
|
||||||
|
|
||||||
|
files: list[InboundFile] = []
|
||||||
|
try:
|
||||||
|
attrs = sftp.listdir_attr(inbound_dir)
|
||||||
|
except IOError as exc:
|
||||||
|
log.warning("SFTP: cannot list %s: %s", inbound_dir, exc)
|
||||||
|
return []
|
||||||
|
|
||||||
|
for attr in sorted(attrs, key=lambda a: a.filename):
|
||||||
|
if attr.st_mode and (attr.st_mode & 0o170000) == 0o040000:
|
||||||
|
# Directory entry — skip.
|
||||||
|
continue
|
||||||
|
if attr.filename.endswith("_warn.txt"):
|
||||||
|
continue
|
||||||
|
cache_path = cache_dir / attr.filename
|
||||||
|
files.append(InboundFile(
|
||||||
|
name=attr.filename,
|
||||||
|
size=attr.st_size or 0,
|
||||||
|
modified_at=datetime.fromtimestamp(attr.st_mtime or 0),
|
||||||
|
local_path=cache_path,
|
||||||
|
))
|
||||||
|
return files
|
||||||
|
|
||||||
|
def _download_inbound_paramiko(self, f: InboundFile) -> Path:
|
||||||
|
"""Download a single ``f`` to ``f.local_path`` (idempotent on size>0)."""
|
||||||
|
with self._connect() as (ssh, sftp):
|
||||||
|
inbound_dir = self._block.paths.get("inbound", "/")
|
||||||
|
remote = f"{inbound_dir.rstrip('/')}/{f.name}"
|
||||||
|
f.local_path.parent.mkdir(parents=True, exist_ok=True)
|
||||||
|
with sftp.open(remote, "rb") as src, open(f.local_path, "wb") as dst:
|
||||||
|
shutil.copyfileobj(src, dst, length=64 * 1024)
|
||||||
|
log.info("SFTP: downloaded %d bytes for %s", f.local_path.stat().st_size, f.name)
|
||||||
|
return f.local_path
|
||||||
|
|
||||||
def _read_file_paramiko(self, remote_path: str) -> bytes:
|
def _read_file_paramiko(self, remote_path: str) -> bytes:
|
||||||
with self._connect() as (ssh, sftp):
|
with self._connect() as (ssh, sftp):
|
||||||
buf = io.BytesIO()
|
buf = io.BytesIO()
|
||||||
|
|||||||
@@ -569,3 +569,167 @@ def backup_status() -> None:
|
|||||||
snap = svc.status()
|
snap = svc.status()
|
||||||
import json
|
import json
|
||||||
click.echo(json.dumps(snap, indent=2, default=str))
|
click.echo(json.dumps(snap, indent=2, default=str))
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# SP-N fix: `cyclone pull-inbound` — targeted inbound pull + process
|
||||||
|
#
|
||||||
|
# Mirrors POST /api/admin/scheduler/pull-inbound. For operators who
|
||||||
|
# want to drive the daily "process today's 999s" workflow from a cron
|
||||||
|
# job or shell script without going through the HTTP API.
|
||||||
|
#
|
||||||
|
# Exit codes:
|
||||||
|
# 0 — processed ≥1 file (or processed 0 with no matches: not an
|
||||||
|
# error, just nothing to do)
|
||||||
|
# 1 — unexpected exception
|
||||||
|
# 2 — SFTP / config error
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
@main.command("pull-inbound")
|
||||||
|
@click.option(
|
||||||
|
"--date", "date_str",
|
||||||
|
required=True,
|
||||||
|
help="Date filter YYYYMMDD — only files whose 8-digit timestamp "
|
||||||
|
"substring matches are downloaded and processed.",
|
||||||
|
)
|
||||||
|
@click.option(
|
||||||
|
"--block", "sftp_block_name",
|
||||||
|
default="dzinesco",
|
||||||
|
show_default=True,
|
||||||
|
help="Name of the SftpBlock in config/payers.yaml to use.",
|
||||||
|
)
|
||||||
|
@click.option(
|
||||||
|
"--file-types", "file_types_csv",
|
||||||
|
default=None,
|
||||||
|
help="Comma-separated whitelist (default: 999,TA1).",
|
||||||
|
)
|
||||||
|
@click.option(
|
||||||
|
"--limit", default=2000, show_default=True, type=int,
|
||||||
|
)
|
||||||
|
def pull_inbound(
|
||||||
|
date_str: str,
|
||||||
|
sftp_block_name: str,
|
||||||
|
file_types_csv: str | None,
|
||||||
|
limit: int,
|
||||||
|
) -> None:
|
||||||
|
"""List, date-filter, download, and process inbound MFT files.
|
||||||
|
|
||||||
|
Bypasses the alphabetical full-listing pass so a daily pull of
|
||||||
|
~400 files takes seconds, not hours. Files already in the local
|
||||||
|
cache are skipped (idempotent).
|
||||||
|
"""
|
||||||
|
import asyncio as _asyncio
|
||||||
|
from cyclone import db as db_mod
|
||||||
|
from cyclone import scheduler as scheduler_mod
|
||||||
|
from cyclone.edi.filenames import ALLOWED_FILE_TYPES
|
||||||
|
from cyclone.clearhouse import SftpClient
|
||||||
|
from cyclone.providers import SftpBlock
|
||||||
|
|
||||||
|
# Validate the date filter.
|
||||||
|
if not (len(date_str) == 8 and date_str.isdigit()):
|
||||||
|
click.echo(f"--date must be YYYYMMDD, got {date_str!r}", err=True)
|
||||||
|
sys.exit(2)
|
||||||
|
|
||||||
|
db_mod.init_db()
|
||||||
|
|
||||||
|
# Find the SftpBlock. The dzinesco clearhouse singleton is seeded
|
||||||
|
# by store.ensure_clearhouse_seeded() (SP9) and carries the
|
||||||
|
# production SFTP block; that's the only one we have at the
|
||||||
|
# moment. For multi-provider SFTP, a config-loader is the right
|
||||||
|
# thing — out of scope for this CLI which is the daily-pull path.
|
||||||
|
from cyclone import store as store_mod
|
||||||
|
store_mod.store.ensure_clearhouse_seeded()
|
||||||
|
clearhouse = store_mod.store.get_clearhouse()
|
||||||
|
if clearhouse is None or clearhouse.name != sftp_block_name:
|
||||||
|
# Fall back: try the named block, but v1 only ships the
|
||||||
|
# dzinesco singleton.
|
||||||
|
if clearhouse is None:
|
||||||
|
click.echo(
|
||||||
|
f"No clearhouse seeded — cannot find SftpBlock "
|
||||||
|
f"{sftp_block_name!r}.",
|
||||||
|
err=True,
|
||||||
|
)
|
||||||
|
sys.exit(2)
|
||||||
|
click.echo(
|
||||||
|
f"SftpBlock {sftp_block_name!r} not seeded; only "
|
||||||
|
f"{clearhouse.name!r} is available.",
|
||||||
|
err=True,
|
||||||
|
)
|
||||||
|
sys.exit(2)
|
||||||
|
block: SftpBlock = clearhouse.sftp_block
|
||||||
|
|
||||||
|
if file_types_csv:
|
||||||
|
wanted = {t.strip().upper() for t in file_types_csv.split(",") if t.strip()}
|
||||||
|
unknown = wanted - ALLOWED_FILE_TYPES
|
||||||
|
if unknown:
|
||||||
|
click.echo(
|
||||||
|
f"file_types {sorted(unknown)!r} not in {sorted(ALLOWED_FILE_TYPES)}",
|
||||||
|
err=True,
|
||||||
|
)
|
||||||
|
sys.exit(2)
|
||||||
|
else:
|
||||||
|
wanted = {"999", "TA1"}
|
||||||
|
|
||||||
|
# Wire up the scheduler singleton (re-use the same pipeline the
|
||||||
|
# HTTP endpoint uses, including the dedup via processed_inbound_files).
|
||||||
|
scheduler_mod.configure_scheduler(
|
||||||
|
block, sftp_block_name=sftp_block_name, force=True,
|
||||||
|
)
|
||||||
|
sched = scheduler_mod.get_scheduler()
|
||||||
|
client = SftpClient(block)
|
||||||
|
|
||||||
|
async def _run() -> dict:
|
||||||
|
from cyclone.edi.filenames import parse_inbound_filename
|
||||||
|
|
||||||
|
all_files = await _asyncio.to_thread(client.list_inbound_names)
|
||||||
|
matched: list = []
|
||||||
|
for f in all_files:
|
||||||
|
if f.name.find(date_str) == -1:
|
||||||
|
continue
|
||||||
|
try:
|
||||||
|
parsed = parse_inbound_filename(f.name)
|
||||||
|
except ValueError:
|
||||||
|
continue
|
||||||
|
if parsed.file_type not in wanted:
|
||||||
|
continue
|
||||||
|
matched.append(f)
|
||||||
|
if len(matched) >= limit:
|
||||||
|
break
|
||||||
|
|
||||||
|
download_errors: list[str] = []
|
||||||
|
downloaded = 0
|
||||||
|
for f in matched:
|
||||||
|
try:
|
||||||
|
await _asyncio.to_thread(client.download_inbound, f)
|
||||||
|
downloaded += 1
|
||||||
|
except Exception as exc: # noqa: BLE001
|
||||||
|
download_errors.append(f"{f.name}: {type(exc).__name__}: {exc}")
|
||||||
|
|
||||||
|
tick = await sched.process_inbound_files(matched)
|
||||||
|
return {
|
||||||
|
"listed": len(all_files),
|
||||||
|
"matched": len(matched),
|
||||||
|
"downloaded": downloaded,
|
||||||
|
"download_errors": download_errors,
|
||||||
|
"processed": tick.files_processed,
|
||||||
|
"skipped": tick.files_skipped,
|
||||||
|
"errored": tick.files_errored,
|
||||||
|
}
|
||||||
|
|
||||||
|
try:
|
||||||
|
summary = _asyncio.run(_run())
|
||||||
|
except Exception as exc: # noqa: BLE001
|
||||||
|
click.echo(f"pull-inbound failed: {type(exc).__name__}: {exc}", err=True)
|
||||||
|
sys.exit(1)
|
||||||
|
|
||||||
|
click.echo(
|
||||||
|
f"date={date_str} file_types={sorted(wanted)} block={sftp_block_name} "
|
||||||
|
f"listed={summary['listed']} matched={summary['matched']} "
|
||||||
|
f"downloaded={summary['downloaded']} processed={summary['processed']} "
|
||||||
|
f"skipped={summary['skipped']} errored={summary['errored']}"
|
||||||
|
)
|
||||||
|
if summary["download_errors"]:
|
||||||
|
click.echo("download errors:", err=True)
|
||||||
|
for e in summary["download_errors"]:
|
||||||
|
click.echo(f" {e}", err=True)
|
||||||
|
|||||||
@@ -8,8 +8,11 @@ Outbound (we send):
|
|||||||
Example: tp11525703-837P-20260620132243505-1of1.x12
|
Example: tp11525703-837P-20260620132243505-1of1.x12
|
||||||
|
|
||||||
Inbound (HPE sends to our FromHPE):
|
Inbound (HPE sends to our FromHPE):
|
||||||
TP{tpid}-{orig_tx}_M{tracking}-{ts}-1of1_{file_type}.x12
|
[Tt][Pp]{tpid}-{orig_tx}_M{tracking}-{ts}-1of1_{file_type}.x12
|
||||||
Example: TP11525703-837P_M019048402-20260520231513488-1of1_999.x12
|
Example: tp11525703-837P_M019048402-20260520231513488-1of1_999.x12
|
||||||
|
(legacy / prodfiles may use uppercase TP; the regex is case-insensitive
|
||||||
|
on the prefix to accept both — Gainwell's filer has used both
|
||||||
|
casings over time.)
|
||||||
|
|
||||||
Both use Mountain Time (MT) timestamps with 17-digit millisecond precision
|
Both use Mountain Time (MT) timestamps with 17-digit millisecond precision
|
||||||
(yyyymmddhhmmssSSS = 4+2+2+2+2+2+3 = 17 digits). Sequence is always "1of1"
|
(yyyymmddhhmmssSSS = 4+2+2+2+2+2+3 = 17 digits). Sequence is always "1of1"
|
||||||
@@ -39,16 +42,23 @@ OUTBOUND_RE = re.compile(
|
|||||||
r"^tp(?P<tpid>\d+)-(?P<tx>[A-Z0-9]+)-(?P<ts>\d{17})-1of1\.(?P<ext>[A-Za-z0-9]+)$"
|
r"^tp(?P<tpid>\d+)-(?P<tx>[A-Z0-9]+)-(?P<ts>\d{17})-1of1\.(?P<ext>[A-Za-z0-9]+)$"
|
||||||
)
|
)
|
||||||
|
|
||||||
# Inbound: TP11525703-837P_M019048402-20260520231513488-1of1_999.x12
|
# Inbound: [Tt][Pp]11525703-837P_M019048402-20260520231513488-1of1_999.x12
|
||||||
|
# - prefix: literal "TP" or "tp" (case-insensitive — Gainwell's
|
||||||
|
# production filer has used both)
|
||||||
# - tpid: 1+ digits (inside TP<...>)
|
# - tpid: 1+ digits (inside TP<...>)
|
||||||
# - orig_tx: 1+ alnum
|
# - orig_tx: 1+ uppercase alnum
|
||||||
# - track: M + 1+ alnum (e.g. M019048402) — the M is part of the
|
# - track: M + 1+ uppercase alnum (e.g. M019048402) — the M is
|
||||||
# tracking value, not a separator.
|
# part of the tracking value, not a separator.
|
||||||
# - ts: 17 digits
|
# - ts: 17 digits
|
||||||
# - seq: literal "1of1"
|
# - seq: literal "1of1"
|
||||||
# - ft: 1+ alnum (e.g. 999, TA1, 271, 277, 277CA, 820, 834, 835, ENCR)
|
# - ft: 1+ uppercase alnum (e.g. 999, TA1, 271, 277, 277CA,
|
||||||
|
# 820, 834, 835, ENCR)
|
||||||
|
#
|
||||||
|
# Case insensitivity is scoped to the ``TP`` prefix via ``(?i:TP)``
|
||||||
|
# — the rest of the pattern is case-sensitive so we still reject a
|
||||||
|
# stray ``837p`` or ``m019048402`` (they'd be invalid HCPF).
|
||||||
INBOUND_RE = re.compile(
|
INBOUND_RE = re.compile(
|
||||||
r"^TP(?P<tpid>\d+)-(?P<orig_tx>[A-Z0-9]+)_(?P<tracking>M[A-Z0-9]+)"
|
r"^(?i:TP)(?P<tpid>\d+)-(?P<orig_tx>[A-Z0-9]+)_(?P<tracking>M[A-Z0-9]+)"
|
||||||
r"-(?P<ts>\d{17})-1of1_(?P<file_type>[A-Z0-9]+)\.(?P<ext>x12)$"
|
r"-(?P<ts>\d{17})-1of1_(?P<file_type>[A-Z0-9]+)\.(?P<ext>x12)$"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -116,7 +126,9 @@ def parse_inbound_filename(name: str) -> InboundFilename:
|
|||||||
"""Parse an inbound HCPF filename.
|
"""Parse an inbound HCPF filename.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
name: Filename like "TP11525703-837P_M019048402-20260520231513488-1of1_999.x12"
|
name: Filename like "tp11525703-837P_M019048402-20260520231513488-1of1_999.x12"
|
||||||
|
(case-insensitive on the ``TP`` prefix; both ``TP`` and
|
||||||
|
``tp`` are accepted.)
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
InboundFilename with tpid, orig_tx, tracking, ts, file_type, ext.
|
InboundFilename with tpid, orig_tx, tracking, ts, file_type, ext.
|
||||||
|
|||||||
@@ -494,6 +494,44 @@ class Scheduler:
|
|||||||
finally:
|
finally:
|
||||||
self._tick_in_progress = False
|
self._tick_in_progress = False
|
||||||
|
|
||||||
|
async def process_inbound_files(
|
||||||
|
self, files: list[InboundFile],
|
||||||
|
) -> TickResult:
|
||||||
|
"""Run the per-file processing pipeline over a pre-fetched list.
|
||||||
|
|
||||||
|
Unlike :meth:`tick`, this does **not** call SFTP — the caller
|
||||||
|
is expected to have already downloaded (or arranged the local
|
||||||
|
copy of) each ``InboundFile.local_path``. Used by the
|
||||||
|
``/api/admin/scheduler/pull-inbound`` admin endpoint and the
|
||||||
|
``cyclone pull-inbound`` CLI command to process a date-filtered
|
||||||
|
subset of the inbound MFT path without paying the cost of a
|
||||||
|
full poll.
|
||||||
|
|
||||||
|
Honors ``_stop_event`` between files. Concurrent calls are
|
||||||
|
coalesced the same way :meth:`tick` does, so a slow handler
|
||||||
|
can't be stampeded by a parallel admin invocation.
|
||||||
|
"""
|
||||||
|
while self._tick_in_progress:
|
||||||
|
await asyncio.sleep(0.05)
|
||||||
|
self._tick_in_progress = True
|
||||||
|
try:
|
||||||
|
started = datetime.now(timezone.utc)
|
||||||
|
result = TickResult(started_at=started, files_seen=len(files))
|
||||||
|
for f in files:
|
||||||
|
if self._stop_event.is_set():
|
||||||
|
break
|
||||||
|
await self._handle_one(f, result)
|
||||||
|
result.finished_at = datetime.now(timezone.utc)
|
||||||
|
self._last_tick = result
|
||||||
|
self._last_poll_at = result.finished_at
|
||||||
|
self._poll_count += 1
|
||||||
|
self._total_processed += result.files_processed
|
||||||
|
self._total_skipped += result.files_skipped
|
||||||
|
self._total_errored += result.files_errored
|
||||||
|
return result
|
||||||
|
finally:
|
||||||
|
self._tick_in_progress = False
|
||||||
|
|
||||||
# ---- Internals --------------------------------------------------------
|
# ---- Internals --------------------------------------------------------
|
||||||
|
|
||||||
async def _run(self) -> None:
|
async def _run(self) -> None:
|
||||||
|
|||||||
@@ -107,6 +107,48 @@ def test_parse_inbound_277():
|
|||||||
assert parsed.file_type == "277"
|
assert parsed.file_type == "277"
|
||||||
|
|
||||||
|
|
||||||
|
def test_parse_inbound_lowercase_tp_prefix_999():
|
||||||
|
# Gainwell's production filer uses lowercase `tp` for inbound 999/TA1.
|
||||||
|
# The inbound regex must accept both casings on the TP prefix.
|
||||||
|
name = "tp11525703-837P_M019048402-20260520231513488-1of1_999.x12"
|
||||||
|
parsed = parse_inbound_filename(name)
|
||||||
|
assert parsed.tpid == "11525703"
|
||||||
|
assert parsed.orig_tx == "837P"
|
||||||
|
assert parsed.tracking == "M019048402"
|
||||||
|
assert parsed.file_type == "999"
|
||||||
|
assert parsed.ext == "x12"
|
||||||
|
|
||||||
|
|
||||||
|
def test_parse_inbound_lowercase_tp_prefix_ta1():
|
||||||
|
name = "tp11525703-837P_M019044969-20260520180505477-1of1_TA1.x12"
|
||||||
|
parsed = parse_inbound_filename(name)
|
||||||
|
assert parsed.file_type == "TA1"
|
||||||
|
|
||||||
|
|
||||||
|
def test_is_inbound_filename_accepts_both_cases():
|
||||||
|
# is_inbound_filename() is the fast path used by the scheduler to
|
||||||
|
# filter the listing. It must accept both casings.
|
||||||
|
upper = "TP11525703-837P_M019048402-20260520231513488-1of1_999.x12"
|
||||||
|
lower = "tp11525703-837P_M019048402-20260520231513488-1of1_999.x12"
|
||||||
|
assert is_inbound_filename(upper)
|
||||||
|
assert is_inbound_filename(lower)
|
||||||
|
|
||||||
|
|
||||||
|
def test_parse_inbound_rejects_mixed_case_tracking():
|
||||||
|
# Tracking value must stay uppercase alnum; the case-insensitive
|
||||||
|
# flag is intentionally scoped to the TP prefix by the file_type
|
||||||
|
# and timestamp constraints, so a mixed-case tracking should still
|
||||||
|
# be rejected (it'd be invalid HCPF).
|
||||||
|
# We exercise the obvious "totally lowercase" rejection to confirm
|
||||||
|
# the rest of the pattern is still strict.
|
||||||
|
with pytest.raises(ValueError, match="Not a valid HCPF inbound"):
|
||||||
|
# Lowercase orig_tx; the orig_tx class is [A-Z0-9]+ so it
|
||||||
|
# must be uppercase.
|
||||||
|
parse_inbound_filename(
|
||||||
|
"tp11525703-837p_M019048402-20260520231513488-1of1_999.x12"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
def test_parse_inbound_rejects_missing_tp_prefix():
|
def test_parse_inbound_rejects_missing_tp_prefix():
|
||||||
with pytest.raises(ValueError, match="Not a valid HCPF inbound"):
|
with pytest.raises(ValueError, match="Not a valid HCPF inbound"):
|
||||||
parse_inbound_filename("11525703-837P_M019048402-20260520231513488-1of1_999.x12")
|
parse_inbound_filename("11525703-837P_M019048402-20260520231513488-1of1_999.x12")
|
||||||
@@ -157,8 +199,10 @@ def test_is_outbound_filename():
|
|||||||
|
|
||||||
def test_is_inbound_filename():
|
def test_is_inbound_filename():
|
||||||
assert is_inbound_filename("TP11525703-837P_M019048402-20260520231513488-1of1_999.x12")
|
assert is_inbound_filename("TP11525703-837P_M019048402-20260520231513488-1of1_999.x12")
|
||||||
# Lowercase tp prefix is the outbound shape, not inbound
|
# Lowercase tp prefix is now accepted too — Gainwell's filer has
|
||||||
assert not is_inbound_filename("tp11525703-837P_M019048402-20260520231513488-1of1_999.x12")
|
# used both casings on inbound 999/TA1 files.
|
||||||
|
assert is_inbound_filename("tp11525703-837P_M019048402-20260520231513488-1of1_999.x12")
|
||||||
|
# Outbound shape is still rejected (no tracking/ts/file_type).
|
||||||
assert not is_inbound_filename("11525703-837P-20260620132243505-1of1.x12")
|
assert not is_inbound_filename("11525703-837P-20260620132243505-1of1.x12")
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -285,3 +285,61 @@ class TestRoutedFileTypes:
|
|||||||
|
|
||||||
def test_handlers_cover_all_routed_types(self):
|
def test_handlers_cover_all_routed_types(self):
|
||||||
assert set(HANDLERS.keys()) == ROUTED_FILE_TYPES
|
assert set(HANDLERS.keys()) == ROUTED_FILE_TYPES
|
||||||
|
|
||||||
|
|
||||||
|
class TestProcessInboundFiles:
|
||||||
|
"""Scheduler.process_inbound_files() — date-filtered pull path.
|
||||||
|
|
||||||
|
The /api/admin/scheduler/pull-inbound endpoint and the
|
||||||
|
``cyclone pull-inbound`` CLI both call this. It must:
|
||||||
|
* process each file in the provided list (no SFTP listdir)
|
||||||
|
* dedupe via ``processed_inbound_files`` (idempotent on rerun)
|
||||||
|
* not touch SFTP at all — files are expected to be on local
|
||||||
|
disk at ``f.local_path``
|
||||||
|
"""
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_processes_provided_files_without_listdir(
|
||||||
|
self, sftp_block, _drop_file, tmp_path,
|
||||||
|
):
|
||||||
|
from cyclone.clearhouse import InboundFile
|
||||||
|
from cyclone.scheduler import STATUS_OK
|
||||||
|
|
||||||
|
# Drop a known 999 file on disk (the stub scheduler doesn't
|
||||||
|
# need SFTP — it reads from staging dir).
|
||||||
|
_drop_file(
|
||||||
|
"TP11525703-837P_M019048402-20260520231513488-1of1_999.x12",
|
||||||
|
b"not a real 999 -- handler will parse_error",
|
||||||
|
)
|
||||||
|
sched = _make_scheduler(sftp_block, tmp_path)
|
||||||
|
|
||||||
|
# Build the InboundFile records manually — caller is
|
||||||
|
# responsible for staging (mirrors the targeted-pull flow).
|
||||||
|
path = tmp_path / "staging" / "FromHPE" / \
|
||||||
|
"TP11525703-837P_M019048402-20260520231513488-1of1_999.x12"
|
||||||
|
files = [InboundFile(
|
||||||
|
name=path.name,
|
||||||
|
size=path.stat().st_size,
|
||||||
|
modified_at=datetime.fromtimestamp(path.stat().st_mtime, tz=timezone.utc),
|
||||||
|
local_path=path,
|
||||||
|
)]
|
||||||
|
|
||||||
|
result = await sched.process_inbound_files(files)
|
||||||
|
# The file's bytes are intentionally invalid — we only care
|
||||||
|
# that process_inbound_files invokes the handler and
|
||||||
|
# records the outcome (here: error from parse).
|
||||||
|
assert result.files_seen == 1
|
||||||
|
assert result.files_seen == result.files_processed + result.files_errored
|
||||||
|
# Idempotent: a second call is a no-op (already-processed dedup).
|
||||||
|
result2 = await sched.process_inbound_files(files)
|
||||||
|
assert result2.files_seen == 1
|
||||||
|
# Either skipped (because the prior call recorded it as error)
|
||||||
|
# or error — both prove the dedup branch fired. We just check
|
||||||
|
# it's not re-processed successfully.
|
||||||
|
with db.SessionLocal()() as s:
|
||||||
|
rows = s.query(ProcessedInboundFile).filter_by(
|
||||||
|
sftp_block_name="test-block",
|
||||||
|
name=path.name,
|
||||||
|
).all()
|
||||||
|
assert len(rows) == 1
|
||||||
|
assert rows[0].status in (STATUS_OK, STATUS_ERROR, STATUS_SKIPPED)
|
||||||
@@ -259,6 +259,66 @@ class TestRealModeListInbound:
|
|||||||
assert len(files) == 1
|
assert len(files) == 1
|
||||||
assert files[0].name == "real.x12"
|
assert files[0].name == "real.x12"
|
||||||
|
|
||||||
|
def test_list_skips_warn_txt_files(self, monkeypatch, tmp_path: Path):
|
||||||
|
# Gainwell's MFT drops advisory *_warn.txt files in the same
|
||||||
|
# inbound dir. They're text-format side-channel notes (not X12
|
||||||
|
# envelopes) and must be skipped at list time.
|
||||||
|
real_attr = MagicMock()
|
||||||
|
real_attr.filename = "TP123-837P_M456-20260520231513488-1of1_999.x12"
|
||||||
|
real_attr.st_mode = 0o100644
|
||||||
|
real_attr.st_size = 1024
|
||||||
|
real_attr.st_mtime = 1718899200
|
||||||
|
|
||||||
|
warn_attr = MagicMock()
|
||||||
|
warn_attr.filename = "TP123-837P-202606181208100000-1of1_warn.txt"
|
||||||
|
warn_attr.st_mode = 0o100644
|
||||||
|
warn_attr.st_size = 200
|
||||||
|
warn_attr.st_mtime = 1718899200
|
||||||
|
|
||||||
|
mock_ssh, mock_sftp = _make_mock_paramiko(
|
||||||
|
monkeypatch, sftp_attrs=[real_attr, warn_attr],
|
||||||
|
)
|
||||||
|
|
||||||
|
def _open(path, mode="rb"):
|
||||||
|
m = MagicMock()
|
||||||
|
m.__enter__.return_value = io.BytesIO(b"x12 content")
|
||||||
|
return m
|
||||||
|
|
||||||
|
mock_sftp.open.side_effect = _open
|
||||||
|
|
||||||
|
block = _block(staging_dir=str(tmp_path / "staging"))
|
||||||
|
client = SftpClient(block)
|
||||||
|
files = client.list_inbound()
|
||||||
|
|
||||||
|
names = [f.name for f in files]
|
||||||
|
assert "TP123-837P_M456-20260520231513488-1of1_999.x12" in names
|
||||||
|
assert not any(n.endswith("_warn.txt") for n in names)
|
||||||
|
assert len(files) == 1
|
||||||
|
|
||||||
|
def test_list_inbound_names_does_not_download(self, monkeypatch, tmp_path: Path):
|
||||||
|
# list_inbound_names() must do a metadata-only SFTP listing —
|
||||||
|
# no sftp.open() / no file written to the cache.
|
||||||
|
attr = MagicMock()
|
||||||
|
attr.filename = "TP123-837P_M456-20260520231513488-1of1_999.x12"
|
||||||
|
attr.st_mode = 0o100644
|
||||||
|
attr.st_size = 1024
|
||||||
|
attr.st_mtime = 1718899200
|
||||||
|
|
||||||
|
mock_ssh, mock_sftp = _make_mock_paramiko(
|
||||||
|
monkeypatch, sftp_attrs=[attr],
|
||||||
|
)
|
||||||
|
|
||||||
|
block = _block(staging_dir=str(tmp_path / "staging"))
|
||||||
|
client = SftpClient(block)
|
||||||
|
files = client.list_inbound_names()
|
||||||
|
|
||||||
|
# sftp.open() must NOT have been called (no download).
|
||||||
|
mock_sftp.open.assert_not_called()
|
||||||
|
assert len(files) == 1
|
||||||
|
# The InboundFile.local_path is set to the planned cache path
|
||||||
|
# but the file itself doesn't exist yet.
|
||||||
|
assert not files[0].local_path.exists()
|
||||||
|
|
||||||
|
|
||||||
class TestRealModeReadFile:
|
class TestRealModeReadFile:
|
||||||
def test_read_returns_bytes(self, monkeypatch):
|
def test_read_returns_bytes(self, monkeypatch):
|
||||||
|
|||||||
Reference in New Issue
Block a user