khushalps/api_utils_converse_v2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

(20250121) With APIs to send, list, and update tags on WhatsApp messages (through Nimbus).

Browse Source
This commit is contained in:
khushal
2025-01-21 15:27:53 +05:30
parent f94d9e1a4b
commit cadbc19f28
13 changed files with 498 additions and 182 deletions
Download Patch File Download Diff File Expand all files Collapse all files
models/api/message/chat/list.py
+8 -8
View File
@@ -6,11 +6,11 @@
DATE:
Tuesday, 3rd Dec., 2024.
Tuesday, 21st Jan., 2025.
OBJECTIVE:
To provide a structure to query the full payload of an email.
To provide a structure to query the full payload of chat message(s).
REFERENCES:
@@ -75,10 +75,10 @@ REGEX_SESSION_TOKEN = r"^[a-f0-9]{8}-[a-f0-9]{4}-[1-5][a-f0-9]{3}-[89ab][a-f0-9]
# *****************************************************************************************************************
class SMSListRequestHeaders(BaseModel):
class ChatMessageListRequestHeaders(BaseModel):
sessionToken: str = Field(
description = "the session token of the user who is requesting the service",
description = "The session token of the user who is requesting the service.",
pattern = REGEX_SESSION_TOKEN,
frozen = True,
alias = "X-Session-Token"
@@ -99,10 +99,10 @@ class SMSListRequestHeaders(BaseModel):
# ---------------------------------------------------------------------------------------------------------------------
class SMSListRequestData(BaseModel):
class ChatMessageListRequestData(BaseModel):
tokenKeys: str | List[str] = Field(
description = "the token identifier(s) that tell you which auth-tokens were used for fetching those messages",
description = "The token identifier(s) that tell you which auth-tokens were used for fetching those messages.",
frozen = True,
)
@@ -115,14 +115,14 @@ class SMSListRequestData(BaseModel):
)
fromCount: int = Field(
description = "the no. of mails to skip before picking mails to list; useful for pagination",
description = "The no. of mails to skip before picking mails to list; useful for pagination.",
ge = 0,
default = 0,
frozen = True
)
tags: List[Any] | None = Field(
description = "any no. of tags that you want to filter by",
description = "Any no. of tags that you want to filter by.",
default = None
)
models/api/message/chat/send.py
+9 -10
View File
@@ -6,11 +6,11 @@
DATE:
Monday, 9th Dec., 2024.
Tuesday, 21st Jan., 2025.
OBJECTIVE:
To provide a structure to receive API calls to send SMS messages from various third-party clients.
To provide a structure to receive API calls to send chat messages from various third-party clients.
REFERENCES:
@@ -46,11 +46,10 @@ from utils_v2.date_time import date_time
# Models:
from models.core.message import CoreMessageModel
from utils_v2.sms.models.sms_message import SentSMSMessageModel
from models.message.sms.send import (
NimbusSMSIndiaMessage,
SavvyBulkSMSKenyaMessage,
SMSSendOneResult,
SMSSendManyResults
from models.message.chat.send import (
NimbusWhatsAppMessage,
ChatSendOneResult,
ChatSendManyResults
)
# To work with date and time:
@@ -88,7 +87,7 @@ REGEX_SESSION_TOKEN = r"^[a-f0-9]{8}-[a-f0-9]{4}-[1-5][a-f0-9]{3}-[89ab][a-f0-9]
# *****************************************************************************************************************
class SMSSendRequestHeaders(BaseModel):
class ChatSendRequestHeaders(BaseModel):
sessionToken: str | None = Field(
description = "the session token of the user who is requesting the service",
@@ -113,10 +112,10 @@ class SMSSendRequestHeaders(BaseModel):
# ---------------------------------------------------------------------------------------------------------------------
class SMSSendRequestData(BaseModel):
class ChatSendRequestData(BaseModel):
tokenKey: ObjectId
message: List[NimbusSMSIndiaMessage] | List[SavvyBulkSMSKenyaMessage]
message: List[NimbusWhatsAppMessage]
tags: List[Any] | None = Field(default = None, validate_default = True)
# ┏┓ ┏•
models/api/message/chat/tags.py
+8 -8
View File
@@ -6,11 +6,11 @@
DATE:
Thursday, 19th Dec., 2024.
Tuesday, 21st Jan., 2025.
OBJECTIVE:
To provide a structure to work with the tags on SMS messages.
To provide a structure to work with the tags on chat messages.
REFERENCES:
@@ -75,10 +75,10 @@ REGEX_SESSION_TOKEN = r"^[a-f0-9]{8}-[a-f0-9]{4}-[1-5][a-f0-9]{3}-[89ab][a-f0-9]
# *****************************************************************************************************************
class SMSUpdateTagsRequestHeaders(BaseModel):
class ChatMessageUpdateTagsRequestHeaders(BaseModel):
sessionToken: str = Field(
description = "the session token of the user who is requesting the service",
description = "The session token of the user who is requesting the service.",
pattern = REGEX_SESSION_TOKEN,
frozen = True,
alias = "X-Session-Token"
@@ -99,21 +99,21 @@ class SMSUpdateTagsRequestHeaders(BaseModel):
# ---------------------------------------------------------------------------------------------------------------------
class SMSUpdateTagsRequestData(BaseModel):
class ChatMessageUpdateTagsRequestData(BaseModel):
messageId: str = Field(
description = "the mail identifier (Mongo ObjectId) of the document that holds the mail",
description = "The message identifier (Mongo ObjectId) of the document that holds the chat message.",
frozen = True
)
unsetTags: List[Any] | None = Field(
description = "the list of tags to remove from the mail",
description = "The list of tags to remove from the message.",
frozen = True,
default = None
)
setTags: List[Any] | None = Field(
description = "the list of tags to add to the mail",
description = "The list of tags to add to the message.",
frozen = True,
default = None
)
models/message/chat/send.py
+63 -50
View File
@@ -6,11 +6,13 @@
DATE:
Monday, 9th Dec., 2024.
Tuesday, 21st Jan., 2025.
OBJECTIVE:
To provide a structure to receive API calls to send SMS messages from various third-party clients.
To provide a structure to receive API calls to send chat messages from various third-party clients. At the time
of creating this file we are starting with Nimbus IT's unofficial WhatsApp services. We intend to add Telegram's
official APIs soon after.
REFERENCES:
@@ -36,7 +38,7 @@ sys.path.append(".")
sys.path.append("..")
# For making data behaviour_models:
from pydantic import BaseModel, Field, field_validator, PastDatetime
from pydantic import BaseModel, Field, field_validator, AwareDatetime, PastDatetime
from typing import Optional, Literal, Union, List, Any
# My utils:
@@ -82,24 +84,42 @@ REGEX_SESSION_TOKEN = r"^[a-f0-9]{8}-[a-f0-9]{4}-[1-5][a-f0-9]{3}-[89ab][a-f0-9]
# *****************************************************************************************************************
class NimbusSMSIndiaMessage(BaseModel):
class NimbusWhatsAppMessage(BaseModel):
recipientNo: str = Field(
description = "the phone no. of the target recipient",
pattern = r"\+?\d{0,3}\s*\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4}",
description = "The phone no. of the target recipient(s)",
# pattern = r"\+?\d{0,3}\s*\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4}",
frozen = True
)
text: str = Field(
description = "the actual text that you want to send",
message: str = Field(
description = "The actual text that you want to send.",
min_length = 1,
frozen = True
)
templateId: str = Field(
description = "the id of the template that you are trying to use to send the message",
min_length = 1,
frozen = True
pdfUrl: str | None = Field(
description = "A PDF file to send with your message.",
frozen = True,
default = None
)
image0Url: str | None = Field(
description = "An image file to send with your message.",
frozen = True,
default = None
)
image1Url: str | None = Field(
description = "An image file to send with your message.",
frozen = True,
default = None
)
scheduleTs: AwareDatetime | None = Field(
description = "The time (UTC) at which the message needs to be sent. Null for immediate delivery.",
frozen = True,
default = None
)
# ┏┓ ┏•
@@ -114,6 +134,25 @@ class NimbusSMSIndiaMessage(BaseModel):
# ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓
# ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗
@field_validator("scheduleTs", mode = "before")
def to_datetime(cls, value):
if not isinstance(value, datetime.datetime):
value = date_time.parse_date_time(
input_value = value,
date_formats = [
"%Y%m%d",
"%Y-%m-%d",
"%Y-%m-%d %H:%M:%S",
"%Y-%m-%d %H:%M:%S%z",
"%Y-%m-%dT%H:%M:%S",
"%Y-%m-%dT%H:%M:%S%z"
]
)
if value:
value = date_time.as_if_timezone(value, date_time.TIMEZONE_UTC)
value = date_time.to_timezone(value, date_time.TIMEZONE_IST)
return value
@field_validator("recipientNo", mode = "before")
def validate_contact_nos(cls, value):
value = regex.replace(text = str(value), pattern = r"[^\d]", substitute_text = "")
@@ -125,46 +164,20 @@ class NimbusSMSIndiaMessage(BaseModel):
# ---------------------------------------------------------------------------------------------------------------------
class SavvyBulkSMSKenyaMessage(BaseModel):
recipientNo: str = Field(
description = "the phone no. of the target recipient",
min_length = 1,
frozen = True
)
text: str = Field(
description = "the actual text that you want to send",
min_length = 1,
frozen = True
)
# ┏┓ ┏•
# ┃ ┏┓┏┓╋┓┏┓
# ┗┛┗┛┛┗┛┗┗┫
# ┛
class Config:
extra = "forbid"
# ---------------------------------------------------------------------------------------------------------------------
class SMSSendOneResult(BaseModel):
class ChatSendOneResult(BaseModel):
success: bool = Field(
description = "whether, or not, the sms was successfully sent",
description = "Whether, or not, the chat message was successfully sent.",
default = False
)
message: str | None = Field(
description = "a brief message to summarize the result of the process",
description = "A brief message to summarize the result of the process.",
default = None
)
smsMessage: NimbusSMSIndiaMessage | SavvyBulkSMSKenyaMessage = Field(
description = "the actual data of the sms",
chatMessage: CoreMessageModel = Field(
description = "The actual data of the chat message.",
default = None
)
@@ -180,30 +193,30 @@ class SMSSendOneResult(BaseModel):
# ---------------------------------------------------------------------------------------------------------------------
class SMSSendManyResults(BaseModel):
class ChatSendManyResults(BaseModel):
totalCount: int = Field(
description = "the total no. of mails that were to be sync'd",
description = "The total no. of messages that were attempted.",
default = 0
)
successCount: int = Field(
description = "the no. of mails that were successfully sync'd",
description = "The no. of chat messages that were successfully sent.",
default = 0
)
failureCount: int = Field(
description = "the no. of mails that were successfully sync'd",
description = "The no. of chat messages that were NOT sent.",
default = 0
)
message: str = Field(
description = "a brief message to summarize the results of the process",
description = "A brief message to summarize the results of the process.",
default = None
)
smsMessages: List[CoreMessageModel] = Field(
description = "the actual data of the sms",
chatMessages: List[CoreMessageModel] = Field(
description = "The actual data of the chat messages.",
default = []
)