diff --git a/scripts/parse_importinfo_shipments.py b/scripts/parse_importinfo_shipments.py index d441a07..52b74a0 100755 --- a/scripts/parse_importinfo_shipments.py +++ b/scripts/parse_importinfo_shipments.py @@ -81,8 +81,8 @@ class InputSpec: class TableParser(HTMLParser): def __init__(self): super().__init__() - self.headers = [] self.rows = [] + self._current_headers = [] self._in_row = False self._in_cell = False self._cell_tag = "" @@ -105,19 +105,21 @@ class TableParser(HTMLParser): self._row_has_header = True else: self._row_has_data = True + elif self._in_cell and tag == "br": + self._current_cell_parts.append(" ") def handle_endtag(self, tag): if self._in_cell and tag == self._cell_tag: - value = html.unescape("".join(self._current_cell_parts)) + value = html.unescape(" ".join(self._current_cell_parts)) self._current_cells.append(" ".join(value.split())) self._in_cell = False self._cell_tag = "" self._current_cell_parts = [] elif self._in_row and tag == "tr": if self._row_has_header and self._current_cells: - self.headers = self._current_cells + self._current_headers = self._current_cells elif self._row_has_data and self._current_cells: - self.rows.append(self._current_cells) + self.rows.append((list(self._current_headers), self._current_cells)) self._in_row = False def handle_data(self, data): @@ -132,9 +134,9 @@ def normalize_header(value): def parse_rows(spec): parser = TableParser() parser.feed(spec.path.read_text(encoding="utf-8", errors="ignore")) - fields = [normalize_header(header) for header in parser.headers] rows = [] - for parsed_row in parser.rows: + for headers, parsed_row in parser.rows: + fields = [normalize_header(header) for header in headers] record = {field: "" for field in OUTPUT_FIELDS} record["source"] = "importinfo" record["query"] = spec.query diff --git a/tests/test_customs_shipments.py b/tests/test_customs_shipments.py index 0d405bc..80caa8f 100644 --- a/tests/test_customs_shipments.py +++ b/tests/test_customs_shipments.py @@ -49,6 +49,32 @@ IMPORTINFO_HTML = textwrap.dedent( ) +def run_parser(args, reports): + return subprocess.run( + [ + "python3", + str(ROOT / "scripts" / "parse_importinfo_shipments.py"), + *args, + "--report-dir", + str(reports), + ], + cwd=ROOT, + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + check=False, + ) + + +def read_rows(reports): + return list( + csv.DictReader( + (reports / "customs-shipments.tsv").read_text(encoding="utf-8").splitlines(), + delimiter="\t", + ) + ) + + class CustomsShipmentParserTests(unittest.TestCase): def test_extracts_relevant_importinfo_rows(self): with TemporaryDirectory() as tmp: @@ -57,33 +83,12 @@ class CustomsShipmentParserTests(unittest.TestCase): html.write_text(IMPORTINFO_HTML, encoding="utf-8") reports = tmp_path / "reports" - result = subprocess.run( - [ - "python3", - str(ROOT / "scripts" / "parse_importinfo_shipments.py"), - "--input", - f"ceva-valve={html}", - "--report-dir", - str(reports), - ], - cwd=ROOT, - text=True, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - check=False, - ) + result = run_parser(["--input", f"ceva-valve={html}"], reports) self.assertEqual("", result.stderr) self.assertEqual(0, result.returncode) - rows = list( - csv.DictReader( - (reports / "customs-shipments.tsv") - .read_text(encoding="utf-8") - .splitlines(), - delimiter="\t", - ) - ) + rows = read_rows(reports) self.assertEqual(1, len(rows)) self.assertEqual("SNHBSHALAX264014", rows[0]["house_bol"]) self.assertEqual("GAME CONSOLE", rows[0]["commodity"]) @@ -126,36 +131,94 @@ class CustomsShipmentParserTests(unittest.TestCase): ) reports = tmp_path / "reports" - result = subprocess.run( - [ - "python3", - str(ROOT / "scripts" / "parse_importinfo_shipments.py"), - "--input", - f"ceva-valve={html}", - "--report-dir", - str(reports), - ], - cwd=ROOT, - text=True, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - check=False, - ) + result = run_parser(["--input", f"ceva-valve={html}"], reports) self.assertEqual(0, result.returncode) - rows = list( - csv.DictReader( - (reports / "customs-shipments.tsv") - .read_text(encoding="utf-8") - .splitlines(), - delimiter="\t", - ) - ) + rows = read_rows(reports) self.assertEqual( ["SNHBSHALAX264014", "SNHBSHALAX264015"], [row["house_bol"] for row in rows], ) + def test_ignores_later_unrelated_table_headers(self): + with TemporaryDirectory() as tmp: + tmp_path = Path(tmp) + html = tmp_path / "extra-table.html" + html.write_text( + IMPORTINFO_HTML.replace( + "", + textwrap.dedent( + """ + + + +
NameDescription
PROMETHEAN INC.CHROMEBOX HTS:
+ + """ + ), + ), + encoding="utf-8", + ) + reports = tmp_path / "reports" + + result = run_parser(["--input", f"ceva-valve={html}"], reports) + + self.assertEqual("", result.stderr) + self.assertEqual(0, result.returncode) + rows = read_rows(reports) + self.assertEqual(["SNHBSHALAX264014"], [row["house_bol"] for row in rows]) + + def test_matches_relevance_terms_split_by_inline_markup(self): + with TemporaryDirectory() as tmp: + tmp_path = Path(tmp) + html = tmp_path / "split-cells.html" + html.write_text( + IMPORTINFO_HTML.replace( + "CEVA C/O VALVE CORPORATIONGAME CONSOLE", + "CEVA C/O VALVE CORPORATIONGAME
CONSOLE", + ), + encoding="utf-8", + ) + reports = tmp_path / "reports" + + result = run_parser(["--input", f"ceva-valve={html}"], reports) + + self.assertEqual("", result.stderr) + self.assertEqual(0, result.returncode) + rows = read_rows(reports) + self.assertEqual(1, len(rows)) + self.assertEqual("CEVA C/O VALVE CORPORATION", rows[0]["consignee"]) + self.assertEqual("GAME CONSOLE", rows[0]["commodity"]) + + def test_manifest_inputs_dedupe_duplicate_bols(self): + with TemporaryDirectory() as tmp: + tmp_path = Path(tmp) + first = tmp_path / "first.html" + second = tmp_path / "second.html" + manifest = tmp_path / "manifest.tsv" + first.write_text(IMPORTINFO_HTML, encoding="utf-8") + second.write_text(IMPORTINFO_HTML, encoding="utf-8") + manifest.write_text( + "\n".join( + ( + "slug\tsource_url\thtml_path", + f"ceva-valve\thttps://example.test/ceva\t{first}", + f"ceva-valve\thttps://example.test/ceva\t{second}", + ) + ) + + "\n", + encoding="utf-8", + ) + reports = tmp_path / "reports" + + result = run_parser(["--manifest", str(manifest)], reports) + + self.assertEqual("", result.stderr) + self.assertEqual(0, result.returncode) + rows = read_rows(reports) + self.assertEqual(["SNHBSHALAX264014"], [row["house_bol"] for row in rows]) + self.assertEqual("https://example.test/ceva", rows[0]["source_url"]) + if __name__ == "__main__": unittest.main()