First version using the scripts that I already had.

.github/workflows/publish-to-pypi.yml

@@ -0,0 +1,34 @@
+name: Publish Python package to PyPI
+
+on:
+  push:
+    tags:
+      - 'v*'
+
+jobs:
+  build-and-publish:
+    runs-on: ubuntu-latest
+
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: '3.11'
+
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install setuptools wheel twine
+
+      - name: Build package
+        run: |
+          python setup.py sdist bdist_wheel
+
+      - name: Publish to PyPI
+        env:
+          TWINE_USERNAME: __token__
+          TWINE_PASSWORD: ${{ secrets.PYPI_API_TOKEN }}
+        run: |
+          twine upload dist/* --verbose

.gitignore

@@ -1,3 +1,5 @@
+/.idea
+
 # Byte-compiled / optimized / DLL files
 __pycache__/
 *.py[cod]

README.md

@@ -1,2 +1,14 @@
 # whatsapp-wrapper
+
 Integration layer for WhatsApp Cloud API with Firestore for easy message storage and management.
+
+## WhatsApp Documentation
+
+- [API Endpoint References](https://developers.facebook.com/docs/whatsapp/cloud-api/reference)
+    - [Messages](https://developers.facebook.com/docs/whatsapp/cloud-api/reference/messages)
+- [API Error Codes](https://developers.facebook.com/docs/whatsapp/cloud-api/support/error-codes)
+- [Override Callback URL](https://developers.facebook.com/docs/whatsapp/embedded-signup/webhooks/#overriding-the-callback-url)
+
+## Graph API Documentation
+
+- [WhatsApp Business Account](https://developers.facebook.com/docs/graph-api/reference/whats-app-business-account/)

requirements.txt

@@ -0,0 +1,4 @@
+wheel
+twine
+requests
+firestore-wrapper

setup.py

@@ -0,0 +1,39 @@
+from setuptools import find_packages, setup
+
+setup(
+    name='whatsapp_wrapper',
+    version='0.1.0',
+    author='Antonio Ventilii',
+    author_email='[email protected]',
+    license='MIT',
+    description='Integration layer for WhatsApp Cloud API with Firestore for easy message storage and management.',
+    long_description=open('README.md').read(),
+    long_description_content_type='text/markdown',
+    url='https://github.com/AntonioVentilii/whatsapp-wrapper',
+    project_urls={
+        'Source Code': 'https://github.com/AntonioVentilii/whatsapp-wrapper',
+        'Issue Tracker': 'https://github.com/AntonioVentilii/whatsapp-wrapper/issues',
+    },
+    packages=find_packages(),
+    classifiers=[
+        'Development Status :: 3 - Alpha',
+        'Intended Audience :: Developers',
+        'Topic :: Software Development :: Libraries :: Python Modules',
+        'License :: OSI Approved :: MIT License',
+        'Programming Language :: Python :: 3',
+        'Programming Language :: Python :: 3.8',
+        'Programming Language :: Python :: 3.9',
+        'Programming Language :: Python :: 3.10',
+        'Programming Language :: Python :: 3.11',
+    ],
+    keywords='WhatsApp, Firestore, API, integration, messaging',
+    install_requires=[
+        'requests',
+        'firestore-wrapper'
+    ],
+    python_requires='>=3.8',
+    entry_points={
+        'console_scripts': [
+        ],
+    },
+)

whatsapp_wrapper/__init__.py

@@ -0,0 +1 @@
+from .core import WhatsAppAPI

whatsapp_wrapper/configs.py

@@ -0,0 +1,22 @@
+import os
+import tempfile
+
+BASE_URL = 'https://graph.facebook.com'
+
+LAST_API_VERSION = 'v19.0'
+
+"""
+When CONTEXT_AS_CHAIN is True, the user needs to reply to the last message in the conversation chain of answer and 
+replies, in order to continue the conversation. If the user sends a new message without replying, the conversation 
+chain will be reset.
+When CONTEXT_AS_CHAIN is False and CONTEXT_HISTORY_DURATION is a positive non-null number, the bot considers the entire
+conversation history as context for generating a reply, starting from the message that was CONTEXT_HISTORY_DURATION 
+seconds ago.
+In all other cases, the bot will generate a reply without any context.
+"""
+CONTEXT_AS_CHAIN = True
+CONTEXT_HISTORY_DURATION = 24 * 60 * 60  # 24 hours
+
+CACHE_DIR = os.path.join(tempfile.gettempdir(), "whatsapp_wrapper_cache")
+if not os.path.exists(CACHE_DIR):
+    os.makedirs(CACHE_DIR)

whatsapp_wrapper/core.py

@@ -0,0 +1,197 @@
+from __future__ import annotations
+
+import time
+import warnings
+from typing import Any, Callable, Optional
+
+import requests
+from requests import Response
+
+from .configs import BASE_URL, LAST_API_VERSION
+from .exceptions import WhatsAppAPIWarning
+from .media_utilities import check_media_type_supported, save_media_to_temp_cache
+from .message_object import MessageObject
+from .whatsapp_db import DatabaseConfig, WhatsAppDB, configure_database
+
+ErrorHandlerType = Callable[[Response, dict], Optional[Any]]
+
+
+def _default_error_handler(response: Response, data: dict):
+    status_code = response.status_code
+    error_message = (
+        f"Request was failed with status code: {status_code}."
+        f" Data: {data}"
+    )
+    raise Exception(error_message)
+
+
+class WhatsAppAPI:
+    BASE_URL = BASE_URL
+    DEFAULT_API_VERSION = LAST_API_VERSION
+
+    _MESSAGE_URI = 'messages'
+    _MEDIA_URI = 'media'
+
+    def __init__(self, mobile_id: str, api_token: str, version: str = LAST_API_VERSION,
+                 database_config: DatabaseConfig = None, error_handler: ErrorHandlerType = None):
+        self.mobile_id = mobile_id
+        self.bearer_token = api_token
+        self.version = version or self.DEFAULT_API_VERSION
+        if database_config is None:
+            self._db = None
+            txt = "Database configuration is not provided. Chats will not be saved."
+            warnings.warn(txt, WhatsAppAPIWarning)
+        else:
+            self._db = configure_database(database_config)
+        self.error_handler: Callable[[Response, dict | None], Any | None] = error_handler or _default_error_handler
+
+    @property
+    def db(self) -> WhatsAppDB:
+        return self._db
+
+    @property
+    def base_url(self):
+        return f'{self.BASE_URL}/{self.version}'
+
+    @property
+    def mobile_url(self):
+        return f'{self.base_url}/{self.mobile_id}'
+
+    @property
+    def headers(self):
+        return {'Authorization': f'Bearer {self.bearer_token}'}
+
+    def request(self, method: str, uri: str, params: dict = None, data: dict = None, base_url: str = None,
+                headers: dict = None) -> Response:
+        method = method.upper()
+        if method not in ['GET', 'POST']:
+            raise ValueError(f"Invalid method: {method}")
+        elif method == 'GET':
+            if data:
+                raise ValueError(f"Data is not allowed for GET requests")
+        elif method == 'POST':
+            if params:
+                raise ValueError(f"Params is not allowed for POST requests")
+            if not data:
+                raise ValueError(f"Data is required for POST requests")
+
+        base_url = base_url or self.base_url
+        url = f"{base_url}/{uri}"
+        headers = headers or self.headers
+
+        if method == 'GET':
+            r = requests.request(method, url, headers=headers, params=params)
+        else:
+            r = requests.request(method, url, headers=headers, json=data)
+
+        if r.status_code != 200:
+            r = self.error_handler(r, data)
+
+        return r
+
+    def get_request(self, uri: str, params: dict = None, base_url: str = None, headers: dict = None):
+        return self.request('GET', uri=uri, params=params, base_url=base_url, headers=headers)
+
+    def post_request(self, uri: str, data: dict = None, base_url: str = None, headers: dict = None):
+        return self.request('POST', uri=uri, data=data, base_url=base_url, headers=headers)
+
+    def get_media_details(self, media_id: str) -> dict:
+        r = self.get_request(media_id)
+        ret = r.json()
+        return ret
+
+    def get_media_data(self, media_url: str) -> bytes:
+        r = requests.get(media_url, headers=self.headers)
+        if r.status_code != 200:
+            raise Exception(f"Failed to get media data! Status code: {r.status_code} - Reason: {r.reason}")
+        ret = r.content
+        return ret
+
+    def get_media_file(self, media_url: str, mime_type: str) -> str:
+        check_media_type_supported(mime_type, raise_error=True)
+        media = self.get_media_data(media_url)
+        file = save_media_to_temp_cache(mime_type, media)
+        return file
+
+    def get_media(self, media_id: str) -> str:
+        metadata = self.get_media_details(media_id)
+        url = metadata['url']
+        mime_type = metadata['mime_type']
+        file = self.get_media_file(url, mime_type)
+        return file
+
+    def save_message_to_db(self, data: dict, wa_id: str):
+        if self.db is None:
+            txt = f"Database configuration is not provided. Message {data} will not be saved."
+            warnings.warn(txt, WhatsAppAPIWarning)
+        else:
+            self.db.add_chat_message(data, wa_id)
+
+    def _send_message(self, data: dict, save_to_db: bool = True) -> dict:
+        base_url = self.mobile_url
+        uri = self._MESSAGE_URI
+        r = self.post_request(uri, data=data, base_url=base_url)
+        ret = r.json()
+        ret['data'] = data
+        if save_to_db:
+            ret['data']['id'] = ret['messages'][0]['id']
+            ret['data']['timestamp'] = time.time()
+            self.save_message_to_db(ret['data'], data['to'])
+        return ret
+
+    def mark_as_read(self, message_id: str) -> dict:
+        data = {
+            "messaging_product": "whatsapp",
+            "status": "read",
+            "message_id": message_id,
+        }
+        ret = self._send_message(data)
+        return ret
+
+    def send_text(self, to: str, message: str, preview_url: bool = False,
+                  reply_to_message_id: str = None, save_to_db: bool = True) -> dict:
+        message_obj = MessageObject(to, reply_to_message_id=reply_to_message_id)
+        payload = message_obj.text(message, preview_url=preview_url)
+        ret = self._send_message(payload, save_to_db=save_to_db)
+        return ret
+
+    def send_reaction(self, to: str, reply_to_message_id: str, emoji: str, save_to_db: bool = True) -> dict:
+        message_obj = MessageObject(to, reply_to_message_id=reply_to_message_id)
+        payload = message_obj.reaction(emoji)
+        ret = self._send_message(payload, save_to_db=save_to_db)
+        return ret
+
+    def send_audio(self, to: str, audio_id_or_url: str, reply_to_message_id: str = None,
+                   save_to_db: bool = True) -> dict:
+        message_obj = MessageObject(to, reply_to_message_id=reply_to_message_id)
+        payload = message_obj.audio(audio_id_or_url)
+        ret = self._send_message(payload, save_to_db=save_to_db)
+        return ret
+
+    def send_document(self, to: str, document_id_or_url: str, caption: str = None, filename: str = None,
+                      reply_to_message_id: str = None, save_to_db: bool = True) -> dict:
+        message_obj = MessageObject(to, reply_to_message_id=reply_to_message_id)
+        payload = message_obj.document(document_id_or_url, caption=caption, filename=filename)
+        ret = self._send_message(payload, save_to_db=save_to_db)
+        return ret
+
+    def send_image(self, to: str, image_id_or_url: str, caption: str = None, reply_to_message_id: str = None,
+                   save_to_db: bool = True) -> dict:
+        message_obj = MessageObject(to, reply_to_message_id=reply_to_message_id)
+        payload = message_obj.image(image_id_or_url, caption=caption)
+        ret = self._send_message(payload, save_to_db=save_to_db)
+        return ret
+
+    def send_sticker(self, to: str, sticker_id_or_url: str, reply_to_message_id: str = None,
+                     save_to_db: bool = True) -> dict:
+        message_obj = MessageObject(to, reply_to_message_id=reply_to_message_id)
+        payload = message_obj.sticker(sticker_id_or_url)
+        ret = self._send_message(payload, save_to_db=save_to_db)
+        return ret
+
+    def send_video(self, to: str, video_id_or_url: str, caption: str = None, reply_to_message_id: str = None,
+                   save_to_db: bool = True) -> dict:
+        message_obj = MessageObject(to, reply_to_message_id=reply_to_message_id)
+        payload = message_obj.video(video_id_or_url, caption=caption)
+        ret = self._send_message(payload, save_to_db=save_to_db)
+        return ret

whatsapp_wrapper/exceptions.py

@@ -0,0 +1,2 @@
+class WhatsAppAPIWarning(Warning):
+    pass

whatsapp_wrapper/media_utilities.py

@@ -0,0 +1,66 @@
+import os
+import uuid
+from datetime import datetime, timedelta
+from mimetypes import guess_extension
+
+from .configs import CACHE_DIR
+
+supported_types = {
+    'audio': ['audio/aac', 'audio/mp4', 'audio/mpeg', 'audio/amr', 'audio/ogg'],
+    'document': ['text/plain', 'application/pdf', 'application/vnd.ms-powerpoint', 'application/msword',
+                 'application/vnd.ms-excel', 'application/vnd.openxmlformats-officedocument.wordprocessingml.document',
+                 'application/vnd.openxmlformats-officedocument.presentationml.presentation',
+                 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet'],
+    'image': ['image/jpeg', 'image/png'],
+    'video': ['video/mp4', 'video/3gp'],
+    'sticker': ['image/webp'],
+}
+flat_supported_types = [item for sublist in supported_types.values() for item in sublist]
+
+type_to_extension = {
+    'audio/ogg': '.ogg',
+}
+
+
+def check_media_type_supported(mime_type: str, raise_error: bool = False) -> bool:
+    check = mime_type in flat_supported_types
+    if not check and raise_error:
+        raise ValueError(f"MIME type '{mime_type}' is not supported.")
+    return check
+
+
+def delete_old_files_in_cache(cache_dir: str = None):
+    cache_dir = cache_dir or CACHE_DIR
+
+    if not os.path.exists(cache_dir):
+        print("Cache directory does not exist.")
+        return
+
+    threshold_date = datetime.now() - timedelta(weeks=1)
+
+    for filename in os.listdir(cache_dir):
+        file_path = os.path.join(cache_dir, filename)
+
+        mod_time = os.path.getmtime(file_path)
+        file_date = datetime.fromtimestamp(mod_time)
+
+        if file_date < threshold_date:
+            os.remove(file_path)
+            print(f"Deleted old file: {filename}")
+
+
+def save_media_to_temp_cache(mime_type: str, binary_data: bytes) -> str:
+    cache_dir = CACHE_DIR
+    delete_old_files_in_cache(cache_dir)
+
+    file_extension = type_to_extension.get(mime_type, guess_extension(mime_type) or '.bin')
+    while True:
+        file_name = f"{uuid.uuid4()}{file_extension}"
+        file_path = os.path.join(cache_dir, file_name)
+        if not os.path.exists(file_path):
+            break
+
+    with open(file_path, "wb") as file:
+        file.write(binary_data)
+
+    return file_path