entry_reports.py

# type: ignore (for mirror)

import time
from datetime import UTC, datetime
from typing import Literal

import requests
from requests import HTTPError
from vdbpy.api.edits import get_edits_by_entry, get_most_recent_edit_by_user_id
from vdbpy.api.entries import (
    get_cached_entry_version,
    get_entry_from_link,
    get_entry_link,
)
from vdbpy.api.messaging import send_message
from vdbpy.api.users import (
    find_user_by_username_1d,
    get_cached_username_by_id_1d,
    get_username_by_id,
)
from vdbpy.config import WEBSITE
from vdbpy.types.shared import EntryTuple, EntryType
from vdbpy.utils.console import get_boolean_or_none
from vdbpy.utils.data import add_s
from vdbpy.utils.date import parse_date
from vdbpy.utils.files import (
    get_lines,
    remove_line_from_file,
    save_file,
)
from vdbpy.utils.logger import get_logger
from vdbpy.utils.network import fetch_json

from mod_types import (
    EntryReport,
    EntryWithVersionIds,
    MikuModReports,
    ReportResult,
    ReportType,
    RuleModules,
)

logger = get_logger()
REPORT_URL = f"{WEBSITE}/api/admin/reports"
WIKI_RULES_URL = "https://wiki.vocadb.net/rules"


def get_open_entry_reports(session: requests.Session) -> list[EntryReport]:
    reports: list[EntryReport] = []
    url = f"{REPORT_URL}?status=Open"
    report_request = fetch_json(session=session, url=url)
    for r in report_request:
        report = EntryReport(
            report_id=r["id"],
            entry_type=r["entry"]["entryType"],
            entry_id=r["entry"]["id"],
            date=parse_date(r["created"]),
            report_type=r["reportTypeName"],
            notes=r["notes"],
            author=r["user"]["name"] if "user" in r else "anon",
        )
        reports.append(report)
    return reports


def create_entry_report(
    session: requests.Session,
    entry_type: EntryType,
    entry_id: int,
    version_number: int,
    report_msg: str,
    report_type: ReportType = "Other",
) -> bool:
    report_data = {
        f"{entry_type.lower()}Id": entry_id,
        "versionNumber": version_number,
        "reportType": report_type,
        "notes": report_msg,
    }

    if entry_type == "Tag":
        report_url = (
            f"{WEBSITE}/api/{add_s(entry_type)}/{entry_id}/reports?"
            f"reportType={report_type}&notes={report_msg}&versionNumber={version_number}"
        )
        report_data = {}
    else:
        report_url = f"{WEBSITE}/{entry_type}/CreateReport"

    r = session.post(
        report_url,
        data=report_data,
        timeout=10,
    )

    time.sleep(2)
    try:
        r.raise_for_status()
        return True
    except HTTPError:
        return False


def close_entry_report(session: requests.Session, report_id: int) -> bool:
    url = f"{WEBSITE}/api/admin/reports/{report_id}"
    r = session.delete(url)
    return r.status_code == 204


def truncate_mikumod_report_message(report_msg: str, max_length: int = 400) -> str:
    if len(report_msg) <= max_length:
        return report_msg
    logger.info(f"Truncating MikuMod report message, {len(report_msg)=}")
    base_msg, entry_list = report_msg.split("Identical mistakes on: ")
    assert base_msg  # noqa: S101
    assert entry_list  # noqa: S101
    assert len(base_msg) < max_length  # noqa: S101

    domains_removed = entry_list.replace(WEBSITE + "/", "")
    if len(base_msg + domains_removed) <= max_length:
        logger.info(
            f"New length (domains removed): {len(domains_removed)}: {domains_removed}"
        )
        return domains_removed

    logger.warning(
        "Too long report message, removing entry links until under the limit..."
    )
    while len(base_msg + domains_removed) > max_length:
        logger.info(
            f"Length: {len(base_msg + domains_removed)}: {base_msg + domains_removed}"
        )
        assert domains_removed[0] == "(", domains_removed  # noqa: S101
        assert domains_removed[-1] == ")", domains_removed  # noqa: S101
        entry_ids = domains_removed.split(", ")
        domains_removed = ", ".join(entry_ids[:-1]) + ")"

    return base_msg + domains_removed


def parse_mikumod_reports(
    reports: list[EntryReport],
    rule_id_mapping: dict[str, int],
    already_reported_entries: set[EntryTuple],
) -> MikuModReports:
    mikumod_reports_by_user_id: MikuModReports = {}
    for report in reports:
        if report.author != "MikuMod":
            logger.debug(f"Skipping non-MikuMod report by user id {report.author}")
            continue

        report_data = report.notes.split(maxsplit=2)
        if len(report_data) != 3:
            logger.info(f"Malformatted MikuMod report: {report_data}")
            _ = input("Press enter to continue...")
            continue

        wiki_page, user_mention, v_version_number_and_msg = report_data
        try:
            rule_name = wiki_page.split("/")[-1]
            rule_id = rule_id_mapping[rule_name]
            _, username = user_mention.split("@")
            user_id = find_user_by_username_1d(username)[1]
            if not user_id:
                logger.warning(f"Could not find user id for user {username}")
                logger.warning("Handle the report manually")
                _ = input("Press enter to continue...")
                continue
            _, version_id = v_version_number_and_msg.split()[0].split("v")

            mikumod_reports_by_user_id.setdefault(user_id, {})
            mikumod_reports_by_user_id[user_id].setdefault(rule_id, [])

            relevant_entries: list[EntryTuple] = []
            if "Identical mistakes on: (" in report.notes:
                other_entries = report.notes.split("Identical mistakes on: (")[1]
                if ")" not in other_entries:
                    other_entries = ", ".join(other_entries.split(", ")[:-1]) + ")"
                other_entries = other_entries.split(")")[0]
                for entry_link in other_entries.split(", "):
                    try:
                        if entry_link.startswith(WEBSITE):
                            entry = get_entry_from_link(entry_link)
                        else:
                            entry = get_entry_from_link(f"{WEBSITE}/{entry_link}")
                        relevant_entries.append(entry)
                        already_reported_entries.add(entry)
                    except IndexError:
                        logger.warning(f"Malformatted MikuMod report: {report.notes}")
                        _ = input("Press enter to continue...")
                        continue
                logger.debug(
                    f"{len(relevant_entries)} relevant entries: {relevant_entries}"
                )

            entry_report_with_version_id_and_relevant_entries = (
                report,
                int(version_id),
                relevant_entries,
            )
            mikumod_reports_by_user_id[user_id][rule_id].append(
                entry_report_with_version_id_and_relevant_entries
            )

        except (KeyError, ValueError):
            logger.debug(f"  Malformatted MikuMod report: {report_data}")
            continue

    return mikumod_reports_by_user_id


def generate_report_msg(
    rule_id: int, version_id: int, user_id: int, rule_modules_by_rule_id: RuleModules
) -> str:
    rule_name, rule_module = rule_modules_by_rule_id[rule_id]
    report_msg = f"{WIKI_RULES_URL}/{rule_name}"
    report_msg += f" @{get_cached_username_by_id_1d(user_id)}"
    report_msg += f" v{version_id}"
    report_msg += f" {rule_module.MSG}"
    if report_msg[-1] != ".":
        report_msg += "."
    report_msg += " [MikuMod report]"
    logger.debug(f"Generated report message: {report_msg}")
    return truncate_mikumod_report_message(report_msg)


def rule_id_already_reported_for_user_id(
    rule_id: int, user_id: int, reports: MikuModReports
) -> bool:
    if user_id not in reports or rule_id not in reports[user_id]:
        return False

    logger.info(f"  Rule {rule_id} already reported for u/{user_id}")
    return True


def recheck_mikumod_reports(
    session: requests.Session,
    rules: RuleModules,
    mikumod_reports: MikuModReports,
    warned_users_file: str,
    mikumod_user_id: int,
) -> None:
    """Recheck reports opened by MikuMod.

    Close handled reports.
    Warn if report ignored.
    """
    warned_user_ids = list(map(int, get_lines(warned_users_file)))
    for counter, (user_id, reports_by_rule_id) in enumerate(mikumod_reports.items()):
        reported_username = get_cached_username_by_id_1d(user_id)
        logger.info(
            f"   {counter + 1}/{len(mikumod_reports)}: Rechecking reports for"
            f" {reported_username} (U/{user_id})"
        )
        for rule_id, reports in reports_by_rule_id.items():
            for report, _, relevant_entries in reports:
                if not report:
                    logger.debug("Not a report, continuing...")
                    continue
                rule_name, rule_module = rules[rule_id]
                entry_link = get_entry_link(report.entry_type, report.entry_id)
                logger.info(f"   Rechecking {entry_link} for R{rule_id} {rule_name}...")
                entry_versions = get_edits_by_entry(report.entry_type, report.entry_id)
                if not entry_versions:
                    logger.warning(f"    The reported {entry_link} has been deleted!")
                    _ = input(
                        "Close the report manually and press enter to continue..."
                    )
                    continue
                version_data = get_cached_entry_version(
                    entry_versions[0].entry_type, entry_versions[0].version_id
                )
                check_result = rule_module.check_entry_version_for_rule(version_data)

                all_entries_valid: bool = True
                if check_result == "Valid":
                    logger.info("    Rule check passed")
                else:
                    logger.info(f"    Rule check: {check_result}")
                    all_entries_valid = False

                if relevant_entries and not all_entries_valid:
                    all_entries_valid = False
                    logger.info(f"    {len(relevant_entries)} relevant entries found")
                    for relevant_entry_counter, entry in enumerate(relevant_entries):
                        entry_link = get_entry_link(*entry)
                        logger.info(f"     {entry_link}")
                        entry_versions = get_edits_by_entry(*entry)
                        if not entry_versions:
                            logger.warning(
                                f"       The reported {entry_link} has been deleted:"
                            )
                            continue
                        version_data = get_cached_entry_version(
                            entry_versions[0].entry_type, entry_versions[0].version_id
                        )
                        check_result = rule_module.check_entry_version_for_rule(
                            version_data
                        )
                        if check_result == "Valid":
                            logger.info(
                                "       "
                                f"{relevant_entry_counter + 1}/{len(relevant_entries)}"
                                f" Rule check passed"
                            )
                        else:
                            logger.info(
                                "       "
                                f"{relevant_entry_counter + 1}/{len(relevant_entries)}"
                                f" Rule check: {check_result}, skipping the rest..."
                            )
                            break
                        all_entries_valid = True

                if all_entries_valid:
                    if close_entry_report(session, report.report_id):
                        logger.info("    Report closed.")
                    else:
                        logger.warning("    Couldn't close report!")
                        _ = input("Investigate. Press enter to continue...")
                    remove_line_from_file(warned_users_file, str(user_id))
                    break
                most_recent_user_edit = get_most_recent_edit_by_user_id(user_id)
                diff_to_report_date_in_hours = int(
                    (most_recent_user_edit.edit_date - report.date).total_seconds()
                    // 3600
                )
                today = datetime.now(tz=UTC)
                diff_to_today_in_hours = int(
                    (today - report.date).total_seconds() // 3600
                )
                if diff_to_report_date_in_hours > 24:
                    logger.warning(
                        f"    User {get_cached_username_by_id_1d(user_id)} has ignored"
                        f" the report - new edits after 24 hours:"
                    )
                    logger.warning(
                        f"    - Reported on: {str(report.date).split('.')[0]}"
                    )
                    logger.warning(
                        f"    - Most recent edit on:"
                        f" {str(most_recent_user_edit.edit_date).split('.')[0]}"
                        f" (+ {diff_to_report_date_in_hours} hours)"
                    )
                    if user_id in warned_user_ids:
                        logger.warning("    - User already warned. Skipping warning.")
                        continue
                    warned = warn_about_ignoring_a_report(
                        session, user_id, report, mikumod_user_id
                    )
                    if warned:
                        save_file(warned_users_file, str(user_id), append=True)
                        warned_user_ids.append(user_id)
                else:
                    logger.info(
                        f"    Last user ({get_cached_username_by_id_1d(user_id)})"
                        f" edit was {diff_to_today_in_hours} hours ago."
                    )


def verify_entry_report(
    entry_with_version_ids: EntryWithVersionIds,
) -> Literal["Error", "Deleted"] | int:
    entry, version_id, most_recent_version_id = entry_with_version_ids
    logger.debug(
        f"Verifying report for {get_entry_link(*entry)}, "
        f"relevant version = {version_id}"
        f", most recent version = {most_recent_version_id}"
    )
    entry_edits = get_edits_by_entry(*entry)
    if not entry_edits:
        logger.warning(
            f"    Skipping entry report, {get_entry_link(*entry)} has been deleted"
        )
        return "Deleted"
    most_recent_edit = entry_edits[0]
    if most_recent_edit.version_id != most_recent_version_id:
        logger.warning(
            f"    Skipping entry report, entry {get_entry_link(*entry)} has been edited"
            f" since the most recent report"
        )
        logger.warning(
            f"    Most recent edit: v{most_recent_edit.version_id} "
            f"!= v{most_recent_version_id}"
        )
        return "Error"

    version_number = [v.version_id for v in entry_edits][::-1].index(version_id)
    logger.debug(f"    Version number for v{version_id} report is {version_number}")
    return version_number


def report_version_id_with_msg(
    session: requests.Session,
    entry_with_version_ids: EntryWithVersionIds,
    report_msg: str,
    prompt_for_report: bool = True,
) -> ReportResult:
    entry, version_number, _ = entry_with_version_ids
    entry_type, entry_id = entry

    logger.warning(f"   Reporting {get_entry_link(*entry)} v{version_number}")
    logger.warning(f"    Report message: {truncate_mikumod_report_message(report_msg)}")

    verification_result_or_version_id = verify_entry_report(entry_with_version_ids)
    if verification_result_or_version_id in ["Deleted", "Error"]:
        return verification_result_or_version_id

    version_number: int = verification_result_or_version_id  # type: ignore

    if prompt_for_report:
        answer = get_boolean_or_none("Send entry report?")
        if answer is None:
            return "Temporary skip"
        if answer is False:
            return "Permanent skip"

    reported = create_entry_report(
        session,
        entry_type=entry_type,
        entry_id=entry_id,
        version_number=version_number,
        report_msg=report_msg,
    )
    if not reported:
        logger.warning("      Couldn't create a new entry report!")
        _ = input("Investigate. Press enter to continue...")
        return "Error"
    return "Success"


def warn_about_ignoring_a_report(
    session: requests.Session, user_id: int, report: EntryReport, mikumod_user_id: int
) -> bool:
    message = (
        f"Please do not ignore entry reports."
        f" {get_entry_link(report.entry_type, report.entry_id)} was reported on"
        f" {str(report.date).split('.')[0]}: '{report.notes}'"
    )

    username = get_username_by_id(user_id)
    cached_username = get_cached_username_by_id_1d(user_id)

    if username != cached_username:
        logger.warning(
            f"User '{cached_username}' has changed their name to '{username}'"
        )
        _ = input("Press enter to skip the report...")
        return False

    logger.warning(message)
    answer = get_boolean_or_none("Send message?")
    if answer is None:
        return False
    if answer is False:
        return False

    send_message(
        session,
        receiver=cached_username,
        subject="Entry report ignored",
        message=message,
        sender_id=mikumod_user_id,
        prompt=False,
    )
    print("-" * 39)  # noqa: T201
    logger.warning("Make sure to push changes on 'warned_users.txt' to GitHub")
    logger.warning(" to prevent duplicate warnings")
    print("-" * 39)  # noqa: T201
    return True