A high-level Twitch stream checker with state tracking and change detection. Monitor multiple Twitch streamers efficiently with asynchronous requests and smart state management.
- π Asynchronous - Built with
aiohttpfor high-performance concurrent checks - π State Tracking - Tracks online/offline transitions with configurable cooldowns
- π― Change Detection - Detects when streamers go live or go offline
- π¦ Batch Processing - Checks up to 100 streamers per API call
- πΎ Persistence - Save and restore monitoring state between sessions
- β Validation - Automatically filters out non-existent usernames
- π‘οΈ Robust - Handles rate limits and automatic token refresh
- π Typed - Full type annotations for better development experience
pip install twitch-checkerimport asyncio
from twitch_checker import TwitchChecker
async def main():
# Initialize with your Twitch API credentials
checker = TwitchChecker(
client_id="your_client_id",
client_secret="your_client_secret",
logins=["shroud", "ninja", "pokimane"]
)
# Check current status
statuses = await checker.check_logins()
for status in statuses:
print(f"{status.login}: {'LIVE' if status.is_live else 'offline'} {status.change or ''}")
asyncio.run(main())export TWITCH_CLIENT_ID="your_client_id"
export TWITCH_CLIENT_SECRET="your_client_secret"checker = TwitchChecker(
client_id="your_client_id",
client_secret="your_client_secret"
)Get Twitch API Credentials:
- Go to Twitch Console
- Create a new application
- Copy Client ID and generate Client Secret
checker = TwitchChecker(
client_id=CLIENT_ID,
client_secret=CLIENT_SECRET,
logins=["streamer1", "streamer2", "streamer3"]
)
# Check once
statuses = await checker.check_logins()
# Check periodically
while True:
statuses = await checker.check_logins()
for status in statuses:
if status.change == "UP":
print(f"π {status.login} just went live!")
elif status.change == "DOWN":
print(f"π΄ {status.login} went offline")
await asyncio.sleep(60) # Check every minute# Wait 5 minutes before reporting offline status
checker = TwitchChecker(
client_id=CLIENT_ID,
client_secret=CLIENT_SECRET,
logins=streamers,
cooldown_seconds=300 # 5 minutes
)This prevents false offline notifications when a streamer briefly disconnects or has internet issues.
# Check which streamers exist and their current status
classification = await checker.classify_logins([
"existing_streamer", "nonexistent_user", "another_streamer"
])
# Result:
# {
# "existing_streamer": "exists_and_live",
# "nonexistent_user": "does_not_exist",
# "another_streamer": "exists_but_not_live"
# }# Save state to file
checker.export_json("state.json")
# Later, restore state
checker = TwitchChecker.from_json(
"state.json",
client_id=CLIENT_ID,
client_secret=CLIENT_SECRET
)__init__(client_id, client_secret, logins=None, *, cooldown_seconds=0, checked_existence=None)
client_id: Twitch API Client IDclient_secret: Twitch API Client Secretlogins: Initial list of streamer logins to monitorcooldown_seconds: Delay before reporting offline status (prevents false positives)checked_existence: Pre-validated usernames (internal use)
Methods:
async check_logins() -> Set[StreamerStatus]: Check all monitored streamersasync classify_logins(logins) -> Dict[str, str]: Categorize streamers by existence/live statusasync batch_user_exists(logins) -> Set[str]: Check which usernames existexport_json(path) -> str: Save state to JSON fileclassmethod from_json(path, *, client_id, client_secret): Load state from JSON file
login: Streamer's usernameis_live: Current live status (True/False)change: Status change ("UP", "DOWN", or None)data: Raw Twitch API stream data
import discord
from twitch_checker import TwitchChecker
class TwitchCog(discord.Cog):
def __init__(self, bot):
self.bot = bot
self.checker = TwitchChecker(CLIENT_ID, CLIENT_SECRET)
self.checker.logins = ["favorite_streamers..."]
async def twitch_loop(self):
while True:
statuses = await self.checker.check_logins()
for status in statuses:
if status.change == "UP":
channel = self.bot.get_channel(NOTIFICATION_CHANNEL_ID)
await channel.send(
f"π΄ **{status.login}** just went live!\n"
f"https://twitch.tv/{status.login}"
)
await asyncio.sleep(120)try:
statuses = await checker.check_logins()
except Exception as e:
print(f"Error checking streamers: {e}")
# Handle rate limits, authentication errors, etc.Contributions are welcome! Please feel free to submit a Pull Request.
- Fork the repository
- Create a feature branch (
git checkout -b feature/amazing-feature) - Commit your changes (
git commit -m 'Add some amazing feature') - Push to the branch (
git push origin feature/amazing-feature) - Open a Pull Request
This project is licensed under the MIT License - see the LICENSE file for details.
If you encounter any problems or have questions:
- Check the Twitch API Documentation
- Open an issue on GitHub
- Ensure your API credentials are correct and have the necessary permissions
- Twitch for providing the API
- aiohttp for async HTTP requests
- The Python community for excellent tooling and support
These files provide:
**CHANGELOG.md:**
- Professional format following Keep a Changelog standards
- Clear breakdown of what's included in the initial release
- Semantic versioning ready for future updates
**README.md:**
- Professional branding with badges (will work once published)
- Clear installation and setup instructions
- Multiple usage examples for different scenarios
- Comprehensive API documentation
- Practical integration examples (like Discord bots)
- Troubleshooting and support sections
- Contributing guidelines
The README is designed to help users get started quickly while also providing depth for advanced use cases. It covers everything from basic monitoring to persistent state management and error handling.