nix

configuration files that power my machines
Log | Files | Refs | README | LICENSE

tw (2325B)

      1 #!/usr/bin/env python3
      2 import sys
      3 import os
      4 import urllib.request
      5 import urllib.error
      6 from typing import cast
      7 from http.client import HTTPResponse
      8 from concurrent.futures import ThreadPoolExecutor, as_completed
      9 
     10 
     11 def read_channels_file(filepath: str) -> list[tuple[str, str]]:
     12     channels: list[tuple[str, str]] = []
     13     if not os.path.isfile(filepath):
     14         return channels
     15 
     16     with open(filepath, "r") as f:
     17         for line in f:
     18             line = line.strip()
     19             if not line:
     20                 continue
     21             parts = line.split(None, 1)
     22             channel = parts[0]
     23             description = parts[1] if len(parts) > 1 else ""
     24             channels.append((channel, description))
     25 
     26     return channels
     27 
     28 
     29 def check_channel_status(channel: str, description: str) -> tuple[str, str, bool]:
     30     try:
     31         url = f"https://www.twitch.tv/{channel}"
     32         req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
     33         with cast(HTTPResponse, urllib.request.urlopen(req, timeout=10)) as response:
     34             content: str = response.read().decode("utf-8", errors="ignore")
     35             is_online: bool = "isLiveBroadcast" in content
     36             return (channel, description, is_online)
     37     except (urllib.error.URLError, urllib.error.HTTPError, TimeoutError):
     38         return (channel, description, False)
     39 
     40 
     41 def main():
     42     channels_file = os.path.expanduser("~/.config/twitch/channels")
     43     if not os.path.isfile(channels_file):
     44         print("Error: ~/.config/twitch/channels not found", file=sys.stderr)
     45         sys.exit(1)
     46 
     47     channels = read_channels_file(channels_file)
     48     if not channels:
     49         return
     50 
     51     max_workers = 10
     52 
     53     with ThreadPoolExecutor(max_workers=max_workers) as executor:
     54         future_to_channel = {
     55             executor.submit(check_channel_status, channel, description): (
     56                 channel,
     57                 description,
     58             )
     59             for channel, description in channels
     60         }
     61 
     62         for future in as_completed(future_to_channel):
     63             channel, description, is_online = future.result()
     64 
     65             if is_online:
     66                 if description:
     67                     print(f"{channel} ({description}) is online")
     68                 else:
     69                     print(f"{channel} is online")
     70 
     71 
     72 if __name__ == "__main__":
     73     main()