0
# Nextcord Webhooks
1
2
Webhook creation and management for external integrations and message delivery outside of bot connections, providing flexible communication channels.
3
4
## Webhook Class
5
6
Core webhook functionality for sending messages and managing webhook properties.
7
8
### Async Webhook { .api }
9
10
```python
11
import nextcord
12
from nextcord import Webhook, SyncWebhook, WebhookMessage
13
from typing import Optional, List, Union, Any, Dict
14
import aiohttp
15
16
class Webhook:
17
"""Represents an async Discord webhook.
18
19
Webhooks allow external services to send messages to Discord channels
20
without needing a bot connection.
21
22
Attributes
23
----------
24
id: int
25
The webhook's ID.
26
type: WebhookType
27
The type of webhook.
28
token: Optional[str]
29
The webhook's token.
30
guild_id: Optional[int]
31
The guild ID the webhook belongs to.
32
channel_id: Optional[int]
33
The channel ID the webhook belongs to.
34
user: Optional[User]
35
The user who created this webhook.
36
name: Optional[str]
37
The webhook's name.
38
avatar: Optional[str]
39
The webhook's avatar hash.
40
source_guild: Optional[PartialWebhookGuild]
41
The source guild for channel follower webhooks.
42
source_channel: Optional[PartialWebhookChannel]
43
The source channel for channel follower webhooks.
44
url: str
45
The webhook's URL.
46
"""
47
48
@classmethod
49
def partial(
50
cls,
51
id: int,
52
token: str,
53
*,
54
session: Optional[aiohttp.ClientSession] = None
55
) -> Webhook:
56
"""Create a partial webhook from ID and token.
57
58
Parameters
59
----------
60
id: int
61
The webhook ID.
62
token: str
63
The webhook token.
64
session: Optional[aiohttp.ClientSession]
65
The session to use for requests.
66
67
Returns
68
-------
69
Webhook
70
A partial webhook instance.
71
"""
72
...
73
74
@classmethod
75
def from_url(
76
cls,
77
url: str,
78
*,
79
session: Optional[aiohttp.ClientSession] = None
80
) -> Webhook:
81
"""Create a webhook from a Discord webhook URL.
82
83
Parameters
84
----------
85
url: str
86
The webhook URL.
87
session: Optional[aiohttp.ClientSession]
88
The session to use for requests.
89
90
Returns
91
-------
92
Webhook
93
A webhook instance from the URL.
94
"""
95
...
96
97
async def fetch(self, *, prefer_auth: bool = True) -> Webhook:
98
"""Fetch the webhook's information.
99
100
Parameters
101
----------
102
prefer_auth: bool
103
Whether to prefer authenticated requests.
104
105
Returns
106
-------
107
Webhook
108
The updated webhook.
109
"""
110
...
111
112
async def delete(self, *, reason: Optional[str] = None) -> None:
113
"""Delete the webhook.
114
115
Parameters
116
----------
117
reason: Optional[str]
118
The reason for deleting the webhook.
119
"""
120
...
121
122
async def edit(
123
self,
124
*,
125
name: Optional[str] = MISSING,
126
avatar: Optional[bytes] = MISSING,
127
reason: Optional[str] = None
128
) -> Webhook:
129
"""Edit the webhook.
130
131
Parameters
132
----------
133
name: Optional[str]
134
The webhook's new name.
135
avatar: Optional[bytes]
136
The webhook's new avatar as bytes.
137
reason: Optional[str]
138
The reason for editing the webhook.
139
140
Returns
141
-------
142
Webhook
143
The updated webhook.
144
"""
145
...
146
147
async def send(
148
self,
149
content: Optional[str] = MISSING,
150
*,
151
username: Optional[str] = MISSING,
152
avatar_url: Optional[str] = MISSING,
153
tts: bool = False,
154
embed: Optional[nextcord.Embed] = MISSING,
155
embeds: Optional[List[nextcord.Embed]] = MISSING,
156
file: Optional[nextcord.File] = MISSING,
157
files: Optional[List[nextcord.File]] = MISSING,
158
allowed_mentions: Optional[nextcord.AllowedMentions] = MISSING,
159
thread: Optional[nextcord.abc.Snowflake] = MISSING,
160
wait: bool = False,
161
suppress_embeds: bool = False
162
) -> Optional[WebhookMessage]:
163
"""Send a message via the webhook.
164
165
Parameters
166
----------
167
content: Optional[str]
168
The message content.
169
username: Optional[str]
170
Override the webhook's username.
171
avatar_url: Optional[str]
172
Override the webhook's avatar URL.
173
tts: bool
174
Whether the message should be text-to-speech.
175
embed: Optional[nextcord.Embed]
176
An embed to send.
177
embeds: Optional[List[nextcord.Embed]]
178
A list of embeds to send (max 10).
179
file: Optional[nextcord.File]
180
A file to send.
181
files: Optional[List[nextcord.File]]
182
A list of files to send (max 10).
183
allowed_mentions: Optional[nextcord.AllowedMentions]
184
Controls which mentions are processed.
185
thread: Optional[nextcord.abc.Snowflake]
186
The thread to send the message to.
187
wait: bool
188
Whether to wait for the message and return it.
189
suppress_embeds: bool
190
Whether to suppress embeds in the message.
191
192
Returns
193
-------
194
Optional[WebhookMessage]
195
The sent message if wait=True, otherwise None.
196
"""
197
...
198
199
# Basic webhook usage examples
200
async def webhook_examples():
201
"""Examples of webhook usage."""
202
203
# Create webhook from URL
204
webhook_url = "https://discord.com/api/webhooks/123456789/abcdef123456"
205
webhook = nextcord.Webhook.from_url(webhook_url)
206
207
# Send simple message
208
await webhook.send("Hello from webhook!")
209
210
# Send message with custom username and avatar
211
await webhook.send(
212
"Custom webhook message",
213
username="Custom Bot",
214
avatar_url="https://example.com/avatar.png"
215
)
216
217
# Send message with embed
218
embed = nextcord.Embed(
219
title="Webhook Embed",
220
description="This message was sent via webhook",
221
color=nextcord.Color.blue()
222
)
223
embed.add_field(name="Field", value="Value", inline=False)
224
225
await webhook.send(embed=embed)
226
227
# Send message and get response
228
message = await webhook.send(
229
"Message with response",
230
wait=True # Wait for the message to be sent and return it
231
)
232
233
if message:
234
print(f"Sent message with ID: {message.id}")
235
236
# Webhook management
237
async def manage_webhook(channel: nextcord.TextChannel):
238
"""Create and manage webhooks."""
239
240
# Create a new webhook
241
webhook = await channel.create_webhook(
242
name="My Bot Webhook",
243
avatar=None, # Can provide avatar bytes here
244
reason="Created for bot notifications"
245
)
246
247
print(f"Created webhook: {webhook.url}")
248
249
# Edit webhook properties
250
await webhook.edit(
251
name="Updated Bot Webhook",
252
reason="Updated webhook name"
253
)
254
255
# Use the webhook
256
await webhook.send("Webhook created and configured!")
257
258
# Get webhook info
259
webhook_info = await webhook.fetch()
260
print(f"Webhook name: {webhook_info.name}")
261
print(f"Created by: {webhook_info.user}")
262
263
# Delete webhook when done
264
await webhook.delete(reason="No longer needed")
265
```
266
267
### Sync Webhook { .api }
268
269
```python
270
class SyncWebhook:
271
"""Represents a synchronous Discord webhook.
272
273
This is useful for applications that don't use async/await
274
or need to send webhook messages from synchronous code.
275
276
All methods are synchronous versions of the async Webhook class.
277
"""
278
279
@classmethod
280
def partial(
281
cls,
282
id: int,
283
token: str,
284
*,
285
session: Optional[Any] = None
286
) -> SyncWebhook:
287
"""Create a partial sync webhook from ID and token."""
288
...
289
290
@classmethod
291
def from_url(
292
cls,
293
url: str,
294
*,
295
session: Optional[Any] = None
296
) -> SyncWebhook:
297
"""Create a sync webhook from a Discord webhook URL."""
298
...
299
300
def send(
301
self,
302
content: Optional[str] = MISSING,
303
*,
304
username: Optional[str] = MISSING,
305
avatar_url: Optional[str] = MISSING,
306
tts: bool = False,
307
embed: Optional[nextcord.Embed] = MISSING,
308
embeds: Optional[List[nextcord.Embed]] = MISSING,
309
file: Optional[nextcord.File] = MISSING,
310
files: Optional[List[nextcord.File]] = MISSING,
311
allowed_mentions: Optional[nextcord.AllowedMentions] = MISSING,
312
thread: Optional[nextcord.abc.Snowflake] = MISSING,
313
wait: bool = False
314
) -> Optional[WebhookMessage]:
315
"""Send a message via the webhook synchronously."""
316
...
317
318
def fetch(self, *, prefer_auth: bool = True) -> SyncWebhook:
319
"""Fetch the webhook's information synchronously."""
320
...
321
322
def edit(
323
self,
324
*,
325
name: Optional[str] = MISSING,
326
avatar: Optional[bytes] = MISSING,
327
reason: Optional[str] = None
328
) -> SyncWebhook:
329
"""Edit the webhook synchronously."""
330
...
331
332
def delete(self, *, reason: Optional[str] = None) -> None:
333
"""Delete the webhook synchronously."""
334
...
335
336
# Synchronous webhook usage
337
def sync_webhook_example():
338
"""Example of synchronous webhook usage."""
339
import requests
340
341
# Create sync webhook from URL
342
webhook_url = "https://discord.com/api/webhooks/123456789/abcdef123456"
343
webhook = nextcord.SyncWebhook.from_url(webhook_url)
344
345
# Send message synchronously
346
webhook.send("Synchronous webhook message!")
347
348
# Send with embed
349
embed = nextcord.Embed(
350
title="Sync Webhook",
351
description="Sent synchronously",
352
color=nextcord.Color.green()
353
)
354
355
message = webhook.send(embed=embed, wait=True)
356
if message:
357
print(f"Message sent with ID: {message.id}")
358
359
# Integration with web frameworks (Flask example)
360
from flask import Flask, request, jsonify
361
362
app = Flask(__name__)
363
364
@app.route('/notify', methods=['POST'])
365
def send_discord_notification():
366
"""Flask endpoint to send Discord notifications via webhook."""
367
try:
368
data = request.json
369
370
# Get webhook URL from environment or config
371
webhook_url = "your_webhook_url_here"
372
webhook = nextcord.SyncWebhook.from_url(webhook_url)
373
374
# Create notification embed
375
embed = nextcord.Embed(
376
title=data.get('title', 'Notification'),
377
description=data.get('message', ''),
378
color=nextcord.Color.blue()
379
)
380
381
if data.get('url'):
382
embed.url = data['url']
383
384
if data.get('fields'):
385
for field in data['fields']:
386
embed.add_field(
387
name=field['name'],
388
value=field['value'],
389
inline=field.get('inline', False)
390
)
391
392
# Send notification
393
webhook.send(
394
embed=embed,
395
username=data.get('username', 'Notification Bot'),
396
avatar_url=data.get('avatar_url')
397
)
398
399
return jsonify({'success': True})
400
401
except Exception as e:
402
return jsonify({'success': False, 'error': str(e)}), 500
403
```
404
405
## Webhook Message
406
407
Messages sent through webhooks with editing and deletion capabilities.
408
409
### WebhookMessage Class { .api }
410
411
```python
412
class WebhookMessage:
413
"""Represents a message sent by a webhook.
414
415
Attributes
416
----------
417
id: int
418
The message ID.
419
type: MessageType
420
The message type.
421
content: str
422
The message content.
423
channel_id: int
424
The channel ID the message was sent in.
425
author: WebhookAuthor
426
The webhook author information.
427
embeds: List[Embed]
428
The message embeds.
429
attachments: List[Attachment]
430
The message attachments.
431
edited_at: Optional[datetime.datetime]
432
When the message was last edited.
433
flags: MessageFlags
434
The message flags.
435
thread: Optional[Thread]
436
The thread the message was sent in.
437
"""
438
439
async def edit(
440
self,
441
*,
442
content: Optional[str] = MISSING,
443
embed: Optional[nextcord.Embed] = MISSING,
444
embeds: Optional[List[nextcord.Embed]] = MISSING,
445
file: Optional[nextcord.File] = MISSING,
446
files: Optional[List[nextcord.File]] = MISSING,
447
attachments: Optional[List[nextcord.Attachment]] = MISSING,
448
allowed_mentions: Optional[nextcord.AllowedMentions] = MISSING
449
) -> WebhookMessage:
450
"""Edit the webhook message.
451
452
Parameters
453
----------
454
content: Optional[str]
455
The new message content.
456
embed: Optional[nextcord.Embed]
457
The new embed for the message.
458
embeds: Optional[List[nextcord.Embed]]
459
The new embeds for the message.
460
file: Optional[nextcord.File]
461
A new file to add to the message.
462
files: Optional[List[nextcord.File]]
463
New files to add to the message.
464
attachments: Optional[List[nextcord.Attachment]]
465
Existing attachments to keep.
466
allowed_mentions: Optional[nextcord.AllowedMentions]
467
Controls which mentions are processed.
468
469
Returns
470
-------
471
WebhookMessage
472
The edited message.
473
"""
474
...
475
476
async def delete(self, *, delay: Optional[float] = None) -> None:
477
"""Delete the webhook message.
478
479
Parameters
480
----------
481
delay: Optional[float]
482
Number of seconds to wait before deleting.
483
"""
484
...
485
486
@property
487
def jump_url(self) -> str:
488
"""str: The jump URL for this message."""
489
...
490
491
@property
492
def created_at(self) -> datetime.datetime:
493
"""datetime.datetime: When the message was created."""
494
...
495
496
# Webhook message management examples
497
async def webhook_message_management():
498
"""Examples of managing webhook messages."""
499
500
webhook = nextcord.Webhook.from_url("webhook_url_here")
501
502
# Send message and keep reference
503
message = await webhook.send(
504
"This message will be edited later",
505
wait=True # Required to get message object
506
)
507
508
if message:
509
# Edit the message after 5 seconds
510
await asyncio.sleep(5)
511
await message.edit(content="This message has been edited!")
512
513
# Add an embed to the message
514
embed = nextcord.Embed(
515
title="Updated Message",
516
description="This embed was added later",
517
color=nextcord.Color.green()
518
)
519
await message.edit(embed=embed)
520
521
# Delete the message after 10 seconds
522
await message.delete(delay=10)
523
524
# Advanced webhook patterns
525
class WebhookNotificationSystem:
526
"""Advanced webhook notification system."""
527
528
def __init__(self, webhook_url: str):
529
self.webhook = nextcord.Webhook.from_url(webhook_url)
530
self.message_cache = {}
531
532
async def send_status_update(
533
self,
534
service: str,
535
status: str,
536
details: Optional[str] = None
537
) -> Optional[WebhookMessage]:
538
"""Send or update a service status message."""
539
540
# Determine embed color based on status
541
color_map = {
542
'operational': nextcord.Color.green(),
543
'degraded': nextcord.Color.orange(),
544
'outage': nextcord.Color.red(),
545
'maintenance': nextcord.Color.blue()
546
}
547
548
embed = nextcord.Embed(
549
title=f"π§ {service} Status",
550
description=status.title(),
551
color=color_map.get(status.lower(), nextcord.Color.gray()),
552
timestamp=datetime.now()
553
)
554
555
if details:
556
embed.add_field(name="Details", value=details, inline=False)
557
558
# Check if we have an existing message for this service
559
if service in self.message_cache:
560
try:
561
# Update existing message
562
message = self.message_cache[service]
563
updated_message = await message.edit(embed=embed)
564
return updated_message
565
except nextcord.NotFound:
566
# Message was deleted, remove from cache
567
del self.message_cache[service]
568
569
# Send new message
570
message = await self.webhook.send(
571
embed=embed,
572
username=f"{service} Monitor",
573
wait=True
574
)
575
576
if message:
577
self.message_cache[service] = message
578
579
return message
580
581
async def send_alert(
582
self,
583
title: str,
584
message: str,
585
severity: str = 'info',
586
ping_role: Optional[str] = None
587
) -> Optional[WebhookMessage]:
588
"""Send an alert notification."""
589
590
severity_config = {
591
'info': {'emoji': 'βΉοΈ', 'color': nextcord.Color.blue()},
592
'warning': {'emoji': 'β οΈ', 'color': nextcord.Color.orange()},
593
'error': {'emoji': 'β', 'color': nextcord.Color.red()},
594
'critical': {'emoji': 'π¨', 'color': nextcord.Color.from_rgb(139, 0, 0)}
595
}
596
597
config = severity_config.get(severity, severity_config['info'])
598
599
embed = nextcord.Embed(
600
title=f"{config['emoji']} {title}",
601
description=message,
602
color=config['color'],
603
timestamp=datetime.now()
604
)
605
606
embed.add_field(name="Severity", value=severity.upper(), inline=True)
607
608
content = ""
609
if ping_role and severity in ['error', 'critical']:
610
content = f"<@&{ping_role}>"
611
612
return await self.webhook.send(
613
content=content,
614
embed=embed,
615
username="Alert System",
616
wait=True
617
)
618
619
async def send_log_entry(
620
self,
621
level: str,
622
service: str,
623
message: str,
624
extra_data: Optional[Dict[str, Any]] = None
625
):
626
"""Send a log entry via webhook."""
627
628
level_colors = {
629
'DEBUG': nextcord.Color.light_gray(),
630
'INFO': nextcord.Color.blue(),
631
'WARNING': nextcord.Color.orange(),
632
'ERROR': nextcord.Color.red(),
633
'CRITICAL': nextcord.Color.from_rgb(139, 0, 0)
634
}
635
636
embed = nextcord.Embed(
637
title=f"π {level} - {service}",
638
description=f"```\n{message}\n```",
639
color=level_colors.get(level, nextcord.Color.default()),
640
timestamp=datetime.now()
641
)
642
643
if extra_data:
644
for key, value in extra_data.items():
645
embed.add_field(
646
name=key.replace('_', ' ').title(),
647
value=str(value)[:1024], # Discord field value limit
648
inline=True
649
)
650
651
await self.webhook.send(
652
embed=embed,
653
username=f"{service} Logger"
654
)
655
656
# Usage example
657
async def notification_system_example():
658
"""Example usage of the webhook notification system."""
659
660
webhook_url = "your_webhook_url_here"
661
notifier = WebhookNotificationSystem(webhook_url)
662
663
# Send initial status
664
await notifier.send_status_update(
665
service="API Server",
666
status="operational",
667
details="All systems running normally"
668
)
669
670
# Update status (will edit the existing message)
671
await notifier.send_status_update(
672
service="API Server",
673
status="degraded",
674
details="Experiencing higher than normal latency"
675
)
676
677
# Send alert
678
await notifier.send_alert(
679
title="High CPU Usage",
680
message="Server CPU usage has exceeded 80% for the past 5 minutes",
681
severity="warning"
682
)
683
684
# Send log entry
685
await notifier.send_log_entry(
686
level="ERROR",
687
service="Database",
688
message="Connection timeout occurred",
689
extra_data={
690
"query_time": "15.3s",
691
"affected_users": 42,
692
"retry_count": 3
693
}
694
)
695
```
696
697
## Webhook Management
698
699
Advanced webhook management including creation, monitoring, and cleanup.
700
701
### Guild Webhook Operations { .api }
702
703
```python
704
# Guild-level webhook management
705
async def manage_guild_webhooks(guild: nextcord.Guild):
706
"""Manage webhooks at the guild level."""
707
708
# Get all webhooks in the guild
709
webhooks = await guild.webhooks()
710
711
print(f"Found {len(webhooks)} webhooks in {guild.name}")
712
713
for webhook in webhooks:
714
print(f"- {webhook.name} (ID: {webhook.id})")
715
print(f" Channel: #{webhook.channel.name if webhook.channel else 'Unknown'}")
716
print(f" Created by: {webhook.user}")
717
print(f" Type: {webhook.type}")
718
print()
719
720
# Channel-specific webhook operations
721
async def manage_channel_webhooks(channel: nextcord.TextChannel):
722
"""Manage webhooks for a specific channel."""
723
724
# Get webhooks for this channel
725
webhooks = await channel.webhooks()
726
727
print(f"Webhooks in #{channel.name}:")
728
729
if not webhooks:
730
print("No webhooks found.")
731
return
732
733
for webhook in webhooks:
734
# Get detailed info
735
webhook_info = await webhook.fetch()
736
737
print(f"Name: {webhook_info.name}")
738
print(f"ID: {webhook_info.id}")
739
print(f"Token: {'***' + webhook_info.token[-4:] if webhook_info.token else 'No token'}")
740
print(f"Created by: {webhook_info.user}")
741
print(f"Avatar: {webhook_info.avatar}")
742
print("---")
743
744
# Webhook monitoring and cleanup
745
class WebhookManager:
746
"""Comprehensive webhook management system."""
747
748
def __init__(self, bot):
749
self.bot = bot
750
self.webhook_usage = {} # Track webhook usage
751
752
async def audit_webhooks(self, guild: nextcord.Guild) -> Dict[str, Any]:
753
"""Audit all webhooks in a guild."""
754
webhooks = await guild.webhooks()
755
756
audit_results = {
757
'total_webhooks': len(webhooks),
758
'active_webhooks': 0,
759
'inactive_webhooks': 0,
760
'unknown_webhooks': 0,
761
'webhook_details': []
762
}
763
764
for webhook in webhooks:
765
try:
766
# Fetch detailed information
767
webhook_info = await webhook.fetch()
768
769
details = {
770
'id': webhook_info.id,
771
'name': webhook_info.name,
772
'channel': webhook_info.channel.name if webhook_info.channel else 'Unknown',
773
'created_by': str(webhook_info.user) if webhook_info.user else 'Unknown',
774
'type': str(webhook_info.type),
775
'has_token': webhook_info.token is not None,
776
'status': 'active' # Assume active if fetchable
777
}
778
779
audit_results['active_webhooks'] += 1
780
audit_results['webhook_details'].append(details)
781
782
except nextcord.NotFound:
783
# Webhook was deleted or is inaccessible
784
audit_results['inactive_webhooks'] += 1
785
audit_results['webhook_details'].append({
786
'id': webhook.id,
787
'name': webhook.name or 'Unknown',
788
'status': 'inactive',
789
'error': 'Not found or no access'
790
})
791
792
except Exception as e:
793
audit_results['unknown_webhooks'] += 1
794
audit_results['webhook_details'].append({
795
'id': webhook.id,
796
'name': webhook.name or 'Unknown',
797
'status': 'error',
798
'error': str(e)
799
})
800
801
return audit_results
802
803
async def cleanup_unused_webhooks(
804
self,
805
guild: nextcord.Guild,
806
dry_run: bool = True
807
) -> Dict[str, Any]:
808
"""Clean up unused or orphaned webhooks."""
809
810
webhooks = await guild.webhooks()
811
cleanup_results = {
812
'total_checked': len(webhooks),
813
'deleted': [],
814
'kept': [],
815
'errors': []
816
}
817
818
for webhook in webhooks:
819
try:
820
webhook_info = await webhook.fetch()
821
822
# Check if webhook should be deleted
823
should_delete = False
824
reason = ""
825
826
# Check if channel still exists
827
if not webhook_info.channel:
828
should_delete = True
829
reason = "Channel no longer exists"
830
831
# Check if created by a bot that's no longer in the guild
832
elif webhook_info.user and webhook_info.user.bot:
833
if webhook_info.user not in guild.members:
834
should_delete = True
835
reason = "Created by bot no longer in guild"
836
837
# Check for specific naming patterns that indicate unused webhooks
838
elif webhook_info.name and any(pattern in webhook_info.name.lower() for pattern in ['test', 'temp', 'unused']):
839
should_delete = True
840
reason = "Appears to be temporary/test webhook"
841
842
if should_delete:
843
if not dry_run:
844
await webhook.delete(reason=f"Cleanup: {reason}")
845
cleanup_results['deleted'].append({
846
'id': webhook_info.id,
847
'name': webhook_info.name,
848
'reason': reason
849
})
850
else:
851
cleanup_results['deleted'].append({
852
'id': webhook_info.id,
853
'name': webhook_info.name,
854
'reason': f"Would delete: {reason}"
855
})
856
else:
857
cleanup_results['kept'].append({
858
'id': webhook_info.id,
859
'name': webhook_info.name
860
})
861
862
except Exception as e:
863
cleanup_results['errors'].append({
864
'webhook_id': webhook.id,
865
'error': str(e)
866
})
867
868
return cleanup_results
869
870
async def create_managed_webhook(
871
self,
872
channel: nextcord.TextChannel,
873
name: str,
874
avatar_path: Optional[str] = None,
875
reason: Optional[str] = None
876
) -> nextcord.Webhook:
877
"""Create a webhook with proper management tracking."""
878
879
# Read avatar if provided
880
avatar_bytes = None
881
if avatar_path:
882
try:
883
with open(avatar_path, 'rb') as f:
884
avatar_bytes = f.read()
885
except FileNotFoundError:
886
print(f"Avatar file not found: {avatar_path}")
887
888
# Create webhook
889
webhook = await channel.create_webhook(
890
name=name,
891
avatar=avatar_bytes,
892
reason=reason or f"Managed webhook created by {self.bot.user}"
893
)
894
895
# Track the webhook
896
self.webhook_usage[webhook.id] = {
897
'created_at': datetime.now(),
898
'channel_id': channel.id,
899
'guild_id': channel.guild.id,
900
'created_by_bot': True,
901
'last_used': None,
902
'use_count': 0
903
}
904
905
return webhook
906
907
def track_webhook_usage(self, webhook_id: int):
908
"""Track usage of a managed webhook."""
909
if webhook_id in self.webhook_usage:
910
self.webhook_usage[webhook_id]['last_used'] = datetime.now()
911
self.webhook_usage[webhook_id]['use_count'] += 1
912
913
async def get_webhook_statistics(self, guild: nextcord.Guild) -> Dict[str, Any]:
914
"""Get comprehensive webhook statistics for a guild."""
915
916
webhooks = await guild.webhooks()
917
918
stats = {
919
'total_webhooks': len(webhooks),
920
'webhooks_by_channel': {},
921
'webhooks_by_type': {},
922
'bot_created_webhooks': 0,
923
'user_created_webhooks': 0,
924
'managed_webhooks': len([wh_id for wh_id in self.webhook_usage if wh_id in [wh.id for wh in webhooks]])
925
}
926
927
for webhook in webhooks:
928
try:
929
webhook_info = await webhook.fetch()
930
931
# Count by channel
932
channel_name = webhook_info.channel.name if webhook_info.channel else 'Unknown'
933
stats['webhooks_by_channel'][channel_name] = stats['webhooks_by_channel'].get(channel_name, 0) + 1
934
935
# Count by type
936
webhook_type = str(webhook_info.type)
937
stats['webhooks_by_type'][webhook_type] = stats['webhooks_by_type'].get(webhook_type, 0) + 1
938
939
# Count by creator type
940
if webhook_info.user and webhook_info.user.bot:
941
stats['bot_created_webhooks'] += 1
942
else:
943
stats['user_created_webhooks'] += 1
944
945
except Exception:
946
pass # Skip webhooks we can't access
947
948
return stats
949
950
# Command examples for webhook management
951
class WebhookCommands:
952
"""Slash commands for webhook management."""
953
954
def __init__(self, bot, webhook_manager: WebhookManager):
955
self.bot = bot
956
self.webhook_manager = webhook_manager
957
958
@nextcord.slash_command(description="Audit server webhooks")
959
async def audit_webhooks(self, interaction: nextcord.Interaction):
960
"""Audit all webhooks in the server."""
961
if not interaction.user.guild_permissions.manage_webhooks:
962
await interaction.response.send_message(
963
"β You need 'Manage Webhooks' permission to use this command.",
964
ephemeral=True
965
)
966
return
967
968
await interaction.response.defer()
969
970
audit_results = await self.webhook_manager.audit_webhooks(interaction.guild)
971
972
embed = nextcord.Embed(
973
title="π Webhook Audit Results",
974
color=nextcord.Color.blue()
975
)
976
977
embed.add_field(
978
name="Summary",
979
value=f"**Total:** {audit_results['total_webhooks']}\n"
980
f"**Active:** {audit_results['active_webhooks']}\n"
981
f"**Inactive:** {audit_results['inactive_webhooks']}\n"
982
f"**Errors:** {audit_results['unknown_webhooks']}",
983
inline=False
984
)
985
986
if audit_results['webhook_details']:
987
# Show first 10 webhooks
988
webhook_list = []
989
for webhook in audit_results['webhook_details'][:10]:
990
status_emoji = {"active": "β ", "inactive": "β", "error": "β οΈ"}
991
emoji = status_emoji.get(webhook['status'], "β")
992
webhook_list.append(f"{emoji} **{webhook['name']}** (#{webhook.get('channel', 'Unknown')})")
993
994
embed.add_field(
995
name="Webhooks",
996
value="\n".join(webhook_list) +
997
(f"\n... and {len(audit_results['webhook_details']) - 10} more"
998
if len(audit_results['webhook_details']) > 10 else ""),
999
inline=False
1000
)
1001
1002
await interaction.followup.send(embed=embed)
1003
1004
@nextcord.slash_command(description="Get webhook statistics")
1005
async def webhook_stats(self, interaction: nextcord.Interaction):
1006
"""Get comprehensive webhook statistics."""
1007
if not interaction.user.guild_permissions.manage_webhooks:
1008
await interaction.response.send_message(
1009
"β You need 'Manage Webhooks' permission to use this command.",
1010
ephemeral=True
1011
)
1012
return
1013
1014
await interaction.response.defer()
1015
1016
stats = await self.webhook_manager.get_webhook_statistics(interaction.guild)
1017
1018
embed = nextcord.Embed(
1019
title="π Webhook Statistics",
1020
color=nextcord.Color.green()
1021
)
1022
1023
embed.add_field(
1024
name="Overview",
1025
value=f"**Total Webhooks:** {stats['total_webhooks']}\n"
1026
f"**Bot Created:** {stats['bot_created_webhooks']}\n"
1027
f"**User Created:** {stats['user_created_webhooks']}\n"
1028
f"**Managed by Bot:** {stats['managed_webhooks']}",
1029
inline=False
1030
)
1031
1032
if stats['webhooks_by_channel']:
1033
channels = list(stats['webhooks_by_channel'].items())[:5] # Top 5 channels
1034
channel_text = "\n".join([f"#{channel}: {count}" for channel, count in channels])
1035
embed.add_field(name="By Channel (Top 5)", value=channel_text, inline=True)
1036
1037
if stats['webhooks_by_type']:
1038
type_text = "\n".join([f"{wh_type}: {count}" for wh_type, count in stats['webhooks_by_type'].items()])
1039
embed.add_field(name="By Type", value=type_text, inline=True)
1040
1041
await interaction.followup.send(embed=embed)
1042
```
1043
1044
This comprehensive documentation covers all aspects of nextcord's webhook system, providing developers with the tools needed to implement flexible external integrations and messaging solutions.