<filename>examples/synchronous.py
import time
import requests.sessions
from typing import Optional, List
from Weverse import WeverseClientSync, InvalidToken, BeingRateLimited, models
from dotenv import load_dotenv
from os import getenv


"""
synchronous.py


Synchronous Examples for Weverse.

This file should be able to run if a token is supplied.
If you would like to view more specific information,
then it would be suggested to look at the API Docs at https://weverse.readthedocs.io/en/latest/ 
or run this file in a debugger.

In order to get a token: https://weverse.readthedocs.io/en/latest/api.html#get-account-token
In order to set a token, Rename .env.example to .env and set WEVERSE_AUTH to the token (recommended). 
You may also set the variable weverse_token to the token (not recommended).

There is also now the availability to login using a usernane and password instead of a token.
This file will display both auth methods.

If you are running in an Asynchronous environment, go take a look at the Asynchronous examples file.
"""


if __name__ == '__main__':
    ...


def get_formatted_time(seconds):
    """Turn seconds into days, hours, minutes, and seconds.

    Not related to Weverse.
    :param seconds: Amount of seconds to convert.
    """
    seconds = round(seconds)
    minute, sec = divmod(seconds, 60)
    hour, minute = divmod(minute, 60)
    day, hour = divmod(hour, 24)

    return f"{f'{day}d ' if day else ''}" \
           f"{f'{hour}h ' if hour else ''}" \
           f"{f'{minute}m ' if minute else ''}" \
           f"{f'{sec}s' if sec else ''}" \
           f"{f'0s' if seconds < 1 else ''}"


load_dotenv()  # loads up .env in memory

weverse_token = getenv("WEVERSE_AUTH") or "fake_token"

web_session = requests.sessions.session()  # create a web session for Weverse to use (Not needed, but recommended)


def on_new_notifications(notifications: List[models.Notification]):
    """
    Decides what to do with a list of Notifications.
    This is a hook method. Whenever there is a notification, this method will be called.
    """
    for notification in notifications:
        handle_notification(notification)


weverse_username = getenv("WEVERSE_USERNAME") or None
weverse_password = getenv("WEVERSE_PASSWORD") or None

client_kwargs = {
    "verbose": True,  # Will print warning messages for links that have failed to connect or were not found.
    "web_session": web_session,  # Existing web session
    "authorization": weverse_token,  # Auth Token to connect to Weverse. Not needed if you are using the login method.
    "username": weverse_username,  # Username if you are using the login method.
    "password": <PASSWORD>,  # Password if you are using the login method.
    "hook": on_new_notifications  # The method that will receive new notifications.
}

weverse_client = WeverseClientSync(**client_kwargs)


# start the client process.
start_kwargs = {
    # Will load ALL followed community posts. If following majority communities, this will take more than 10 minutes
    # to complete. If you are not in need of OLDER posts, you can set it to False and it should load within a few
    # seconds.
    "create_old_posts": True,
    # Will load up the past 20 notifications. This is a quick process, so you may as well load them up.
    "create_notifications": True,
    # Will load ALL media posts. This will take a very long time if enabled.
    "create_media": True
}

# You can actually check if the token works before starting the cache process and receiving an exception.
# Do not check if you are using a regular login.
# token_works = weverse_client.check_token_works()


try:
    print("Weverse cache is currently being created.")
    start_time = time.time()  # Start Time for Weverse Cache
    weverse_client.start(**start_kwargs)  # start the client process to all needed data.
    end_time = time.time()  # The time when Weverse Cache has loaded.
    print(f"Weverse cache has loaded after {get_formatted_time(end_time - start_time)}")
except InvalidToken:
    raise RuntimeError("An Invalid Token was passed in")
except BeingRateLimited:
    raise RuntimeError("Weverse Client is being rate-limited.")  # Note that this issue has never occurred yet.
except Exception as e:
    # No other specific exceptions are raised, But we shouldn't ignore any errors.
    raise RuntimeError(f"A different error has occurred. {e}")

# We now have all of the data we would want to modify.


def loop_notifications():
    """Check for notifications on a loop.

    THIS IS THE MANUAL WAY TO CHECK FOR NOTIFICATIONS.
    IT IS RECOMMENDED TO PASS IN A HOOK TO THE CLIENT INSTEAD.

    """
    continue_loop = True
    while continue_loop:
        # This is a loop running to check for notifications.
        new_notifications = check_new_notifications()
        if not new_notifications:
            continue

        for new_notification in new_notifications:
            handle_notification(new_notification)
        continue_loop = False  # no longer want to check for notifications.


# loop the notifications
if not weverse_client._hook:  # we are just checking to see if we passed in a hook method
    loop_notifications()


def check_new_notifications() -> Optional[List[models.Notification]]:
    """Check for new weverse notifications.

    This should be called in a loop to constantly check for notifications.
    """
    if not weverse_client.cache_loaded:
        # This is more of an issue with async, but we should always make sure the cache is loaded.
        return

    if not weverse_client.check_new_user_notifications():
        # This will check if there are new notifications and will process them into objects for us.
        return

    # They should now be loaded into notification cache, but just to confirm nothing went wrong, we will check.
    if not weverse_client.user_notifications:
        return

    # We can now get the latest notifications
    # get_new_notifications is new in V1.0.3. Separates new notifications from older ones.
    latest_notifications = weverse_client.get_new_notifications()
    return latest_notifications


def handle_notification(new_notification: models.Notification):
    # Two ways to determine the community name.
    community_name = new_notification.community_name or new_notification.bold_element
    if not community_name:
        # In the case a notification faults from Weverse, this is a safety measure.
        return

    # We do not know the type of notification it is because it is usually hidden inside the message.
    # The wrapper does not automatically check for this, so we must call the method.
    notification_type = weverse_client.determine_notification_type(new_notification.message)

    possible_notification_types = ["comment", "post", "media", "announcement"]

    # The content ID is the ID of the actual post we are looking for.
    content_id = new_notification.contents_id

    # In the below conditions, you should do any logic needed for knowing the types...
    if notification_type == possible_notification_types[0]:  # comment
        comment = weverse_client.get_comment_by_id(content_id)
        ...  # Do stuff with the comment here
    elif notification_type == possible_notification_types[1]:  # post
        post = weverse_client.get_post_by_id(content_id)
        ...  # Do stuff with the post here
    elif notification_type == possible_notification_types[2]:  # media
        media = weverse_client.get_media_by_id(content_id)
        ...  # Do stuff with the media here
    elif notification_type == possible_notification_types[3]:  # announcement
        announcement = weverse_client.get_announcement_by_id(content_id)
        ...  # Do stuff with the announcement here
    else:  # No Type Found (Perhaps a new type was created but wrapper is not updated)
        print(f"{new_notification.id} has an unrecognized type.")

    ...  # do more stuff if you want to.





<filename>examples/synchronous.py
import time
import requests.sessions
from typing import Optional, List
from Weverse import WeverseClientSync, InvalidToken, BeingRateLimited, models
from dotenv import load_dotenv
from os import getenv


"""
synchronous.py


Synchronous Examples for Weverse.

This file should be able to run if a token is supplied.
If you would like to view more specific information,
then it would be suggested to look at the API Docs at https://weverse.readthedocs.io/en/latest/ 
or run this file in a debugger.

In order to get a token: https://weverse.readthedocs.io/en/latest/api.html#get-account-token
In order to set a token, Rename .env.example to .env and set WEVERSE_AUTH to the token (recommended). 
You may also set the variable weverse_token to the token (not recommended).

There is also now the availability to login using a usernane and password instead of a token.
This file will display both auth methods.

If you are running in an Asynchronous environment, go take a look at the Asynchronous examples file.
"""


if __name__ == '__main__':
    ...


def get_formatted_time(seconds):
    """Turn seconds into days, hours, minutes, and seconds.

    Not related to Weverse.
    :param seconds: Amount of seconds to convert.
    """
    seconds = round(seconds)
    minute, sec = divmod(seconds, 60)
    hour, minute = divmod(minute, 60)
    day, hour = divmod(hour, 24)

    return f"{f'{day}d ' if day else ''}" \
           f"{f'{hour}h ' if hour else ''}" \
           f"{f'{minute}m ' if minute else ''}" \
           f"{f'{sec}s' if sec else ''}" \
           f"{f'0s' if seconds < 1 else ''}"


load_dotenv()  # loads up .env in memory

weverse_token = getenv("WEVERSE_AUTH") or "fake_token"

web_session = requests.sessions.session()  # create a web session for Weverse to use (Not needed, but recommended)


def on_new_notifications(notifications: List[models.Notification]):
    """
    Decides what to do with a list of Notifications.
    This is a hook method. Whenever there is a notification, this method will be called.
    """
    for notification in notifications:
        handle_notification(notification)


weverse_username = getenv("WEVERSE_USERNAME") or None
weverse_password = getenv("WEVERSE_PASSWORD") or None

client_kwargs = {
    "verbose": True,  # Will print warning messages for links that have failed to connect or were not found.
    "web_session": web_session,  # Existing web session
    "authorization": weverse_token,  # Auth Token to connect to Weverse. Not needed if you are using the login method.
    "username": weverse_username,  # Username if you are using the login method.
    "password": <PASSWORD>,  # Password if you are using the login method.
    "hook": on_new_notifications  # The method that will receive new notifications.
}

weverse_client = WeverseClientSync(**client_kwargs)


# start the client process.
start_kwargs = {
    # Will load ALL followed community posts. If following majority communities, this will take more than 10 minutes
    # to complete. If you are not in need of OLDER posts, you can set it to False and it should load within a few
    # seconds.
    "create_old_posts": True,
    # Will load up the past 20 notifications. This is a quick process, so you may as well load them up.
    "create_notifications": True,
    # Will load ALL media posts. This will take a very long time if enabled.
    "create_media": True
}

# You can actually check if the token works before starting the cache process and receiving an exception.
# Do not check if you are using a regular login.
# token_works = weverse_client.check_token_works()


try:
    print("Weverse cache is currently being created.")
    start_time = time.time()  # Start Time for Weverse Cache
    weverse_client.start(**start_kwargs)  # start the client process to all needed data.
    end_time = time.time()  # The time when Weverse Cache has loaded.
    print(f"Weverse cache has loaded after {get_formatted_time(end_time - start_time)}")
except InvalidToken:
    raise RuntimeError("An Invalid Token was passed in")
except BeingRateLimited:
    raise RuntimeError("Weverse Client is being rate-limited.")  # Note that this issue has never occurred yet.
except Exception as e:
    # No other specific exceptions are raised, But we shouldn't ignore any errors.
    raise RuntimeError(f"A different error has occurred. {e}")

# We now have all of the data we would want to modify.


def loop_notifications():
    """Check for notifications on a loop.

    THIS IS THE MANUAL WAY TO CHECK FOR NOTIFICATIONS.
    IT IS RECOMMENDED TO PASS IN A HOOK TO THE CLIENT INSTEAD.

    """
    continue_loop = True
    while continue_loop:
        # This is a loop running to check for notifications.
        new_notifications = check_new_notifications()
        if not new_notifications:
            continue

        for new_notification in new_notifications:
            handle_notification(new_notification)
        continue_loop = False  # no longer want to check for notifications.


# loop the notifications
if not weverse_client._hook:  # we are just checking to see if we passed in a hook method
    loop_notifications()


def check_new_notifications() -> Optional[List[models.Notification]]:
    """Check for new weverse notifications.

    This should be called in a loop to constantly check for notifications.
    """
    if not weverse_client.cache_loaded:
        # This is more of an issue with async, but we should always make sure the cache is loaded.
        return

    if not weverse_client.check_new_user_notifications():
        # This will check if there are new notifications and will process them into objects for us.
        return

    # They should now be loaded into notification cache, but just to confirm nothing went wrong, we will check.
    if not weverse_client.user_notifications:
        return

    # We can now get the latest notifications
    # get_new_notifications is new in V1.0.3. Separates new notifications from older ones.
    latest_notifications = weverse_client.get_new_notifications()
    return latest_notifications


def handle_notification(new_notification: models.Notification):
    # Two ways to determine the community name.
    community_name = new_notification.community_name or new_notification.bold_element
    if not community_name:
        # In the case a notification faults from Weverse, this is a safety measure.
        return

    # We do not know the type of notification it is because it is usually hidden inside the message.
    # The wrapper does not automatically check for this, so we must call the method.
    notification_type = weverse_client.determine_notification_type(new_notification.message)

    possible_notification_types = ["comment", "post", "media", "announcement"]

    # The content ID is the ID of the actual post we are looking for.
    content_id = new_notification.contents_id

    # In the below conditions, you should do any logic needed for knowing the types...
    if notification_type == possible_notification_types[0]:  # comment
        comment = weverse_client.get_comment_by_id(content_id)
        ...  # Do stuff with the comment here
    elif notification_type == possible_notification_types[1]:  # post
        post = weverse_client.get_post_by_id(content_id)
        ...  # Do stuff with the post here
    elif notification_type == possible_notification_types[2]:  # media
        media = weverse_client.get_media_by_id(content_id)
        ...  # Do stuff with the media here
    elif notification_type == possible_notification_types[3]:  # announcement
        announcement = weverse_client.get_announcement_by_id(content_id)
        ...  # Do stuff with the announcement here
    else:  # No Type Found (Perhaps a new type was created but wrapper is not updated)
        print(f"{new_notification.id} has an unrecognized type.")

    ...  # do more stuff if you want to.





synchronous.py   Synchronous Examples for Weverse.  This file should be able to run if a token is supplied. If you would like to view more specific information, then it would be suggested to look at the API Docs at https://weverse.readthedocs.io/en/latest/  or run this file in a debugger.  In order to get a token: https://weverse.readthedocs.io/en/latest/api.html#get-account-token In order to set a token, Rename .env.example to .env and set WEVERSE_AUTH to the token (recommended).  You may also set the variable weverse_token to the token (not recommended).  There is also now the availability to login using a usernane and password instead of a token. This file will display both auth methods.  If you are running in an Asynchronous environment, go take a look at the Asynchronous examples file. Turn seconds into days, hours, minutes, and seconds.      Not related to Weverse.     :param seconds: Amount of seconds to convert. # loads up .env in memory # create a web session for Weverse to use (Not needed, but recommended) Decides what to do with a list of Notifications.     This is a hook method. Whenever there is a notification, this method will be called. # Will print warning messages for links that have failed to connect or were not found. # Existing web session # Auth Token to connect to Weverse. Not needed if you are using the login method. # Username if you are using the login method. # Password if you are using the login method. # The method that will receive new notifications. # start the client process. # Will load ALL followed community posts. If following majority communities, this will take more than 10 minutes # to complete. If you are not in need of OLDER posts, you can set it to False and it should load within a few # seconds. # Will load up the past 20 notifications. This is a quick process, so you may as well load them up. # Will load ALL media posts. This will take a very long time if enabled. # You can actually check if the token works before starting the cache process and receiving an exception. # Do not check if you are using a regular login. # token_works = weverse_client.check_token_works() # Start Time for Weverse Cache # start the client process to all needed data. # The time when Weverse Cache has loaded. # Note that this issue has never occurred yet. # No other specific exceptions are raised, But we shouldn't ignore any errors. # We now have all of the data we would want to modify. Check for notifications on a loop.      THIS IS THE MANUAL WAY TO CHECK FOR NOTIFICATIONS.     IT IS RECOMMENDED TO PASS IN A HOOK TO THE CLIENT INSTEAD. # This is a loop running to check for notifications. # no longer want to check for notifications. # loop the notifications # we are just checking to see if we passed in a hook method Check for new weverse notifications.      This should be called in a loop to constantly check for notifications. # This is more of an issue with async, but we should always make sure the cache is loaded. # This will check if there are new notifications and will process them into objects for us. # They should now be loaded into notification cache, but just to confirm nothing went wrong, we will check. # We can now get the latest notifications # get_new_notifications is new in V1.0.3. Separates new notifications from older ones. # Two ways to determine the community name. # In the case a notification faults from Weverse, this is a safety measure. # We do not know the type of notification it is because it is usually hidden inside the message. # The wrapper does not automatically check for this, so we must call the method. # The content ID is the ID of the actual post we are looking for. # In the below conditions, you should do any logic needed for knowing the types... # comment # Do stuff with the comment here # post # Do stuff with the post here # media # Do stuff with the media here # announcement # Do stuff with the announcement here # No Type Found (Perhaps a new type was created but wrapper is not updated) # do more stuff if you want to.