jubilant-system/hooks.py

"""
for list of hooks refer to doc.txt, search by "hooks (don't mismatch with DB hooks)"
common structure:
1. Exists notify function for every hook type that calls from appropriate places
2. Notify function call every appropriate hook

Should we have separate triggers for update and insert cases?
"""
import json
import sqlite3
import typing

import sql_requests
import utils
from EDMCLogging import get_main_logger

logger = get_main_logger()

properly_delete_hooks: list[typing.Callable] = list()
insert_data_hooks: list[typing.Callable] = list()

SPECIAL_SQUADRONS: list = list()

try:

    with open('SPECIAL_SQUADRONS.txt', mode='r', encoding='utf-8') as spec_squads_file:
        for line in spec_squads_file.readlines():
            SPECIAL_SQUADRONS.append(int(line.strip()))

except FileNotFoundError:
    pass

logger.debug(f'Specials squadrons: {json.dumps(SPECIAL_SQUADRONS)}')


def notify_properly_delete(squad_id: int, db_conn: sqlite3.Connection) -> None:
    """Notifies all properly delete hooks, calls before deleting

    :param squad_id:
    :param db_conn:
    :return:
    """
    # logger.debug(f'Notifying properly delete squad hooks for {squad_id} ID')

    for hook in properly_delete_hooks:
        hook(squad_id, db_conn)


def notify_insert_data(squad_info: dict, db_conn: sqlite3.Connection) -> None:
    """Notifies all insert data hooks, calls after inserting

    :param squad_info:
    :param db_conn:
    :return:
    """

    # logger.debug(f'Notifying insert data hooks for {squad_info["id"]} ID')
    for hook in insert_data_hooks:
        hook(squad_info, db_conn)


def detect_new_ru_squads(squad_info: dict, db_conn: sqlite3.Connection) -> None:
    """Sends alert if it was firstly discovered ru squad with >5 members

    :param squad_info:
    :param db_conn:
    :return:
    """

    """
    Determine if we just discover squad, not updating it
    check if is 32 in tag (means russian squad)
    check if members count satisfies low threshold
    send alert  
    """
    MEMBERS_LOW_THRESHOLD: int = 1

    if db_conn.execute(sql_requests.check_if_squad_exists_in_db, (squad_info['id'],)).fetchone()[0] == 1:
        # squad has only one record, it means squad just discovered

        if 32 in squad_info['userTags']:
            # it's russian squad

            if squad_info['memberCount'] > MEMBERS_LOW_THRESHOLD:

                if len(squad_info['motd']) == 0:
                    motd = ''

                else:
                    motd = f"`{squad_info['motd']}`"
                previous_season_sum: int = squad_info['previous_season_trade_score'] + \
                    squad_info['previous_season_combat_score'] + \
                    squad_info['previous_season_exploration_score'] + \
                    squad_info['previous_season_cqc_score'] + \
                    squad_info['previous_season_bgs_score'] + \
                    squad_info['previous_season_powerplay_score'] + \
                    squad_info['previous_season_aegis_score']
                current_season_sum: int = squad_info['current_season_trade_score'] + \
                    squad_info['current_season_combat_score'] + \
                    squad_info['current_season_exploration_score'] + \
                    squad_info['current_season_cqc_score'] + \
                    squad_info['current_season_bgs_score'] + \
                    squad_info['current_season_powerplay_score'] + \
                    squad_info['current_season_aegis_score']
                message = f"""
New russian squad with more then {MEMBERS_LOW_THRESHOLD} members: {squad_info['name']}
members: {squad_info['memberCount']}
tag: {squad_info['tag']}
created: {squad_info['created']}
platform: {squad_info['platform']}
owner: {squad_info['ownerName']}
tags:\n{utils.humanify_resolved_user_tags(utils.resolve_user_tags(squad_info['userTags']))}
activity:
    previous season sum: {previous_season_sum}
    current season sum: {current_season_sum}
motd: {motd}"""

                utils.notify_discord(message)


def detect_important_changes_ru_squads(squad_info: dict, db_conn: sqlite3.Connection) -> None:
    """Alert if something important changed for a russian squad

    :param squad_info: FDEV authored dict
    :param db_conn:
    :return:
    """

    """
    how it should works?
    Works only with updated (not firstly inserted) squads
    1. Detect if squad just set/took ru tag
    2. Detect if squad changed their members count
    3. Detect if squad changed their motd
    4. Detect if squad changed their ownership
    5. Detect if squad changed their minor faction
    6. Tags changes
    7. Notify discord
    """

    squad_id = squad_info['id']
    if db_conn.execute(sql_requests.check_if_squad_exists_in_db, (squad_id,)).fetchone()[0] == 1:
        # we just discover squad, not updating
        return

    new_tags_raw, old_tags_raw = new_old_diff('user_tags', db_conn, squad_id)
    new_tags: list = json.loads(new_tags_raw)
    old_tags: list = json.loads(old_tags_raw)

    message: str = str()
    message_start = str()

    # detect if squad should be observed as specially stated, should be done in separate function,
    # but I don't feel be able to write a new function today
    if squad_id in SPECIAL_SQUADRONS:
        isImportant = True
        message_start += 'Special squadron\n'

    else:
        isImportant = False

    # let's find out situation with RU tag
    if 32 not in new_tags and 32 not in old_tags:
        # squadron wasn't russian, isn't russian
        if not isImportant:
            return

        else:
            message_start += "Squadron isn't russian\n"

    elif 32 in new_tags and 32 in old_tags:
        # squadron was russian, is russian
        pass

    elif 32 in new_tags and 32 not in old_tags:
        # squadron wasn't russian and is russian
        message = message + 'Squadron become russian\n'

    elif 32 not in new_tags and 32 in old_tags:
        # squadron was russian, isn't russian
        message = message + 'Squadron stop being russian\n'

    # let's clarify situation with members count
    new_members, old_members = new_old_diff('member_count', db_conn, squad_id)

    if new_members == old_members:
        # count wasn't changed
        pass

    else:
        message = message + f'Members count changed {old_members} -> {new_members}\n'

    # let's check motd
    new_motd, old_motd = new_old_news_diff('motd', db_conn, squad_id)  # TODO: make easier and better

    if new_motd == old_motd:
        # motd wasn't changed
        pass

    else:
        message = message + f'Motd changed, old:\n```\n{old_motd}\n```\nnew:\n```\n{new_motd}\n```\n'

    # let's check ownership
    new_owner, old_owner = new_old_diff('owner_name', db_conn, squad_id)

    if new_owner == old_owner:
        # the same owner
        pass

    else:
        message = message + f'Ownership changed: {old_owner} -> {new_owner}\n'

    # let's check minor faction
    new_faction, old_faction = new_old_diff('faction_name', db_conn, squad_id)

    if new_faction == old_faction:
        # the same faction
        pass

    else:
        message = message + f'Minor faction changed: {old_faction} -> {new_faction}\n'

    # let's check tags changes
    if new_tags == old_tags:
        # nothing changed
        pass

    else:
        message += f"```diff\n{tags_diff2str(new_tags, old_tags)}```"

    if len(message) != 0:
        message = message_start + message
        utils.notify_discord(f'State changing for RU squad `{squad_info["name"]}` {squad_info["tag"]}\n'
                             f'platform: {squad_info["platform"]}\nmembers: {squad_info["memberCount"]}\n'
                             f'created: {squad_info["created"]}\nowner: {squad_info["ownerName"]}\n' + message)

    return


def detect_removing_ru_squads(squad_id: int, db_conn: sqlite3.Connection):
    """Send alert to discord if was removed russian squad

    :param squad_id:
    :param db_conn:
    :return:
    """

    """
    Detect if removing squad was russian
    send alert 
    """
    important = db_conn.execute(sql_requests.select_important_before_delete, (squad_id,)).fetchone()

    if important is None:
        return  # squad doesn't exists in our DB so we can't figure out anything about it

    name = important[0]
    platform = important[1]
    members = important[2]
    tag = important[3]
    user_tags = json.loads(important[4])
    created = important[5]
    owner_name = important[6]

    if 32 in user_tags:
        # ru squad
        message = f'Deleted RU squad `{name}`\nplatform: {platform}, members: {members}, tag: {tag}, ' \
                  f'created: {created}, owner: `{owner_name}`'
        utils.notify_discord(message)

    return


# should I move it to utils?
def new_old_diff(column: str, db_conn: sqlite3.Connection, squad_id: int) -> typing.Tuple[typing.Any, typing.Any]:
    """Returns last two record for squad_id for specified column

    :param column:
    :param db_conn:
    :param squad_id:
    :return:
    """
    sql_req: sqlite3.Cursor = db_conn.execute(sql_requests.select_old_new.format(column=column), (squad_id,))

    new = sql_req.fetchone()[0]
    old = sql_req.fetchone()[0]

    return new, old


def new_old_news_diff(column: str, db_conn: sqlite3.Connection, squad_id: int) -> typing.Tuple[typing.Any, typing.Any]:
    """Returns last two record for squad_id for specified column

    :param column:
    :param db_conn:
    :param squad_id:
    :return:
    """
    sql_req: sqlite3.Cursor = db_conn.execute(sql_requests.select_new_old_news.format(column=column), (squad_id,))

    try:
        new = sql_req.fetchone()[0]

    except TypeError:
        new = None

    try:
        old = sql_req.fetchone()[0]

    except TypeError:
        old = None

    return new, old


def tags_diff2str(new_tags_ids: list, old_tags_ids: list) -> str:
    """Compares two list of tags, new and old, and returns it in diff like str

    :param new_tags_ids: list ids of new tags
    :param old_tags_ids: list ids of old tags
    :return: diff like str
    """

    resolved_tags: dict[str, list[str]] = dict()

    removed_tags_ids: list = list(set(old_tags_ids) - set(new_tags_ids))
    added_tags_ids: list = list(set(new_tags_ids) - set(old_tags_ids))

    tags_union_ids: list = list(set(new_tags_ids).union(set(old_tags_ids)))

    for tag_id in tags_union_ids:
        collection_name, tag_name = utils.resolve_user_tag(tag_id)

        if tag_id in removed_tags_ids:
            resolved_tags = utils.append_to_list_in_dict(resolved_tags, collection_name, f'-   {tag_name}')

        elif tag_id in added_tags_ids:
            resolved_tags = utils.append_to_list_in_dict(resolved_tags, collection_name, f'+   {tag_name}')

        else:  # tag_id not in added_tags_ids and not in removed_tags_ids - nothing changed
            resolved_tags = utils.append_to_list_in_dict(resolved_tags, collection_name, f'    {tag_name}')

    return utils.humanify_resolved_user_tags(resolved_tags, do_tabulate=False)


insert_data_hooks.append(detect_new_ru_squads)
insert_data_hooks.append(detect_important_changes_ru_squads)
properly_delete_hooks.append(detect_removing_ru_squads)