rules.py

# type: ignore (for mirror)
# ruff: noqa: S101

import importlib.util
import inspect
from collections import defaultdict, deque
from pathlib import Path
from types import ModuleType
from typing import TYPE_CHECKING, Any

import requests
from bs4 import BeautifulSoup
from vdbpy.config import LOCAL_WIKI_URL, WIKI_URL
from vdbpy.types.shared import EntryTuple
from vdbpy.utils.logger import get_logger
from vdbpy.utils.network import fetch_text

from mod_types import RuleModules, RuleTableRow

if TYPE_CHECKING:
    from vdbpy.types.shared import EntryStatus

logger = get_logger()


def topo_sort_graph(graph: dict[int, list[int]], all_rules: set[int]) -> list[int]:
    adj: defaultdict[int, list[int]] = defaultdict(list)
    indegree: defaultdict[int, int] = defaultdict(int)

    nodes = set(all_rules)
    for deps in graph.values():
        nodes.update(deps)

    for node, deps in graph.items():
        for dep in deps:
            adj[dep].append(node)
            indegree[node] += 1
        indegree[node] += 0

    for n in nodes:
        indegree[n] += 0

    queue = deque([n for n in nodes if indegree[n] == 0])
    topo_result: list[int] = []

    while queue:
        n = queue.popleft()
        topo_result.append(n)
        for nxt in adj[n]:
            indegree[nxt] -= 1
            if indegree[nxt] == 0:
                queue.append(nxt)

    if len(topo_result) != len(nodes):
        raise ValueError("Cycle detected in rule dependency graph")

    dependent_rules = set(graph.keys())
    independent_rules = [r for r in topo_result if r not in dependent_rules]
    dep_sorted_rules = [r for r in topo_result if r in dependent_rules]

    return independent_rules + dep_sorted_rules


def validate_rule_check_module(module: ModuleType, rulefile: Path) -> bool:
    required_rule_check_attrs = [
        "check_entry_version_for_rule",
        "test",
        "FIELDS",
        "ENTRY_TYPES",
        "MSG",
        "COMPLETE",
        "AUTOMATICALLY_FIXED",
    ]
    for attr in required_rule_check_attrs:
        if not hasattr(module, attr):
            logger.warning(f"Rule check module {rulefile} does not include '{attr}'")
            return False

    if module.AUTOMATICALLY_FIXED in (True, "Partially"):
        if not hasattr(module, "autofix") or not callable(module.autofix):
            logger.warning(
                f"Rule check module {rulefile} with AUTOMATICALLY_FIXED="
                f"{module.AUTOMATICALLY_FIXED} requires 'autofix()'"
            )
            return False

        correct_return_type = bool
        if inspect.signature(module.autofix).return_annotation != correct_return_type:
            logger.warning(f"Incorrect return type for autofix() on {rulefile}")
            return False

        correct_params = {
            "session": requests.Session,
            "entry": EntryTuple,
            "base_update_note": str,
            "prompt": bool,
            "args": Any,
        }
        for key, _type in correct_params.items():
            if inspect.signature(module.autofix).parameters[key].annotation != _type:
                logger.warning(f"autofix() '{key}' param is not {_type} on {rulefile}")
                return False

    if hasattr(module, "TAG_ID"):
        tag_id = module.TAG_ID
        if not hasattr(module, "find_relevant_entries"):
            msg = (
                f"Rule check module {rulefile} with tag id {tag_id} requires"
                " 'find_relevant_entries()'"
            )
            logger.warning(msg)
            return False

    return True


def get_rule_modules_by_id(path: Path, selected_rule_id: int = 0) -> RuleModules:
    """Collect rule ids, rule names and rule modules from rule checks directory."""
    logger.debug(f"Loading rule modules from '{path}'")
    rule_dependency_graph: dict[int, list[int]] = {}
    rule_checks: RuleModules = {}
    python_files = list(Path(path).glob("*.py"))
    logger.debug(f"Found {len(python_files)} python files...")
    for rulefile in python_files:
        logger.debug(rulefile)
        if rulefile.name == "__init__.py":
            logger.debug("Skipping __init__.py")
            continue

        if rulefile.name.startswith("0_"):
            logger.debug("Skipping template file")
            continue

        rule_id_str, rule_name = rulefile.name.split(".py")[0].split("_")
        if selected_rule_id and int(rule_id_str) != selected_rule_id:
            logger.debug(f"Skipping rule module {rule_id_str}")
            continue
        rule_id = int(rule_id_str)
        spec = importlib.util.spec_from_file_location(rulefile.stem, rulefile)
        module = None
        if spec and spec.loader:
            module = importlib.util.module_from_spec(spec)
            spec.loader.exec_module(module)
        if not module:
            logger.warning(f"Could not import rule check module {rulefile}")
            continue

        valid_rule_check_module = validate_rule_check_module(module, rulefile)
        if not valid_rule_check_module:
            logger.debug("Invalid rule module, continuing...")
            continue

        if hasattr(module, "ASSUME_VALID_FOR_RULE_ID"):
            rule_dependency_graph[rule_id] = module.ASSUME_VALID_FOR_RULE_ID

        rule_checks[rule_id] = (rule_name, module)

    all_rule_ids = set(rule_checks.keys())
    for rule_id, deps in rule_dependency_graph.items():
        for dep in deps:
            assert dep in all_rule_ids, (
                f"Missing dependency {dep} for {rule_id}, {all_rule_ids=}"
            )
    order = topo_sort_graph(rule_dependency_graph, all_rule_ids)

    return {
        rule_id: rule_checks[rule_id] for rule_id in order if rule_id in rule_checks
    }


def print_missing_rule_modules(
    existing_rule_ids: set[int], rule_table: dict[int, RuleTableRow]
) -> None:
    missing_rule_ids = set(rule_table.keys()) - existing_rule_ids
    logger.info("Missing rules:")
    for missing_rule_id in missing_rule_ids:
        missing_rule_fields = rule_table[missing_rule_id]
        if missing_rule_fields.rule_mikumodded == "Planned":
            logger.info(f"- {missing_rule_fields.rule_name}")


def get_rule_table() -> dict[int, RuleTableRow]:
    rows_to_return: dict[int, RuleTableRow] = {}
    try:
        html = fetch_text(f"{LOCAL_WIKI_URL}/rules/table")
    except requests.exceptions.ConnectionError:
        html = fetch_text(f"{WIKI_URL}/rules/table")
    soup = BeautifulSoup(html, "html.parser")
    table = soup.find("table")
    if table:
        header = table.find("thead")
        assert header
        header_text = [e.get_text(strip=True) for e in header.find_all("th")]
        assert header_text == [
            "ID",
            "Date",
            "Name",
            "E.Type",
            "E.Status",
            "Context",
            "Tag",
            "MikuModded",
            "Complete",
            "Autofix",
            "List",
        ]
        tbody = table.find("tbody")
        assert tbody
        rows = tbody.find_all("tr")
        for row in rows:
            cells = row.find_all("td")
            assert len(cells) == 11, f"Expected 11 table cells but got {len(cells)}"
            rule_id = int(cells[0].get_text(strip=True))
            rule_name = cells[2].get_text(strip=True).lower().replace(" ", "-")
            rule_entry_type: list[str] = cells[3].get_text(strip=True).split(", ")
            rule_entry_status: EntryStatus = cells[4].get_text(strip=True)  # type: ignore
            tag_id_value = cells[6].get_text(strip=True)
            rule_tag_id = int(tag_id_value) if tag_id_value.isnumeric() else 0
            rule_complete = cells[8].get_text(strip=True).lower() == "true"
            match cells[7].get_text(strip=True).lower():
                case "true":
                    rule_mikumodded = True
                case "false":
                    rule_mikumodded = False
                case _:
                    rule_mikumodded = "Planned"
            match cells[9].get_text(strip=True).lower():
                case "true":
                    rule_autofixed = True
                case "false":
                    rule_autofixed = False
                case _:
                    rule_autofixed = "Partially"
            rows_to_return[rule_id] = RuleTableRow(
                rule_name,
                rule_entry_type,
                rule_entry_status,
                rule_tag_id,
                rule_mikumodded,
                rule_complete,
                rule_autofixed,
            )
    return rows_to_return