78 lines
2.8 KiB
Python
78 lines
2.8 KiB
Python
#Kemoverse - a gacha-style bot for the Fediverse.
|
|
#Copyright © 2025 Waifu
|
|
#
|
|
#This program is free software: you can redistribute it and/or modify
|
|
#it under the terms of the GNU Affero General Public License as
|
|
#published by the Free Software Foundation, either version 3 of the
|
|
#License, or (at your option) any later version.
|
|
#
|
|
#This program is distributed in the hope that it will be useful,
|
|
#but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
#MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
#GNU Affero General Public License for more details.
|
|
#
|
|
#You should have received a copy of the GNU Affero General Public License
|
|
#along with this program. If not, see https://www.gnu.org/licenses/.
|
|
|
|
import time
|
|
import traceback
|
|
import config
|
|
from notification import process_fediverse_notification
|
|
from db_utils import get_config, set_config, connect, setup_administrators
|
|
from fediverse_factory import get_fediverse_service
|
|
|
|
from config import USE_WHITELIST
|
|
|
|
def stream_notifications():
|
|
# Initialize database connection
|
|
connect()
|
|
|
|
# Setup default administrators
|
|
setup_administrators()
|
|
|
|
# Initialize the Fediverse service
|
|
fediverse_service = get_fediverse_service(config.INSTANCE_TYPE)
|
|
|
|
# Get the last seen notification ID from the database
|
|
last_seen_id = get_config("last_seen_notif_id")
|
|
|
|
# Show whitelist status
|
|
whitelist_status = "enabled" if USE_WHITELIST else "disabled"
|
|
print(f'Instance whitelisting: {whitelist_status}')
|
|
|
|
print('Listening for notifications...')
|
|
while True:
|
|
try:
|
|
# Get notifications from the fediverse service
|
|
notifications = fediverse_service.get_notifications(since_id=last_seen_id)
|
|
|
|
if notifications:
|
|
new_last_seen_id = last_seen_id
|
|
|
|
for notification in notifications:
|
|
notif_id = notification.id
|
|
|
|
# Skip old or same ID notifications
|
|
if last_seen_id is not None and notif_id <= last_seen_id:
|
|
continue
|
|
|
|
# Process the notification using the abstracted processor
|
|
process_fediverse_notification(notification, fediverse_service)
|
|
|
|
# Update only if this notif_id is greater
|
|
if new_last_seen_id is None or notif_id > new_last_seen_id:
|
|
new_last_seen_id = notif_id
|
|
|
|
# Save the latest seen ID
|
|
if new_last_seen_id and new_last_seen_id != last_seen_id:
|
|
set_config("last_seen_notif_id", new_last_seen_id)
|
|
last_seen_id = new_last_seen_id
|
|
|
|
time.sleep(5)
|
|
|
|
except Exception as e:
|
|
print(f"An exception has occured: {e}\n{traceback.format_exc()}")
|
|
time.sleep(5)
|
|
|
|
if __name__ == "__main__":
|
|
stream_notifications()
|