Hello from Python SDK!
", + sender_email=SENDER_EMAIL, + reply_email=REPLY_EMAIL, + sender_name=SENDER_NAME, + title="Python Email Test" + ) + run("18 Email.send_single", test_18) + + # 19 — Email.send (1 recipient) + def test_19(): + client.email.send( + [EmailAccount(first_name=fn1, last_name=ln1, email=email1)], + "Python SDK Email 1", "Hello 1!
", + SENDER_EMAIL, REPLY_EMAIL, SENDER_NAME, "Python Email Test" + ) + run("19 Email.send (1 recipient)", test_19) + + # 20 — Email.send (2 recipients) + def test_20(): + client.email.send( + [ + EmailAccount(first_name=fn1, last_name=ln1, email=email1), + EmailAccount(first_name=fn2, last_name=ln2, email=email2), + ], + "Python SDK Email 2", "Hello 2!
", + SENDER_EMAIL, REPLY_EMAIL, SENDER_NAME, "Python Email Test" + ) + run("20 Email.send (2 recipients)", test_20) + + # 21 — Email.send (3 recipients) + def test_21(): + client.email.send( + [ + EmailAccount(first_name=fn1, last_name=ln1, email=email1), + EmailAccount(first_name=fn2, last_name=ln2, email=email2), + EmailAccount(first_name=fn3, last_name=ln3, email=email3), + ], + "Python SDK Email 3", "Hello 3!
", + SENDER_EMAIL, REPLY_EMAIL, SENDER_NAME, "Python Email Test" + ) + run("21 Email.send (3 recipients)", test_21) + + # 22 — Email.send_campaign (direct campaign object) + def test_22(): + campaign = EmailCampaign( + subject="Python SDK Campaign Test", + title="Python Email Campaign", + message="Campaign email from Python SDK!
", + sender_email=SENDER_EMAIL, + reply_email=REPLY_EMAIL, + sender_name=SENDER_NAME, + accounts=[ + EmailAccount(first_name=fn1, last_name=ln1, email=email1), + EmailAccount(first_name=fn2, last_name=ln2, email=email2), + ], + ) + client.email.send_campaign(campaign) + run("22 Email.send_campaign", test_22) + + # ── Webhook Tests (23-29) ──────────────────────────────────────────────────── + print("\n--- Webhook ---") + + SECRET = "test-webhook-secret-python" + registered_webhook_id = None + + # 23 — Webhook.register + def test_23(): + nonlocal registered_webhook_id + config = WebhookConfig( + url=webhook_url, + events=[WebhookEventType.MESSAGE_SENT], + secret=SECRET + ) + resp = client.webhook.register(config) + wid = resp.id + if not wid: + raise RuntimeError("webhook ID is empty after register") + registered_webhook_id = str(wid) + run("23 Webhook.register", test_23) + + # 24 — Webhook.list + def test_24(): + hooks = client.webhook.list() + if not isinstance(hooks, list) or len(hooks) == 0: + raise RuntimeError("expected at least one webhook, got 0") + run("24 Webhook.list", test_24) + + # 25 — Webhook.update + def test_25(): + if not registered_webhook_id: + raise RuntimeError("no webhook ID from test 23") + client.webhook.update( + registered_webhook_id, + {"url": webhook_url + "?updated=1", "secret": "updated-secret-python"} + ) + run("25 Webhook.update", test_25) + + # 26 — Webhook.verify_signature (valid) + def test_26(): + event_hash = "abc123eventHash" + sig = hmac_sha256_base64(SECRET, f"{client_id}:{event_hash}") + ok = client.webhook.verify_signature(sig, client_id, event_hash, SECRET) + if not ok: + raise RuntimeError("expected valid signature to return True") + run("26 Webhook.verify_signature (valid)", test_26) + + # 27 — Webhook.verify_signature (invalid) + def test_27(): + ok = client.webhook.verify_signature("invalidsig==", client_id, "somehash", SECRET) + if ok: + raise RuntimeError("expected invalid signature to return False") + run("27 Webhook.verify_signature (invalid)", test_27) + + # 28 — Webhook.create_handler (parses a webhook event) + def test_28(): + received_events = [] + handler = Webhook_create_handler(client, received_events) + payload = { + "eventType": "message.sent", + "data": {"to": "+15005550001"}, + "eventHash": "abc123", + } + result = handler(payload) + if not result.get("received"): + raise RuntimeError("handler did not return received=True") + if len(received_events) == 0: + raise RuntimeError("on_event callback was not called") + event = received_events[0] + if not event.event_type: + raise RuntimeError("eventType is empty after parsing") + run("28 Webhook.create_handler (parse event)", test_28) + + # 29 — Webhook.delete + def test_29(): + if not registered_webhook_id: + raise RuntimeError("no webhook ID from test 23") + client.webhook.delete(registered_webhook_id) + run("29 Webhook.delete", test_29) + + # ── Contact Tests (30-31) ──────────────────────────────────────────────────── + print("\n--- Contact ---") + + # 30 — Contact.set_do_not_text(True) + def test_30(): + client.contact.set_do_not_text(True, phone=phone1) + run("30 Contact.set_do_not_text(True)", test_30) + + # 31 — Contact.set_do_not_text(False) + def test_31(): + client.contact.set_do_not_text(False, phone=phone1) + run("31 Contact.set_do_not_text(False)", test_31) + + # ── Brand Tests (32-36) ────────────────────────────────────────────────────── + print("\n--- Brands ---") + + brand_id = None + + # 32 — Brand.create + def test_32(): + nonlocal brand_id + resp = client.brands.create({ + "legalCompanyName": "Test Company LLC", + "entityType": "PRIVATE_PROFIT", + "taxId": "123456789", + "taxIdCountry": "US", + "country": "US", + "verticalType": "TECHNOLOGY", + "websiteUrl": "https://example.com", + "street": "123 Main St", + "city": "Miami", + "state": "FL", + "postalCode": "33101", + "contactFirstName": fn1, + "contactLastName": ln1, + "contactEmail": email1, + "contactPhone": phone1, + }) + if not resp.get("id"): + raise RuntimeError("Invalid brand id") + brand_id = resp["id"] + run("32 Brand.create", test_32) + + # 33 — Brand.get + def test_33(): + if not brand_id: + raise RuntimeError("dependency test 32 failed") + resp = client.brands.get(brand_id) + if resp.get("id") != brand_id: + raise RuntimeError("Brand id mismatch") + run("33 Brand.get", test_33) + + # 34 — Brand.list + def test_34(): + resp = client.brands.list() + if not isinstance(resp, list): + raise RuntimeError("Expected a list") + run("34 Brand.list", test_34) + + # 35 — Brand.update + def test_35(): + if not brand_id: + raise RuntimeError("dependency test 32 failed") + resp = client.brands.update(brand_id, {"city": "Orlando"}) + if resp.get("id") != brand_id: + raise RuntimeError("Brand id mismatch after update") + run("35 Brand.update", test_35) + + # 36 — Brand.delete + def test_36(): + if not brand_id: + raise RuntimeError("dependency test 32 failed") + client.brands.delete(brand_id) + run("36 Brand.delete", test_36) + + # ── Campaign Tests (37-42) ──────────────────────────────────────────────────── + print("\n--- Campaigns ---") + + campaign_brand_id = None + campaign_id = None + + # 37 — Campaign setup: create brand + def test_37(): + nonlocal campaign_brand_id + resp = client.brands.create({ + "legalCompanyName": "Campaign Test LLC", + "entityType": "PRIVATE_PROFIT", + "taxId": "987654321", + "taxIdCountry": "US", + "country": "US", + "verticalType": "TECHNOLOGY", + "websiteUrl": "https://example.com", + "street": "456 Test Ave", + "city": "Miami", + "state": "FL", + "postalCode": "33101", + "contactFirstName": fn1, + "contactLastName": ln1, + "contactEmail": email1, + "contactPhone": phone1, + }) + if not resp.get("id"): + raise RuntimeError("Invalid brand id") + campaign_brand_id = resp["id"] + run("37 Campaign setup — Brand.create", test_37) + + # 38 — Campaign.create + def test_38(): + nonlocal campaign_id + if not campaign_brand_id: + raise RuntimeError("dependency test 37 failed") + resp = client.campaigns.create({ + "brandId": campaign_brand_id, + "useCase": "MARKETING", + "description": "Integration test campaign for automated testing", + "messageFlow": "Customers opt-in via website form at https://example.com/sms-signup", + "hasEmbeddedLinks": False, + "hasEmbeddedPhone": False, + "isAgeGated": False, + "isDirectLending": False, + "optInKeywords": ["START", "YES"], + "optInMessage": "You have opted in to receive messages. Reply STOP to unsubscribe.", + "optInProofUrl": "https://example.com/opt-in-proof", + "helpKeywords": ["HELP", "INFO"], + "helpMessage": "For help reply HELP or call 1-800-555-0000.", + "optOutKeywords": ["STOP", "END"], + "optOutMessage": "You have been unsubscribed. Reply START to opt back in. STOP", + "sampleMessages": [ + "Hello ${firstName}, this is a test message. Reply STOP to unsubscribe.", + "Reminder: your appointment is tomorrow. Reply HELP for assistance.", + ], + }) + if not resp.get("id"): + raise RuntimeError("Invalid campaign id") + campaign_id = resp["id"] + run("38 Campaign.create", test_38) + + # 39 — Campaign.get + def test_39(): + if not campaign_id: + raise RuntimeError("dependency test 38 failed") + resp = client.campaigns.get(campaign_id) + if resp.get("id") != campaign_id: + raise RuntimeError("Campaign id mismatch") + run("39 Campaign.get", test_39) + + # 40 — Campaign.list + def test_40(): + resp = client.campaigns.list() + if not isinstance(resp, list): + raise RuntimeError("Expected a list") + run("40 Campaign.list", test_40) + + # 41 — Campaign.update + def test_41(): + if not campaign_id: + raise RuntimeError("dependency test 38 failed") + resp = client.campaigns.update(campaign_id, {"description": "Updated integration test campaign description"}) + if resp.get("id") != campaign_id: + raise RuntimeError("Campaign id mismatch after update") + run("41 Campaign.update", test_41) + + # 42 — Campaign.delete + cleanup brand + def test_42(): + if not campaign_id: + raise RuntimeError("dependency test 38 failed") + client.campaigns.delete(campaign_id) + if campaign_brand_id: + client.brands.delete(campaign_brand_id) + run("42 Campaign.delete", test_42) + + # ── Contact Validator ───────────────────────────────────────────────────────── + + # 43 — ContactValidator.validate_email + def test_43(): + resp = client.contact_validator.validate_email(email1) + if not resp.status: + raise RuntimeError("status is empty") + run("43 ContactValidator.validate_email", test_43) + + # 44 — ContactValidator.validate_emails + def test_44(): + resp = client.contact_validator.validate_emails([email1, email2]) + if resp.summary.total != 2: + raise RuntimeError(f"expected summary.total=2, got {resp.summary.total}") + run("44 ContactValidator.validate_emails", test_44) + + # 45 — ContactValidator.validate_phone + def test_45(): + resp = client.contact_validator.validate_phone(phone1) + if not resp.status: + raise RuntimeError("status is empty") + run("45 ContactValidator.validate_phone", test_45) + + # 46 — ContactValidator.validate_phones + def test_46(): + resp = client.contact_validator.validate_phones([{"phone": phone1}, {"phone": phone2}]) + if resp.summary.total != 2: + raise RuntimeError(f"expected summary.total=2, got {resp.summary.total}") + run("46 ContactValidator.validate_phones", test_46) + + # ── Cleanup & Results ───────────────────────────────────────────────────────── + os.unlink(png_path) + + print("\n==============================================") + print(f" RESULTS: {passed} passed, {failed} failed") + print("==============================================") + + summary = json.dumps({"sdk": "python", "passed": passed, "failed": failed, "total": passed + failed}) + print(f"\nSUMMARY_JSON: {summary}") + + sys.exit(1 if failed > 0 else 0) + + +def Webhook_create_handler(client: CCAI, received_events: list): + """Helper to call Webhook.create_handler and capture parsed events.""" + def on_event(event: WebhookEvent): + received_events.append(event) + + return client.webhook.create_handler({"on_event": on_event}) + + +if __name__ == "__main__": + main() diff --git a/pyproject.toml b/pyproject.toml index 354ead9..1eb466d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "ccai-python" -version = "1.0.1" +version = "1.1.0" description = "Python client for CloudContactAI API" readme = "README.md" requires-python = ">=3.10" diff --git a/src/ccai_python/__init__.py b/src/ccai_python/__init__.py index 13db69f..d1ba9c9 100644 --- a/src/ccai_python/__init__.py +++ b/src/ccai_python/__init__.py @@ -11,6 +11,17 @@ from .email_service import Email, EmailAccount, EmailCampaign, EmailResponse, EmailOptions from .webhook import Webhook, WebhookConfig, WebhookEventType, WebhookEvent from .contact_service import Contact, ContactDoNotTextRequest, ContactDoNotTextResponse +from .brand_service import Brand +from .campaign_service import Campaign as CampaignService +from .contact_validator_service import ( + ContactValidator, + EmailValidationResult, + PhoneValidationResult, + ValidationSummary, + BulkEmailValidationResult, + BulkPhoneValidationResult, + PhoneInput, +) __all__ = [ 'CCAI', @@ -32,7 +43,16 @@ 'WebhookEvent', 'Contact', 'ContactDoNotTextRequest', - 'ContactDoNotTextResponse' + 'ContactDoNotTextResponse', + 'Brand', + 'CampaignService', + 'ContactValidator', + 'EmailValidationResult', + 'PhoneValidationResult', + 'ValidationSummary', + 'BulkEmailValidationResult', + 'BulkPhoneValidationResult', + 'PhoneInput', ] __version__ = '1.0.1' diff --git a/src/ccai_python/brand_service.py b/src/ccai_python/brand_service.py new file mode 100644 index 0000000..28b8e98 --- /dev/null +++ b/src/ccai_python/brand_service.py @@ -0,0 +1,91 @@ +"""Brand service for managing brand registrations via CloudContactAI API""" + +from typing import Any, Dict, List, Optional +import re + + +ENTITY_TYPES = {"PRIVATE_PROFIT", "PUBLIC_PROFIT", "NON_PROFIT", "GOVERNMENT", "SOLE_PROPRIETOR"} +VERTICAL_TYPES = { + "AUTOMOTIVE", "AGRICULTURE", "BANKING", "COMMUNICATION", "CONSTRUCTION", "EDUCATION", + "ENERGY", "ENTERTAINMENT", "GOVERNMENT", "HEALTHCARE", "HOSPITALITY", "INSURANCE", + "LEGAL", "MANUFACTURING", "NON_PROFIT", "PROFESSIONAL", "REAL_ESTATE", "RETAIL", + "TECHNOLOGY", "TRANSPORTATION", +} +TAX_ID_COUNTRIES = {"US", "CA", "GB", "AU"} +STOCK_EXCHANGES = {"NASDAQ", "NYSE", "AMEX", "TSX", "LON", "JPX", "HKEX", "OTHER"} + + +def _validate_brand(data: Dict[str, Any], is_create: bool = True) -> None: + errors: List[Dict[str, str]] = [] + + if is_create: + for field in [ + "legalCompanyName", "entityType", "taxId", "taxIdCountry", "country", + "verticalType", "websiteUrl", "street", "city", "state", "postalCode", + "contactFirstName", "contactLastName", "contactEmail", "contactPhone", + ]: + if not data.get(field): + errors.append({"field": field, "message": f"{field} is required"}) + + if "entityType" in data and data["entityType"] not in ENTITY_TYPES: + errors.append({"field": "entityType", "message": "Invalid entity type"}) + + if "verticalType" in data and data["verticalType"] not in VERTICAL_TYPES: + errors.append({"field": "verticalType", "message": "Invalid vertical type"}) + + if "taxIdCountry" in data and data["taxIdCountry"] not in TAX_ID_COUNTRIES: + errors.append({"field": "taxIdCountry", "message": "Invalid tax ID country"}) + + if "stockExchange" in data and data["stockExchange"] and data["stockExchange"] not in STOCK_EXCHANGES: + errors.append({"field": "stockExchange", "message": "Invalid stock exchange"}) + + if "websiteUrl" in data and data.get("websiteUrl"): + url = data["websiteUrl"] + if not url.startswith("http://") and not url.startswith("https://"): + errors.append({"field": "websiteUrl", "message": "Website URL must start with http:// or https://"}) + + if "contactEmail" in data and data.get("contactEmail"): + if not re.match(r"^[^@\s]+@[^@\s]+\.[^@\s]+$", data["contactEmail"]): + errors.append({"field": "contactEmail", "message": "Invalid email format"}) + + if "taxId" in data and data.get("taxId") and "taxIdCountry" in data: + if data["taxIdCountry"] in ("US", "CA") and not re.match(r"^\d{9}$", data["taxId"]): + errors.append({"field": "taxId", "message": f"Tax ID must be exactly 9 digits for {data['taxIdCountry']}"}) + + if data.get("entityType") == "PUBLIC_PROFIT": + if not data.get("stockSymbol"): + errors.append({"field": "stockSymbol", "message": "Stock symbol is required for PUBLIC_PROFIT entities"}) + if not data.get("stockExchange"): + errors.append({"field": "stockExchange", "message": "Stock exchange is required for PUBLIC_PROFIT entities"}) + + if errors: + raise ValueError({"message": "Validation failed", "errors": errors}) + + +class Brand: + """Brand service for managing brand registrations""" + + def __init__(self, ccai): + self._ccai = ccai + + def create(self, data: Dict[str, Any]) -> Dict[str, Any]: + """Create a new brand""" + _validate_brand(data, is_create=True) + return self._ccai.custom_request("post", "/v1/brands", data, base_url=self._ccai.compliance_base_url) + + def get(self, brand_id: int) -> Dict[str, Any]: + """Get a brand by ID""" + return self._ccai.custom_request("get", f"/v1/brands/{brand_id}", base_url=self._ccai.compliance_base_url) + + def list(self) -> List[Dict[str, Any]]: + """List all brands for the authenticated account""" + return self._ccai.custom_request("get", "/v1/brands", base_url=self._ccai.compliance_base_url) + + def update(self, brand_id: int, data: Dict[str, Any]) -> Dict[str, Any]: + """Partially update a brand""" + _validate_brand(data, is_create=False) + return self._ccai.custom_request("patch", f"/v1/brands/{brand_id}", data, base_url=self._ccai.compliance_base_url) + + def delete(self, brand_id: int) -> None: + """Delete a brand""" + self._ccai.custom_request("delete", f"/v1/brands/{brand_id}", base_url=self._ccai.compliance_base_url) diff --git a/src/ccai_python/campaign_service.py b/src/ccai_python/campaign_service.py new file mode 100644 index 0000000..6a4b8e9 --- /dev/null +++ b/src/ccai_python/campaign_service.py @@ -0,0 +1,125 @@ +"""Campaign service for managing campaign registrations via CloudContactAI API""" + +from typing import Any, Dict, List + + +CAMPAIGN_USE_CASES = { + "TWO_FACTOR_AUTHENTICATION", "ACCOUNT_NOTIFICATION", "CUSTOMER_CARE", "DELIVERY_NOTIFICATION", + "FRAUD_ALERT", "HIGHER_EDUCATION", "LOW_VOLUME_MIXED", "MARKETING", "MIXED", + "POLLING_VOTING", "PUBLIC_SERVICE_ANNOUNCEMENT", "SECURITY_ALERT", +} +CAMPAIGN_SUB_USE_CASES = { + "TWO_FACTOR_AUTHENTICATION", "ACCOUNT_NOTIFICATION", "CUSTOMER_CARE", "DELIVERY_NOTIFICATION", + "FRAUD_ALERT", "MARKETING", "POLLING_VOTING", +} + +MIXED_USE_CASES = {"MIXED", "LOW_VOLUME_MIXED"} + +REQUIRED_FIELDS = [ + "brandId", "useCase", "description", "messageFlow", "hasEmbeddedLinks", + "hasEmbeddedPhone", "isAgeGated", "isDirectLending", "optInKeywords", + "optInMessage", "optInProofUrl", "helpKeywords", "helpMessage", + "optOutKeywords", "optOutMessage", "sampleMessages", +] + + +def _validate_campaign(data: Dict[str, Any], is_create: bool = True) -> None: + errors: List[Dict[str, str]] = [] + + if is_create: + for field in REQUIRED_FIELDS: + if not data.get(field) and data.get(field) is not False: + errors.append({"field": field, "message": f"{field} is required"}) + + if "useCase" in data and data["useCase"] not in CAMPAIGN_USE_CASES: + errors.append({"field": "useCase", "message": "Invalid use case"}) + + use_case = data.get("useCase") + sub_use_cases = data.get("subUseCases") + + if use_case in MIXED_USE_CASES: + if not sub_use_cases or not isinstance(sub_use_cases, list) or not (2 <= len(sub_use_cases) <= 3): + errors.append({"field": "subUseCases", "message": "MIXED/LOW_VOLUME_MIXED requires 2-3 sub use cases"}) + elif sub_use_cases: + for suc in sub_use_cases: + if suc not in CAMPAIGN_SUB_USE_CASES: + errors.append({"field": "subUseCases", "message": f"Invalid sub use case: {suc}"}) + elif use_case and sub_use_cases: + errors.append({"field": "subUseCases", "message": "subUseCases should be empty for non-MIXED use cases"}) + + sample_messages = data.get("sampleMessages") + if sample_messages is not None: + if not isinstance(sample_messages, list) or not (2 <= len(sample_messages) <= 5): + errors.append({"field": "sampleMessages", "message": "sampleMessages must contain 2-5 items"}) + elif sample_messages: + opt_out_keywords = data.get("optOutKeywords") or [] + help_keywords = data.get("helpKeywords") or [] + + has_opt_out = any( + "Reply STOP" in msg or any(f"Reply {kw}" in msg for kw in opt_out_keywords) + for msg in sample_messages + ) + if not has_opt_out: + errors.append({"field": "sampleMessages", "message": "At least one sample must contain 'Reply STOP' or 'Reply {optOutKeyword}'"}) + + has_help = any( + "Reply HELP" in msg or any(f"Reply {kw}" in msg for kw in help_keywords) + for msg in sample_messages + ) + if not has_help: + errors.append({"field": "sampleMessages", "message": "At least one sample must contain 'Reply HELP' or 'Reply {helpKeyword}'"}) + + opt_out_message = data.get("optOutMessage") + if opt_out_message is not None: + opt_out_keywords = data.get("optOutKeywords") or [] + if "STOP" not in opt_out_message and not any(kw in opt_out_message for kw in opt_out_keywords): + errors.append({"field": "optOutMessage", "message": "optOutMessage must contain 'STOP' or at least one optOutKeyword"}) + + help_message = data.get("helpMessage") + if help_message is not None: + help_keywords = data.get("helpKeywords") or [] + if "HELP" not in help_message and not any(kw in help_message for kw in help_keywords): + errors.append({"field": "helpMessage", "message": "helpMessage must contain 'HELP' or at least one helpKeyword"}) + + opt_in_proof_url = data.get("optInProofUrl") + if opt_in_proof_url is not None: + if not opt_in_proof_url.startswith("http://") and not opt_in_proof_url.startswith("https://"): + errors.append({"field": "optInProofUrl", "message": "optInProofUrl must start with http:// or https://"}) + + for link_field in ("termsLink", "privacyLink"): + val = data.get(link_field) + if val: + if not val.startswith("http://") and not val.startswith("https://"): + errors.append({"field": link_field, "message": f"{link_field} must start with http:// or https://"}) + + if errors: + raise ValueError({"message": "Validation failed", "errors": errors}) + + +class Campaign: + """Campaign service for managing campaign registrations""" + + def __init__(self, ccai): + self._ccai = ccai + + def create(self, data: Dict[str, Any]) -> Dict[str, Any]: + """Create a new campaign""" + _validate_campaign(data, is_create=True) + return self._ccai.custom_request("post", "/v1/campaigns", data, base_url=self._ccai.compliance_base_url) + + def get(self, campaign_id: int) -> Dict[str, Any]: + """Get a campaign by ID""" + return self._ccai.custom_request("get", f"/v1/campaigns/{campaign_id}", base_url=self._ccai.compliance_base_url) + + def list(self) -> List[Dict[str, Any]]: + """List all campaigns for the authenticated account""" + return self._ccai.custom_request("get", "/v1/campaigns", base_url=self._ccai.compliance_base_url) + + def update(self, campaign_id: int, data: Dict[str, Any]) -> Dict[str, Any]: + """Partially update a campaign""" + _validate_campaign(data, is_create=False) + return self._ccai.custom_request("patch", f"/v1/campaigns/{campaign_id}", data, base_url=self._ccai.compliance_base_url) + + def delete(self, campaign_id: int) -> None: + """Delete a campaign""" + self._ccai.custom_request("delete", f"/v1/campaigns/{campaign_id}", base_url=self._ccai.compliance_base_url) diff --git a/src/ccai_python/ccai.py b/src/ccai_python/ccai.py index e81e0ed..08a7dcb 100644 --- a/src/ccai_python/ccai.py +++ b/src/ccai_python/ccai.py @@ -16,6 +16,9 @@ from .email_service import Email from .webhook import Webhook from .contact_service import Contact +from .brand_service import Brand +from .campaign_service import Campaign as CampaignService +from .contact_validator_service import ContactValidator class CCAIConfig(BaseModel): @@ -55,11 +58,13 @@ class CCAI: PROD_BASE_URL = "https://core.cloudcontactai.com/api" PROD_EMAIL_URL = "https://email-campaigns.cloudcontactai.com/api/v1" PROD_FILES_URL = "https://files.cloudcontactai.com" + PROD_COMPLIANCE_URL = "https://compliance.cloudcontactai.com/api" # Test environment URLs TEST_BASE_URL = "https://core-test-cloudcontactai.allcode.com/api" TEST_EMAIL_URL = "https://email-campaigns-test-cloudcontactai.allcode.com/api/v1" TEST_FILES_URL = "https://files-test-cloudcontactai.allcode.com" + TEST_COMPLIANCE_URL = "https://compliance-test-cloudcontactai.allcode.com/api" def __init__( self, @@ -68,6 +73,7 @@ def __init__( base_url: Optional[str] = None, email_base_url: Optional[str] = None, file_base_url: Optional[str] = None, + compliance_base_url: Optional[str] = None, use_test: bool = False ) -> None: if not client_id: @@ -85,6 +91,9 @@ def __init__( resolved_files = self._resolve_url( file_base_url, "CCAI_FILES_BASE_URL", self.PROD_FILES_URL, self.TEST_FILES_URL, use_test ) + self._compliance_base_url = self._resolve_url( + compliance_base_url, "CCAI_COMPLIANCE_BASE_URL", self.PROD_COMPLIANCE_URL, self.TEST_COMPLIANCE_URL, use_test + ) self._config = CCAIConfig( client_id=client_id, @@ -100,6 +109,9 @@ def __init__( self.email = Email(self) self.webhook = Webhook(self) self.contact = Contact(self) + self.brands = Brand(self) + self.campaigns = CampaignService(self) + self.contact_validator = ContactValidator(self) def _resolve_url( self, @@ -137,6 +149,10 @@ def email_base_url(self) -> str: def file_base_url(self) -> str: return self._config.file_base_url + @property + def compliance_base_url(self) -> str: + return self._compliance_base_url + @property def use_test(self) -> bool: return self._config.use_test @@ -164,6 +180,8 @@ def request( timeout=timeout ) response.raise_for_status() + if response.status_code == 204: + return cast(Dict[str, Any], {}) return cast(Dict[str, Any], response.json()) except requests.HTTPError as e: if e.response is not None: @@ -205,6 +223,8 @@ def custom_request( timeout=timeout ) response.raise_for_status() + if response.status_code == 204: + return cast(Dict[str, Any], {}) return cast(Dict[str, Any], response.json()) except requests.HTTPError as e: if e.response is not None: diff --git a/src/ccai_python/contact_validator_service.py b/src/ccai_python/contact_validator_service.py new file mode 100644 index 0000000..75b35bb --- /dev/null +++ b/src/ccai_python/contact_validator_service.py @@ -0,0 +1,76 @@ +""" +contact_validator_service.py - Email and phone contact validation via CloudContactAI + +:license: MIT +:copyright: 2025 CloudContactAI LLC +""" + +from typing import Any, Dict, List, Optional +from pydantic import BaseModel + + +class EmailValidationResult(BaseModel): + contact: str + type: str + status: str + metadata: Dict[str, Any] = {} + + +class PhoneValidationResult(BaseModel): + contact: str + type: str + status: str + metadata: Dict[str, Any] = {} + + +class ValidationSummary(BaseModel): + total: int + valid: int + invalid: int + risky: int + landline: int = 0 + + +class BulkEmailValidationResult(BaseModel): + results: List[EmailValidationResult] + summary: ValidationSummary + + +class BulkPhoneValidationResult(BaseModel): + results: List[PhoneValidationResult] + summary: ValidationSummary + + +class PhoneInput(BaseModel): + phone: str + countryCode: Optional[str] = None + + +class ContactValidator: + """Service for validating email addresses and phone numbers""" + + def __init__(self, ccai) -> None: + self.ccai = ccai + + def validate_email(self, email: str) -> EmailValidationResult: + """Validate a single email address""" + response = self.ccai.request('POST', '/v1/contact-validator/email', {'email': email}) + return EmailValidationResult(**response) + + def validate_emails(self, emails: List[str]) -> BulkEmailValidationResult: + """Validate multiple email addresses (up to 50)""" + response = self.ccai.request('POST', '/v1/contact-validator/emails', {'emails': emails}) + return BulkEmailValidationResult(**response) + + def validate_phone(self, phone: str, country_code: Optional[str] = None) -> PhoneValidationResult: + """Validate a single phone number in E.164 format""" + response = self.ccai.request( + 'POST', '/v1/contact-validator/phone', + {'phone': phone, 'countryCode': country_code} + ) + return PhoneValidationResult(**response) + + def validate_phones(self, phones: List[Dict[str, Any]]) -> BulkPhoneValidationResult: + """Validate multiple phone numbers (up to 50)""" + response = self.ccai.request('POST', '/v1/contact-validator/phones', {'phones': phones}) + return BulkPhoneValidationResult(**response) diff --git a/src/ccai_python/examples/brand_example.py b/src/ccai_python/examples/brand_example.py new file mode 100644 index 0000000..c3557dd --- /dev/null +++ b/src/ccai_python/examples/brand_example.py @@ -0,0 +1,72 @@ +""" +Brand registration example using the CCAI Python client + +:license: MIT +:copyright: 2025 CloudContactAI LLC +""" + +from ccai_python import CCAI + +# Initialize the client +ccai = CCAI( + client_id="YOUR-CLIENT-ID", + api_key="API-KEY-TOKEN" +) + + +def brand_examples(): + """Example of managing brands""" + try: + # Create a brand + print("Creating a brand...") + brand = ccai.brands.create({ + "legalCompanyName": "Collect.org Inc.", + "dba": "Collect", + "entityType": "NON_PROFIT", + "taxId": "123456789", + "taxIdCountry": "US", + "country": "US", + "verticalType": "NON_PROFIT", + "websiteUrl": "https://www.collect.org", + "street": "123 Main Street", + "city": "San Francisco", + "state": "CA", + "postalCode": "94105", + "contactFirstName": "Jane", + "contactLastName": "Doe", + "contactEmail": "jane@collect.org", + "contactPhone": "+14155551234", + }) + print(f"Brand created with ID: {brand['id']}") + + # Get brand by ID + print("\nFetching brand by ID...") + fetched = ccai.brands.get(brand["id"]) + print(f"Brand: {fetched['legalCompanyName']}, Score: {fetched.get('websiteMatchScore')}") + + # List all brands + print("\nListing all brands...") + brands = ccai.brands.list() + print(f"Found {len(brands)} brand(s)") + + # Update a brand + print("\nUpdating brand...") + updated = ccai.brands.update(brand["id"], { + "street": "456 Oak Avenue", + "city": "Los Angeles", + "contactEmail": "admin@collect.org", + }) + print(f"Brand updated: {updated['street']}, {updated['city']}") + + # Delete a brand + print("\nDeleting brand...") + ccai.brands.delete(brand["id"]) + print("Brand deleted successfully") + + except Exception as error: + print(f"Error: {error}") + raise + + +if __name__ == "__main__": + brand_examples() diff --git a/src/ccai_python/examples/campaign_example.py b/src/ccai_python/examples/campaign_example.py new file mode 100644 index 0000000..0f62f60 --- /dev/null +++ b/src/ccai_python/examples/campaign_example.py @@ -0,0 +1,81 @@ +""" +Campaign registration example using the CCAI Python client + +:license: MIT +:copyright: 2025 CloudContactAI LLC +""" + +from ccai_python import CCAI + +# Initialize the client +ccai = CCAI( + client_id="YOUR-CLIENT-ID", + api_key="API-KEY-TOKEN" +) + + +def campaign_examples(): + """Example of managing campaigns""" + try: + # Create a campaign (assumes brand ID 1 exists) + print("Creating a campaign...") + campaign = ccai.campaigns.create({ + "brandId": 1, + "useCase": "MIXED", + "subUseCases": ["CUSTOMER_CARE", "TWO_FACTOR_AUTHENTICATION", "ACCOUNT_NOTIFICATION"], + "description": "This campaign handles security codes and support for Collect.org.", + "messageFlow": "Users opt-in via our signup form checkbox at https://collect.org/signup", + "termsLink": "https://collect.org/terms", + "privacyLink": "https://collect.org/privacy", + "hasEmbeddedLinks": True, + "hasEmbeddedPhone": False, + "isAgeGated": False, + "isDirectLending": False, + "optInKeywords": ["START", "JOIN"], + "optInMessage": "Welcome to Collect.org! Msg&Data rates may apply. Reply STOP to cancel.", + "optInProofUrl": "https://collect.org/images/opt-in-proof.png", + "helpKeywords": ["HELP", "INFO"], + "helpMessage": "Collect.org: For help email support@collect.org. Reply STOP to cancel.", + "optOutKeywords": ["STOP", "UNSUBSCRIBE"], + "optOutMessage": "Collect.org: You have been unsubscribed. STOP received.", + "sampleMessages": [ + "Your Collect.org security code is 554321. Reply STOP to cancel.", + "Hi [Name], your ticket #[ID] has been updated. Reply HELP for more info." + ] + }) + print(f"Campaign created with ID: {campaign['id']}, fee: ${campaign['monthlyFee']}/mo") + + # Get campaign by ID + print("\nFetching campaign by ID...") + fetched = ccai.campaigns.get(campaign["id"]) + print(f"Campaign: {fetched['useCase']}, Brand: {fetched['brandId']}") + + # List all campaigns + print("\nListing all campaigns...") + campaigns = ccai.campaigns.list() + print(f"Found {len(campaigns)} campaign(s)") + + # Update a campaign + print("\nUpdating campaign...") + updated = ccai.campaigns.update(campaign["id"], { + "description": "Updated campaign description for Collect.org messaging.", + "sampleMessages": [ + "Your Collect.org code is 123456. Reply STOP to opt-out.", + "Your support ticket has been resolved. Reply HELP for more info.", + "Your payment of $50.00 was received. Reply STOP to cancel." + ] + }) + print(f"Campaign updated: {updated['description']}") + + # Delete a campaign + print("\nDeleting campaign...") + ccai.campaigns.delete(campaign["id"]) + print("Campaign deleted successfully") + + except Exception as error: + print(f"Error: {error}") + raise + + +if __name__ == "__main__": + campaign_examples()