Update configuration and enhance newsletter generation features
- Change WIKI_API_URL in .env.example and config to point to the new API endpoint. - Set COOKIE_SECURE to false for local development in .env.example and config. - Improve the newsletter generation logic to handle new and edited articles separately. - Add display options for showing date, user, and category in the newsletter output. - Enhance error handling for Wiki API requests and update templates for better user feedback. - Update CSS for new UI elements and improve overall layout in dashboard and login pages.
This commit is contained in:
@@ -1,6 +1,38 @@
|
||||
from datetime import datetime
|
||||
from datetime import datetime, timedelta
|
||||
from html import escape
|
||||
from typing import Any
|
||||
from typing import Any, TypedDict
|
||||
from urllib.parse import quote
|
||||
|
||||
WIKI_ARTICLE_BASE = "https://www.thomas-krenn.com/de/wiki/"
|
||||
WIKI_HOME_URL = "https://www.thomas-krenn.com/de/wiki/Hauptseite"
|
||||
ORG_NAME = "Thomas-Krenn.AG"
|
||||
LINE = "=" * 78
|
||||
SUBLINE = "-" * 78
|
||||
|
||||
|
||||
class DisplayOptions(TypedDict):
|
||||
show_date: bool
|
||||
show_user: bool
|
||||
show_category: bool
|
||||
|
||||
|
||||
DEFAULT_DISPLAY: DisplayOptions = {
|
||||
"show_date": True,
|
||||
"show_user": True,
|
||||
"show_category": True,
|
||||
}
|
||||
|
||||
|
||||
def parse_display_options(
|
||||
show_date: str | None = None,
|
||||
show_user: str | None = None,
|
||||
show_category: str | None = None,
|
||||
) -> DisplayOptions:
|
||||
return {
|
||||
"show_date": show_date == "true",
|
||||
"show_user": show_user == "true",
|
||||
"show_category": show_category == "true",
|
||||
}
|
||||
|
||||
|
||||
def filter_articles(
|
||||
@@ -17,60 +49,280 @@ def filter_articles(
|
||||
]
|
||||
|
||||
|
||||
def create_plain_text(articles: list[dict[str, Any]], title: str) -> str:
|
||||
lines = [title, "=" * len(title), ""]
|
||||
if not articles:
|
||||
lines.append("Keine Artikel im gewählten Zeitraum gefunden.")
|
||||
return "\n".join(lines)
|
||||
def split_articles_by_type(articles: list[dict[str, Any]]) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
|
||||
sorted_articles = sorted(articles, key=lambda a: a.get("timestamp", ""), reverse=True)
|
||||
new_articles: list[dict[str, Any]] = []
|
||||
edited_articles: list[dict[str, Any]] = []
|
||||
new_titles: set[str] = set()
|
||||
|
||||
for idx, item in enumerate(articles, start=1):
|
||||
stamp = _format_ts(item.get("timestamp"))
|
||||
cat = ", ".join(item.get("categories", [])) or "Keine Kategorie"
|
||||
comment = item.get("comment") or "-"
|
||||
for item in sorted_articles:
|
||||
title = item.get("title", "")
|
||||
if not title or item.get("type") != "new":
|
||||
continue
|
||||
if title not in new_titles:
|
||||
new_articles.append(item)
|
||||
new_titles.add(title)
|
||||
|
||||
seen_edited: set[str] = set()
|
||||
for item in sorted_articles:
|
||||
title = item.get("title", "")
|
||||
if not title or title in new_titles or title in seen_edited:
|
||||
continue
|
||||
edited_articles.append(item)
|
||||
seen_edited.add(title)
|
||||
|
||||
return new_articles, edited_articles
|
||||
|
||||
|
||||
def create_subject(days: int, category: str = "") -> str:
|
||||
base = f"Interner Wiki-Newsletter – {ORG_NAME} ({days} Tage)"
|
||||
if category.strip():
|
||||
return f"{base} | Kategorie: {category.strip()}"
|
||||
return base
|
||||
|
||||
|
||||
def create_plain_text(
|
||||
articles: list[dict[str, Any]],
|
||||
days: int,
|
||||
display: DisplayOptions | None = None,
|
||||
category: str = "",
|
||||
) -> str:
|
||||
display = display or DEFAULT_DISPLAY
|
||||
new_articles, edited_articles = split_articles_by_type(articles)
|
||||
period_start, period_end = _period_range(days)
|
||||
created_at = datetime.now().strftime("%d.%m.%Y %H:%M")
|
||||
|
||||
lines = [
|
||||
LINE,
|
||||
f"{ORG_NAME.upper()} – INTERNER WIKI-NEWSLETTER",
|
||||
LINE,
|
||||
f"Zeitraum: {period_start} bis {period_end} ({days} Tage)",
|
||||
f"Erstellt am: {created_at}",
|
||||
]
|
||||
if category.strip():
|
||||
lines.append(f"Kategorie: {category.strip()}")
|
||||
lines.extend(
|
||||
[
|
||||
"",
|
||||
"Liebe Kolleginnen und Kollegen,",
|
||||
"",
|
||||
"im Thomas-Krenn-Wiki wurden im ausgewählten Zeitraum folgende Artikel",
|
||||
"neu veröffentlicht bzw. aktualisiert:",
|
||||
"",
|
||||
]
|
||||
)
|
||||
|
||||
if not new_articles and not edited_articles:
|
||||
lines.extend(
|
||||
[
|
||||
f"{idx}. {item.get('title', 'Ohne Titel')}",
|
||||
f" Zeit: {stamp}",
|
||||
f" Benutzer: {item.get('user', '-')}",
|
||||
f" Kategorien: {cat}",
|
||||
f" Kommentar: {comment}",
|
||||
SUBLINE,
|
||||
"HINWEIS",
|
||||
SUBLINE,
|
||||
"Im gewählten Zeitraum wurden keine passenden Wiki-Änderungen gefunden.",
|
||||
"",
|
||||
_plain_footer(new_articles, edited_articles),
|
||||
]
|
||||
)
|
||||
return "\n".join(lines).strip()
|
||||
|
||||
lines.extend(_plain_section("NEUE ARTIKEL", new_articles, display, empty_text="Keine neuen Artikel im Zeitraum."))
|
||||
lines.append("")
|
||||
lines.extend(
|
||||
_plain_section(
|
||||
"BEARBEITETE ARTIKEL",
|
||||
edited_articles,
|
||||
display,
|
||||
empty_text="Keine bearbeiteten Artikel im Zeitraum.",
|
||||
)
|
||||
)
|
||||
lines.append("")
|
||||
lines.append(_plain_footer(new_articles, edited_articles))
|
||||
return "\n".join(lines).strip()
|
||||
|
||||
|
||||
def create_html(articles: list[dict[str, Any]], title: str) -> str:
|
||||
if not articles:
|
||||
return f"<h2>{escape(title)}</h2><p>Keine Artikel im gewählten Zeitraum gefunden.</p>"
|
||||
def create_html(
|
||||
articles: list[dict[str, Any]],
|
||||
days: int,
|
||||
display: DisplayOptions | None = None,
|
||||
category: str = "",
|
||||
) -> str:
|
||||
display = display or DEFAULT_DISPLAY
|
||||
new_articles, edited_articles = split_articles_by_type(articles)
|
||||
period_start, period_end = _period_range(days)
|
||||
created_at = datetime.now().strftime("%d.%m.%Y %H:%M")
|
||||
subject = escape(create_subject(days, category))
|
||||
|
||||
blocks = [f"<h2>{escape(title)}</h2>", "<ul>"]
|
||||
for item in articles:
|
||||
article_title = escape(item.get("title", "Ohne Titel"))
|
||||
stamp = escape(_format_ts(item.get("timestamp")))
|
||||
user = escape(item.get("user", "-"))
|
||||
cat = escape(", ".join(item.get("categories", [])) or "Keine Kategorie")
|
||||
comment = escape(item.get("comment") or "-")
|
||||
blocks.append(
|
||||
(
|
||||
"<li>"
|
||||
f"<strong>{article_title}</strong><br>"
|
||||
f"Zeit: {stamp}<br>"
|
||||
f"Benutzer: {user}<br>"
|
||||
f"Kategorien: {cat}<br>"
|
||||
f"Kommentar: {comment}"
|
||||
"</li>"
|
||||
)
|
||||
category_row = ""
|
||||
if category.strip():
|
||||
category_row = f'<p style="margin:4px 0;color:#334155;"><strong>Kategorie:</strong> {escape(category.strip())}</p>'
|
||||
|
||||
if not new_articles and not edited_articles:
|
||||
body = '<p style="margin:16px 0;">Im gewählten Zeitraum wurden keine passenden Wiki-Änderungen gefunden.</p>'
|
||||
else:
|
||||
body = "\n".join(
|
||||
[
|
||||
_html_section("Neue Artikel", new_articles, display, "Keine neuen Artikel im Zeitraum."),
|
||||
_html_section("Bearbeitete Artikel", edited_articles, display, "Keine bearbeiteten Artikel im Zeitraum."),
|
||||
]
|
||||
)
|
||||
blocks.append("</ul>")
|
||||
|
||||
summary = _html_summary(new_articles, edited_articles)
|
||||
return f"""<!DOCTYPE html>
|
||||
<html lang="de">
|
||||
<head><meta charset="utf-8"><title>{subject}</title></head>
|
||||
<body style="margin:0;padding:0;background:#f4f6fa;font-family:Arial,Helvetica,sans-serif;color:#1f2430;">
|
||||
<table role="presentation" width="100%" cellpadding="0" cellspacing="0" style="background:#f4f6fa;padding:24px 0;">
|
||||
<tr><td align="center">
|
||||
<table role="presentation" width="640" cellpadding="0" cellspacing="0" style="background:#ffffff;border:1px solid #dde3ef;border-radius:8px;overflow:hidden;">
|
||||
<tr><td style="background:#0c2c5a;color:#ffffff;padding:20px 24px;">
|
||||
<h1 style="margin:0;font-size:20px;">Interner Wiki-Newsletter</h1>
|
||||
<p style="margin:8px 0 0;font-size:14px;">{escape(ORG_NAME)}</p>
|
||||
</td></tr>
|
||||
<tr><td style="padding:24px;">
|
||||
<p style="margin:0 0 8px;color:#334155;"><strong>Zeitraum:</strong> {period_start} bis {period_end} ({days} Tage)</p>
|
||||
<p style="margin:0 0 8px;color:#334155;"><strong>Erstellt am:</strong> {created_at}</p>
|
||||
{category_row}
|
||||
<p style="margin:20px 0 8px;line-height:1.5;">Liebe Kolleginnen und Kollegen,</p>
|
||||
<p style="margin:0 0 20px;line-height:1.5;">im Thomas-Krenn-Wiki wurden im ausgewählten Zeitraum folgende Artikel neu veröffentlicht bzw. aktualisiert:</p>
|
||||
{body}
|
||||
{summary}
|
||||
<p style="margin:24px 0 0;"><a href="{WIKI_HOME_URL}" style="color:#1b5dbf;">Zum Thomas-Krenn-Wiki</a></p>
|
||||
</td></tr>
|
||||
<tr><td style="background:#f8fafc;padding:16px 24px;border-top:1px solid #e6ebf4;font-size:12px;color:#64748b;line-height:1.5;">
|
||||
Dieser Newsletter wurde automatisch erstellt und ist nur für den internen Gebrauch bei {escape(ORG_NAME)} bestimmt.
|
||||
</td></tr>
|
||||
</table>
|
||||
</td></tr>
|
||||
</table>
|
||||
</body>
|
||||
</html>"""
|
||||
|
||||
|
||||
def _plain_section(
|
||||
heading: str,
|
||||
items: list[dict[str, Any]],
|
||||
display: DisplayOptions,
|
||||
empty_text: str,
|
||||
) -> list[str]:
|
||||
lines = [SUBLINE, f"{heading} ({len(items)})", SUBLINE, ""]
|
||||
if not items:
|
||||
lines.append(f" {empty_text}")
|
||||
lines.append("")
|
||||
return lines
|
||||
|
||||
for idx, item in enumerate(items, start=1):
|
||||
lines.extend(_plain_item(idx, item, display))
|
||||
return lines
|
||||
|
||||
|
||||
def _plain_item(idx: int, item: dict[str, Any], display: DisplayOptions) -> list[str]:
|
||||
title = item.get("title", "Ohne Titel")
|
||||
url = _article_url(item.get("title", ""))
|
||||
meta_parts: list[str] = []
|
||||
if display["show_date"]:
|
||||
meta_parts.append(_format_ts(item.get("timestamp")))
|
||||
if display["show_user"]:
|
||||
meta_parts.append(str(item.get("user", "-")))
|
||||
if display["show_category"]:
|
||||
cat = ", ".join(item.get("categories", [])) or "Keine Kategorie"
|
||||
meta_parts.append(cat)
|
||||
|
||||
lines = [f" {idx}. {title}", f" {url}"]
|
||||
if meta_parts:
|
||||
lines.append(f" ({' | '.join(meta_parts)})")
|
||||
lines.append("")
|
||||
return lines
|
||||
|
||||
|
||||
def _plain_footer(new_articles: list[dict[str, Any]], edited_articles: list[dict[str, Any]]) -> str:
|
||||
total = len(new_articles) + len(edited_articles)
|
||||
return "\n".join(
|
||||
[
|
||||
SUBLINE,
|
||||
"ZUSAMMENFASSUNG",
|
||||
SUBLINE,
|
||||
f" Neue Artikel: {len(new_articles)}",
|
||||
f" Bearbeitete Artikel: {len(edited_articles)}",
|
||||
f" Gesamt: {total}",
|
||||
"",
|
||||
f"Wiki: {WIKI_HOME_URL}",
|
||||
"",
|
||||
SUBLINE,
|
||||
f"Dieser Newsletter wurde automatisch erstellt und ist nur für den internen",
|
||||
f"Gebrauch bei {ORG_NAME} bestimmt.",
|
||||
LINE,
|
||||
]
|
||||
)
|
||||
|
||||
|
||||
def _html_section(heading: str, items: list[dict[str, Any]], display: DisplayOptions, empty_text: str) -> str:
|
||||
blocks = [
|
||||
f'<h2 style="margin:24px 0 8px;font-size:16px;color:#0c2c5a;border-bottom:2px solid #dde3ef;padding-bottom:6px;">{escape(heading)} ({len(items)})</h2>'
|
||||
]
|
||||
if not items:
|
||||
blocks.append(f'<p style="margin:8px 0 0;color:#64748b;">{escape(empty_text)}</p>')
|
||||
return "\n".join(blocks)
|
||||
|
||||
blocks.append('<table role="presentation" width="100%" cellpadding="0" cellspacing="0" style="margin-top:8px;">')
|
||||
for idx, item in enumerate(items, start=1):
|
||||
blocks.append(_html_item_row(idx, item, display))
|
||||
blocks.append("</table>")
|
||||
return "\n".join(blocks)
|
||||
|
||||
|
||||
def _html_item_row(idx: int, item: dict[str, Any], display: DisplayOptions) -> str:
|
||||
title = escape(item.get("title", "Ohne Titel"))
|
||||
url = escape(_article_url(item.get("title", "")))
|
||||
meta_parts: list[str] = []
|
||||
if display["show_date"]:
|
||||
meta_parts.append(f"<strong>Datum:</strong> {escape(_format_ts(item.get('timestamp')))}")
|
||||
if display["show_user"]:
|
||||
meta_parts.append(f"<strong>Benutzer:</strong> {escape(str(item.get('user', '-')))}")
|
||||
if display["show_category"]:
|
||||
cat = escape(", ".join(item.get("categories", [])) or "Keine Kategorie")
|
||||
meta_parts.append(f"<strong>Kategorie:</strong> {cat}")
|
||||
meta_html = f'<div style="font-size:12px;color:#64748b;margin-top:4px;">{" | ".join(meta_parts)}</div>' if meta_parts else ""
|
||||
|
||||
return f"""<tr>
|
||||
<td style="padding:10px 0;border-bottom:1px solid #e6ebf4;vertical-align:top;width:28px;color:#64748b;">{idx}.</td>
|
||||
<td style="padding:10px 0;border-bottom:1px solid #e6ebf4;vertical-align:top;">
|
||||
<a href="{url}" style="color:#1b5dbf;font-weight:bold;text-decoration:none;">{title}</a>
|
||||
{meta_html}
|
||||
</td>
|
||||
</tr>"""
|
||||
|
||||
|
||||
def _html_summary(new_articles: list[dict[str, Any]], edited_articles: list[dict[str, Any]]) -> str:
|
||||
total = len(new_articles) + len(edited_articles)
|
||||
return f"""<table role="presentation" width="100%" cellpadding="0" cellspacing="0" style="margin-top:24px;background:#f8fafc;border:1px solid #e6ebf4;border-radius:6px;">
|
||||
<tr><td style="padding:14px 16px;font-size:14px;line-height:1.6;">
|
||||
<strong>Zusammenfassung</strong><br>
|
||||
Neue Artikel: {len(new_articles)}<br>
|
||||
Bearbeitete Artikel: {len(edited_articles)}<br>
|
||||
Gesamt: {total}
|
||||
</td></tr>
|
||||
</table>"""
|
||||
|
||||
|
||||
def _article_url(title: str) -> str:
|
||||
return f"{WIKI_ARTICLE_BASE}{quote(title.replace(' ', '_'))}"
|
||||
|
||||
|
||||
def article_url(title: str) -> str:
|
||||
return _article_url(title)
|
||||
|
||||
|
||||
def _period_range(days: int) -> tuple[str, str]:
|
||||
end = datetime.now()
|
||||
start = end - timedelta(days=days)
|
||||
return start.strftime("%d.%m.%Y"), end.strftime("%d.%m.%Y")
|
||||
|
||||
|
||||
def _format_ts(value: str | None) -> str:
|
||||
if not value:
|
||||
return "-"
|
||||
try:
|
||||
dt = datetime.strptime(value, "%Y-%m-%dT%H:%M:%SZ")
|
||||
return dt.strftime("%d.%m.%Y %H:%M UTC")
|
||||
return dt.strftime("%d.%m.%Y %H:%M")
|
||||
except ValueError:
|
||||
return value
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import smtplib
|
||||
from datetime import datetime
|
||||
from email.mime.multipart import MIMEMultipart
|
||||
from email.mime.text import MIMEText
|
||||
from typing import Sequence
|
||||
|
||||
@@ -40,7 +41,9 @@ def send_newsletter(
|
||||
_log(db, subject, recipients, "failed", "SMTP unvollständig konfiguriert.", scheduled)
|
||||
return
|
||||
|
||||
msg = MIMEText(html_content, "html", "utf-8")
|
||||
msg = MIMEMultipart("alternative")
|
||||
msg.attach(MIMEText(text_content, "plain", "utf-8"))
|
||||
msg.attach(MIMEText(html_content, "html", "utf-8"))
|
||||
msg["Subject"] = subject
|
||||
msg["From"] = settings["from_email"]
|
||||
msg["To"] = ", ".join(recipients)
|
||||
|
||||
@@ -6,6 +6,14 @@ import httpx
|
||||
|
||||
from app.core.config import settings
|
||||
|
||||
USER_AGENT = "TK-Newsletter-Admin/1.0 (internal; contact@thomas-krenn.com)"
|
||||
|
||||
|
||||
class WikiFetchError(Exception):
|
||||
def __init__(self, message: str) -> None:
|
||||
self.message = message
|
||||
super().__init__(message)
|
||||
|
||||
|
||||
class WikiService:
|
||||
def __init__(self) -> None:
|
||||
@@ -17,7 +25,7 @@ class WikiService:
|
||||
rcstart = now.strftime("%Y-%m-%dT%H:%M:%SZ")
|
||||
rcend = start.strftime("%Y-%m-%dT%H:%M:%SZ")
|
||||
|
||||
params = {
|
||||
params: dict[str, str] = {
|
||||
"action": "query",
|
||||
"format": "json",
|
||||
"list": "recentchanges",
|
||||
@@ -34,18 +42,31 @@ class WikiService:
|
||||
|
||||
rows: list[dict[str, Any]] = []
|
||||
cont: dict[str, Any] = {}
|
||||
async with httpx.AsyncClient(timeout=20.0) as client:
|
||||
while True:
|
||||
response = await client.get(self.api_url, params={**params, **cont})
|
||||
response.raise_for_status()
|
||||
payload = response.json()
|
||||
rows.extend(payload.get("query", {}).get("recentchanges", []))
|
||||
if "continue" not in payload:
|
||||
break
|
||||
cont = payload["continue"]
|
||||
headers = {"User-Agent": USER_AGENT}
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=30.0, headers=headers, follow_redirects=True) as client:
|
||||
while True:
|
||||
response = await client.get(self.api_url, params={**params, **cont})
|
||||
if response.status_code >= 400:
|
||||
raise WikiFetchError(
|
||||
f"Wiki-API nicht erreichbar (HTTP {response.status_code}). "
|
||||
f"Prüfe WIKI_API_URL in der Konfiguration."
|
||||
)
|
||||
payload = response.json()
|
||||
if "error" in payload:
|
||||
code = payload["error"].get("code", "unknown")
|
||||
info = payload["error"].get("info", "Unbekannter API-Fehler")
|
||||
raise WikiFetchError(f"Wiki-API-Fehler ({code}): {info}")
|
||||
|
||||
titles = sorted({item["title"] for item in rows if item.get("title")})
|
||||
categories = await self._get_categories_for_titles(client, titles)
|
||||
rows.extend(payload.get("query", {}).get("recentchanges", []))
|
||||
if "continue" not in payload:
|
||||
break
|
||||
cont = payload["continue"]
|
||||
|
||||
titles = sorted({item["title"] for item in rows if item.get("title")})
|
||||
categories = await self._get_categories_for_titles(client, titles)
|
||||
except httpx.RequestError as exc:
|
||||
raise WikiFetchError(f"Wiki-API-Anfrage fehlgeschlagen: {exc}") from exc
|
||||
|
||||
for row in rows:
|
||||
row["categories"] = categories.get(row.get("title", ""), [])
|
||||
@@ -67,12 +88,15 @@ class WikiService:
|
||||
"titles": "|".join(chunk),
|
||||
}
|
||||
response = await client.get(self.api_url, params=params)
|
||||
response.raise_for_status()
|
||||
if response.status_code >= 400:
|
||||
continue
|
||||
pages = response.json().get("query", {}).get("pages", {})
|
||||
for page in pages.values():
|
||||
title = page.get("title")
|
||||
if not title:
|
||||
continue
|
||||
raw_categories = page.get("categories", [])
|
||||
result[title] = [c.get("title", "").replace("Kategorie:", "") for c in raw_categories if c.get("title")]
|
||||
result[title] = [
|
||||
c.get("title", "").replace("Kategorie:", "") for c in raw_categories if c.get("title")
|
||||
]
|
||||
return result
|
||||
|
||||
Reference in New Issue
Block a user