一个 XiaoqiangClub 自用的消息发送模块

These details have not been verified by PyPI

Project links

Project description

🚀 XQCSendMessage

XQCSendMessage Logo

XQCSendMessage 是一个自用的 Python 消息发送模块，旨在提供一个相对便捷的途径来通过多种渠道（如邮件、钉钉、企业微信等）发送通知或消息。无论您需要同步还是异步发送，XQCSendMessage 都尝试提供简洁、统一的 API。

📖 目录

✨ 主要特性
📤 安装
🚀 快速上手
- ⚙️ 客户端模式 (SyncClient / AsyncClient)
➡️ 直接函数调用模式
📄 许可证
🙏 支持我

✨ 主要特性

多渠道支持:
- 📧 邮件: 支持 SMTP，方便集成邮件服务。
- 🤖 钉钉: 支持 Webhook 机器人，包含签名验证。
- 🏢 企业微信:
  - Webhook 机器人模式。
  - 应用消息 模式，支持更多消息类型。
同步与异步: 提供了同步和异步两种发送模式，以适应不同场景的需求。
统一的 API 尝试: 客户端设计力求简洁，方便在不同发送渠道间切换。
错误处理: 内置了一些异常类，希望能帮助定位发送过程中的问题。
扩展性: 基于抽象基类的设计，方便根据需要添加自定义发送器。

📤 安装

通过 pip 或 poetry 安装 xqcsendmessage:

# 使用 pip
pip install xqcsendmessage

# 使用 poetry
poetry add xqcsendmessage

🚀 快速上手

XQCSendMessage 提供了两种调用模式：

实例化客户端模式：通过 SyncClient 或 AsyncClient 实例化一个客户端，在初始化时配置发送器，然后通过客户端的 send 方法发送消息。这种模式适合需要复用发送器配置，或者在发送过程中动态调整参数的场景。
直接函数调用模式：通过顶层函数直接发送消息，无需显式实例化客户端或发送器。这种模式适合快速发送单条消息，或者在脚本中进行一次性调用的场景。

⚙️ 客户端模式 (`SyncClient` / `AsyncClient`)

XQCSendMessage 的客户端设计力求直观。您只需要选择一个发送器类型，并在客户端初始化时传入配置。

⚙️ `SyncClient` / `AsyncClient` 初始化参数说明

sender_type (Literal["email", "dingtalk", "wecom_webhook", "wecom_app"]): 必须参数，指定要使用的发送器类型。
**config: 关键字参数，用于配置特定发送器。这些参数会传递给底层发送器的 __init__ 方法。同时，您也可以在此处设置消息的默认参数，这些默认参数可以在调用 .send() 方法时被覆盖。
- 邮件 (sender_type="email"):
  - smtp_server (str): SMTP 服务器地址。
  - smtp_port (int): SMTP 服务器端口。
  - sender_email (str): 发件人邮箱。
  - sender_password (str): 发件人邮箱密码或授权码。
  - use_tls (bool, default True): 是否使用 TLS 加密。
  - recipients (List[str], optional): 默认收件人列表。
  - subject (str, optional): 默认邮件主题。
  - subtype (str, default "plain"): 默认邮件内容类型 ("plain" 或 "html")。
- 钉钉 (sender_type="dingtalk"):
  - webhook (str): 钉钉机器人的 Webhook 地址。
  - secret (str, optional): 钉钉机器人的密钥，用于签名。
  - message (Union[str, Dict[str, Any]], optional): 默认消息内容或消息体。
  - at (Dict[str, Any], optional): 默认 @ 人参数，例如 {"isAtAll": True} 或 {"atMobiles": ["139xxxx0000"]}。
- 企业微信 Webhook (sender_type="wecom_webhook"):
  - webhook (str): 企业微信机器人的 Webhook 地址。
  - message (Union[str, Dict[str, Any]], optional): 默认消息内容或消息体。
  - touser (str, optional): 默认接收成员，例如 "@all" 或 "USERID1|USERID2"。
  - toparty (str, optional): 默认接收部门，例如 "PARTYID1|PARTYID2"。
  - totag (str, optional): 默认接收标签，例如 "TAGID1|TAGID2"。
- 企业微信应用 (sender_type="wecom_app"):
  - corpid (str): 企业 ID。
  - corpsecret (str): 应用的 Secret。
  - agentid (int): 应用的 AgentId。
  - message (Union[str, Dict[str, Any]], optional): 默认消息内容或消息体。
  - touser (str, optional): 默认接收成员，例如 "@all" 或 "USERID1|USERID2"。
  - toparty (str, optional): 默认接收部门，例如 "PARTYID1|PARTYID2"。
  - totag (str, optional): 默认接收标签，例如 "TAGID1|TAGID2"。
  - safe (int, optional): 表示是否是保密消息，0 表示否，1 表示是，默认 0。
  - enable_duplicate_check (int, optional): 表示是否开启重复消息检查，0 表示否，1 表示是，默认 0。
  - duplicate_check_interval (int, optional): 表示重复消息检查的时间间隔，单位秒，最大不超过 120。

⚙️ `client.send()` 方法参数说明

message (Optional[Union[str, Dict[str, Any]]], default None): 要发送的消息内容。
- 邮件: 如果是字符串，则作为邮件正文 content；如果是字典，则会引发 SendMessageError。
- 钉钉/企业微信: 如果是字符串，则自动封装为文本消息 {"msgtype": "text", "text": {"content": message}}；如果是字典，则作为完整的消息体发送。
  - msgtype 支持的类型包括：
    - text: 文本消息
    - markdown: Markdown 格式消息
    - image: 图片消息
    - news: 图文消息 (企业微信)
    - file: 文件消息
    - voice: 语音消息
    - video: 视频消息
    - card: 文本卡片消息 (企业微信)
    - interactive_task_card: 互动任务卡片消息 (企业微信)
    - template_card: 模板卡片消息 (企业微信)
    - oa: OA 消息 (钉钉)
    - link: 链接消息 (钉钉)
**kwargs: 关键字参数，用于覆盖客户端初始化时设置的默认消息参数，或者传递给底层发送器的额外参数。
- 邮件: 可以覆盖 subject, content, recipients, subtype。
- 钉钉: 可以覆盖 at 参数 ({"isAtAll": True} 或 {"atMobiles": ["139xxxx0000"]} 等)。
- 企业微信: 可以覆盖 touser, toparty, totag, safe, enable_duplicate_check, duplicate_check_interval 等。

📧 示例 1.1: 发送一封邮件 (同步)

from xqcsendmessage import SyncClient

# --- 配置 ---
SMTP_SERVER = "YOUR_SMTP_SERVER"  # ⚠️ 请替换为你的 SMTP 服务器
SMTP_PORT = 587                   # ⚠️ 请替换为你的 SMTP 端口
SENDER_EMAIL = "YOUR_SENDER_EMAIL"  # ⚠️ 请替换为你的邮箱
SENDER_PASSWORD = "YOUR_SENDER_PASSWORD"    # ⚠️ 请替换为你的邮箱密码或授权码
RECIPIENTS = ["YOUR_RECIPIENT_EMAIL"] # ⚠️ 请替换为收件人邮箱列表

def main_email_sync():
    """主函数"""
    # 1. 创建同步客户端，并在初始化时配置邮件发送器
    client = SyncClient(
        sender_type="email",
        smtp_server=SMTP_SERVER,
        smtp_port=SMTP_PORT,
        sender_email=SENDER_EMAIL,
        sender_password=SENDER_PASSWORD,
        use_tls=True, # 端口 465 通常直接使用 SSL，而不是通过 starttls 升级
        recipients=RECIPIENTS, # 可以在初始化时设置默认接收人
        subtype="plain" # 可以在初始化时设置默认邮件类型
    )

    # 2. 发送邮件
    try:
        result = client.send(
            content="这是一封通过 xqcsendmessage 发送的同步测试邮件。", # 邮件内容
            subject="Hello from xqcsendmessage!", # 邮件主题
            # recipients 和 subtype 可以不传，会使用初始化时的默认值
            # recipients=["override@example.com"], # 也可以在发送时覆盖
            # subtype="html"
        )
        print(f"✅ 邮件发送成功: {result}")
    except Exception as e:
        print(f"🔥 邮件发送失败: {e}")

if __name__ == "__main__":
    main_email_sync()

🤖 示例 1.2: 发送钉钉消息 (同步)

from xqcsendmessage import SyncClient

# --- 配置 ---
# 你的钉钉机器人的 Webhook 和密钥
WEBHOOK_URL = "YOUR_DINGTALK_WEBHOOK_URL"
SECRET = "YOUR_DINGTALK_SECRET"  # 如果没有设置密钥，则为 None

def main_dingtalk_sync():
    """主函数"""
    # 1. 创建同步客户端，并在初始化时配置钉钉发送器
    client = SyncClient(
        sender_type="dingtalk",
        webhook=WEBHOOK_URL,
        secret=SECRET,
        # 可以在初始化时设置默认消息内容和 @ 人参数，这些参数会在 send 方法中作为默认值，但可以被 send 方法中的 kwargs 覆盖
        message="【XQCSendMessage】这是一条来自 XQCSendMessage 的同步测试消息。",
        at={"isAtAll": True} # 默认发送给所有人
    )

    # 2. 发送消息
    try:
        # 发送字符串消息，此时会使用初始化时设置的默认 @ 人参数（发送给所有人）
        result_default_at = client.send(
            "【XQCSendMessage】这是一条来自 XQCSendMessage 的同步测试消息 (使用默认 @ 人)。"
        )
        print(f"✅ 钉钉消息发送成功 (默认 @ 人): {result_default_at}")

        # 发送字符串消息，并覆盖 @ 人参数（指定手机号）
        result_override_at = client.send(
            "【XQCSendMessage】这是一条来自 XQCSendMessage 的同步测试消息 (覆盖 @ 人，指定手机号)。",
            at={"atMobiles": ["13900139000"], "isAtAll": False}
        )
        print(f"✅ 钉钉消息发送成功 (覆盖 @ 人): {result_override_at}")

        # 发送字典消息，并覆盖 @ 人参数（指定手机号）
        result_dict_override_at = client.send(
            {
                "msgtype": "text",
                "text": {"content": "【XQCSendMessage】这是一条来自 XQCSendMessage 的同步测试消息 (字典，覆盖 @ 人，指定手机号)。"},
                "at": {"atMobiles": ["13700137000"], "isAtAll": False}
            }
        )
        print(f"✅ 钉钉消息发送成功 (字典，覆盖 @ 人): {result_dict_override_at}")

    except Exception as e:
        print(f"🔥 钉钉消息发送失败: {e}")

if __name__ == "__main__":
    main_dingtalk_sync()

🏢 示例 1.3: 发送企业微信 Webhook 消息 (同步)

from xqcsendmessage import SyncClient

# --- 配置 ---
# 你的企业微信机器人的 Webhook
WEBHOOK_URL = "YOUR_WECOM_WEBHOOK_URL"

def main_wecom_webhook_sync():
    """主函数"""
    # 1. 创建同步客户端，并在初始化时配置企业微信 Webhook 发送器
    client = SyncClient(
        sender_type="wecom_webhook",
        webhook=WEBHOOK_URL,
        # 可以在初始化时设置默认消息内容和参数，这些参数会在 send 方法中作为默认值，但可以被 send 方法中的 kwargs 覆盖
        message="【XQCSendMessage】这是一条来自 XQCSendMessage 的同步 Webhook 测试消息。",
        touser="@all" # 默认发送给所有人
    )

    # 2. 发送消息
    try:
        # 发送字符串消息，此时会使用初始化时设置的默认参数
        result_default = client.send(
            "【XQCSendMessage】这是一条来自 XQCSendMessage 的同步 Webhook 测试消息 (使用默认参数)。"
        )
        print(f"✅ 企业微信 Webhook 消息发送成功 (默认参数): {result_default}")

        # 发送字符串消息，并覆盖默认参数（例如，指定不同的内容）
        result_override_str = client.send(
            "【XQCSendMessage】这是一条来自 XQCSendMessage 的同步 Webhook 测试消息 (覆盖字符串内容，指定用户)。",
            touser="USERID" # 可以在这里覆盖 touser
        )
        print(f"✅ 企业微信 Webhook 消息发送成功 (覆盖字符串): {result_override_str}")

        # 发送字典消息，并覆盖默认参数
        result_override_dict = client.send(
            {
                "msgtype": "text",
                "text": {"content": "【XQCSendMessage】这是一条来自 XQCSendMessage 的同步 Webhook 测试消息 (覆盖字典内容，指定用户)。"}
            },
            touser="USERID" # 可以在这里覆盖 touser
        )
        print(f"✅ 企业微信 Webhook 消息发送成功 (覆盖字典): {result_override_dict}")

    except Exception as e:
        print(f"🔥 企业微信 Webhook 消息发送失败: {e}")

if __name__ == "__main__":
    main_wecom_webhook_sync()

🏢 示例 1.4: 发送企业微信应用消息 (同步)

from xqcsendmessage import SyncClient

# --- 配置 ---
# 你的企业微信应用配置
CORP_ID = "YOUR_CORP_ID"
CORP_SECRET = "YOUR_CORP_SECRET"
AGENT_ID = 1000002  # 你的应用 AgentId

def main_wecom_app_sync():
    """主函数"""
    # 1. 创建同步客户端，并在初始化时配置企业微信应用发送器
    client = SyncClient(
        sender_type="wecom_app",
        corpid=CORP_ID,
        corpsecret=CORP_SECRET,
        agentid=AGENT_ID,
        # 可以在初始化时设置默认消息内容和参数，这些参数会在 send 方法中作为默认值，但可以被 send 方法中的 kwargs 覆盖
        message="【XQCSendMessage】这是一条通过企业微信应用发送的同步测试消息。",
        touser="@all" # 默认发送给所有人
    )

    # 2. 发送消息
    try:
        # 发送字符串消息，此时会使用初始化时设置的默认参数
        result_default = client.send(
            "【XQCSendMessage】这是一条通过企业微信应用发送的同步测试消息 (使用默认参数)。"
        )
        print(f"✅ 企业微信应用消息发送成功 (默认参数): {result_default}")

        # 发送字符串消息，并覆盖默认参数（例如，指定不同的内容）
        result_override_str = client.send(
            "【XQCSendMessage】这是一条通过企业微信应用发送的同步测试消息 (覆盖字符串内容，指定用户)。",
            touser="USERID" # 可以在这里覆盖 touser
        )
        print(f"✅ 企业微信应用消息发送成功 (覆盖字符串): {result_override_str}")

        # 发送字典消息，并覆盖默认参数
        result_override_dict = client.send(
            {
                "msgtype": "text",
                "text": {"content": "【XQCSendMessage】这是一条通过企业微信应用发送的同步测试消息 (覆盖字典内容，指定用户)。"}
            },
            touser="USERID" # 可以在这里覆盖 touser
        )
        print(f"✅ 企业微信应用消息发送成功 (覆盖字典): {result_override_dict}")

    except Exception as e:
        print(f"🔥 企业微信应用消息发送失败: {e}")

if __name__ == "__main__":
    main_wecom_app_sync()

📧 示例 1.5: 发送邮件 (异步)

import asyncio
from xqcsendmessage import AsyncClient

# --- 配置 ---
SMTP_SERVER = "YOUR_SMTP_SERVER"  # ⚠️ 请替换为你的 SMTP 服务器
SMTP_PORT = 587                   # ⚠️ 请替换为你的 SMTP 端口
SENDER_EMAIL = "YOUR_SENDER_EMAIL"  # ⚠️ 请替换为你的邮箱
SENDER_PASSWORD = "YOUR_SENDER_PASSWORD"    # ⚠️ 请替换为你的邮箱密码或授权码
RECIPIENTS = ["YOUR_RECIPIENT_EMAIL"] # ⚠️ 请替换为收件人邮箱列表

async def main_email_async():
    """主函数"""
    # 1. 创建异步客户端，并在初始化时配置邮件发送器
    client = AsyncClient(
        sender_type="email",
        smtp_server=SMTP_SERVER,
        smtp_port=SMTP_PORT,
        sender_email=SENDER_EMAIL,
        sender_password=SENDER_PASSWORD,
        use_tls=True, # 端口 465 通常直接使用 SSL，而不是通过 starttls 升级
        recipients=RECIPIENTS, # 可以在初始化时设置默认接收人
        subtype="plain" # 可以在初始化时设置默认邮件类型
    )

    # 2. 发送邮件
    try:
        result = await client.send(
            content="这是一封通过 xqcsendmessage 发送的异步测试邮件。", # 邮件内容
            subject="Hello from xqcsendmessage!", # 邮件主题
            # recipients 和 subtype 可以不传，会使用初始化时的默认值
            # recipients=["override@example.com"], # 也可以在发送时覆盖
            # subtype="html"
        )
        print(f"✅ 邮件发送成功: {result}")
    except Exception as e:
        print(f"🔥 邮件发送失败: {e}")

if __name__ == "__main__":
    asyncio.run(main_email_async())

🤖 示例 1.6: 发送钉钉消息 (异步)

import asyncio
from xqcsendmessage import AsyncClient

# --- 配置 ---
# 你的钉钉机器人的 Webhook 和密钥
WEBHOOK_URL = "YOUR_DINGTALK_WEBHOOK_URL"
SECRET = "YOUR_DINGTALK_SECRET"  # 如果没有设置密钥，则为 None

async def main_dingtalk_async():
    """主函数"""
    # 1. 创建异步客户端，并在初始化时配置钉钉发送器
    client = AsyncClient(
        sender_type="dingtalk",
        webhook=WEBHOOK_URL,
        secret=SECRET,
        # 可以在初始化时设置默认消息内容和 @ 人参数，这些参数会在 send 方法中作为默认值，但可以被 send 方法中的 kwargs 覆盖
        message="【XQCSendMessage】这是一条来自 XQCSendMessage 的异步测试消息。",
        at={"isAtAll": True} # 默认发送给所有人
    )

    # 2. 发送消息
    try:
        # 发送字符串消息，此时会使用初始化时设置的默认 @ 人参数（发送给所有人）
        result_default_at = await client.send(
            "【XQCSendMessage】这是一条来自 XQCSendMessage 的异步测试消息 (使用默认 @ 人)。"
        )
        print(f"✅ 钉钉消息发送成功 (默认 @ 人): {result_default_at}")

        # 发送字符串消息，并覆盖 @ 人参数（指定手机号）
        result_override_at = await client.send(
            "【XQCSendMessage】这是一条来自 XQCSendMessage 的异步测试消息 (覆盖 @ 人，指定手机号)。",
            at={"atMobiles": ["13900139000"], "isAtAll": False}
        )
        print(f"✅ 钉钉消息发送成功 (覆盖 @ 人): {result_override_at}")

        # 发送字典消息，并覆盖 @ 人参数（指定手机号）
        result_dict_override_at = await client.send(
            {
                "msgtype": "text",
                "text": {"content": "【XQCSendMessage】这是一条来自 XQCSendMessage 的异步测试消息 (字典，覆盖 @ 人，指定手机号)。"},
                "at": {"atMobiles": ["13700137000"], "isAtAll": False}
            }
        )
        print(f"✅ 钉钉消息发送成功 (字典，覆盖 @ 人): {result_dict_override_at}")

    except Exception as e:
        print(f"🔥 钉钉消息发送失败: {e}")

if __name__ == "__main__":
    asyncio.run(main_dingtalk_async())

🏢 示例 1.7: 发送企业微信 Webhook 消息 (异步)

import asyncio
from xqcsendmessage import AsyncClient

# --- 配置 ---
# 你的企业微信机器人的 Webhook
WEBHOOK_URL = "YOUR_WECOM_WEBHOOK_URL"

async def main_wecom_webhook_async():
    """主函数"""
    # 1. 创建异步客户端，并在初始化时配置企业微信 Webhook 发送器
    client = AsyncClient(
        sender_type="wecom_webhook",
        webhook=WEBHOOK_URL,
        # 可以在初始化时设置默认消息内容和参数，这些参数会在 send 方法中作为默认值，但可以被 send 方法中的 kwargs 覆盖
        message="【XQCSendMessage】这是一条来自 XQCSendMessage 的异步 Webhook 测试消息。",
        touser="@all" # 默认发送给所有人
    )

    # 2. 发送消息
    try:
        # 发送字符串消息，此时会使用初始化时设置的默认参数
        result_default = await client.send(
            "【XQCSendMessage】这是一条来自 XQCSendMessage 的异步 Webhook 测试消息 (使用默认参数)。"
        )
        print(f"✅ 企业微信 Webhook 消息发送成功 (默认参数): {result_default}")

        # 发送字符串消息，并覆盖默认参数（例如，指定不同的内容）
        result_override_str = await client.send(
            "【XQCSendMessage】这是一条来自 XQCSendMessage 的异步 Webhook 测试消息 (覆盖字符串内容，指定用户)。",
            touser="USERID" # 可以在这里覆盖 touser
        )
        print(f"✅ 企业微信 Webhook 消息发送成功 (覆盖字符串): {result_override_str}")

        # 发送字典消息，并覆盖默认参数
        result_override_dict = await client.send(
            {
                "msgtype": "text",
                "text": {"content": "【XQCSendMessage】这是一条来自 XQCSendMessage 的异步 Webhook 测试消息 (覆盖字典内容，指定用户)。"}
            },
            touser="USERID" # 可以在这里覆盖 touser
        )
        print(f"✅ 企业微信 Webhook 消息发送成功 (覆盖字典): {result_override_dict}")

    except Exception as e:
        print(f"🔥 企业微信 Webhook 消息发送失败: {e}")

if __name__ == "__main__":
    asyncio.run(main_wecom_webhook_async())

🏢 示例 1.8: 发送企业微信应用消息 (异步)

import asyncio
from xqcsendmessage import AsyncClient

# --- 配置 ---
# 你的企业微信应用配置
CORP_ID = "YOUR_CORP_ID"
CORP_SECRET = "YOUR_CORP_SECRET"
AGENT_ID = 1000002  # 你的应用 AgentId

async def main_wecom_app_async():
    """主函数"""
    # 1. 创建异步客户端，并在初始化时配置企业微信应用发送器
    client = AsyncClient(
        sender_type="wecom_app",
        corpid=CORP_ID,
        corpsecret=CORP_SECRET,
        agentid=AGENT_ID,
        # 可以在初始化时设置默认消息内容和参数，这些参数会在 send 方法中作为默认值，但可以被 send 方法中的 kwargs 覆盖
        message="【XQCSendMessage】这是一条通过企业微信应用发送的异步测试消息。",
        touser="@all" # 默认发送给所有人
    )

    # 2. 发送消息
    try:
        # 发送字符串消息，此时会使用初始化时设置的默认参数
        result_default = await client.send(
            "【XQCSendMessage】这是一条通过企业微信应用发送的异步测试消息 (使用默认参数)。"
        )
        print(f"✅ 企业微信应用消息发送成功 (默认参数): {result_default}")

        # 发送字符串消息，并覆盖默认参数（例如，指定不同的内容）
        result_override_str = await client.send(
            "【XQCSendMessage】这是一条通过企业微信应用发送的异步测试消息 (覆盖字符串内容，指定用户)。",
            touser="USERID" # 可以在这里覆盖 touser
        )
        print(f"✅ 企业微信应用消息发送成功 (覆盖字符串): {result_override_str}")

        # 发送字典消息，并覆盖默认参数
        result_override_dict = await client.send(
            {
                "msgtype": "text",
                "text": {"content": "【XQCSendMessage】这是一条通过企业微信应用发送的异步测试消息 (覆盖字典内容，指定用户)。"}
            },
            touser="USERID" # 可以在这里覆盖 touser
        )
        print(f"✅ 企业微信应用消息发送成功 (覆盖字典): {result_override_dict}")

    except Exception as e:
        print(f"🔥 企业微信应用消息发送失败: {e}")

if __name__ == "__main__":
    asyncio.run(main_wecom_app_async())

➡️ 直接函数调用模式

xqcsendmessage 提供了便捷的顶层函数，可以直接调用以发送消息，无需显式实例化客户端或发送器。

⚙️ 顶层发送函数参数说明

这些函数将消息内容作为第一个参数，其余参数直接作为关键字参数传入。

send_email / send_email_async:
- message (Union[str, Dict[str, Any]]): 邮件内容。如果为字符串，则作为 content；如果为字典，则可以包含 subject, content, recipients, subtype。
- smtp_server (str): SMTP 服务器地址。
- smtp_port (int): SMTP 服务器端口。
- sender_email (str): 发件人邮箱。
- sender_password (str): 发件人邮箱密码或授权码。
- recipients (List[str]): 收件人列表。
- subject (str, optional): 邮件主题，如果 message 为字符串，则必须提供。
- content (str, optional): 邮件内容，如果 message 为字典且未包含 content，则必须提供。
- subtype (str, default "plain"): 邮件内容类型，"plain" 或 "html"。
- use_tls (bool, default True): 是否使用 TLS 加密。
- **kwargs: 其他可选参数，将传递给底层的 EmailSender 或 AsyncEmailSender。
send_dingtalk / send_dingtalk_async:
- message (Union[str, Dict[str, Any]]): 消息内容，可以是字符串（将作为文本消息发送）或符合钉钉机器人格式的字典。
- webhook (str): 钉钉机器人的 Webhook 地址。
- secret (str, optional): 钉钉机器人的密钥，用于签名。
- **kwargs: 其他可选参数，例如 is_at_all (bool), at_mobiles (List[str]) 等，会合并到消息字典中。
send_wecom_webhook / send_wecom_webhook_async:
- message (Union[str, Dict[str, Any]]): 消息内容，可以是字符串（将作为文本消息发送）或符合企业微信机器人支持的格式。
- webhook (str): 企业微信机器人的 Webhook 地址。
- **kwargs: 其他可选参数，例如 touser, toparty, totag 等，会合并到消息字典中。
send_wecom_app / send_wecom_app_async:
- message (Union[str, Dict[str, Any]]): 消息内容，可以是字符串（将作为文本消息发送）或符合企业微信应用消息支持的格式。
- corpid (str): 企业 ID。
- corpsecret (str): 应用的 Secret。
- agentid (int): 应用的 AgentId。
- **kwargs: 其他可选参数，例如 touser, toparty, totag, safe, enable_duplicate_check, duplicate_check_interval 等，会合并到消息字典中。

📧 示例 2.1: 直接发送一封邮件 (同步)

from xqcsendmessage import send_email

# --- 配置 ---
SMTP_SERVER = "YOUR_SMTP_SERVER"  # ⚠️ 请替换为你的 SMTP 服务器
SMTP_PORT = 587                   # ⚠️ 请替换为你的 SMTP 端口
SENDER_EMAIL = "YOUR_SENDER_EMAIL"  # ⚠️ 请替换为你的邮箱
SENDER_PASSWORD = "YOUR_SENDER_PASSWORD"    # ⚠️ 请替换为你的邮箱密码或授权码
RECIPIENTS = ["YOUR_RECIPIENT_EMAIL"] # ⚠️ 请替换为收件人邮箱列表

# 直接发送邮件
try:
    result = send_email(
        content="这是一封通过 xqcsendmessage 直接发送的同步测试邮件。",
        smtp_server=SMTP_SERVER,
        smtp_port=SMTP_PORT,
        sender_email=SENDER_EMAIL,
        sender_password=SENDER_PASSWORD,
        recipients=RECIPIENTS,
        subject="Hello from xqcsendmessage! (Direct Call)",
        subtype="plain"
    )
    print(f"✅ 直接调用邮件发送成功: {result}")
except Exception as e:
    print(f"🔥 直接调用邮件发送失败: {e}")

🤖 示例 2.2: 直接发送钉钉消息 (异步)

import asyncio
from xqcsendmessage import send_dingtalk_async

# --- 配置 ---
# 你的钉钉机器人的 Webhook 和密钥
WEBHOOK_URL = "YOUR_DINGTALK_WEBHOOK_URL"
SECRET = "YOUR_DINGTALK_SECRET"

async def main_direct_dingtalk():
    # 直接发送钉钉消息
    try:
        # 发送字符串消息，at 相关的参数通过 kwargs 传递
        result = await send_dingtalk_async(
            "【XQCSendMessage】这是一条通过 XQCSendMessage 直接发送的异步钉钉测试消息 (发送给所有人)。",
            webhook=WEBHOOK_URL,
            secret=SECRET,
            is_at_all=True # 发送给所有人
        )
        print(f"✅ 钉钉消息发送成功 (发送给所有人): {result}")

        # 发送字典消息，并覆盖 @ 人参数（指定手机号）
        result_dict_override_at = await send_dingtalk_async(
            {
                "msgtype": "text",
                "text": {"content": "【XQCSendMessage】这是一条通过 XQCSendMessage 直接发送的异步钉钉测试消息 (字典，指定手机号)。"},
                "at": {"atMobiles": ["13700137000"], "isAtAll": False}
            },
            webhook=WEBHOOK_URL,
            secret=SECRET
        )
        print(f"✅ 钉钉消息发送成功 (字典，指定手机号): {result_dict_override_at}")

    except Exception as e:
        print(f"🔥 直接调用钉钉发送失败: {e}")

if __name__ == "__main__":
    asyncio.run(main_direct_dingtalk())

🏢 示例 2.3: 直接发送企业微信 Webhook 消息 (同步)

from xqcsendmessage import send_wecom_webhook

# --- 配置 ---
# 企业微信机器人的 Webhook
WEBHOOK_URL = "YOUR_WECOM_WEBHOOK_URL"

# 直接发送企业微信 Webhook 消息
try:
    # 发送字符串消息
    result_direct_str = send_wecom_webhook(
        "【XQCSendMessage】这是一条通过 XQCSendMessage 直接发送的同步 Webhook 测试消息 (字符串)。",
        webhook=WEBHOOK_URL
    )
    print(f"✅ 直接调用企业微信 Webhook 消息发送成功 (字符串): {result_direct_str}")

    # 发送字典消息，并覆盖 touser 参数
    result_direct_dict_override = send_wecom_webhook(
        {
            "msgtype": "text",
            "text": {"content": "【XQCSendMessage】这是一条通过 XQCSendMessage 直接发送的同步 Webhook 测试消息 (字典，指定用户)。"}
        },
        webhook=WEBHOOK_URL,
        touser="USERID" # 可以在这里覆盖 touser
    )
    print(f"✅ 直接调用企业微信 Webhook 消息发送成功 (字典，指定用户): {result_direct_dict_override}")

except Exception as e:
    print(f"🔥 企业微信 Webhook 消息发送失败: {e}")

🏢 示例 2.4: 直接发送企业微信应用消息 (异步)

import asyncio
from xqcsendmessage import send_wecom_app_async

# --- 配置 ---
# 企业微信应用配置
CORP_ID = "YOUR_CORP_ID"
CORP_SECRET = "YOUR_CORP_SECRET"
AGENT_ID = 1000002  # 你的应用 AgentId

async def main_direct_wecom_app_async():
    # 直接发送企业微信应用消息
    try:
        # 发送字符串消息
        result_direct_str = await send_wecom_app_async(
            "【XQCSendMessage】这是一条通过企业微信应用直接发送的异步测试消息 (字符串)。",
            corpid=CORP_ID,
            corpsecret=CORP_SECRET,
            agentid=AGENT_ID,
            touser="@all"
        )
        print(f"✅ 直接调用企业微信应用消息发送成功 (字符串): {result_direct_str}")

        # 发送字典消息，并覆盖 touser 参数
        result_direct_dict_override = await send_wecom_app_async(
            {
                "msgtype": "text",
                "text": {"content": "【XQCSendMessage】这是一条通过企业微信应用直接发送的异步测试消息 (字典，指定用户)。"},
            },
            corpid=CORP_ID,
            corpsecret=CORP_SECRET,
            agentid=AGENT_ID,
            touser="USERID" # 可以在这里覆盖 touser
        )
        print(f"✅ 直接调用企业微信应用消息发送成功 (字典): {result_direct_dict_override}")

    except Exception as e:
        print(f"🔥 直接调用企业微信应用消息发送失败: {e}")

if __name__ == "__main__":
    asyncio.run(main_direct_wecom_app_async())

📄 许可证

本项目基于 MIT License 开源。

🙏 支持我

如果您觉得 XQCSendMessage 对您有帮助，可以通过以下方式支持我：

Project details

These details have not been verified by PyPI

Project links

Release history Release notifications | RSS feed

0.1.1

Nov 14, 2025

0.1.0

Nov 13, 2025

0.0.3

Nov 12, 2025

0.0.2

Nov 12, 2025

This version

0.0.1

Nov 12, 2025

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

xqcsendmessage-0.0.1.tar.gz (20.9 kB view details)

Uploaded Nov 12, 2025 Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

The dropdown lists show the available interpreters, ABIs, and platforms. Enable javascript to be able to filter the list of wheel files.

xqcsendmessage-0.0.1-py3-none-any.whl (22.7 kB view details)

Uploaded Nov 12, 2025 Python 3

File details

Details for the file xqcsendmessage-0.0.1.tar.gz.

File metadata

Download URL: xqcsendmessage-0.0.1.tar.gz
Upload date: Nov 12, 2025
Size: 20.9 kB
Tags: Source
Uploaded using Trusted Publishing? No
Uploaded via: poetry/2.1.3 CPython/3.13.3 Windows/11

File hashes

Hashes for xqcsendmessage-0.0.1.tar.gz
Algorithm	Hash digest
SHA256	`2bb1458075e3a55f21e590905ca326d5c92be3873ca1b8a31df8fe4d093f8842`
MD5	`592b7f7c3d58f055d97d7aa62036b3dc`
BLAKE2b-256	`094429d70f8a4545e562005097a5c90402800c01841b9b4aa707e20f95a3a899`

See more details on using hashes here.

File details

Details for the file xqcsendmessage-0.0.1-py3-none-any.whl.

File metadata

Download URL: xqcsendmessage-0.0.1-py3-none-any.whl
Upload date: Nov 12, 2025
Size: 22.7 kB
Tags: Python 3
Uploaded using Trusted Publishing? No
Uploaded via: poetry/2.1.3 CPython/3.13.3 Windows/11

File hashes

Hashes for xqcsendmessage-0.0.1-py3-none-any.whl
Algorithm	Hash digest
SHA256	`569d5a04d423572a9fe01a4cc4a1cd40968f5b6d35dff02dadd703c49dacf3fa`
MD5	`2c5501f9d403c2aacbce35f263d3a02d`
BLAKE2b-256	`e0c1faa6d65b5876adc053215e5a178f1f140615d42b70fde9142f058cbdfbe5`

See more details on using hashes here.

xqcsendmessage 0.0.1

Navigation

Verified details

Maintainers

Unverified details

Project links

Meta

Classifiers

Project description

🚀 XQCSendMessage

📖 目录

✨ 主要特性

📤 安装

🚀 快速上手

⚙️ 客户端模式 (SyncClient / AsyncClient)

⚙️ SyncClient / AsyncClient 初始化参数说明

⚙️ client.send() 方法参数说明

📧 示例 1.1: 发送一封邮件 (同步)

🤖 示例 1.2: 发送钉钉消息 (同步)

🏢 示例 1.3: 发送企业微信 Webhook 消息 (同步)

🏢 示例 1.4: 发送企业微信应用消息 (同步)

📧 示例 1.5: 发送邮件 (异步)

🤖 示例 1.6: 发送钉钉消息 (异步)

🏢 示例 1.7: 发送企业微信 Webhook 消息 (异步)

🏢 示例 1.8: 发送企业微信应用消息 (异步)

➡️ 直接函数调用模式

⚙️ 顶层发送函数参数说明

📧 示例 2.1: 直接发送一封邮件 (同步)

🤖 示例 2.2: 直接发送钉钉消息 (异步)

🏢 示例 2.3: 直接发送企业微信 Webhook 消息 (同步)

🏢 示例 2.4: 直接发送企业微信应用消息 (异步)

📄 许可证

🙏 支持我

Project details

Verified details

Maintainers

Unverified details

Project links

Meta

Classifiers

Release history Release notifications | RSS feed

Download files

Source Distribution

Built Distribution

File details

File metadata

File hashes

File details

File metadata

File hashes

⚙️ 客户端模式 (`SyncClient` / `AsyncClient`)

⚙️ `SyncClient` / `AsyncClient` 初始化参数说明

⚙️ `client.send()` 方法参数说明