Receive the Claim File, Check It, Load It Once

Goal

Safely receive an affiliated insurer's claim file over SFTP (completion flag, checksum, atomic rename, rerun safety), validate the header and trailer, load it per file, and then reconcile it against our intake ledger.

Why it matters

File integration incidents happen at boundaries — a file still being written, damage in transit, double loading from a rerun, and numbers that were not checked against each other. All four pass quietly without an error and show up only in closing or customer complaints.

Steps

1. Create the partner's remote directory with python3 /opt/lab/fixtures/eaimw/batch/partner_drop.py --out /root/eaimw/batch/remote --day 20260923, and with sftp -D /usr/lib/openssh/sftp-server -b <배치파일> (batch file), cd into that directory, run ls -l, and save that output as it is to /root/eaimw/batch/remote-ls.txt.
2. Write /root/eaimw/batch/fetch.sh. It reads the environment variables REMOTE_DIR (default /root/eaimw/batch/remote), LOCAL_DIR (default /root/eaimw/batch/inbox) and QUAR_DIR (default /root/eaimw/batch/quarantine), gets the remote listing over SFTP, and only the files that have a <파일>.done flag (the file name followed by .done), namely the .dat files, are received into LOCAL_DIR.
3. Compare the flag contents (<바이트수> <sha256>, the byte count and the sha256) with the received file. If they differ, move that file to QUAR_DIR (do not leave it in LOCAL_DIR), continue processing the rest, and finally end with a non-zero code.
4. While receiving, receive inside LOCAL_DIR as <파일>.part (the file name followed by .part), and when verification is done, rename it to the final name. After receiving, add .fetched to the end of the remote file and flag names over SFTP. If you run it again, it receives nothing and ends with 0.
5. /root/eaimw/batch/check.py <파일> (file): each line 60 bytes (EUC-KR, LF at the line end), the first line H, the last line T, D in between, H date = file name date, and T count and total = D count and total. If it is right, OK <건수> <합계> (OK, the count, the total) and exit 0; if wrong, E <사유> (E and the reason) on the first line and exit 2.
6. /root/eaimw/batch/load.py --db <경로> <파일> (path, file): load only files that passed validation into the SQLite table claims(claim_no 기본키, policy_no, amount, claim_date, insured, src_file) (with claim_no as the primary key), and record them in load_log(file 기본키, records, total, loaded_at) (with file as the primary key). Do not load the same file again (end with 0), and one file is one transaction (all or nothing).
7. /root/eaimw/batch/recon.py --db <경로> --ledger <원장CSV> --out <결과CSV> (path, ledger CSV, result CSV): match claims and the ledger (claim_no,amount) by claim number and write claim_no,status,partner_amount,our_amount in claim number order. status is MATCHED, ONLY_PARTNER, ONLY_OURS or AMOUNT_MISMATCH, and the amount of the missing side is blank. Finally, fetch and load today's file (CLM_401_20260923.dat), and write the result of reconciling it with the ledger made by partner_drop.py --ledger /root/eaimw/batch/ledger.csv --day 20260923 to /root/eaimw/batch/recon.csv.

Notes

• Specification: /opt/lab/fixtures/eaimw/batch/SPEC.md. D line positions (from 0): claim number 1–15, policy number 15–27, amount 27–40, claim date 40–48, name 48–60.
• SFTP batch example: printf 'cd %s\nls -1\n' "$REMOTE_DIR" > /tmp/b; sftp -q -D /usr/lib/openssh/sftp-server -b /tmp/b — command lines starting with sftp> get mixed into the output. get 원격 로컬 (get remote local) and rename 옛이름 새이름 (rename old-name new-name) work the same way.
• Checksum: sha256sum 파일 | cut -d' ' -f1 (with the file name in place of the placeholder), size: stat -c %s 파일.
• Common mistakes: receiving the temporary file into /tmp and then mv-ing it (not atomic if it is a different filesystem), committing per row in the load, and running reconciliation from only one side (you miss records present only on the other side).

Look at the remote over SFTP

Create the remote directory with partner_drop.py and save the output listed with an sftp -D batch to /root/eaimw/batch/remote-ls.txt.

Write two lines, cd and ls -l, in the batch file and run it with sftp -D /usr/lib/openssh/sftp-server -b batch-file. Redirect the output as it is.

Fetch only files that have a flag

/root/eaimw/batch/fetch.sh fetches from REMOTE_DIR into LOCAL_DIR only the .dat files that have a .done.

First get the list of names with ls -1, and for each .dat, see whether the same name + .done is in the list (grep -qx). Fetching is get remote local.

Quarantine it if the checksum differs

Compare the flag's byte count and sha256 with the received file, and if they differ, move it to QUAR_DIR and end with a non-zero code at the end.

Fetch the flag too, read it with read -r size sha, and compare with stat -c %s and sha256sum. Do not stop the rest because one file is broken; just remember the exit code.

Receive under a temporary name and mark the remote

Receive into a .part inside LOCAL_DIR, rename it after verification, and add .fetched to the remote file and flag to make reruns safe.

The temporary file must be in the same place as the final directory for the rename to be atomic. The remote marking is SFTP's rename command.

Check the header against the trailer

/root/eaimw/batch/check.py verifies line length (bytes), H/D/T structure, date, count and total (OK / E and exit 2).

Read the file as rb, split into lines, and check len(line)==60. Compare T's count (1–8) and total (8–23) with the values you counted directly from D. Decode the name with euc_kr.

Load once per file

/root/eaimw/batch/load.py --db loads only validated files in one transaction and blocks reloading with load_log.

Open with isolation_level=None and write BEGIN … COMMIT yourself. If a claim number primary key conflict (IntegrityError) occurs, ROLLBACK to undo up to the previous rows.

Reconcile with our ledger

Reconcile claims and the ledger with /root/eaimw/batch/recon.py, and write the result based on today's file to /root/eaimw/batch/recon.csv.

Go through the sorted union of the claim numbers from both sides. If you go through only one side, you miss records present only on the other. Make the ledger with partner_drop.py --ledger.