teams-link-monitor 0.0a1

No project description provided

teams-link-monitor

A modular, lightweight Python library designed to transition localized scripts, background tasks, and automation workers into resilient, monitored system workflows. teams-link-monitor enables applications to instantly dispatch context-rich error alerts and local desktop state captures directly to Microsoft Teams channels using modern Workflows incoming webhooks.

Features

• Zero-Latency Alerting: Immediately pushes asynchronous diagnostic alerts, replacing sluggish logging-pull or email-polling architectures.
• Modern Adaptive Cards: Dispatches native Microsoft Teams Adaptive Cards conforming to strict schema version 1.4 layout requirements.
• Context-Rich Diagnostics: Isolates exceptions, error scopes, and script execution timestamps cleanly out of the box.
• Visual Proof Validation: Automatically integrates with display-capture capabilities to preserve transient graphical UI anomalies (e.g., unexpected modal locks, frozen processing windows, or silent application crashes).
• Extensible & Framework-Agnostic: Built as an object-oriented utility that seamlessly injects into logging engines like loguru or Python's native logging module.

---

Project Architecture

teams-link-monitor/
│
├── .env
├── .gitignore
├── LICENSE
├── pyproject.toml
├── README.md
│
├── teams_link_monitor/
│   ├── __init__.py
│   ├── notifier.py
│   ├── payload.py
│   └── screenshot.py
│
└── tests/
    ├── conftest.py
    ├── test_notifier.py
    ├── test_payload.py
    └── test_screenshot.py

---

Prerequisites: Microsoft Teams Channel Configuration

To direct failure alerts to a designated Microsoft Teams channel, you must provision an incoming webhook endpoint via the Teams Workflows infrastructure:

1. Open Microsoft Teams and navigate to your target channel.
2. Click the three dots (...) next to the channel name and select Manage channel.
3. Navigate to Apps (or Connectors depending on your active software version) and launch Workflows.
4. Choose the template titled "Post to a channel when an incoming webhook request is received".
5. Assign a descriptive name to your workflow automation (e.g., System Alert Monitor) and conclude the configuration wizard.
6. Copy the unique webhook URL generated by Microsoft Teams. It will follow this structure: [https://yourcompany.webhook.office.com/webhookb2/](https://yourcompany.webhook.office.com/webhookb2/)...

---

Installation

Install teams-link-monitor globally or within your local virtual environment using pip:

pip install teams-link-monitor

Local Configuration Environment

In the root working directory of your execution environment, populate a .env file to map your channel webhook and desired storage path variables:

# Microsoft Teams Workflows Webhook URL
TEAMS_WEBHOOK_URL="https://yourcompany.webhook.office.com/webhookb2/..."

# Target folder destination to dump visual display logs
MONITOR_OUTPUT_DIR="C:/Users/Username/Project_Logs/Failure_Screenshots"

---

Step-by-Step Usage Guide

Recipe 1: Global Catch-All Handler

Ideal for routine processing tasks (like midnight synchronization scripts or batch data transformations). If any unhandled runtime dependency fails, the wrapper captures the active workspace state, shoots an alert, and exits cleanly.

import sys
from teams_link_monitor import TeamsNotifier, capture_desktop_state

# Initializing the notifier automatically scans your local execution environment
notifier = TeamsNotifier()

def main_execution_logic():
    print("Initializing target workflow processing...")
    # Simulate a critical runtime processing failure
    raise ConnectionError("External transaction gateway timed out while waiting for a handshake.")

if __name__ == "__main__":
    SCRIPT_NAME = "Daily_Data_Sync"

    try:
        main_execution_logic()
        print("Script completed successfully.")
        
    except Exception as e:
        print("Critical exception intercepted! Initializing telemetry dispatch...")
        
        # 1. Capture the exact desktop state before script death
        screenshot_name, screenshot_path = capture_desktop_state(SCRIPT_NAME)
        
        # 2. Fire the Adaptive Card payload directly to Teams
        notifier.send_error(
            script_name=SCRIPT_NAME, 
            error_message=str(e), 
            screenshot_path=screenshot_path
        )
        
        # Return a failure tracking code to your systems orchestrator
        sys.exit(1)

Recipe 2: Resilient Worker Daemon Loop

For background worker threads or active stream loops, wrapping operations internally ensures an individual transaction drop issues a channel warning without crashing your long-running system thread.

import time
from teams_link_monitor import TeamsNotifier, capture_desktop_state

notifier = TeamsNotifier()
SCRIPT_NAME = "Continuous_Queue_Listener"

print(f"Spawning {SCRIPT_NAME} orchestration background listener...")

while True:
    try:
        # Long-polling function tracking system changes
        # check_inbound_message_bus()
        time.sleep(10)
        
    except Exception as e:
        # 1. Take a screenshot and notify without killing the persistent loop
        _, file_path = capture_desktop_state(SCRIPT_NAME)
        notifier.send_error(SCRIPT_NAME, str(e), screenshot_path=file_path)
        
        # 2. Add an internal cool-down throttle window to prevent message spamming
        print("Telemetry pushed to Teams. Pausing for 5 minutes before reconnection attempt...")
        time.sleep(300)

Recipe 3: Native loguru Integration

If your architecture relies on structured logging engines like loguru, you can pipe alerts directly into a custom Sink. This keeps telemetry operations separated from your primary code paths.

from loguru import logger
from teams_link_monitor import TeamsNotifier, capture_desktop_state

notifier = TeamsNotifier()

def teams_alert_sink(message):
    record = message.record
    if record["level"].name in ["ERROR", "CRITICAL"]:
        script_ctx = record["extra"].get("script_name", "Generic_Task")
        
        # Pull visual logs and dispatch to Teams hook
        _, pic_path = capture_desktop_state(script_ctx)
        notifier.send_error(script_ctx, record["message"], screenshot_path=pic_path)

# Bind custom sink processor
logger.add(teams_alert_sink, level="ERROR")
logger = logger.bind(script_name="Inventory_Processing_App")

try:
    logger.info("Parsing incoming text database layout...")
    raise ValueError("Target index identifier sequence is malformed or inaccessible.")
except Exception as err:
    logger.critical(f"System processing loop failed: {err}")

---

Versatility Configurations

If you are deploying tasks inside headless microservices or automated cloud containers that do not maintain an active graphical user display layout, simply drop the optional screenshot path parameter:

from teams_link_monitor import TeamsNotifier

notifier = TeamsNotifier()

# Dispatches a clean text card without expecting a localized graphical element
notifier.send_error(
    script_name="Headless_Parser_Service",
    error_message="FileNotFoundError: Core master tracking layout file missing."
)

---

Diagnostic Logging

teams-link-monitor uses loguru for internal diagnostic logging. To avoid hijacking your application's console layout, all internal library logs are disabled by default.

If you need to view internal package diagnostics or troubleshoot network/screenshot failures, explicitly enable the library namespace in your application startup script:

from loguru import logger
import teams_link_monitor

# Enable diagnostic logs for this package
logger.enable("teams_link_monitor")

---

Security & Compliance Note

This module functions strictly as a localized text/JSON payload interpreter and outbound network client wrapper. It communicates exclusively with your explicitly designated Microsoft Teams endpoint and does not demand administrative operating privileges. All visual file captures remain located within storage paths configured strictly by your internal environmental variables.