community.general/plugins/modules/telegram.py

#!/usr/bin/python

# Copyright (c) 2016, Artem Feofanov <artem.feofanov@gmail.com>
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later

from __future__ import annotations


DOCUMENTATION = r"""
module: telegram
author:
  - "Artem Feofanov (@tyouxa)"
  - "Nikolai Lomov (@lomserman)"

short_description: Send notifications using telegram

description:
  - Send notifications using telegram bot, to a verified group or user.
  - Also, the user may try to use any other telegram bot API method, if you specify O(api_method) argument.
notes:
  - You need a telegram account and create telegram bot to use this module.
extends_documentation_fragment:
  - community.general.attributes
attributes:
  check_mode:
    support: full
  diff_mode:
    support: none
options:
  token:
    type: str
    description:
      - Token identifying your telegram bot.
    required: true
  api_method:
    type: str
    description:
      - Bot API method.
      - For reference, see U(https://core.telegram.org/bots/api).
    default: SendMessage
    version_added: 2.0.0
  api_args:
    type: dict
    description:
      - Any parameters for the method.
      - For reference to default method, V(SendMessage), see U(https://core.telegram.org/bots/api#sendmessage).
    version_added: 2.0.0
"""

EXAMPLES = r"""
- name: Send notify to Telegram
  community.general.telegram:
    token: '9999999:XXXXXXXXXXXXXXXXXXXXXXX'
    api_args:
      chat_id: "000000"
      parse_mode: "markdown"
      text: "Your precious application has been deployed: https://example.com"
      disable_web_page_preview: true
      disable_notification: true

- name: Forward message to someone
  community.general.telegram:
    token: '9999999:XXXXXXXXXXXXXXXXXXXXXXX'
    api_method: forwardMessage
    api_args:
      chat_id: "000000"
      from_chat_id: 111111
      disable_notification: true
      message_id: '{{ saved_msg_id }}'
"""

RETURN = r"""
msg:
  description: The message you attempted to send.
  returned: success
  type: str
  sample: "Ansible task finished"
telegram_error:
  description: Error message gotten from Telegram API.
  returned: failure
  type: str
  sample: "Bad Request: message text is empty"
"""

import json
from urllib.parse import quote

from ansible.module_utils.basic import AnsibleModule

# noinspection PyUnresolvedReferences
from ansible.module_utils.urls import fetch_url


def main():
    module = AnsibleModule(
        argument_spec=dict(
            token=dict(type="str", required=True, no_log=True),
            api_args=dict(type="dict"),
            api_method=dict(type="str", default="SendMessage"),
        ),
        supports_check_mode=True,
    )

    token = quote(module.params.get("token"))
    api_args = module.params.get("api_args") or {}
    api_method = module.params.get("api_method")
    # filling backward compatibility args
    api_args["chat_id"] = api_args.get("chat_id")
    api_args["parse_mode"] = api_args.get("parse_mode")
    api_args["text"] = api_args.get("text")

    if api_args["parse_mode"] == "plain":
        del api_args["parse_mode"]

    url = f"https://api.telegram.org/bot{token}/{api_method}"

    if module.check_mode:
        module.exit_json(changed=False)

    response, info = fetch_url(
        module, url, method="POST", data=json.dumps(api_args), headers={"Content-Type": "application/json"}
    )
    if info["status"] == 200:
        module.exit_json(changed=True)
    elif info["status"] == -1:
        # SSL errors, connection problems, etc.
        module.fail_json(msg="Failed to send message", info=info, response=response)
    else:
        body = json.loads(info["body"])
        module.fail_json(
            msg=f"Failed to send message, return status = {info['status']}\nurl = {url}\napi_args = {api_args}",
            telegram_error=body["description"],
        )


if __name__ == "__main__":
    main()