#!/usr/bin/env python3
#
# Copyright (c) 2024 Stormux
# Copyright (c) 2010-2012 The Orca Team
# Copyright (c) 2012 Igalia, S.L.
# Copyright (c) 2005-2010 Sun Microsystems Inc.
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This library 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
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this library; if not, write to the
# Free Software Foundation, Inc., Franklin Street, Fifth Floor,
# Boston MA  02110-1301 USA.
#
# Forked from Orca screen reader.
# Cthulhu project: https://git.stormux.org/storm/cthulhu

"""Provides debug utilities for Cthulhu.  Debugging is managed by a debug
level, which is held in the debugLevel field.  All other methods take
a debug level, which is compared to the current debug level to
determine if the content should be output."""

from __future__ import annotations

__id__        = "$Id$"
__version__   = "$Revision$"
__date__      = "$Date$"
__copyright__ = "Copyright (c) 2005-2008 Sun Microsystems Inc."
__license__   = "LGPL"

import inspect
import traceback
import os
import re
import subprocess
import sys
import types
from typing import Any, Optional, TextIO, Pattern, TYPE_CHECKING
from datetime import datetime

import gi
gi.require_version("Atspi", "2.0")
from gi.repository import Atspi

if TYPE_CHECKING:
    from .ax_object import AXObject as AXObjectType

AXObject: Optional[type[AXObjectType]] = None

def _get_ax_object() -> type[AXObjectType]:
    global AXObject
    if AXObject is None:
        from .ax_object import AXObject as ax_object
        AXObject = ax_object
    return AXObject

# Used to turn off all debugging.
#
LEVEL_OFF = 10000

# Used to describe events of considerable importance and which will prevent
# normal program execution.
#
LEVEL_SEVERE = 1000

# Used to describe events of interest to end users or system managers or which
# indicate potential problems, but which Cthulhu can deal with without crashing.
#
LEVEL_WARNING = 900

# Used to indicate reasonably significant messages that make sense to end users
# and system managers.
#
# For the purposes of Cthulhu, LEVEL_INFO means display the text being sent to
# speech and braille.
#
LEVEL_INFO = 800

# Used to indicate static configuration information to assist in debugging
# problems that may be associated with a particular configuration.
#
# For the purposes of Cthulhu, LEVEL_CONFIGURATION means display the various
# aspects of whether a particular feature (e.g., speech, braille, etc.)
# is enabled or not as well as details about that feature.
#
LEVEL_CONFIGURATION = 700

# Used for lowest volume of detailed tracing information.
#
# For the purposes of Cthulhu, this is braille and keyboard input, script
# activation and deletion, locus of focus changes, and visual changes
# to the locus of focus.
#
LEVEL_FINE = 600

# Used for medium volume of detailed tracing information.
#
# For the purposes of Cthulhu, this is for debugging speech and braille
# generators and tracking the synthesis of device events.
#
LEVEL_FINER = 500

# Used for maximum volume of detailed tracing information.
#
# For the purposes of Cthulhu, this is for tracking all AT-SPI object
# events.  NOTE that one can up the debug level of AT-SPI object
# events by setting the eventDebugLevel.  In addition, one can filter
# events by setting eventDebugFilter to a regular expression that
# matches event type names.
#
LEVEL_FINEST = 400

# Used for all detailed tracing information, even finer than LEVEL_FINEST
#
LEVEL_ALL = 0

debugLevel: int = LEVEL_SEVERE

# The debug file.  If this is not set, then all debug output is done
# via stdout.  If this is set, then all debug output is sent to the
# file.  This can be useful for debugging because one can pass in a
# non-buffered file to better track down hangs.
#
debugFile: Optional[TextIO] = None

# The debug filter should be either None (which means to match all
# events) or a compiled regular expression from the 're' module (see
# http://www.amk.ca/python/howto/regex/).  The regular expression will
# be used as a matching function - if the event type creates a match
# in the regular expression, then it will be considered for output.  A
# typical call to this method might look like:
#
# debug.eventDebugFilter = rc.compile('focus:|window:activate')
#
eventDebugLevel: int = LEVEL_FINEST
eventDebugFilter: Optional[Pattern[str]] = None

# If True, we output debug information for the event queue.  We
# use this in addition to log level to prevent debug logic from
# bogging down event handling.
#
debugEventQueue: bool = False

# What module(s) should be traced if traceit is being used. By default
# we'll just attend to ourself. (And by default, we will not enable
# traceit.) Note that enabling this functionality will drag your system
# to a complete and utter halt and should only be used in extreme
# desperation by developers who are attempting to reproduce a very
# specific, immediate issue. Trust me. :-) Disabling braille monitor in
# this case is also strongly advised.
#
TRACE_MODULES: list[str] = ['cthulhu']

# Specific modules to ignore with traceit.
#
TRACE_IGNORE_MODULES: list[str] = ['traceback', 'linecache', 'locale', 'gettext',
                        'logging', 'UserDict', 'encodings', 'posixpath',
                        'genericpath', 're']

# Specific apps to trace with traceit.
#
TRACE_APPS: list[str] = []

# What AT-SPI event(s) should be traced if traceit is being used. By
# default, we'll trace everything. Examples of what you might wish to
# do to narrow things down include:
#
# TRACE_EVENTS = ['object:state-changed', 'focus:']
#     (for any and all object:state-changed events plus focus: events)
# TRACE_EVENTS = ['object:state-changed:selected']
#     (if you know the exact event type of interest)
#
TRACE_EVENTS: list[str] = []

# What role(s) should be traced if traceit is being used. By
# default, we'll trace everything. An example of what you might wish
# to do to narrow things down, if you know buttons trigger the problem:
#
# TRACE_ROLES = [Atspi.Role.PUSH_BUTTON, Atspi.Role.TOGGLE_BUTTON]
#
TRACE_ROLES: list[Atspi.Role] = []

# Whether or not traceit should only trace the work being done when
# processing an actual event. This is when most bad things happen.
# So we'll default to True.
#
TRACE_ONLY_PROCESSING_EVENTS: bool = True

objEvent: Optional[Atspi.Event] = None

def printException(level: int) -> None:
    """Prints out information regarding the current exception.

    Arguments:
    - level: the accepted debug level
    """

    if level >= debugLevel:
        println(level)
        traceback.print_exc(100, debugFile)
        println(level)

def printStack(level: int) -> None:
    """Prints out the current stack.

    Arguments:
    - level: the accepted debug level
    """

    if level >= debugLevel:
        println(level)
        traceback.print_stack(None, 100, debugFile)
        println(level)

def _asString(obj: Any) -> str:
    AXObject = _get_ax_object()
    if isinstance(obj, Atspi.Accessible):
        result = AXObject.get_role_name(obj)
        name = AXObject.get_name(obj)
        if name:
            result += f": '{name}'"
        if not result:
            result = "DEAD"

        return f"[{result}]"

    if isinstance(obj, Atspi.Event):
        return (
            f"{obj.type} for {_asString(obj.source)} in "
            f"{_asString(AXObject.get_application(obj.source))} "
            f"({obj.detail1}, {obj.detail2}, {_asString(obj.any_data)})"
        )

    if isinstance(obj, (Atspi.Role, Atspi.StateType, Atspi.CollectionMatchType)):
        return obj.value_nick

    if isinstance(obj, list):
        return f"[{', '.join(map(_asString, obj))}]"

    if isinstance(obj, types.FunctionType):
        if hasattr(obj, "__self__"):
            return f"{obj.__module__}.{obj.__self__.__class__.__name__}.{obj.__name__}"
        return f"{obj.__module__}.{obj.__name__}"

    if isinstance(obj, types.FrameType):
        module_name = inspect.getmodulename(obj.f_code.co_filename)
        return f"{module_name}.{obj.f_code.co_name}"

    if isinstance(obj, inspect.FrameInfo):
        module_name = inspect.getmodulename(obj.filename)
        return f"{module_name}.{obj.function}"

    return str(obj)

def _format_tokens(tokens: list[Any]) -> str:
    text = " ".join(map(_asString, tokens))
    text = re.sub(r"[ \u00A0]+", " ", text)
    text = re.sub(r" (?=[,.:)])(?![\n])", "", text)
    return text

def format_log_message(prefix: str, message: str, reason: Optional[str] = None) -> str:
    text = f"{prefix}: {message}"
    if reason:
        text = f"{text} (reason={reason})"
    return text

def print_log(level: int, prefix: str, message: str, reason: Optional[str] = None,
              timestamp: bool = False, stack: bool = False) -> None:
    text = format_log_message(prefix, message, reason)
    printMessage(level, text, timestamp, stack)

def print_log_tokens(level: int, prefix: str, tokens: list[Any], reason: Optional[str] = None,
                     timestamp: bool = False, stack: bool = False) -> None:
    text = format_log_message(prefix, _format_tokens(tokens), reason)
    printMessage(level, text, timestamp, stack)

def printTokens(level: int, tokens: list[Any], timestamp: bool = False, stack: bool = False) -> None:
    if level < debugLevel:
        return

    println(level, _format_tokens(tokens), timestamp, stack)

def print_tokens(level: int, tokens: list[Any], timestamp: bool = False, stack: bool = False) -> None:
    return printTokens(level, tokens, timestamp, stack)

def printMessage(level: int, text: str, timestamp: bool = False, stack: bool = False) -> None:
    if level < debugLevel:
        return

    println(level, text, timestamp, stack)

def print_message(level: int, text: str, timestamp: bool = False, stack: bool = False) -> None:
    return printMessage(level, text, timestamp, stack)

def _stackAsString(max_frames: int = 4) -> str:
    callers = []
    current_module = inspect.getmodule(inspect.currentframe())
    stack = inspect.stack()
    for i in range(1, len(stack)):
        frame = stack[i]
        module = inspect.getmodule(frame[0])
        if module == current_module:
            continue
        if frame.function == 'main':
            continue
        if module is None or module.__name__ is None:
            continue
        callers.append(frame)
        if len(callers) >= max_frames:
            break

    callers.reverse()
    return " > ".join(map(_asString, callers))

def println(level: int, text: str = "", timestamp: bool = False, stack: bool = False) -> None:
    """Prints the text to stderr unless debug is enabled.

    If debug is enabled the text will be redirected to the
    file debugFile.

    Arguments:
    - level: the accepted debug level
    - text: the text to print (default is a blank line)
    """

    if level >= debugLevel:
        text = text.replace("\ufffc", "[OBJ]")
        if timestamp:
            text = text.replace("\n", f"\n{' ' * 18}")
            text = f"{text} - {datetime.now().strftime('%H:%M:%S.%f')}"
        if stack:
            text += f" {_stackAsString()}"

        if debugFile:
            try:
                debugFile.writelines([text, "\n"])
                debugFile.flush()
            except TypeError:
                text = "TypeError when trying to write text"
                debugFile.writelines([text, "\n"])
                debugFile.flush()
            except Exception:
                text = "Exception when trying to write text"
                debugFile.writelines([text, "\n"])
                debugFile.flush()
        else:
            try:
                sys.stderr.writelines([text, "\n"])
                sys.stderr.flush()
            except TypeError:
                text = "TypeError when trying to write text"
                sys.stderr.writelines([text, "\n"])
                sys.stderr.flush()
            except Exception:
                text = "Exception when trying to write text"
                sys.stderr.writelines([text, "\n"])
                sys.stderr.flush()

def printResult(level: int, result: Any = None) -> None:
    """Prints the return result, along with information about the
    method, arguments, and any errors encountered."""

    if level < debugLevel:
        return

    stack = inspect.stack()
    current, prev = stack[1], stack[2]
    frame = current[0]

    # To better print arguments which are accessible objects
    args = inspect.getargvalues(frame)
    for key, value in list(args.locals.items()):
        args.locals[key] = str(value)
    fArgs = str.replace(inspect.formatargvalues(*args), "'", "")

    callString = 'CALL:   %s.%s (line %s) -> %s.%s%s' % (
        inspect.getmodulename(prev[1]), prev[3], prev[2],
        inspect.getmodulename(current[1]), current[3], fArgs)
    string = f'{callString}\nRESULT: {result}'
    println(level, f'{string}')

def printObjectEvent(level: int, event: Atspi.Event, sourceInfo: Optional[str] = None,
                     timestamp: bool = False) -> None:
    """Prints out an Python Event object.  The given level may be
    overridden if the eventDebugLevel is greater.  Furthermore, only
    events with event types matching the eventDebugFilter regular
    expression will be printed.

    Arguments:
    - level: the accepted debug level
    - event: the Python Event to print
    - sourceInfo: additional string to print out
    """

    if eventDebugFilter and not eventDebugFilter.match(event.type):
        return

    level = max(level, eventDebugLevel)

    anydata = event.any_data
    if isinstance(anydata, str) and len(anydata) > 100:
        anydata = f"{anydata[0:100]} (...)"

    text = "OBJECT EVENT: %s (%d, %d, %s)" \
           % (event.type, event.detail1, event.detail2, anydata)
    println(level, text, timestamp)

    if sourceInfo:
        println(level, f"{' ' * 18}{sourceInfo}", timestamp)

def printInputEvent(level: int, string: str, timestamp: bool = False) -> None:
    """Prints out an input event.  The given level may be overridden
    if the eventDebugLevel (see setEventDebugLevel) is greater.

    Arguments:
    - level: the accepted debug level
    - string: the string representing the input event
    """

    println(max(level, eventDebugLevel), string, timestamp)

def printDetails(level: int, indent: str, accessible: Atspi.Accessible,
                 includeApp: bool = True, timestamp: bool = False) -> None:
    """Lists the details of the given accessible with the given
    indentation.

    Arguments:
    - level: the accepted debug level
    - indent: a string containing spaces for indentation
    - accessible: the accessible whose details are to be listed
    - includeApp: if True, include information about the app
    """

    if level >= debugLevel and accessible:
        println(level,
                getAccessibleDetails(level, accessible, indent, includeApp),
                timestamp)

def getAccessibleDetails(level: int, acc: Atspi.Accessible, indent: str = "",
                         includeApp: bool = True) -> str:
    """Returns a string, suitable for printing, that describes the
    given accessible.

    Arguments:
    - indent: A string to prefix the output with
    - includeApp: If True, include information about the app
                  for this accessible.
    """

    if level < debugLevel:
        return ""

    from .ax_utilities_debugging import AXUtilitiesDebugging
    return AXUtilitiesDebugging.object_details_as_string(acc, indent, includeApp)

# The following code originated from the following URL:
# 
# http://www.dalkescientific.com/writings/diary/archive/ \
#                                     2005/04/20/tracing_python_code.html
#
import linecache

def _getFileAndModule(frame: types.FrameType) -> tuple[Optional[str], Optional[str]]:
    filename, module = None, None
    try:
        filename = frame.f_globals["__file__"]
        module = frame.f_globals["__name__"]
    except Exception:
        pass
    else:
        if (filename.endswith(".pyc") or filename.endswith(".pyo")):
            filename = filename[:-1]

    return filename, module

def _shouldTraceIt() -> bool:
    AXObject = _get_ax_object()
    if not objEvent:
        return not TRACE_ONLY_PROCESSING_EVENTS

    eventSource = objEvent.source
    if TRACE_APPS:
        app = AXObject.get_application(eventSource)
        if AXObject.get_name(app) not in TRACE_APPS:
            return False

    if TRACE_ROLES and AXObject.get_role(eventSource) not in TRACE_ROLES:
        return False

    if TRACE_EVENTS and \
       not [x for x in map(objEvent.type.startswith, TRACE_EVENTS) if x]:
        return False

    return True

def traceit(frame: types.FrameType, event: str, arg: Any) -> Optional[object]:
    """Line tracing utility to output all lines as they are executed by
    the interpreter.  This is to be used by sys.settrace and is for
    debugging purposes.

    Arguments:
    - frame: is the current stack frame
    - event: 'call', 'line', 'return', 'exception', 'c_call', 'c_return',
             or 'c_exception'
    - arg:   depends on the event type (see docs for sys.settrace)
    """

    if not _shouldTraceIt():
        return None

    filename, module = _getFileAndModule(frame)
    if not (filename and module):
        return traceit
    if module in TRACE_IGNORE_MODULES:
        return traceit
    if TRACE_MODULES and module.split('.')[0] not in TRACE_MODULES:
        return traceit
    if event not in ['call', 'line', 'return']:
        return traceit

    lineno = frame.f_lineno
    line = linecache.getline(filename, lineno).rstrip()
    output = f'TRACE {module}:{lineno}: {line}'

    if event == 'call':
        argvals = inspect.getargvalues(frame)
        keys = [x for x in argvals[0] if x != 'self']
        try:
            values = list(map(argvals[3].get, keys))
        except TypeError:
            if len(keys) == 1 and isinstance(keys[0], list):
                values = list(map(argvals[3].get, keys[0]))
            else:
                return traceit
        for i, key in enumerate(keys):
            output += f'\n  ARG {key}={values[i]}'

    lineElements = line.strip().split()
    if lineElements and lineElements[0] == 'return':
        if event == 'line':
            return traceit
        output = f'{output} (rv: {arg})'

    println(LEVEL_ALL, output)

    return traceit

def getOpenFDCount(pid: int) -> int:
    procs = subprocess.check_output([ 'lsof', '-w', '-Ff', '-p', str(pid)])
    procs = procs.decode('UTF-8').split('\n')
    files = list(filter(lambda s: s and s[0] == 'f' and s[1:].isdigit(), procs))

    return len(files)

def getCmdline(pid: int) -> str:
    try:
        openFile = os.popen(f'cat /proc/{pid}/cmdline')
        cmdline = openFile.read()
        openFile.close()
    except Exception:
        cmdline = '(Could not obtain cmdline)'
    cmdline = cmdline.replace('\x00', ' ')

    return cmdline

def pidOf(procName: str) -> list[int]:
    openFile = subprocess.Popen(f'pgrep {procName}',
                                shell=True,
                                stdout=subprocess.PIPE).stdout
    pids = openFile.read()
    openFile.close()
    return [int(p) for p in pids.split()]

def examineProcesses(force: bool = False) -> None:
    AXObject = _get_ax_object()
    from .ax_utilities import AXUtilities
    if force:
        level = LEVEL_OFF
    else:
        level = LEVEL_ALL

    desktop = AXUtilities.get_desktop()
    println(level, 'INFO: Desktop has %i apps:' % AXObject.get_child_count(desktop), True)
    for i, app in enumerate(AXObject.iter_children(desktop)):
        pid = AXObject.get_process_id(app)
        cmd = getCmdline(pid)
        fds = getOpenFDCount(pid)
        name = AXObject.get_name(app)
        if name == '':
            name = 'WARNING: Possible hang or dead app'
        println(level, '%3i. %s (pid: %s) %s file descriptors: %i' \
                    % (i+1, name, pid, cmd, fds), True)

    # Other 'suspect' processes which might not show up as accessible apps.
    otherApps = ['apport']
    for app in otherApps:
        pids = pidOf(app)
        if not pids:
            println(level, f'INFO: no pid for {app}', True)
            continue

        for pid in pids:
            cmd = getCmdline(pid)
            fds = getOpenFDCount(pid)
            println(level, 'INFO: %s (pid: %s) %s file descriptors: %i' \
                        % (app, pid, cmd, fds), True)
