BlueMastoFeed/bluemastofeed.py

import os
import time
import feedparser
import json
import logging
import requests
import threading
import smtplib
from bs4 import BeautifulSoup
from io import BytesIO
from mastodon import Mastodon
from atproto import Client
from dotenv import load_dotenv
from http.server import HTTPServer, BaseHTTPRequestHandler
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

load_dotenv()
FEED_URL = os.getenv("FEED_URL")
SEEN_POSTS_FILE = "/data/seen_posts.txt"

MASTODON_BASE_URL = os.getenv("MASTODON_API_BASE_URL")
MASTODON_TOKEN = os.getenv("MASTODON_ACCESS_TOKEN")
BSKY_HANDLE = os.getenv("BSKY_IDENTIFIER")
BSKY_PASSWORD = os.getenv("BSKY_PASSWORD")

# Logging konfigurieren (Standard-Format)
logger = logging.getLogger()
logger.setLevel(logging.INFO)
handler = logging.StreamHandler()  # Log an stdout (Docker-Standard)
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
logger.addHandler(handler)


class HealthHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path == "/health":
            self.send_response(200)
            self.end_headers()
            self.wfile.write(b"OK")
        else:
            self.send_response(404)
            self.end_headers()

def start_health_server():
    server = HTTPServer(("0.0.0.0", 8000), HealthHandler)
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    logger.info("Healthcheck server runs on port 8000.")


def should_send_email(on_success: bool):
    mode = os.getenv("EMAIL_MODE", "errors").lower()
    if mode == "none":
        return False
    if mode == "all":
        return True
    if mode == "errors" and not on_success:
        return True
    return False


def send_status_email(subject, html_content):
    try:
        smtp_host = os.getenv("SMTP_HOST")
        smtp_port = int(os.getenv("SMTP_PORT", 587))
        smtp_user = os.getenv("SMTP_USER")
        smtp_password = os.getenv("SMTP_PASSWORD")
        email_from = os.getenv("EMAIL_FROM")
        email_to = os.getenv("EMAIL_TO")

        msg = MIMEMultipart("alternative")
        msg["Subject"] = subject
        msg["From"] = email_from
        msg["To"] = email_to

        part = MIMEText(html_content, "html")
        msg.attach(part)

        with smtplib.SMTP(smtp_host, smtp_port) as server:
            server.starttls()
            server.login(smtp_user, smtp_password)
            server.sendmail(email_from, email_to, msg.as_string())
        logger.info("Status E-Mail gesendet.")
    except Exception as e:
        logger.error(f"Fehler beim Senden der E-Mail: {e}")


def load_seen_ids():
    os.makedirs(os.path.dirname(SEEN_POSTS_FILE), exist_ok=True)
    if not os.path.exists(SEEN_POSTS_FILE):
        with open(SEEN_POSTS_FILE, "w"): pass
        return set()
    with open(SEEN_POSTS_FILE, "r") as f:
        return set(line.strip() for line in f)

def save_seen_id(post_id):
    with open(SEEN_POSTS_FILE, "a") as f:
        f.write(post_id + "\n")

def post_to_mastodon(message):
    mastodon = Mastodon(access_token=MASTODON_TOKEN, api_base_url=MASTODON_BASE_URL)
    mastodon.toot(message)

def fetch_og_data(url):
    try:
        resp = requests.get(url, timeout=10)
        resp.raise_for_status()
        soup = BeautifulSoup(resp.text, "html.parser")
        og_title = soup.find("meta", property="og:title")
        og_image = soup.find("meta", property="og:image")
        title = og_title["content"] if og_title and og_title.has_attr("content") else None
        image_url = og_image["content"] if og_image and og_image.has_attr("content") else None
        return title, image_url
    except Exception as e:
        logger.error(f"Error loading OG data: {e}")
        return None, None

def post_to_bluesky(message, link):
    client = Client()
    client.login(BSKY_HANDLE, BSKY_PASSWORD)

    title, image_url = fetch_og_data(link)
    text = title or message

    if title and image_url:
        try:
            embed = {
                "$type": "app.bsky.embed.external",
                "external": {
                    "uri": link,
                    "title": title,
                    "description": "",  # Optional: Beschreibung kannst du per OG:description holen
                    "thumb": {
                        "$type": "blob",
                        "ref": None,  # Wird vom Upload ersetzt
                        "mimeType": "",  # Wird vom Upload ersetzt
                        "size": 0  # Wird vom Upload ersetzt
                    }
                }
            }

            # Bild herunterladen und hochladen
            img_resp = requests.get(image_url, timeout=10)
            img_resp.raise_for_status()
            image_bytes = BytesIO(img_resp.content)

            blob = client.upload_blob(image_bytes)
            embed["external"]["thumb"] = blob.blob  # Automatisch ersetzt

            client.send_post(text=text, embed=embed)
            logger.info("Posted with OG preview.")
            return
        except Exception as e:
            logger.error(f"Error uploading OG preview: {e}")

    # Fallback: Nur Text + Link
    client.send_post(f"{text}\n{link}")
    logger.info("Posted without OG preview.")

def main():
    seen_ids = load_seen_ids()
    feed = feedparser.parse(FEED_URL)

    for entry in feed.entries:
        post_id = entry.get("id") or entry.get("link")
        if post_id in seen_ids:
            continue

        title = entry.get("title", "").strip()
        link = entry.get("link", "").strip()
        message = link  # Link alleine posten für Mastodon OG-Vorschau

        logger.info(f"New post: {title}")

        try:
            post_to_mastodon(message)
            time.sleep(2)
            post_to_bluesky(message, link)
            save_seen_id(post_id)
            logger.info("Successfully posted.")

            if should_send_email(on_success=True):
                email_subject = f"✅ Erfolgreich gepostet: {title}"
                email_body = f"""
                <html>
                  <body>
                    <h2>Beitrag erfolgreich gepostet</h2>
                    <p><strong>Titel:</strong> {title}</p>
                    <p><strong>Link:</strong> <a href="{link}">{link}</a></p>
                  </body>
                </html>
                """
                send_status_email(email_subject, email_body)

        except Exception as e:
            logger.error(f"Error posting: {e}")
            
            if should_send_email(on_success=False):
                email_subject = f"❌ Fehler beim Posten: {title}"
                email_body = f"""
                <html>
                  <body>
                    <h2>Fehler beim Posten</h2>
                    <p><strong>Titel:</strong> {title}</p>
                    <p><strong>Link:</strong> <a href="{link}">{link}</a></p>
                    <p><strong>Fehlermeldung:</strong> {str(e)}</p>
                  </body>
                </html>
                """
                send_status_email(email_subject, email_body)

        time.sleep(5)

if __name__ == "__main__":
    INTERVAL_MINUTES = int(os.getenv("INTERVAL_MINUTES", 30))  # Default: 30 Minuten
    logger.info(f"Start feed check every {INTERVAL_MINUTES} minutes.")

    start_health_server()  # HTTP-Healthcheck starten

    while True:
        try:
            main()
        except Exception as e:
            logger.error(f"Error in main execution: {e}")
        logger.info(f"Wait {INTERVAL_MINUTES} minutes until next execution...")
        time.sleep(INTERVAL_MINUTES * 60)