Webhooks

class Webhook:
    """
    Asynchronous Discord webhook client.
    """
    def __init__(self, url: str, *, session: Optional[aiohttp.ClientSession] = None): ...
    
    # Webhook properties
    id: int  # Webhook ID
    type: WebhookType  # Webhook type
    guild_id: Optional[int]  # Guild ID if guild webhook
    channel_id: Optional[int]  # Channel ID if channel webhook
    user: Optional[User]  # User who created webhook
    name: Optional[str]  # Webhook name
    avatar: Optional[str]  # Webhook avatar hash
    token: Optional[str]  # Webhook token
    application_id: Optional[int]  # Application ID for application webhooks
    source_guild: Optional[PartialWebhookGuild]  # Source guild for follower webhooks
    source_channel: Optional[PartialWebhookChannel]  # Source channel for follower webhooks
    url: str  # Webhook URL
    
    # Class methods for creation
    @classmethod
    def from_url(cls, url: str, *, session: Optional[aiohttp.ClientSession] = None) -> Webhook:
        """Create webhook from URL."""
    
    @classmethod
    async def from_state(
        cls,
        data: Dict[str, Any],
        *,
        session: Optional[aiohttp.ClientSession] = None
    ) -> Webhook:
        """Create webhook from Discord API data."""
    
    # Message sending
    async def send(
        self,
        content: Optional[str] = None,
        *,
        username: Optional[str] = None,
        avatar_url: Optional[str] = None,
        tts: bool = False,
        ephemeral: bool = False,
        file: Optional[File] = None,
        files: Optional[List[File]] = None,
        embed: Optional[Embed] = None,
        embeds: Optional[List[Embed]] = None,
        allowed_mentions: Optional[AllowedMentions] = None,
        thread: Optional[Snowflake] = None,
        thread_name: Optional[str] = None,
        wait: bool = False,
        suppress_embeds: bool = False
    ) -> Optional[WebhookMessage]:
        """
        Send a message via webhook.
        
        Parameters:
        - content: Message content
        - username: Override webhook username
        - avatar_url: Override webhook avatar
        - tts: Whether message is text-to-speech
        - ephemeral: Whether message is ephemeral (interaction webhooks only)
        - file: Single file to upload
        - files: Multiple files to upload
        - embed: Single embed to send
        - embeds: Multiple embeds to send
        - allowed_mentions: Mention settings
        - thread: Thread to send message in
        - thread_name: Create new thread with this name
        - wait: Whether to wait for message confirmation
        - suppress_embeds: Whether to suppress link embeds
        
        Returns:
        WebhookMessage if wait=True, None otherwise
        """
    
    # Message management
    async def fetch_message(self, id: int, *, thread: Optional[Snowflake] = None) -> WebhookMessage:
        """
        Fetch a webhook message by ID.
        
        Parameters:
        - id: Message ID to fetch
        - thread: Thread containing the message
        
        Returns:
        WebhookMessage object
        """
    
    async def edit_message(
        self,
        message_id: int,
        *,
        content: Optional[str] = None,
        embed: Optional[Embed] = None,
        embeds: Optional[List[Embed]] = None,
        file: Optional[File] = None,
        files: Optional[List[File]] = None,
        attachments: Optional[List[Attachment]] = None,
        allowed_mentions: Optional[AllowedMentions] = None,
        thread: Optional[Snowflake] = None
    ) -> WebhookMessage:
        """
        Edit a webhook message.
        
        Parameters:
        - message_id: ID of message to edit
        - content: New message content
        - embed: New embed
        - embeds: New embeds
        - file: New file to upload
        - files: New files to upload
        - attachments: Attachments to keep
        - allowed_mentions: Mention settings
        - thread: Thread containing the message
        
        Returns:
        Edited WebhookMessage
        """
    
    async def delete_message(self, message_id: int, *, thread: Optional[Snowflake] = None) -> None:
        """
        Delete a webhook message.
        
        Parameters:
        - message_id: ID of message to delete
        - thread: Thread containing the message
        """
    
    # Webhook management
    async def edit(
        self,
        *,
        name: Optional[str] = None,
        avatar: Optional[bytes] = None,
        channel: Optional[TextChannel] = None,
        reason: Optional[str] = None
    ) -> Webhook:
        """
        Edit webhook properties.
        
        Parameters:
        - name: New webhook name
        - avatar: New webhook avatar
        - channel: New webhook channel
        - reason: Reason for audit log
        
        Returns:
        Updated Webhook object
        """
    
    async def delete(self, *, reason: Optional[str] = None) -> None:
        """
        Delete the webhook.
        
        Parameters:
        - reason: Reason for audit log
        """
    
    # Thread management  
    async def create_thread(
        self,
        *,
        name: str,
        message: Optional[WebhookMessage] = None,
        auto_archive_duration: int = 1440,
        rate_limit_per_user: Optional[int] = None,
        reason: Optional[str] = None
    ) -> Thread:
        """
        Create a thread via webhook.
        
        Parameters:
        - name: Thread name
        - message: Message to start thread from
        - auto_archive_duration: Auto-archive duration in minutes
        - rate_limit_per_user: Slowmode delay in seconds
        - reason: Reason for audit log
        
        Returns:
        Created Thread object
        """

class WebhookMessage:
    """
    Represents a message sent by a webhook.
    """
    def __init__(self, *, data: Dict[str, Any], state, webhook: Webhook): ...
    
    # Message properties (similar to regular Message)
    id: int  # Message ID
    webhook_id: int  # Webhook ID
    channel: PartialMessageable  # Channel message was sent in
    guild: Optional[Guild]  # Guild if sent in guild channel
    content: str  # Message content
    clean_content: str  # Content with mentions resolved
    created_at: datetime  # Message creation timestamp
    edited_at: Optional[datetime]  # Last edit timestamp
    tts: bool  # Whether message is text-to-speech
    mention_everyone: bool  # Whether message mentions @everyone
    mentions: List[User]  # Mentioned users
    channel_mentions: List[GuildChannel]  # Mentioned channels
    role_mentions: List[Role]  # Mentioned roles
    attachments: List[Attachment]  # File attachments
    embeds: List[Embed]  # Rich embeds
    reactions: List[Reaction]  # Message reactions
    pinned: bool  # Whether message is pinned
    type: MessageType  # Message type
    flags: MessageFlags  # Message flags
    thread: Optional[Thread]  # Thread if message is in thread
    components: List[Component]  # UI components
    
    # Message management
    async def edit(
        self,
        *,
        content: Optional[str] = None,
        embed: Optional[Embed] = None,
        embeds: Optional[List[Embed]] = None,
        file: Optional[File] = None,
        files: Optional[List[File]] = None,
        attachments: Optional[List[Attachment]] = None,
        allowed_mentions: Optional[AllowedMentions] = None
    ) -> WebhookMessage:
        """Edit the webhook message."""
    
    async def delete(self, *, delay: Optional[float] = None) -> None:
        """Delete the webhook message."""
    
    # Message interactions
    async def add_reaction(self, emoji: Union[Emoji, Reaction, PartialEmoji, str]) -> None:
        """Add a reaction to the message."""
    
    async def remove_reaction(self, emoji: Union[Emoji, Reaction, PartialEmoji, str], member: Member) -> None:
        """Remove a reaction from the message."""
    
    async def clear_reactions(self) -> None:
        """Clear all reactions from the message."""
    
    async def pin(self, *, reason: Optional[str] = None) -> None:
        """Pin the message."""
    
    async def unpin(self, *, reason: Optional[str] = None) -> None:
        """Unpin the message."""
    
    def to_reference(self) -> MessageReference:
        """Create a reference to this message for replies."""

class PartialWebhookGuild:
    """
    Partial guild information from webhook data.
    """
    def __init__(self, data: Dict[str, Any]): ...
    
    id: int  # Guild ID
    name: str  # Guild name
    icon: Optional[str]  # Guild icon hash

class PartialWebhookChannel:
    """
    Partial channel information from webhook data.
    """
    def __init__(self, data: Dict[str, Any]): ...
    
    id: int  # Channel ID
    name: str  # Channel name
    type: ChannelType  # Channel type

class SyncWebhook:
    """
    Synchronous Discord webhook client.
    """
    def __init__(self, url: str, *, session: Optional[requests.Session] = None): ...
    
    # Same properties as async Webhook
    id: int
    type: WebhookType
    guild_id: Optional[int]
    channel_id: Optional[int]
    user: Optional[User]
    name: Optional[str]
    avatar: Optional[str]
    token: Optional[str]
    application_id: Optional[int]
    source_guild: Optional[PartialWebhookGuild]
    source_channel: Optional[PartialWebhookChannel]
    url: str
    
    # Class methods
    @classmethod
    def from_url(cls, url: str, *, session: Optional[requests.Session] = None) -> SyncWebhook:
        """Create sync webhook from URL."""
    
    # Message sending (synchronous versions)
    def send(
        self,
        content: Optional[str] = None,
        *,
        username: Optional[str] = None,
        avatar_url: Optional[str] = None,
        tts: bool = False,
        file: Optional[File] = None,
        files: Optional[List[File]] = None,
        embed: Optional[Embed] = None,
        embeds: Optional[List[Embed]] = None,
        allowed_mentions: Optional[AllowedMentions] = None,
        thread: Optional[Snowflake] = None,
        wait: bool = False
    ) -> Optional[SyncWebhookMessage]:
        """Send a message via webhook (synchronous)."""
    
    def fetch_message(self, id: int, *, thread: Optional[Snowflake] = None) -> SyncWebhookMessage:
        """Fetch a webhook message by ID (synchronous)."""
    
    def edit_message(
        self,
        message_id: int,
        *,
        content: Optional[str] = None,
        embed: Optional[Embed] = None,
        embeds: Optional[List[Embed]] = None,
        file: Optional[File] = None,
        files: Optional[List[File]] = None,
        attachments: Optional[List[Attachment]] = None,
        allowed_mentions: Optional[AllowedMentions] = None,
        thread: Optional[Snowflake] = None
    ) -> SyncWebhookMessage:
        """Edit a webhook message (synchronous)."""
    
    def delete_message(self, message_id: int, *, thread: Optional[Snowflake] = None) -> None:
        """Delete a webhook message (synchronous)."""
    
    def edit(
        self,
        *,
        name: Optional[str] = None,
        avatar: Optional[bytes] = None,
        reason: Optional[str] = None
    ) -> SyncWebhook:
        """Edit webhook properties (synchronous)."""
    
    def delete(self, *, reason: Optional[str] = None) -> None:
        """Delete the webhook (synchronous)."""

class SyncWebhookMessage:
    """
    Synchronous version of WebhookMessage.
    Same properties and methods as WebhookMessage but synchronous.
    """
    def __init__(self, *, data: Dict[str, Any], state, webhook: SyncWebhook): ...
    
    # Same properties as WebhookMessage
    id: int
    webhook_id: int
    channel: PartialMessageable
    guild: Optional[Guild]
    content: str
    clean_content: str
    created_at: datetime
    edited_at: Optional[datetime]
    attachments: List[Attachment]
    embeds: List[Embed]
    # ... (all other properties)
    
    # Synchronous methods
    def edit(self, **kwargs) -> SyncWebhookMessage:
        """Edit the webhook message (synchronous)."""
    
    def delete(self, *, delay: Optional[float] = None) -> None:
        """Delete the webhook message (synchronous)."""
    
    def add_reaction(self, emoji: Union[Emoji, Reaction, PartialEmoji, str]) -> None:
        """Add a reaction to the message (synchronous)."""
    
    def remove_reaction(self, emoji: Union[Emoji, Reaction, PartialEmoji, str], member: Member) -> None:
        """Remove a reaction from the message (synchronous)."""
    
    def clear_reactions(self) -> None:
        """Clear all reactions from the message (synchronous)."""
    
    def pin(self, *, reason: Optional[str] = None) -> None:
        """Pin the message (synchronous)."""
    
    def unpin(self, *, reason: Optional[str] = None) -> None:
        """Unpin the message (synchronous)."""

class WebhookType(Enum):
    """Webhook type enumeration."""
    incoming = 1  # Incoming webhook
    channel_follower = 2  # Channel follower webhook
    application = 3  # Application webhook (for slash commands)

import discord
import asyncio
import aiohttp

async def send_webhook_message():
    """Send a message via webhook."""
    webhook_url = "https://discord.com/api/webhooks/YOUR_WEBHOOK_URL"
    
    async with aiohttp.ClientSession() as session:
        webhook = discord.Webhook.from_url(webhook_url, session=session)
        
        # Send simple message
        await webhook.send("Hello from webhook!")
        
        # Send message with custom username and avatar
        await webhook.send(
            "Custom webhook message!",
            username="Custom Bot",
            avatar_url="https://example.com/avatar.png"
        )
        
        # Send message with embed
        embed = discord.Embed(
            title="Webhook Embed",
            description="This is an embed sent via webhook",
            color=0x00ff00
        )
        embed.add_field(name="Field 1", value="Value 1", inline=True)
        embed.add_field(name="Field 2", value="Value 2", inline=True)
        embed.set_footer(text="Sent via webhook")
        
        await webhook.send(embed=embed)

# Run the async function
asyncio.run(send_webhook_message())

async def send_webhook_with_file():
    """Send webhook message with file attachment."""
    webhook_url = "https://discord.com/api/webhooks/YOUR_WEBHOOK_URL"
    
    async with aiohttp.ClientSession() as session:
        webhook = discord.Webhook.from_url(webhook_url, session=session)
        
        # Send file from disk
        with open('example.png', 'rb') as f:
            file = discord.File(f, filename='example.png')
            await webhook.send("Check out this image!", file=file)
        
        # Send multiple files
        files = [
            discord.File('file1.txt'),
            discord.File('file2.png', filename='renamed.png')
        ]
        await webhook.send("Multiple files!", files=files)
        
        # Send file with embed
        embed = discord.Embed(title="File Upload")
        embed.set_image(url="attachment://example.png")
        
        with open('example.png', 'rb') as f:
            file = discord.File(f, filename='example.png')
            await webhook.send(embed=embed, file=file)

asyncio.run(send_webhook_with_file())

async def manage_webhook_messages():
    """Demonstrate webhook message management."""
    webhook_url = "https://discord.com/api/webhooks/YOUR_WEBHOOK_URL"
    
    async with aiohttp.ClientSession() as session:
        webhook = discord.Webhook.from_url(webhook_url, session=session)
        
        # Send message and get response
        message = await webhook.send("Initial message", wait=True)
        print(f"Sent message with ID: {message.id}")
        
        # Wait a bit
        await asyncio.sleep(2)
        
        # Edit the message
        await webhook.edit_message(
            message.id,
            content="Edited message!",
            embed=discord.Embed(title="Edited", color=0xff9900)
        )
        
        # Wait and delete
        await asyncio.sleep(2)
        await webhook.delete_message(message.id)
        print("Message deleted")

asyncio.run(manage_webhook_messages())

async def webhook_with_threads():
    """Use webhooks with Discord threads."""
    webhook_url = "https://discord.com/api/webhooks/YOUR_WEBHOOK_URL"
    
    async with aiohttp.ClientSession() as session:
        webhook = discord.Webhook.from_url(webhook_url, session=session)
        
        # Send message and create thread
        message = await webhook.send("Starting a discussion!", wait=True)
        
        # Create thread from message
        thread = await webhook.create_thread(
            name="Discussion Thread",
            message=message,
            auto_archive_duration=1440,  # 24 hours
            reason="Starting discussion"
        )
        
        # Send messages in the thread
        await webhook.send(
            "This is in the thread!",
            thread=thread.id
        )
        
        await webhook.send(
            "Another thread message with embed!",
            thread=thread.id,
            embed=discord.Embed(
                title="Thread Message",
                description="This message is in a thread",
                color=0x9932cc
            )
        )

asyncio.run(webhook_with_threads())

import discord
import requests

def sync_webhook_example():
    """Example using synchronous webhook client."""
    webhook_url = "https://discord.com/api/webhooks/YOUR_WEBHOOK_URL"
    
    # Create synchronous webhook
    webhook = discord.SyncWebhook.from_url(webhook_url)
    
    # Send simple message
    webhook.send("Hello from sync webhook!")
    
    # Send with embed
    embed = discord.Embed(
        title="Sync Webhook",
        description="This was sent synchronously!",
        color=0x00ff00
    )
    
    message = webhook.send(embed=embed, wait=True)
    print(f"Sent message: {message.id}")
    
    # Edit the message
    webhook.edit_message(
        message.id,
        content="Edited sync message!",
        embed=discord.Embed(title="Edited", color=0xff0000)
    )

# Call synchronous function
sync_webhook_example()

import discord
import aiohttp
import asyncio

async def webhook_with_error_handling():
    """Demonstrate proper webhook error handling."""
    webhook_url = "https://discord.com/api/webhooks/YOUR_WEBHOOK_URL"
    
    async with aiohttp.ClientSession() as session:
        webhook = discord.Webhook.from_url(webhook_url, session=session)
        
        try:
            # Try to send a message
            await webhook.send("Test message")
            print("Message sent successfully")
            
        except discord.HTTPException as e:
            if e.status == 404:
                print("Webhook not found - check URL")
            elif e.status == 429:
                print(f"Rate limited - retry after {e.retry_after} seconds")
            else:
                print(f"HTTP error: {e.status} - {e.text}")
                
        except discord.InvalidArgument as e:
            print(f"Invalid argument: {e}")
            
        except Exception as e:
            print(f"Unexpected error: {e}")
        
        # Validate webhook before using
        try:
            # Fetch webhook info to validate
            webhook_info = await webhook.fetch()
            print(f"Using webhook: {webhook_info.name}")
            
        except discord.NotFound:
            print("Webhook does not exist")
            return
        except discord.Forbidden:
            print("No permission to access webhook")
            return
        
        # Send message with validation
        if len("Very long message content" * 100) > 2000:
            print("Message too long, truncating...")
            content = ("Very long message content" * 100)[:1997] + "..."
        else:
            content = "Normal message"
        
        await webhook.send(content)

asyncio.run(webhook_with_error_handling())

import discord
import aiohttp
import asyncio
from datetime import datetime, timezone

class WebhookBot:
    """Advanced webhook bot with multiple features."""
    
    def __init__(self, webhook_url: str):
        self.webhook_url = webhook_url
        self.session = None
        self.webhook = None
    
    async def __aenter__(self):
        self.session = aiohttp.ClientSession()
        self.webhook = discord.Webhook.from_url(self.webhook_url, session=self.session)
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if self.session:
            await self.session.close()
    
    async def send_notification(self, title: str, message: str, color: int = 0x0099ff):
        """Send a notification embed."""
        embed = discord.Embed(
            title=title,
            description=message,
            color=color,
            timestamp=datetime.now(timezone.utc)
        )
        embed.set_footer(text="Notification System")
        
        return await self.webhook.send(embed=embed, wait=True)
    
    async def send_status_update(self, service: str, status: str, details: str = None):
        """Send a service status update."""
        status_colors = {
            "online": 0x00ff00,
            "warning": 0xffaa00,
            "offline": 0xff0000
        }
        
        embed = discord.Embed(
            title=f"🔧 {service} Status Update",
            color=status_colors.get(status.lower(), 0x666666),
            timestamp=datetime.now(timezone.utc)
        )
        
        embed.add_field(name="Status", value=status.title(), inline=True)
        embed.add_field(name="Service", value=service, inline=True)
        embed.add_field(name="Time", value=f"<t:{int(datetime.now().timestamp())}:R>", inline=True)
        
        if details:
            embed.add_field(name="Details", value=details, inline=False)
        
        return await self.webhook.send(embed=embed, wait=True)
    
    async def send_log_message(self, level: str, message: str, extra_data: dict = None):
        """Send a log message with formatting."""
        level_colors = {
            "debug": 0x666666,
            "info": 0x0099ff,
            "warning": 0xffaa00,
            "error": 0xff0000,
            "critical": 0x990000
        }
        
        level_emojis = {
            "debug": "🐛",
            "info": "ℹ️",
            "warning": "⚠️",
            "error": "❌",
            "critical": "🚨"
        }
        
        embed = discord.Embed(
            title=f"{level_emojis.get(level.lower(), '📝')} {level.upper()}",
            description=f"```\n{message}\n```",
            color=level_colors.get(level.lower(), 0x666666),
            timestamp=datetime.now(timezone.utc)
        )
        
        if extra_data:
            for key, value in extra_data.items():
                embed.add_field(name=key.title(), value=str(value), inline=True)
        
        return await self.webhook.send(embed=embed, wait=True)
    
    async def send_metrics(self, metrics: dict):
        """Send system metrics."""
        embed = discord.Embed(
            title="📊 System Metrics",
            color=0x9932cc,
            timestamp=datetime.now(timezone.utc)
        )
        
        for metric, value in metrics.items():
            embed.add_field(name=metric.replace('_', ' ').title(), value=value, inline=True)
        
        return await self.webhook.send(embed=embed, wait=True)

# Usage example
async def main():
    webhook_url = "https://discord.com/api/webhooks/YOUR_WEBHOOK_URL"
    
    async with WebhookBot(webhook_url) as bot:
        # Send various types of messages
        await bot.send_notification(
            "System Alert",
            "Database connection restored",
            color=0x00ff00
        )
        
        await bot.send_status_update(
            "Web Server",
            "online",
            "All systems operational"
        )
        
        await bot.send_log_message(
            "error",
            "Failed to process user request",
            {"user_id": 12345, "endpoint": "/api/users", "error_code": 500}
        )
        
        await bot.send_metrics({
            "cpu_usage": "45%",
            "memory_usage": "2.1GB / 8GB",
            "active_users": 1234,
            "requests_per_minute": 89
        })

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