From 43e3f632ccdb1d53a95a8b39e055dbb1fb4b664e Mon Sep 17 00:00:00 2001 From: sugoidogo Date: Sat, 19 Jul 2025 01:31:50 -0700 Subject: [PATCH] switch to twitch.py and other improvements --- .gitignore | 4 +- astrolobot.py | 261 ++++++++++++++++++++++++----------------------- requirements.txt | 2 +- swisseph | 1 - 4 files changed, 139 insertions(+), 129 deletions(-) mode change 100644 => 100755 astrolobot.py delete mode 160000 swisseph diff --git a/.gitignore b/.gitignore index ed376b0..92bcc8a 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,5 @@ _* .venv -swisseph \ No newline at end of file +ephe +pip +tokens.json \ No newline at end of file diff --git a/astrolobot.py b/astrolobot.py old mode 100644 new mode 100755 index 35ca4fd..866b0df --- a/astrolobot.py +++ b/astrolobot.py @@ -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()) \ No newline at end of file + print(get_transits_formatted()) + main() \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index 6f0bf08..c002e48 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,2 +1,2 @@ pyswisseph -twitchio \ No newline at end of file +twitch.py \ No newline at end of file diff --git a/swisseph b/swisseph deleted file mode 160000 index d8c10a9..0000000 --- a/swisseph +++ /dev/null @@ -1 +0,0 @@ -Subproject commit d8c10a91ffd60282d266b091f2d59e57e7221c05