import io import json import unittest from unittest.mock import patch from parse import amount_value, extract_with_ai, parse_request, validate_extraction, validate_exchange_extraction class FinancialCommandParserTests(unittest.TestCase): def exchange_fields(self): return { "other_center_text": "5805", "center_id": "42", "to_other_amount_text": "663032", "to_other_currency_text": "يورو", "to_other_currency_id": "eur", "to_other_evidence": "له 663032 يورو", "to_source_amount_text": "769117", "to_source_currency_text": "دولار", "to_source_currency_id": "usd", "to_source_evidence": "لنا 769117 دولار", "ambiguous": False, } def test_exchange_two_opposing_entries_from_customer_reply(self): catalogs = {"currencies": [{"id": "eur"}, {"id": "usd"}], "centers": [{"id": "42"}], "companies": []} quote = "الفيل 5805\nله\n663032 يورو\nلنا\n769117 دولار" result = validate_exchange_extraction(self.exchange_fields(), "بدل", quote, catalogs) self.assertEqual(result["to_other_amount"], "663032.0000") self.assertEqual(result["to_source_amount"], "769117.0000") def test_exchange_rejects_invented_amount_and_ambiguous_direction(self): quote = "الفيل 5805\nله 663032 يورو\nلنا 769117 دولار" fields = self.exchange_fields() fields["to_other_amount_text"] = "663033" with self.assertRaises(ValueError): validate_exchange_extraction(fields, "بدل", quote) fields = self.exchange_fields() fields["to_other_evidence"] = "663032 يورو" with self.assertRaises(ValueError): validate_exchange_extraction(fields, "بدل", quote) fields = self.exchange_fields() fields["to_source_currency_text"] = "يورو" with self.assertRaises(ValueError): validate_exchange_extraction(fields, "بدل", quote) def test_exchange_api_uses_separate_strict_schema(self): response = {"status": "completed", "output": [{"content": [{"type": "output_text", "text": json.dumps(self.exchange_fields(), ensure_ascii=False)}]}]} catalogs = {"currencies": [{"id": "eur", "name": "يورو"}, {"id": "usd", "name": "دولار"}], "centers": [{"id": "42", "box": "5805", "name": "الفيل"}], "companies": []} with patch("parse.urllib.request.urlopen", return_value=io.BytesIO(json.dumps(response).encode())) as opened: result = extract_with_ai("sk-test", "بدل", "الفيل 5805\nله 663032 يورو\nلنا 769117 دولار", "exchange", "gpt-5-nano", catalogs) self.assertEqual(result["center_id"], "42") request = json.loads(opened.call_args.args[0].data) self.assertEqual(request["text"]["format"]["name"], "exchange_command_fields") self.assertTrue(request["text"]["format"]["strict"]) self.assertNotIn("webhook", json.dumps(request).lower()) def test_amount_forms(self): for text, expected in ( ("1000", "1000.0000"), ("1.000", "1000.0000"), ("1,000", "1000.0000"), ("ألف", "1000.0000"), ("مية ألف", "100000.0000"), ("1000 الف", "1000000.0000"), ("1,000.50", "1000.5000"), ("1.000,50", "1000.5000"), ): with self.subTest(text=text): self.assertEqual(amount_value(text), expected) def test_transfer_reply_fields(self): fields = { "amount_text": "1500", "currency_text": "دولار", "destination_text": "151", "beneficiary_text": "محمد الأحمد", "company_text": "", "account_text": "", "currency_id": "usd", "center_id": "2", "company_id": "", "ambiguous": False, } result = validate_extraction(fields, "حول لصندوق 151", "1500 دولار\nمحمد الأحمد\nسرمدا", "transfer") self.assertEqual(result["amount"], "1500.0000") def test_transfer_all_fields_in_one_message_without_fixed_order(self): fields = { "amount_text": "1,000", "currency_text": "USD", "destination_text": "151", "beneficiary_text": "محمد الأحمد", "company_text": "", "account_text": "", "currency_id": "usd", "center_id": "2", "company_id": "", "ambiguous": False, } command = "حول\nمحمد الأحمد\nإلى صندوق 151\nUSD 1,000" result = validate_extraction(fields, command, "", "transfer") self.assertEqual(result["amount"], "1000.0000") def test_transfer_fields_can_be_split_in_opposite_direction(self): fields = { "amount_text": "1500", "currency_text": "دولار", "destination_text": "151", "beneficiary_text": "محمد الأحمد", "company_text": "", "account_text": "", "currency_id": "usd", "center_id": "2", "company_id": "", "ambiguous": False, } result = validate_extraction(fields, "حول 1500 دولار لحساب 151", "الاسم: محمد الأحمد", "transfer") self.assertEqual(result["beneficiary_text"], "محمد الأحمد") def test_approval_account_and_currency_alias(self): fields = { "amount_text": "مية الف", "currency_text": "$", "destination_text": "", "beneficiary_text": "", "company_text": "تواصل", "account_text": "100", "currency_id": "usd", "center_id": "", "company_id": "7", "ambiguous": False, } result = validate_extraction(fields, "اعتمد مية الف $ تواصل لحساب 100", "", "approval") self.assertEqual(result["amount"], "100000.0000") def test_approval_fields_can_be_entirely_in_quoted_message(self): fields = { "amount_text": "1000", "currency_text": "USD", "destination_text": "", "beneficiary_text": "محمد الأحمد", "company_text": "تواصل", "account_text": "100", "currency_id": "usd", "center_id": "", "company_id": "7", "ambiguous": False, } quoted = "المستفيد محمد الأحمد\nحساب 100 لدى تواصل\n1000 USD" result = validate_extraction(fields, "اعتمد", quoted, "approval") self.assertEqual(result["amount"], "1000.0000") def test_invented_or_ambiguous_data_is_rejected(self): fields = { "amount_text": "1000", "currency_text": "USD", "destination_text": "151", "beneficiary_text": "محمد الأحمد", "company_text": "", "account_text": "", "currency_id": "usd", "center_id": "2", "company_id": "", "ambiguous": False, } with self.assertRaises(ValueError): validate_extraction(fields, "حول 151", "1000 دولار\nمحمد الأحمد", "transfer") fields["currency_text"] = "دولار" fields["ambiguous"] = True with self.assertRaises(ValueError): validate_extraction(fields, "حول 151", "1000 دولار\nمحمد الأحمد", "transfer") def test_numeric_substring_and_ambiguous_separator_are_rejected(self): fields = { "amount_text": "100", "currency_text": "دولار", "destination_text": "151", "beneficiary_text": "محمد الأحمد", "company_text": "", "account_text": "", "currency_id": "usd", "center_id": "2", "company_id": "", "ambiguous": False, } with self.assertRaises(ValueError): validate_extraction(fields, "حول 151", "1000 دولار\nمحمد الأحمد", "transfer") with self.assertRaises(ValueError): amount_value("1000.000") with self.assertRaises(ValueError): amount_value("1000 151") fields["amount_text"] = "الف" with self.assertRaises(ValueError): validate_extraction(fields, "حول 151", "الفين دولار\nمحمد الأحمد", "transfer") def test_ai_request_uses_structured_output_without_webhook_payload(self): response = {"status": "completed", "output": [{"content": [{"type": "output_text", "text": json.dumps({ "amount_text": "1000", "currency_text": "USD", "destination_text": "151", "beneficiary_text": "محمد الأحمد", "company_text": "", "account_text": "", "currency_id": "usd", "center_id": "2", "company_id": "", "ambiguous": False, })}]}]} catalogs = {"currencies": [{"id": "usd", "name": "دولار"}], "centers": [{"id": "2", "box": "151", "name": "سرمدا"}], "companies": []} with patch("parse.urllib.request.urlopen", return_value=io.BytesIO(json.dumps(response).encode())) as opened: result = extract_with_ai("sk-test", "حول 151", "1000 USD\nمحمد الأحمد", "transfer", "gpt-5-nano", catalogs) self.assertEqual(result["amount_text"], "1000") request = json.loads(opened.call_args.args[0].data) self.assertEqual(request["text"]["format"]["type"], "json_schema") self.assertFalse(request["store"]) self.assertNotIn("webhook", json.dumps(request).lower()) self.assertIn('"id":"usd"', request["input"]) self.assertIn('"box":"151"', request["input"]) def test_catalog_ids_are_advisory_but_amount_and_destination_must_be_grounded(self): fields = {"amount_text": "1000", "currency_text": "دولار", "destination_text": "151", "beneficiary_text": "محمد الأحمد", "company_text": "", "account_text": "", "currency_id": "usd", "center_id": "2", "company_id": "", "ambiguous": False} catalogs = {"currencies": [{"id": "usd"}], "centers": [{"id": "2"}], "companies": []} self.assertEqual(validate_extraction(fields, "حول 151", "1000 دولار محمد الأحمد", "transfer", catalogs)["amount"], "1000.0000") fields["currency_id"] = "fake" fields["center_id"] = "fake" fields["company_id"] = "7" self.assertEqual(validate_extraction(fields, "حول 151", "1000 دولار محمد الأحمد", "transfer", catalogs)["amount"], "1000.0000") fields["destination_text"] = "1000" with self.assertRaises(ValueError): validate_extraction(fields, "حول 1000 دولار محمد الأحمد", "", "transfer", catalogs) def test_reported_approval_and_exchange_without_ai_request(self): catalogs = {"currencies": [ {"id": "usd", "name": "دولار", "aliases": ["$", "usd"]}, {"id": "eur", "name": "يورو", "aliases": ["eur"]}], "centers": [{"id": "42", "box": "510", "name": "مركز 510"}], "companies": []} with patch("parse.urllib.request.urlopen", side_effect=AssertionError("unexpected AI request")): approval = parse_request({"mode": "approval", "command": "اعتمد 100 دولار لحساب 510", "quoted": "", "api_key": "", "model": "gpt-5-nano", **catalogs}) exchange = parse_request({"mode": "exchange", "command": "بدل حساب 510\nله\n\n663032 يورو\n\nلنا\n\n769117 دولار", "quoted": "", "api_key": "", "model": "gpt-5-nano", **catalogs}) self.assertEqual(approval["data"]["amount"], "100.0000") self.assertEqual(approval["data"]["account_text"], "510") self.assertEqual(exchange["data"]["other_center_text"], "510") self.assertEqual(exchange["data"]["to_other_amount"], "663032.0000") self.assertEqual(exchange["data"]["to_source_amount"], "769117.0000") def test_reported_exchange_split_between_reply_and_command(self): catalogs = {"currencies": [{"id": "eur", "name": "يورو"}, {"id": "usd", "name": "دولار"}], "centers": [{"id": "42", "box": "510", "name": "مركز 510"}], "companies": []} result = parse_request({"mode": "exchange", "command": "بدل حساب 510", "quoted": "له 663032 يورو\nلنا 769117 دولار", "api_key": "", "model": "gpt-5-nano", **catalogs}) self.assertEqual(result["data"]["to_other_currency_text"], "يورو") self.assertEqual(result["data"]["to_source_currency_text"], "دولار") quoted_result = parse_request({"mode": "exchange", "command": "بدل", "quoted": "حساب 510\nله\n663032 يورو\nلنا\n769117 دولار", "api_key": "", "model": "gpt-5-nano", **catalogs}) self.assertEqual(quoted_result["data"]["other_center_text"], "510") def test_exchange_does_not_ignore_conflicting_box_in_command(self): catalogs = {"currencies": [{"id": "eur", "name": "يورو"}, {"id": "usd", "name": "دولار"}], "centers": [{"id": "42", "box": "510", "name": "مركز 510"}, {"id": "43", "box": "511", "name": "مركز 511"}], "companies": []} with self.assertRaisesRegex(ValueError, "أكثر من رقم صندوق"): parse_request({"mode": "exchange", "command": "بدل حساب 511", "quoted": "حساب 510\nله 663032 يورو\nلنا 769117 دولار", "api_key": "", "model": "gpt-5-nano", **catalogs}) def test_transfer_reply_uses_original_beneficiary_and_rejects_two_money_pairs(self): catalogs = {"currencies": [{"id": "usd", "name": "دولار"}, {"id": "eur", "name": "يورو"}], "centers": [{"id": "2", "box": "151", "name": "سرمدا الأحمد", "city": "سرمدا"}], "companies": []} result = parse_request({"mode": "transfer", "command": "حول 151", "quoted": "1500 دولار\nمحمد الأحمد\nسرمدا", "api_key": "", "model": "gpt-5-nano", **catalogs}) self.assertEqual(result["data"]["beneficiary_text"], "محمد الأحمد") with self.assertRaisesRegex(ValueError, "أكثر من مبلغ"): parse_request({"mode": "transfer", "command": "حول", "quoted": "حساب 510\nله 663032 يورو\nلنا 769117 دولار", "api_key": "", "model": "gpt-5-nano", **catalogs}) def test_ambiguous_request_still_requires_ai_configuration(self): catalogs = {"currencies": [{"id": "usd", "name": "دولار"}], "centers": [{"id": "2", "box": "151", "name": "سرمدا"}], "companies": []} with self.assertRaisesRegex(ValueError, "غير مهيأة"): parse_request({"mode": "transfer", "command": "حول 151", "quoted": "1500 دولار", "api_key": "", "model": "gpt-5-nano", **catalogs}) def test_ai_cannot_use_box_number_as_amount(self): catalogs = {"currencies": [{"id": "usd", "name": "دولار"}], "centers": [{"id": "42", "box": "510", "name": "مركز 510"}], "companies": []} fields = {"amount_text": "510", "currency_text": "دولار", "destination_text": "", "beneficiary_text": "", "company_text": "", "account_text": "510", "currency_id": "usd", "center_id": "42", "company_id": "", "ambiguous": False} with self.assertRaises(ValueError): validate_extraction(fields, "اعتمد 100 دولار لحساب 510", "", "approval", catalogs) def test_twenty_center_candidates_are_allowed_but_not_twenty_one(self): catalogs = {"currencies": [{"id": "usd", "name": "دولار"}], "centers": [{"id": str(index), "box": str(100 + index), "name": "مركز " + str(index)} for index in range(1, 21)], "companies": []} request = {"mode": "approval", "command": "اعتمد 100 دولار لحساب 101", "quoted": "", "api_key": "", "model": "gpt-5-nano", **catalogs} self.assertEqual(parse_request(request)["data"]["amount"], "100.0000") request["centers"] = catalogs["centers"] + [{"id": "21", "box": "121", "name": "مركز 21"}] with self.assertRaisesRegex(ValueError, "قائمة العملة أو الوجهة غير صالحة"): parse_request(request) def test_transfer_one_message_matches_name_number_or_both(self): catalogs = {"currencies": [{"id": "usd", "name": "دولار"}], "centers": [{"id": "11", "box": "111", "name": "النورين"}], "companies": []} for destination in ("النورين 111", "نوورين 111", "نورين", "111"): with self.subTest(destination=destination): result = parse_request({"mode": "transfer", "command": "حول 100 دولار محمد محمود " + destination, "quoted": "", "api_key": "", "model": "gpt-5-nano", **catalogs}) self.assertEqual(result["data"]["center_id"], "11") self.assertEqual(result["data"]["beneficiary_text"], "محمد محمود") self.assertEqual(result["data"]["amount"], "100.0000") result = parse_request({"mode": "transfer", "command": "حول 100 دولار محمد محمود إلى النورين 111", "quoted": "", "api_key": "", "model": "gpt-5-nano", **catalogs}) self.assertEqual(result["data"]["beneficiary_text"], "محمد محمود") self.assertEqual(result["data"]["destination_text"], "النورين 111") def test_transfer_rejects_conflicting_name_and_box(self): catalogs = {"currencies": [{"id": "usd", "name": "دولار"}], "centers": [{"id": "11", "box": "111", "name": "النورين"}, {"id": "12", "box": "112", "name": "النوران"}], "companies": []} with self.assertRaisesRegex(ValueError, "اسم الوجهة لا يطابق رقم الصندوق"): parse_request({"mode": "transfer", "command": "حول 100 دولار محمد محمود النوران 111", "quoted": "", "api_key": "", "model": "gpt-5-nano", **catalogs}) def test_exchange_center_name_and_box_and_external_approval_account(self): catalogs = {"currencies": [{"id": "usd", "name": "دولار"}, {"id": "eur", "name": "يورو"}], "centers": [{"id": "11", "box": "111", "name": "النورين"}], "companies": [{"id": "7", "name": "تواصل"}]} exchange = parse_request({"mode": "exchange", "command": "بدل نوورين 111\nله 150 يورو\nلنا 160 دولار", "quoted": "", "api_key": "", "model": "gpt-5-nano", **catalogs}) self.assertEqual(exchange["data"]["center_id"], "11") self.assertEqual(exchange["data"]["to_other_amount"], "150.0000") approval = parse_request({"mode": "approval", "command": "اعتمد 100 دولار تواصل صندوق 100", "quoted": "", "api_key": "", "model": "gpt-5-nano", **catalogs}) self.assertEqual(approval["data"]["company_id"], "7") self.assertEqual(approval["data"]["account_text"], "100") self.assertEqual(approval["data"]["center_id"], "") def test_exchange_accepts_center_name_without_box(self): catalogs = {"currencies": [{"id": "usd", "name": "دولار"}, {"id": "eur", "name": "يورو"}], "centers": [{"id": "11", "box": "111", "name": "النورين"}], "companies": []} for name in ("النورين", "نوورين", "نورين"): with self.subTest(name=name): result = parse_request({"mode": "exchange", "command": "بدل " + name + "\nله 150 يورو\nلنا 160 دولار", "quoted": "", "api_key": "", "model": "gpt-5-nano", **catalogs}) self.assertEqual(result["data"]["center_id"], "11") self.assertEqual(result["data"]["other_center_text"], name) def test_exchange_reported_lahum_variants_are_deterministic(self): catalogs = {"currencies": [{"id": "usd", "name": "دولار"}, {"id": "eur", "name": "يورو"}], "centers": [{"id": "534", "box": "534", "name": "الوزاز"}], "companies": []} cases = ( ("بدل\nصندوق 534\nلنا 100 دولار\nلهم 95 يورو", "95.0000", "100.0000", False), ("بدل\nصندوق 534 وزاز\nلنا 100 دولار\nلهم 95 يورو", "95.0000", "100.0000", True), ("بدل حساب 534\nله\n663.032 يورو\nلنا\n769117 دولار", "663032.0000", "769117.0000", False), ("بدل\nحساب 534 وزاز\nلنا 100 دولار\nلهم 95 يورو", "95.0000", "100.0000", True), ("بدل حساب 534\nلنا\n663.032 يورو\nلهم\n769117 دولار", "769117.0000", "663032.0000", False), ("بدل للحساب 534\nلنا 100 دولار\nلهم 95 يورو", "95.0000", "100.0000", False), ("بدل للصندوق 534\nلنا 100 دولار\nلهم 95 يورو", "95.0000", "100.0000", False), ) with patch("parse.urllib.request.urlopen", side_effect=AssertionError("unexpected AI request")): for command, to_other, to_source, has_name in cases: with self.subTest(command=command): result = parse_request({"mode": "exchange", "command": command, "quoted": "", "api_key": "", "model": "gpt-5-nano", **catalogs})["data"] self.assertIn("534", result["other_center_text"]) self.assertEqual("وزاز" in result["other_center_text"], has_name) self.assertEqual(result["to_other_amount"], to_other) self.assertEqual(result["to_source_amount"], to_source) def test_exchange_ai_cannot_replace_explicit_box_with_catalog_name_or_other_number(self): catalogs = {"currencies": [{"id": "usd", "name": "دولار"}, {"id": "eur", "name": "يورو"}], "centers": [{"id": "534", "box": "534", "name": "الوزاز"}], "companies": []} command = "بدل لحساب 534 وزاز\nنرسل له 95 يورو\nنقبض منه 100 دولار" fields = {"other_center_text": "الوزاز", "center_id": "534", "to_other_amount_text": "95", "to_other_currency_text": "يورو", "to_other_currency_id": "eur", "to_other_evidence": "نرسل له 95 يورو", "to_source_amount_text": "100", "to_source_currency_text": "دولار", "to_source_currency_id": "usd", "to_source_evidence": "نقبض منه 100 دولار", "ambiguous": False} request = {"mode": "exchange", "command": command, "quoted": "", "api_key": "sk-test-12345678901234567890", "model": "gpt-5-nano", **catalogs} with patch("parse.extract_with_ai", return_value=fields): result = parse_request(request)["data"] self.assertEqual(result["other_center_text"], "لحساب 534 وزاز") self.assertEqual(result["to_other_amount"], "95.0000") self.assertEqual(result["to_source_amount"], "100.0000") with patch("parse.extract_with_ai", return_value={**fields, "other_center_text": "535"}): with self.assertRaisesRegex(ValueError, "لا يطابق النص الأصلي"): parse_request(request) wrong_center_request = {**request, "centers": catalogs["centers"] + [{"id": "535", "box": "535", "name": "مركز آخر"}]} with patch("parse.extract_with_ai", return_value={**fields, "center_id": "535"}): with self.assertRaisesRegex(ValueError, "لا يطابق رقم الصندوق"): parse_request(wrong_center_request) with self.assertRaisesRegex(ValueError, "أكثر من رقم صندوق"): parse_request({**request, "command": "بدل للحساب 534\nللصندوق 535\n" "نرسل له 95 يورو\nنقبض منه 100 دولار"}) if __name__ == "__main__": unittest.main()