Updated API call

main
Ebbe Baß 2025-10-28 14:58:08 +01:00
parent 758784d004
commit cf3411cded
1 changed files with 107 additions and 23 deletions

View File

@ -1,7 +1,30 @@
import requests #!/usr/bin/env python3
"""
client/daily_client.py
Fetch a quote from the API and post it to a Microsoft Teams channel via a webhook.
Behavior:
- Ensures a `.env` file exists next to this script with sensible defaults (if missing).
- Loads KEY=VALUE pairs from the `.env` into the environment (if not already set).
- Requests a random quote from the API using a Bearer token.
- Posts the quote to Teams as a MessageCard (title: "Daily Quote").
Exit codes:
0 = success
1 = no quote returned
2 = failed to fetch quote
3 = failed to post to Teams
"""
import os import os
import sys
import json
import requests
import argparse
from pathlib import Path from pathlib import Path
# Ensure a .env file exists next to this script and load variables from it. # Ensure a .env file exists next to this script and load variables from it.
def ensure_env_file(env_path: str, defaults: dict): def ensure_env_file(env_path: str, defaults: dict):
"""Create an env file at env_path with defaults if it doesn't exist.""" """Create an env file at env_path with defaults if it doesn't exist."""
@ -16,6 +39,7 @@ def ensure_env_file(env_path: str, defaults: dict):
except Exception as e: except Exception as e:
print(f"Failed to create env file {env_path}: {e}") print(f"Failed to create env file {env_path}: {e}")
def load_env_file(env_path: str): def load_env_file(env_path: str):
"""Load KEY=VALUE lines from env_path into os.environ if not already set. """Load KEY=VALUE lines from env_path into os.environ if not already set.
@ -40,6 +64,7 @@ def load_env_file(env_path: str):
except Exception as e: except Exception as e:
print(f"Failed to read env file {env_path}: {e}") print(f"Failed to read env file {env_path}: {e}")
# Path to .env in the same folder as this script # Path to .env in the same folder as this script
_env_path = os.path.join(os.path.dirname(__file__), '.env') _env_path = os.path.join(os.path.dirname(__file__), '.env')
_defaults = { _defaults = {
@ -51,6 +76,7 @@ _defaults = {
ensure_env_file(_env_path, _defaults) ensure_env_file(_env_path, _defaults)
load_env_file(_env_path) load_env_file(_env_path)
# Where to fetch a quote from (API) # Where to fetch a quote from (API)
API_URL = os.environ.get('SCHLOTER_API_URL', _defaults['SCHLOTER_API_URL']) API_URL = os.environ.get('SCHLOTER_API_URL', _defaults['SCHLOTER_API_URL'])
@ -60,6 +86,7 @@ TEAMS_WEBHOOK_URL = os.environ.get('TEAMS_WEBHOOK_URL', _defaults['TEAMS_WEBHOOK
# Static token for the API (can be set via environment variable) # Static token for the API (can be set via environment variable)
API_TOKEN = os.environ.get('SCHLOTER_API_TOKEN', _defaults['SCHLOTER_API_TOKEN']) API_TOKEN = os.environ.get('SCHLOTER_API_TOKEN', _defaults['SCHLOTER_API_TOKEN'])
def get_quote(): def get_quote():
"""Fetch a quote from the API using a Bearer token. """Fetch a quote from the API using a Bearer token.
@ -69,30 +96,87 @@ def get_quote():
resp = requests.get(API_URL, headers=headers, timeout=10) resp = requests.get(API_URL, headers=headers, timeout=10)
resp.raise_for_status() resp.raise_for_status()
data = resp.json() data = resp.json()
return data.get("quote", "No quote found.") # API returns {'id': ..., 'quote': '...'}
return data.get("quote")
def send_teams_message(webhook_url, title, message, color="0076D7"):
"""
Sends a formatted message to a Microsoft Teams channel using a webhook as a MessageCard.
Treats HTTP 200 and 202 as success and prints response details.
"""
headers = {"Content-Type": "application/json"}
payload = {
"@type": "MessageCard",
"@context": "https://schema.org/extensions",
"themeColor": color,
"summary": title,
"sections": [
{
"activityTitle": f"**{title}**",
"text": message
}
]
}
try:
response = requests.post(webhook_url, headers=headers, data=json.dumps(payload), timeout=10)
# Microsoft Teams incoming webhooks may return 200 or 202 for success
if response.status_code in (200, 202):
text = response.text.strip()
if text:
print(f"✅ Message sent successfully! Response: {response.status_code} - {text}")
else:
print(f"✅ Message sent successfully! Response: {response.status_code}")
return True
else:
print(f"❌ Failed to send message: {response.status_code} - {response.text}")
return False
except Exception as e:
print(f"⚠️ Error sending message: {e}")
return False
def post_to_teams(quote): def post_to_teams(quote):
"""Post the quote to a Teams Workflows (incoming webhook) URL. """Wrapper to post the daily quote as a MessageCard to Teams."""
title = "Daily Quote"
Teams incoming webhooks accept a simple JSON payload like {"text": "..."}. return send_teams_message(TEAMS_WEBHOOK_URL, title=title, message=quote)
This function posts that payload and raises on failure.
"""
payload = {"text": quote}
headers = {"Content-Type": "application/json"}
try:
resp = requests.post(TEAMS_WEBHOOK_URL, json=payload, headers=headers, timeout=10)
resp.raise_for_status()
except requests.RequestException as e:
print(f"Failed to post to Teams webhook: {e}")
raise
else:
print("Posted to Teams webhook successfully.")
if __name__ == "__main__": if __name__ == "__main__":
# Fetch a quote and post it to Teams. Use env vars to override defaults. # Allow a debug/dry-run mode so we don't have to call the schloter API.
quote = get_quote() p = argparse.ArgumentParser(description="Fetch quote (or use local debug quote) and post to Teams webhook")
print("Quote fetched:", quote) p.add_argument('--no-api', action='store_true', help='Do not call the schloter API; use a local quote instead')
# Uncomment the next line to actually send to Teams (ensure TEAMS_WEBHOOK_URL is set) p.add_argument('--local-quote', help='Local quote text to use when --no-api is set')
# post_to_teams(quote) args = p.parse_args()
# ...existing code from schloters_dailys.py...
no_api_env = os.environ.get('SCHLOTER_DEBUG_NO_API', '').lower() in ('1', 'true', 'yes')
use_no_api = args.no_api or no_api_env
if use_no_api:
# Choose local quote from CLI, env, or a sensible default
quote = args.local_quote or os.environ.get('SCHLOTER_LOCAL_QUOTE') or 'Debug quote: no API call performed.'
print('DEBUG: Running without calling schloter API; using local quote.')
else:
try:
quote = get_quote()
except Exception as e:
print(f"Failed to fetch quote from API: {e}")
sys.exit(2)
if not quote:
print("No quote available")
sys.exit(1)
print("Quote to send:", quote)
ok = post_to_teams(quote)
if not ok:
print("Failed to post quote to Teams")
sys.exit(3)
print("Quote successfully posted to Teams.")
sys.exit(0)