142 lines
4.7 KiB
Python
142 lines
4.7 KiB
Python
#!/usr/bin/env python3
|
|
"""TeamPulse — Teams meeting chat audit tool.
|
|
|
|
Usage:
|
|
.venv/bin/python main.py [meeting-url]
|
|
|
|
If no URL is given, the clipboard is checked for a Teams meeting link.
|
|
If neither yields a URL, navigate manually in the browser window.
|
|
Post !start "Presenter Name" and !stop in the chat to define a time window.
|
|
A Markdown memo is saved to the current directory when the window closes.
|
|
"""
|
|
import platform
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
_TEAMS_URL_RE = re.compile(r'https://teams(?:\.live)?\.microsoft\.com/\S+')
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent / "src"))
|
|
|
|
from playwright.sync_api import sync_playwright
|
|
|
|
from teampulse.auth import create_context, ensure_logged_in
|
|
from teampulse.memo import generate_memo, save_memo
|
|
from teampulse.models import AuditEntry
|
|
from teampulse.monitor import Monitor
|
|
from teampulse.resolver import Resolver
|
|
|
|
CACHE_PATH = Path.home() / ".teampulse" / "cache.json"
|
|
TEAMS_URL = "https://teams.microsoft.com"
|
|
|
|
|
|
def _read_clipboard() -> str:
|
|
try:
|
|
system = platform.system()
|
|
if system == "Darwin":
|
|
return subprocess.check_output(["pbpaste"]).decode("utf-8").strip()
|
|
elif system == "Linux":
|
|
return subprocess.check_output(["xclip", "-selection", "clipboard", "-o"]).decode("utf-8").strip()
|
|
elif system == "Windows":
|
|
return subprocess.check_output(
|
|
["powershell", "-command", "Get-Clipboard"], text=True
|
|
).strip()
|
|
except Exception:
|
|
pass
|
|
return ""
|
|
|
|
|
|
def _extract_teams_url(text: str) -> str | None:
|
|
m = _TEAMS_URL_RE.search(text)
|
|
return m.group(0) if m else None
|
|
|
|
|
|
def _get_meeting_url() -> str | None:
|
|
if len(sys.argv) > 1:
|
|
url = _extract_teams_url(sys.argv[1])
|
|
if url:
|
|
print(f"Meeting-URL aus Argument: {url[:60]}...")
|
|
return url
|
|
|
|
clip = _read_clipboard()
|
|
url = _extract_teams_url(clip)
|
|
if url:
|
|
print(f"Meeting-URL aus Clipboard: {url[:60]}...")
|
|
return url
|
|
|
|
return None
|
|
|
|
|
|
def main():
|
|
meeting_url = _get_meeting_url()
|
|
|
|
with sync_playwright() as playwright:
|
|
print("Starte Browser...")
|
|
context = create_context(playwright, headless=False)
|
|
ensure_logged_in(context)
|
|
|
|
page = context.new_page()
|
|
|
|
# Always load Teams main first so user-name selectors are available
|
|
page.goto(TEAMS_URL)
|
|
|
|
monitor = Monitor(page=page, current_user="")
|
|
print("Lese eingeloggten Nutzer...")
|
|
current_user = monitor.get_current_user_display_name()
|
|
monitor._current_user = current_user
|
|
print(f"Eingeloggt als: {current_user}")
|
|
|
|
if meeting_url:
|
|
print("Navigiere zur Meeting-URL...")
|
|
page.goto(meeting_url)
|
|
try:
|
|
page.wait_for_selector(
|
|
"[data-tid='channel-pane-message'], [data-tid='chat-pane-message']",
|
|
timeout=6000,
|
|
)
|
|
print("Chat direkt erreicht.")
|
|
except Exception:
|
|
# Join-Lobby or redirect — try navigating to Teams chat list instead.
|
|
# The meeting chat is accessible from Chat sidebar without joining the call.
|
|
print("Join-Lobby erkannt — navigiere zur Chat-Übersicht...")
|
|
page.goto(TEAMS_URL + "/#/conversations/")
|
|
try:
|
|
page.wait_for_selector(
|
|
"[data-tid='channel-pane-message'], [data-tid='chat-pane-message']",
|
|
timeout=5000,
|
|
)
|
|
print("Chat erkannt.")
|
|
except Exception:
|
|
print("Bitte im Browser-Fenster zum laufenden Meeting-Chat navigieren:")
|
|
print(" Linke Sidebar → Chat → laufendes Meeting anklicken\n")
|
|
else:
|
|
print("\nBitte im Browser-Fenster zum Meeting-Chat navigieren:")
|
|
print(" Linke Sidebar → Chat → laufendes Meeting anklicken\n")
|
|
|
|
print("\nPoste '!start Name' im Chat um zu beginnen.\n")
|
|
|
|
window = monitor.run()
|
|
|
|
resolver = Resolver(cache_path=CACHE_PATH, page=page)
|
|
print(f"\nLöse {len(window.entries)} E-Mail-Adresse(n) auf...")
|
|
resolved_entries = []
|
|
for entry in window.entries:
|
|
email = resolver.resolve(entry.display_name)
|
|
print(f" {entry.display_name} → {email}")
|
|
resolved_entries.append(AuditEntry(display_name=entry.display_name, email=email))
|
|
|
|
window.entries = resolved_entries
|
|
|
|
memo_content = generate_memo(window)
|
|
path = save_memo(memo_content)
|
|
|
|
print(f"\nMemo gespeichert: {path}")
|
|
print("\n" + memo_content)
|
|
|
|
context.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|