Initial commit: Clean project structure without secrets in history
This commit is contained in:
Michaelis
163
web_check.py
Normal file
163
web_check.py
Normal file
@@ -0,0 +1,163 @@
|
||||
import cloudscraper
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import difflib
|
||||
import sys
|
||||
from bs4 import BeautifulSoup
|
||||
from urllib.parse import urljoin
|
||||
from datetime import datetime
|
||||
from dotenv import load_dotenv
|
||||
|
||||
# --- KONFIGURATION ---
|
||||
# Laden der Umgebungsvariablen aus .env Datei
|
||||
load_dotenv()
|
||||
|
||||
ADMIN_USER = os.getenv("ADMIN_USER")
|
||||
ADMIN_PW = os.getenv("ADMIN_PW")
|
||||
NTFY_BASE = os.getenv("NTFY_BASE")
|
||||
|
||||
JSON_DATEI = "./results.json"
|
||||
GEMINI_SESSION = "1"
|
||||
|
||||
# ntfy-Themen (Öffentlich)
|
||||
TOPIC_PUBLIC = "polit-scraper-public"
|
||||
TOPIC_ADMIN = "polit-scraper-admin"
|
||||
|
||||
# UTF-8 für Konsole erzwingen
|
||||
sys.stdout.reconfigure(encoding='utf-8')
|
||||
|
||||
# Cloudscraper initialisieren (Chrome-Emulation)
|
||||
scraper = cloudscraper.create_scraper(
|
||||
browser={'browser': 'chrome', 'platform': 'windows', 'desktop': True}
|
||||
)
|
||||
|
||||
def analysiere_mit_gemini(text, links):
|
||||
"""KI-Analyse: Text und Links an Gemini senden, JSON extrahieren."""
|
||||
links_text = "\nRelevante Links: " + ", ".join(links) if links else ""
|
||||
anweisung = f"Analysiere diesen Text: {text}{links_text}"
|
||||
befehl = ["gemini", "-r", GEMINI_SESSION, "-p", anweisung]
|
||||
try:
|
||||
prozess = subprocess.run(befehl, capture_output=True, text=True, encoding='utf-8')
|
||||
antwort_roh = prozess.stdout.strip()
|
||||
start = antwort_roh.find('{')
|
||||
ende = antwort_roh.rfind('}') + 1
|
||||
if start != -1 and ende != -1:
|
||||
return json.loads(antwort_roh[start:ende])
|
||||
return {"relevant": False}
|
||||
except Exception as e:
|
||||
print(f" [!] KI-Fehler: {e}")
|
||||
return {"relevant": False}
|
||||
|
||||
def sende_public(titel, nachricht, klick_url, category="info", priority=3):
|
||||
"""Öffentlicher ntfy-Versand mit Priorität und Klick-URL."""
|
||||
if not NTFY_BASE: return
|
||||
url = f"{NTFY_BASE}/{TOPIC_PUBLIC}"
|
||||
emoji_mapping = {
|
||||
"demo": "rotating_light,loudspeaker",
|
||||
"socialmedia": "detective,no_entry",
|
||||
"polizei": "police_car,warning",
|
||||
"stadt": "cityscape,newspaper",
|
||||
"solidaritaet": "fist,heart",
|
||||
"info": "newspaper"
|
||||
}
|
||||
tags = emoji_mapping.get(category.lower(), "newspaper")
|
||||
headers = {
|
||||
"Title": titel.encode('utf-8'),
|
||||
"Tags": tags,
|
||||
"Click": klick_url,
|
||||
"Priority": str(priority)
|
||||
}
|
||||
try:
|
||||
import requests
|
||||
requests.post(url, data=nachricht.encode("utf-8"), headers=headers,
|
||||
auth=(ADMIN_USER, ADMIN_PW), timeout=5)
|
||||
except Exception as e:
|
||||
print(f" [!] Fehler Public-Versand: {e}")
|
||||
|
||||
def sende_admin(titel, nachricht):
|
||||
"""Admin-Benachrichtigung bei technischen Fehlern."""
|
||||
if not NTFY_BASE: return
|
||||
url = f"{NTFY_BASE}/{TOPIC_ADMIN}"
|
||||
headers = {"Title": titel.encode('utf-8'), "Tags": "warning,skull", "Priority": "4"}
|
||||
try:
|
||||
import requests
|
||||
requests.post(url, data=nachricht.encode("utf-8"), headers=headers,
|
||||
auth=(ADMIN_USER, ADMIN_PW), timeout=5)
|
||||
except Exception as e:
|
||||
print(f" [!] Fehler Admin-Versand: {e}")
|
||||
|
||||
def lade_gedaechtnis():
|
||||
if not os.path.exists(JSON_DATEI): return {}
|
||||
with open(JSON_DATEI, "r", encoding='utf-8') as f:
|
||||
try: return json.load(f)
|
||||
except: return {}
|
||||
|
||||
def speichere_gedaechtnis(daten):
|
||||
with open(JSON_DATEI, "w", encoding='utf-8') as f:
|
||||
json.dump(daten, f, indent=4, ensure_ascii=False)
|
||||
|
||||
def extrahiere_links(soup, basis_url):
|
||||
links = []
|
||||
for a in soup.find_all('a', href=True):
|
||||
links.append(urljoin(basis_url, a['href']))
|
||||
return list(set(links))
|
||||
|
||||
def verarbeite_website(url, gedaechtnis):
|
||||
try:
|
||||
antwort = scraper.get(url, timeout=15)
|
||||
antwort.encoding = 'utf-8'
|
||||
status = antwort.status_code
|
||||
|
||||
if status != 200:
|
||||
print(f" [!] Seite fehlerhaft: {url} (Status: {status})")
|
||||
sende_admin("Fehler bei Webseite", f"{url} lieferte Status {status}.")
|
||||
return
|
||||
|
||||
soup = BeautifulSoup(antwort.text, "html.parser")
|
||||
hauptbereich = soup.find("main") or soup.find("article") or soup.body or soup
|
||||
neuer_text = hauptbereich.get_text(separator=" ", strip=True)
|
||||
|
||||
alter_eintrag = gedaechtnis.get(url, {})
|
||||
alter_text = alter_eintrag.get("inhalt", "")
|
||||
|
||||
if alter_text != neuer_text:
|
||||
print(f" [NEU] Änderung auf {url} gefunden. Ich frage die KI...")
|
||||
alt_z = alter_text.splitlines()
|
||||
neu_z = neuer_text.splitlines()
|
||||
diff = [z[2:] for z in difflib.ndiff(alt_z, neu_z) if z.startswith("+ ")]
|
||||
text_fuer_ki = "\n".join(diff)
|
||||
alle_links = extrahiere_links(hauptbereich, url)
|
||||
|
||||
if text_fuer_ki.strip():
|
||||
analyse = analysiere_mit_gemini(text_fuer_ki, alle_links)
|
||||
if analyse.get("relevant") == True:
|
||||
print(f" [!] KI bestätigt Relevanz: {analyse.get('titel')}")
|
||||
klick_url = analyse.get("direct_link", url)
|
||||
sende_public(
|
||||
titel=analyse.get("titel", "Polit-Update"),
|
||||
nachricht=analyse.get("nachricht", "Neu auf der Seite"),
|
||||
klick_url=klick_url,
|
||||
category=analyse.get("category", "info"),
|
||||
priority=analyse.get("priority", 3)
|
||||
)
|
||||
|
||||
gedaechtnis[url] = {"inhalt": neuer_text, "zeit": str(datetime.now()), "status": status}
|
||||
else:
|
||||
print(f" [OK] Keine Änderungen auf {url}.")
|
||||
|
||||
except Exception as e:
|
||||
print(f" ❌ Fehler bei {url}: {e}")
|
||||
sende_admin("Programmfehler", f"{url}: {e}")
|
||||
|
||||
# --- HAUPTPROGRAMM ---
|
||||
gedaechtnis = lade_gedaechtnis()
|
||||
with open("./urls.txt", "r", encoding='utf-8') as datei:
|
||||
for zeile in datei:
|
||||
adresse = zeile.strip()
|
||||
if not adresse or adresse.startswith("#"): continue
|
||||
print(f"Ich prüfe: {adresse}")
|
||||
verarbeite_website(adresse, gedaechtnis)
|
||||
|
||||
speichere_gedaechtnis(gedaechtnis)
|
||||
print("\nFertig. Gute Nacht!")
|
||||
Reference in New Issue
Block a user