139 lines
4.6 KiB
Python
139 lines
4.6 KiB
Python
import requests
|
|
import os
|
|
import json
|
|
from pathlib import Path
|
|
import datetime
|
|
|
|
# Ensure a .env file exists next to this script and load variables from it.
|
|
def ensure_env_file(env_path: str, defaults: dict):
|
|
"""Create an env file at env_path with defaults if it doesn't exist."""
|
|
p = Path(env_path)
|
|
if not p.exists():
|
|
try:
|
|
p.parent.mkdir(parents=True, exist_ok=True)
|
|
with p.open('w', encoding='utf-8') as f:
|
|
for k, v in defaults.items():
|
|
f.write(f"{k}={v}\n")
|
|
print(f"Created env file: {env_path}")
|
|
except Exception as e:
|
|
print(f"Failed to create env file {env_path}: {e}")
|
|
|
|
def load_env_file(env_path: str):
|
|
"""Load KEY=VALUE lines from env_path into os.environ if not already set.
|
|
|
|
Supports simple lines, ignores blank lines and lines starting with '#'.
|
|
"""
|
|
p = Path(env_path)
|
|
if not p.exists():
|
|
return
|
|
try:
|
|
with p.open('r', encoding='utf-8') as f:
|
|
for raw in f:
|
|
line = raw.strip()
|
|
if not line or line.startswith('#'):
|
|
continue
|
|
if '=' not in line:
|
|
continue
|
|
k, v = line.split('=', 1)
|
|
k = k.strip()
|
|
v = v.strip().strip('"').strip("'")
|
|
if k and k not in os.environ:
|
|
os.environ[k] = v
|
|
except Exception as e:
|
|
print(f"Failed to read env file {env_path}: {e}")
|
|
|
|
# Path to .env in the same folder as this script
|
|
_env_path = os.path.join(os.path.dirname(__file__), '.env')
|
|
_defaults = {
|
|
'SCHLOTER_API_URL': 'http://schloter.api.ping-mee.de/quotes/random',
|
|
'TEAMS_WEBHOOK_URL': 'https://outlook.office.com/webhook/REPLACE_WITH_YOUR_WEBHOOK',
|
|
'SCHLOTER_API_TOKEN': '?2%?fK+@%Y9wy!f6'
|
|
}
|
|
|
|
ensure_env_file(_env_path, _defaults)
|
|
load_env_file(_env_path)
|
|
|
|
# Where to fetch a quote from (API)
|
|
API_URL = os.environ.get('SCHLOTER_API_URL', _defaults['SCHLOTER_API_URL'])
|
|
|
|
# Microsoft Teams Incoming Webhook URL (Workflows / connectors)
|
|
TEAMS_WEBHOOK_URL = os.environ.get('TEAMS_WEBHOOK_URL', _defaults['TEAMS_WEBHOOK_URL'])
|
|
|
|
# Static token for the API (can be set via environment variable)
|
|
API_TOKEN = os.environ.get('SCHLOTER_API_TOKEN', _defaults['SCHLOTER_API_TOKEN'])
|
|
|
|
def get_quote():
|
|
"""Fetch a quote from the API using a Bearer token.
|
|
|
|
Returns the quote string or raises on HTTP error.
|
|
"""
|
|
headers = {"Authorization": f"Bearer {API_TOKEN}"}
|
|
resp = requests.get(API_URL, headers=headers, timeout=10)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
return data.get("quote", "No quote found.")
|
|
|
|
def post_to_teams(quote):
|
|
"""Post the quote to a Teams Workflows (incoming webhook) URL.
|
|
|
|
To satisfy Power Automate flows that expect triggerBody()['attachments'],
|
|
this posts both a plain 'text' property and an 'attachments' array containing
|
|
an Adaptive Card. Returns True on success (200 or 202), False otherwise.
|
|
"""
|
|
headers = {"Content-Type": "application/json"}
|
|
|
|
adaptive_card = {
|
|
"type": "AdaptiveCard",
|
|
"body": [
|
|
{
|
|
"type": "TextBlock",
|
|
"size": "Medium",
|
|
"weight": "Bolder",
|
|
"text": "${title}"
|
|
},
|
|
{
|
|
"type": "TextBlock",
|
|
"text": quote,
|
|
"wrap": True
|
|
}
|
|
],
|
|
"$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
|
|
"version": "1.6"
|
|
}
|
|
|
|
payload = {
|
|
# keep text for flows that read triggerBody()['text']
|
|
"text": quote,
|
|
# attachments array for flows that expect triggerBody()['attachments']
|
|
"attachments": [
|
|
{
|
|
"contentType": "application/vnd.microsoft.card.adaptive",
|
|
"content": adaptive_card
|
|
}
|
|
]
|
|
}
|
|
|
|
try:
|
|
resp = requests.post(TEAMS_WEBHOOK_URL, headers=headers, data=json.dumps(payload), timeout=10)
|
|
except requests.RequestException as e:
|
|
print(f"Network error posting to Teams webhook: {e}")
|
|
return False
|
|
|
|
# Don't raise; print detailed failure info for debugging
|
|
if resp.status_code in (200, 202):
|
|
print(f"Posted to Teams webhook successfully (status {resp.status_code}).")
|
|
if resp.text:
|
|
print("Response body:", resp.text)
|
|
return True
|
|
else:
|
|
print(f"Failed to post to Teams webhook: {resp.status_code}")
|
|
print("Response body:", resp.text)
|
|
return False
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# Fetch a quote and post it to Teams. Use env vars to override defaults.
|
|
quote = "Anwesenheitskontrolle: "+datetime.date+" "+get_quote()
|
|
post_to_teams(quote)
|
|
print("Quote fetched:", quote)
|