from __future__ import annotations import io import unittest import zipfile from openpyxl import Workbook from openpyxl.styles import PatternFill from booking_ingestion.excel import ( BookingExcelError, PROCESSOR_VERSION, REVIEW_CONFIRMED, REVIEW_PENDING, build_excel_parse_result, parse_excel_bytes, validate_excel_filename, ) def workbook_bytes(sheets: list[tuple[str, list[list[object]]]]) -> bytes: workbook = Workbook() first = True for title, rows in sheets: worksheet = workbook.active if first else workbook.create_sheet() first = False worksheet.title = title for row in rows: worksheet.append(row) output = io.BytesIO() workbook.save(output) workbook.close() return output.getvalue() def booking_rows(*rows: list[object]) -> bytes: return workbook_bytes( [ ( "BOOKING", [ ["Booking export"], ["NO", "Tour Code", "Tour Days", "โรงแรม", "备注"], *rows, ], ), ( "ROOM_TYPE_MASTER", [["Pattern", "Category"], ["【U-TWN8.5】", "U-TWN"]], ), ] ) class BookingExcelParserTests(unittest.TestCase): def test_extracts_tour_code_room_items_and_review_rows(self) -> None: payload = booking_rows( [ 1, "\t\nLT260715LD", "2-2-1", "17-19 酒店 (【U-TWN8.5】 17,【U-เตียงเสริม13】 2)(ออฟฟิศจ่าย)【ยืนยันเเล้ว】;", ], [ 2, "LLT260715AC", "2-2", "17-19 酒店 (Sup【高级房TWN】 12,Sup【高级房DBL】 9)", ], [3, "LLT260715AD", "2-3", "17-20 酒店 (【U-TWN13】"], [4, "LLT260720HB", "2-2", "20-22 酒店 (【U-TWN1.1】 6)"], [5, "HD260721VIPA", "2-2", "23-26 酒店 (【拼จU-4ค24】 2)"], [6, "IGNORED", "2-2", "23-26 酒店 (【surcharge】 1)"], ) document = parse_excel_bytes(payload) self.assertEqual(document.worksheets, ("BOOKING",)) self.assertEqual([row.group_code_raw for row in document.rows], [ "LT260715LD", "LLT260715AC", "LLT260715AD", "LLT260720HB", "HD260721VIPA", ]) self.assertEqual(document.extracted_item_count, 7) self.assertEqual(document.pending_item_count, 2) mixed = document.rows[0] self.assertEqual( [ (item.room_type_code, item.quantity, item.review_status) for item in mixed.room_items ], [ ("U-TWN", 17, REVIEW_CONFIRMED), (None, 2, REVIEW_PENDING), ], ) self.assertEqual(mixed.room_items[1].room_type_raw, "U-เตียงเสริม13") split = document.rows[1] self.assertEqual( [(item.room_type_code, item.quantity) for item in split.room_items], [("TWN", 12), ("DBL", 9)], ) self.assertEqual(document.rows[2].room_items[0].quantity, 1) self.assertEqual(document.rows[2].room_items[0].room_type_code, "U-TWN") self.assertEqual(document.rows[3].room_items[0].room_type_code, "U-TWN") self.assertEqual(document.rows[4].room_items[0].quantity, 2) self.assertIsNone(document.rows[4].room_items[0].room_type_code) parsed = build_excel_parse_result(mixed) self.assertEqual(parsed["status"], "needs_review") self.assertEqual(len(parsed["room_items"]), 2) self.assertEqual(parsed["warnings"][0]["code"], "BOOKING_ROOM_TYPE_REVIEW_REQUIRED") self.assertEqual(parsed["processor_version"], "2.1.0") self.assertEqual(PROCESSOR_VERSION, "2.1.0") def test_bilingual_headers_accept_lian_tai_and_qbd_variants(self) -> None: payload = workbook_bytes( [ ( "AUTO_CALC", [["BLOCK_CODE", "RATE_CODE"], ["LT-NEW", "RACK"]], ), ( "LianTai", [ ["Booking export"], [ "Tour Code \n主团号", "Tour Days", "Notes", "วางข้อมูลที่นี่ / 酒店明细", ], ["LT-NEW", "", "", "酒店 (【U-TWN8.5】 2)"], ], ), ( "QBD", [ ["QBD export"], [None], [ "Month", "NO.", "Tour Code \n主团号", "Tour Days", "Raw Hotel Detail / วางข้อมูลที่นี่ / 酒店明细", ], ["", 1, "QBD-NEW", "", "酒店 (【U-DBL12】 3)"], ], ), ] ) document = parse_excel_bytes(payload) self.assertEqual(document.worksheets, ("LianTai", "QBD")) self.assertEqual( [row.group_code_raw for row in document.rows], ["LT-NEW", "QBD-NEW"] ) self.assertEqual(document.extracted_item_count, 2) self.assertEqual(document.pending_item_count, 0) def test_legacy_tour_code_aliases_remain_supported(self) -> None: for tour_header in ( "Tour Code", "Group Code", "Gourp Code", "团号", "团队代码", ): with self.subTest(tour_header=tour_header): payload = workbook_bytes( [ ( "Legacy", [ [tour_header, "โรงแรม"], ["LEGACY", "酒店 (【U-TWN8.5】 1)"], ], ) ] ) document = parse_excel_bytes(payload) self.assertEqual(document.worksheets, ("Legacy",)) self.assertEqual(document.rows[0].group_code_raw, "LEGACY") def test_lookalike_bilingual_header_is_rejected(self) -> None: payload = workbook_bytes( [ ( "Data", [ [ "Tour Code / 主团号（备用）", "Raw Hotel Detail / วางข้อมูลที่นี่ / 酒店明细", ], ["SYN", "酒店 (【U-TWN8.5】 1)"], ], ) ] ) with self.assertRaises(BookingExcelError) as raised: parse_excel_bytes(payload) self.assertEqual(raised.exception.code, "BOOKING_EXCEL_HEADERS_MISSING") self.assertIn("Tour Code/主团号", raised.exception.safe_message) self.assertIn("酒店明细", raised.exception.safe_message) def test_duplicate_recognized_headers_are_ambiguous(self) -> None: payload = workbook_bytes( [ ( "Data", [ [ "Tour Code", "Tour Code / 主团号", "วางข้อมูลที่นี่ / 酒店明细", ], ["SYN", "SYN", "酒店 (【U-TWN8.5】 1)"], ], ) ] ) with self.assertRaises(BookingExcelError) as raised: parse_excel_bytes(payload) self.assertEqual(raised.exception.code, "BOOKING_EXCEL_HEADERS_AMBIGUOUS") self.assertIn("Tour Code/主团号", raised.exception.safe_message) self.assertIn("酒店明细", raised.exception.safe_message) def test_physically_last_tour_row_replaces_earlier_and_cancel_removes_it(self) -> None: payload = booking_rows( [1, "SAME", "", "酒店 (【U-TWN8.5】 5)"], [2, "CANCELLED", "", "酒店 (【U-TWN8.5】 3)"], [3, "SAME", "", "酒店 (【U-DBL12】 2)"], [4, "CANCELLED", "", "取消"], ) document = parse_excel_bytes(payload) self.assertEqual(len(document.rows), 1) self.assertEqual(document.rows[0].group_code_key, "SAME") self.assertEqual(document.rows[0].room_items[0].room_type_code, "U-DBL") self.assertEqual(document.rows[0].room_items[0].quantity, 2) def test_yellow_formatting_alone_does_not_cancel_a_tour(self) -> None: workbook = Workbook() worksheet = workbook.active worksheet.title = "Booking" worksheet.append(["NO", "Tour Code", "Tour Days", "โรงแรม", "备注"]) worksheet.append([1, "YELLOW", "", "酒店 (【U-TWN8.5】 3)", ""]) worksheet.append([2, "YELLOW", "", "酒店 (【U-TWN8.5】 4)", ""]) yellow = PatternFill(fill_type="solid", fgColor="FFFF00") for cell in worksheet[3]: if cell.value not in (None, ""): cell.fill = yellow output = io.BytesIO() workbook.save(output) workbook.close() document = parse_excel_bytes(output.getvalue()) self.assertEqual(len(document.rows), 1) self.assertEqual(document.rows[0].group_code_key, "YELLOW") self.assertEqual(document.rows[0].room_items[0].quantity, 4) def test_unbracketed_room_description_requires_review(self) -> None: payload = booking_rows( [1, "SUITE", "", "酒店 (Family Suite Two b/r Garden View 2)"], ) document = parse_excel_bytes(payload) item = document.rows[0].room_items[0] self.assertEqual(item.room_type_raw, "Family Suite") self.assertEqual(item.quantity, 2) self.assertEqual(item.review_status, REVIEW_PENDING) def test_filename_accepts_xlsx_only(self) -> None: self.assertEqual(validate_excel_filename("预订报表.XLSX"), "预订报表.XLSX") for invalid in ("report.xls", "report.xlsm", "../report.xlsx", ""): with self.subTest(invalid=invalid), self.assertRaises(BookingExcelError): validate_excel_filename(invalid) def test_non_xlsx_and_macro_payloads_are_rejected(self) -> None: with self.assertRaises(BookingExcelError) as invalid: parse_excel_bytes(b"not an xlsx") self.assertEqual(invalid.exception.code, "BOOKING_EXCEL_INVALID") source = booking_rows([1, "SYN", "", "酒店 (【U-TWN8.5】 1)"]) original = zipfile.ZipFile(io.BytesIO(source)) output = io.BytesIO() with original, zipfile.ZipFile(output, "w", zipfile.ZIP_DEFLATED) as target: for info in original.infolist(): target.writestr(info, original.read(info.filename)) target.writestr("xl/vbaProject.bin", b"macro") with self.assertRaises(BookingExcelError) as macro: parse_excel_bytes(output.getvalue()) self.assertEqual(macro.exception.code, "BOOKING_EXCEL_MACRO_UNSUPPORTED") def test_workbook_without_tour_and_hotel_headers_is_rejected(self) -> None: payload = workbook_bytes( [("Data", [["Group Code", "Room Type", "Quantity"], ["SYN", "DBL", 1]])] ) with self.assertRaises(BookingExcelError) as raised: parse_excel_bytes(payload) self.assertEqual(raised.exception.code, "BOOKING_EXCEL_HEADERS_MISSING") def test_required_formula_is_rejected(self) -> None: formula = booking_rows([1, "SYN", "", "=\"酒店 (【U-TWN8.5】 1)\""]) with self.assertRaises(BookingExcelError) as raised: parse_excel_bytes(formula) self.assertEqual(raised.exception.code, "BOOKING_EXCEL_FORMULA_UNSUPPORTED") if __name__ == "__main__": unittest.main(verbosity=2)