Add customs shipment source helper

scripts/check_customs_shipments.sh

@@ -0,0 +1,72 @@
+#!/bin/sh
+set -eu
+
+if [ "$#" -lt 1 ]; then
+    echo "usage: $0 RUN_DIR" >&2
+    exit 1
+fi
+
+RUN_DIR=$1
+OUT_DIR=$RUN_DIR/api/customs
+REPORT_DIR=$RUN_DIR/reports
+SCRIPT_DIR=$(CDPATH= cd "$(dirname "$0")" && pwd)
+
+mkdir -p "$OUT_DIR" "$REPORT_DIR"
+
+ERROR_FILE=$REPORT_DIR/customs-shipments-errors.txt
+: > "$ERROR_FILE"
+
+IMPORTINFO_USER_AGENT=${IMPORTINFO_USER_AGENT:-"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0 Safari/537.36"}
+
+MANIFEST=$OUT_DIR/importinfo-inputs.tsv
+: > "$MANIFEST"
+
+fetch_importinfo() {
+    slug=$1
+    query=$2
+    url=$3
+    tmp=$OUT_DIR/$slug.html.tmp
+    out=$OUT_DIR/$slug.html
+
+    if curl \
+        -A "$IMPORTINFO_USER_AGENT" \
+        -H "Accept: text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8" \
+        -H "Accept-Language: en-US,en;q=0.9" \
+        --retry 2 \
+        --retry-delay 2 \
+        --max-time 45 \
+        -fsSL "$url" > "$tmp" 2>/dev/null
+    then
+        mv "$tmp" "$out"
+        printf '%s\t%s\t%s\n' "$query" "$url" "$out" >> "$MANIFEST"
+        return 0
+    fi
+
+    rm -f "$tmp" "$out"
+    printf 'failed to fetch %s\n' "$url" >> "$ERROR_FILE"
+    return 1
+}
+
+fetch_importinfo \
+    "importinfo-ceva-valve" \
+    "ceva-valve" \
+    "https://www.importinfo.com/search?s=CEVA%20C%2FO%20VALVE%20CORPORATION" || true
+fetch_importinfo \
+    "importinfo-ingram-valve" \
+    "ingram-valve" \
+    "https://www.importinfo.com/search?s=INGRAM%20MICRO%20C%2FO%20VALVE%20CORPORATION" || true
+fetch_importinfo \
+    "importinfo-tech-front-game-console" \
+    "tech-front-game-console" \
+    "https://www.importinfo.com/search?s=TECH-FRONT%20GAME%20CONSOLE%20VALVE" || true
+fetch_importinfo \
+    "importinfo-valve-corporation" \
+    "valve-corporation-game-console" \
+    "https://www.importinfo.com/search?s=VALVE%20CORPORATION%20GAME%20CONSOLE" || true
+
+python3 "$SCRIPT_DIR/parse_importinfo_shipments.py" \
+    --manifest "$MANIFEST" \
+    --report-dir "$REPORT_DIR"
+
+echo "Saved customs shipment pages to $OUT_DIR"
+echo "Saved customs shipment report to $REPORT_DIR/customs-shipments.md"

tests/test_customs_shipments.py

@@ -1,4 +1,6 @@
 import csv
+import os
+import stat
 import subprocess
 import textwrap
 import unittest
@@ -75,6 +77,11 @@ def read_rows(reports):
     )
 
 
+def write_executable(path, content):
+    path.write_text(textwrap.dedent(content).lstrip(), encoding="utf-8")
+    path.chmod(path.stat().st_mode | stat.S_IXUSR)
+
+
 class CustomsShipmentParserTests(unittest.TestCase):
     def test_extracts_relevant_importinfo_rows(self):
         with TemporaryDirectory() as tmp:
@@ -220,5 +227,78 @@ class CustomsShipmentParserTests(unittest.TestCase):
             self.assertEqual("https://example.test/ceva", rows[0]["source_url"])
 
 
+class CheckCustomsShipmentsShellTests(unittest.TestCase):
+    def test_fetches_pages_and_runs_parser(self):
+        with TemporaryDirectory() as tmp:
+            tmp_path = Path(tmp)
+            run_dir = tmp_path / "run"
+            fake_bin = tmp_path / "bin"
+            fake_bin.mkdir()
+            write_executable(
+                fake_bin / "curl",
+                f"""
+                #!/bin/sh
+                cat <<'HTML'
+                {IMPORTINFO_HTML}
+                HTML
+                """,
+            )
+            env = os.environ.copy()
+            env["PATH"] = f"{fake_bin}{os.pathsep}{env['PATH']}"
+
+            result = subprocess.run(
+                [str(ROOT / "scripts" / "check_customs_shipments.sh"), str(run_dir)],
+                cwd=ROOT,
+                env=env,
+                text=True,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                check=False,
+            )
+
+            self.assertEqual("", result.stderr)
+            self.assertEqual(0, result.returncode)
+            self.assertTrue(
+                (run_dir / "api" / "customs" / "importinfo-ceva-valve.html").exists()
+            )
+            key_lines = (
+                run_dir / "reports" / "customs-shipments-key-lines.txt"
+            ).read_text(encoding="utf-8")
+            self.assertIn("SNHBSHALAX264014", key_lines)
+
+    def test_records_blocked_fetches_without_failing_run(self):
+        with TemporaryDirectory() as tmp:
+            tmp_path = Path(tmp)
+            run_dir = tmp_path / "run"
+            fake_bin = tmp_path / "bin"
+            fake_bin.mkdir()
+            write_executable(
+                fake_bin / "curl",
+                """
+                #!/bin/sh
+                exit 22
+                """,
+            )
+            env = os.environ.copy()
+            env["PATH"] = f"{fake_bin}{os.pathsep}{env['PATH']}"
+
+            result = subprocess.run(
+                [str(ROOT / "scripts" / "check_customs_shipments.sh"), str(run_dir)],
+                cwd=ROOT,
+                env=env,
+                text=True,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                check=False,
+            )
+
+            self.assertEqual("", result.stderr)
+            self.assertEqual(0, result.returncode)
+            errors = (run_dir / "reports" / "customs-shipments-errors.txt").read_text(
+                encoding="utf-8"
+            )
+            self.assertIn("failed to fetch", errors)
+
+
 if __name__ == "__main__":
     unittest.main()