switch to twitch.py and other improvements

2025-07-19 01:31:50 -07:00
parent 4102e8e672
commit 43e3f632cc
4 changed files with 139 additions and 129 deletions
@@ -1,3 +1,5 @@
 _*
 .venv
-swisseph
+ephe
+pip
+tokens.json
@@ -1,29 +1,26 @@
+#!/usr/bin/env python3.11
+
 from datetime import datetime,timezone
-import webbrowser
 from urllib.parse import quote_plus
 from urllib.request import urlretrieve
 from threading import Thread
 from os import makedirs,path
-import http.server
-import json
+import json,sys,webbrowser,importlib

-def download_files():
-    if not path.exists(script_path()+'swisseph/ephe/'):
-        makedirs(script_path()+'swisseph/ephe/')
-    #if not path.exists(script_path()+'swisseph/ephe/sefstars.txt'):
-    #    print('downloading sefstars.txt')
-    #    urlretrieve('https://github.com/aloistr/swisseph/raw/refs/heads/master/ephe/sefstars.txt',script_path()+'swisseph/ephe/sefstars.txt')
-    if not path.exists(script_path()+'swisseph/ephe/seas_18.se1'):
-        print('downloading seas_18.se1')
-        urlretrieve('https://github.com/aloistr/swisseph/raw/refs/heads/master/ephe/seas_18.se1',script_path()+'swisseph/ephe/seas_18.se1')
-    #if not path.exists(script_path()+'swisseph/ephe/semo_18.se1'):
-    #    print('downloadingh semo_18.se1')
-    #    urlretrieve('https://github.com/aloistr/swisseph/raw/refs/heads/master/ephe/semo_18.se1',script_path()+'swisseph/ephe/semo_18.se1')
-    #if not path.exists(script_path()+'swisseph/ephe/sepl_18.se1'):
-    #    print('downloading sepl_18.se1')
-    #    urlretrieve('https://github.com/aloistr/swisseph/raw/refs/heads/master/ephe/sepl_18.se1',script_path()+'swisseph/ephe/sepl_18.se1')
-    print('swiss ephemeris files ready')
+def dependency_setup():
+    print('checking dependencies')
+    from pip._internal.cli.main import main as pip
+    pip(['install','-qq','pyswisseph','twitch.py','--target',script_path()+'pip'])
+    sys.path.append(script_path()+'pip')

+    if not path.exists(script_path()+'ephe/seas_18.se1'):
+        print('downloading database')
+        makedirs(script_path()+'ephe')
+        urlretrieve('https://github.com/aloistr/swisseph/raw/refs/heads/master/ephe/seas_18.se1',script_path()+'ephe/seas_18.se1')
+
+    print('dependencies ready')
+
+# Astrology functions

 zodiac_signs = ["Aries", "Taurus", "Gemini", "Cancer", "Leo", "Virgo", 
    "Libra", "Scorpio", "Sagittarius", "Capricorn", "Aquarius", "Pisces"]
@@ -31,7 +28,7 @@ zodiac_signs = ["Aries", "Taurus", "Gemini", "Cancer", "Leo", "Virgo",
 def get_planet_info(date):
    import swisseph as swe
    # make sure location is set in THIS thread
-    swe.set_ephe_path(script_path()+'swisseph/ephe/')
+    swe.set_ephe_path(script_path()+'ephe/')
    swe.set_sid_mode(swe.SIDM_LAHIRI)
    # Get current UTC time
    julian_day = swe.julday(date.year, date.month, date.day)
@@ -92,21 +89,21 @@ def get_planet_info_formatted(date=datetime.utcnow()):
            case 0:
                info=''
            case 1:
-                info=planets['retrograde'][0]+' is in Retrograde'
+                info=planets['retrograde'][0]+' is in Retrograde\n'
            case 2:
-                info=' and '.join(planets['retrograde'])+' are in Retrograde'
+                info=' and '.join(planets['retrograde'])+' are in Retrograde\n'
            case _:
-                info=', '.join(planets['retrograde'][0:-1])+', and '+planets['retrograde'][-1]+' are in Retrograde'
+                info=', '.join(planets['retrograde'][0:-1])+', and '+planets['retrograde'][-1]+' are in Retrograde\n'

    positions_info=[]
    for planet,position in planets['positions'].items():
        positions_info.append(planet+' is in '+zodiac_signs[position])

-    info+='\n'+', '.join(positions_info[0:2])
-    info+='\n'+', '.join(positions_info[2:5])
-    info+='\n'+', '.join(positions_info[5:8])
-    info+='\n'+', '.join(positions_info[8:11])
-    info+='\n'+', '.join(positions_info[11:13])
+    info+=', '.join(positions_info[0:2])+'\n'
+    info+=', '.join(positions_info[2:5])+'\n'
+    info+=', '.join(positions_info[5:8])+'\n'
+    info+=', '.join(positions_info[8:11])+'\n'
+    info+=', '.join(positions_info[11:13])+'\n'
    return info

 def get_transits():
@@ -136,126 +133,138 @@ def get_transits_formatted():
    transits=get_transits()
    info=''
    for planet,transit in transits.items():
-        info+='\n'+planet+' is '+transit
+        info+=planet+' is '+transit+'\n'
    return info

-async def load_tokens():
-    global obs_settings
-    obs_tokens=obspython.obs_data_get_obj(obs_settings,'tokens')
-    obs_tokens=obspython.obs_data_get_json(obs_tokens)
-    if not obs_tokens:
-        return
-    obs_tokens=json.loads(obs_tokens)
-    for token in obs_tokens.values():
-        await client.add_token(token['access'],token['refresh'])
-
-async def save_tokens():
-    global obs_settings
-    tokens={}
-    for user_id in client.tokens:
-        token={
-            'access':client.tokens[user_id]['token'],
-            'refresh':client.tokens[user_id]['refresh'],
-        }
-        tokens[user_id]=token
-    obs_tokens=json.dumps(tokens)
-    obs_tokens=obspython.obs_data_create_from_json(obs_tokens)
-    obspython.obs_data_set_obj(obs_settings,'tokens',obs_tokens)
-
-async def setup_hook():
-    import twitchio
-    for user_id in client.tokens:
-        await client.subscribe_websocket(twitchio.eventsub.ChatMessageSubscription(broadcaster_user_id=user_id,user_id=user_id),token_for=user_id)
-        print('connected to channel '+user_id)
-    print('astrolobot setup complete')
-
-async def event_message(event):
-    if(event.text=='!astrology'):
-        info = get_planet_info_formatted()
-        for line in info.splitlines(False):
-            if line:
-                await event.broadcaster.send_message(line,event.broadcaster)
-    if(event.text=='!transits'):
-        info = get_transits_formatted()
-        for line in info.splitlines(False):
-            if line:
-                await event.broadcaster.send_message(line,event.broadcaster)
-
-async def event_oauth_authorized(event):
-    global client
-    client.add_token(event['access_token'],event['refresh_token'])
-    await setup_hook()
-
-def main():
-    import twitchio
-    global client, client_id, client_secret, adapter, scopes
-    #chdir(script_path())
-    scopes = twitchio.Scopes(['user:read:chat','user:write:chat','user:bot','channel:bot'])
-    adapter= twitchio.web.AiohttpAdapter(port=8523)
-    client = twitchio.Client(client_id=client_id,client_secret=client_secret,scopes=scopes,adapter=adapter)
-    client.event_message=event_message
-    client.event_oauth_authorized=event_oauth_authorized
-    client.setup_hook=setup_hook
-    client.save_tokens=save_tokens
-    client.load_tokens=load_tokens
-    client.run()
-
-twitchio_thread=Thread(target=main,daemon=True)
-
-def get_client_info(*args):
-    webbrowser.open('https://dev.twitch.tv/console/apps')
+# Twitch functions

 def login(*args):
-    global adapter, scopes
-    webbrowser.open(adapter.get_authorization_url(scopes=scopes),2,True)
+    global device_code
+    webbrowser.open('https://www.twitch.tv/activate?public=true&device-code='+device_code)
+
+def save_tokens(access_token,refresh_token):
+    with open(script_path()+'tokens.json','w') as file:
+        json.dump(
+            {
+                'access_token':access_token,
+                'refresh_token':refresh_token,
+            },
+            file
+        )
+
+def load_tokens():
+    try:
+        with open(script_path()+'tokens.json','r') as file:
+            return json.load(file)
+    except FileNotFoundError:
+        return {}
+
+def main():
+    from twitch import Client
+    from twitch.types import eventsub
+    from twitch.ext.oauth import DeviceAuthFlow, Scopes
+    global client
+    client = Client(client_id='h08yimv2stqxci85tpfh5k7t16an2u')
+    tokens=load_tokens().values()
+    DeviceAuthFlow(
+        client=client,
+        scopes=[
+            Scopes.CHANNEL_BOT,
+            Scopes.USER_BOT,
+            Scopes.USER_READ_CHAT,
+            Scopes.USER_WRITE_CHAT,
+        ],
+        wrap_run=len(tokens)==0
+    )
+
+    @client.event
+    async def on_code(code: str):
+        global device_code, obs_settings
+        device_code=code
+        if obs_settings == None:
+            login()
+
+    @client.event
+    async def on_auth(access_token: str, refresh_token: str):
+        save_tokens(access_token,refresh_token)
+
+    @client.event
+    async def on_ready() -> None:
+        print('astrolobot ready')
+
+    @client.event
+    async def on_chat_message(data: eventsub.chat.MessageEvent):
+        if data['message']['text']=='!astrology':
+            for line in get_planet_info_formatted().splitlines(False):
+                await client.channel.chat.send_message(line,data['message_id'])
+            return
+        if data['message']['text']=='!transits':
+            for line in get_transits_formatted().splitlines(False):
+                await client.channel.chat.send_message(line,data['message_id'])
+
+    client.run(*tokens)
+
+# OBS functions
+
+obs_settings=None
+
+def script_main():
+    dependency_setup()
+    main()
+    
+def obs_load_tokens():
+    import obspython
+    access_token=obspython.obs_data_get_string(obs_settings,'access_token')
+    refresh_token=obspython.obs_data_get_string(obs_settings,'refresh_token')
+    if not (access_token or refresh_token):
+        print('astrolobot: login required',file=sys.stderr)
+        return {}
+    else:
+        return {
+            'access_token':access_token,
+            'refresh_token':refresh_token,
+        }
+
+def obs_save_tokens(access_token,refresh_token):
+    import obspython
+    obspython.obs_data_set_string(obs_settings,'access_token',access_token)
+    obspython.obs_data_set_string(obs_settings,'refresh_token',refresh_token)

 # https://docs.obsproject.com/scripting#script-function-exports

 def script_description():
-    return 'adds the !astrology command to your twitch channel, providing the current star signs of the planets and whether they are in retrograde'
+    return 'adds the !astrology and !transits commands to your twitch channel, providing the current and upcoming star signs and retrograde status of the planets'

 def script_properties():
    import obspython
    properties=obspython.obs_properties_create()
-    obspython.obs_properties_set_flags(properties,obspython.OBS_PROPERTIES_DEFER_UPDATE)
-    obspython.obs_properties_add_text(properties,'client_id','Client ID',obspython.OBS_TEXT_PASSWORD)
-    obspython.obs_properties_add_text(properties,'client_secret','Client Secret',obspython.OBS_TEXT_PASSWORD)
-    obspython.obs_properties_add_button(properties,'client_info','Find Client Info',get_client_info)
    obspython.obs_properties_add_button(properties,'login','Login to Twitch',login)
    return properties

 def script_load(settings):
-    import sys
-    global obs_settings
-    global save
-
-    sys.path.append(script_path())
-
-    try:
-        import twitchio, swisseph
-    except:
-        import pip
-        pip.main(['install','-qqq','pyswisseph','twitchio','twitchio[starlette]','--target',script_path()])
-
-    download_files()
-
+    global obs_settings, save_tokens, load_tokens
    obs_settings=settings
+    save_tokens=obs_save_tokens
+    load_tokens=obs_load_tokens
+    def isatty():
+        return False
+    sys.stdout.isatty=isatty
+    Thread(target=script_main,daemon=True).start()

 def script_unload():
-    global client
-    import asyncio
-    asyncio.run(client.close())
+    try:
+        import asyncio
+        global client
+        asyncio.run(client.close())
+    except NameError or ImportError:
+        pass

-def script_update(settings):
-    global client_id,client_secret
-    client_id=obspython.obs_data_get_string(settings,'client_id')
-    client_secret=obspython.obs_data_get_string(settings,'client_secret')
-    if client_id and client_secret:
-        twitchio_thread.start()
+# test functions

 if __name__ == '__main__':
    def script_path():
        return path.dirname(__file__)+'/'
-    download_files()
+    dependency_setup()
    print(get_planet_info_formatted())
-    print(get_transits_formatted())
+    print(get_transits_formatted())
+    main()
@@ -1,2 +1,2 @@
 pyswisseph
-twitchio
+twitch.py