From 0ca838e33575337f0743f49ff93d1f7ca83e0d37 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Hauke=20Z=C3=BChl?= Date: Sat, 9 May 2026 16:51:50 +0200 Subject: [PATCH] Umbau zum Modul mit einer Klasse 'Muell' --- python/Muell/__init__.py | 204 +++++++++++++++++++++++++++++++++++++++ python/muell.py | 185 +---------------------------------- 2 files changed, 209 insertions(+), 180 deletions(-) create mode 100644 python/Muell/__init__.py diff --git a/python/Muell/__init__.py b/python/Muell/__init__.py new file mode 100644 index 0000000..5815a39 --- /dev/null +++ b/python/Muell/__init__.py @@ -0,0 +1,204 @@ +''' +Um die Werte für kommune und strasse zu erfahren, rufe https://www.egst.de/de/abfallabholung/ per +Browser auf und ermittle die gesendeten Daten in der Entwicklerkonsole deines Browsers. + +Die Angaben hier im Skript stehen für Hörstel als Kommune und Im Wiesengrund als Strasse + +Um die Daten an Telegram zu senden, musst du dir ein Skript/Programm schreiben, dass die API +von Telegram verwendet und Daten an dich oder deine Gruppe senden kann! + +Beispiel für eine Konfigurationsdatei: +kommune: 2601 +strasse: 2146 +telegram: + tgReceiver: + pathBot: '~/bin/YMBot' +signal: + path: "~/bin/signal-cli" + account: "+49" + group_id: "" +''' + +import csv +import os +import re +import requests +import httplib2 +import urllib +import yaml + +from datetime import datetime, timedelta +from pathlib import Path + +class Muell: + + # Variablen + __key = 'e21758b9c711463552fb9c70ac7d4273' + __modus = 'd6c5855a62cf32a4dadbc2831f0f295f' + + __host = 'api.abfall.io' + + __url = f'https://{__host}/?key={__key}&modus={__modus}&waction=export_csv' + __export_als = f"{{'action':'{__url}','target':''}}" + + __current_year = datetime.today().year + __zeitraum = f'{__current_year}0101-{__current_year}1231' + + __postdata = { + 'f_id_abfalltyp_0': '50', + 'f_id_abfalltyp_1': '161', + 'f_id_abfalltyp_2': '53', + 'f_id_abfalltyp_3': '187', + 'f_id_abfalltyp_4': '169', + 'f_abfallarten_index_max': '5', + 'f_abfallarten': '50,161,53,187', + 'f_zeitraum': __zeitraum, + 'f_export_als': __export_als, + } + + __headers = { + 'User-Agent': 'Mozilla/5.0 (Linux; x68_64; x64; rv:88.0) Gecko/20100101 Firefox/88.0', + 'Content-Type': 'application/x-www-form-urlencoded', + } + + # Methoden + def __get_index_positions(self, list_of_elems, element): + ''' Returns the indexes of all occurrences of given element in + the list- listOfElements ''' + index_pos_list = [] + for i in range(len(list_of_elems)): + if list_of_elems[i] == element: + index_pos_list.append(i) + + return index_pos_list + + def __readConfig(self): + ''' Liest ~/.muell.yaml und speichert Daten in Dictionary ''' + + home = Path.home() # home ist ohne / am Ende! + config = {} + with open(f'{home}/.muell.yaml', 'r') as config_file: + config = yaml.safe_load(config_file) + + return config + + def __init(self, key, modus, host, headers): + ''' Initialisiert das System und sucht entsprechende Daten heraus ''' + + url = f'https://{host}/?key={key}&modus={modus}&waction=init' + http = httplib2.Http() + (resp, content) = http.request(url, "POST", headers = headers, body = urllib.parse.urlencode(self.__postdata)) + + result = re.findall(r"", str(content)) + + datum = result[1].split(' ') + + name = None + value = None + for i in datum: + i = i.split('=') + if i[0] == 'name': + name = i[1][1:-1] + if i[0] == 'value': + value = i[1][1:-1] + + return name, value + + def __read_file(self, jahr): + ''' Liest Daten aus muell.csv und liefert ggf. Ergebnis zurück''' + + antwort_liste = list() + headline = list() + try: + with open(f'muell{jahr}.csv', 'r', encoding="latin1") as f: + csv_reader = csv.reader(f, delimiter=';') + lines = 0 + antwort_liste = [] + for row in csv_reader: + if lines == 0: + # print(f'Column names are {", ".join(row)}') + headline = row + lines += 1 + else: + if isinstance(row, list): + antwort_liste.append(row) + except FileNotFoundError: + pass + + return (antwort_liste, headline) + + def get_data(self): + (antwort_liste, headline) = self.__read_file(self.__current_year) + config = self.__readConfig() + + if not antwort_liste: + # Keine vernünftigen Daten, ergo mal gucken, was die Webseite ergibt + self.__postdata['f_id_kommune'] = config['kommune'] + self.__postdata['f_id_strasse'] = config['strasse'] + + (name, value) = self.__init(key, modus, host, headers) + if name != None and value != None: + self.__postdata[name] = value + + http = httplib2.Http() + (resp, content) = http.request(url, "POST", headers = headers, body = urllib.parse.urlencode(self.__postdata)) + + antwort = str(content) + f = open(f'muell{self.__current_year}.csv', 'wb') + f.write(content) + f.close() + (antwort_liste, headline) = self.__read_file(self.__current_year) + + tomorrow = (datetime.now() + timedelta(1)).strftime('%d.%m.%Y') + index = set(); + for row in antwort_liste: + try: + if (row.count(tomorrow) > 1): + for i in range(len(row)): + if (row[i] == tomorrow and i < 4): + index.add(i) + else: + pos = row.index(tomorrow) + if pos < 4: + index.add(pos) + except ValueError: + pass + + wird = 'wird' if len(index) < 2 else 'werden' + tonnen = [] + for pos in index: + tonnen.append(headline[pos]) + + return (tonnen, wird) + + def ausgabe_telegram(self, tonnen, wird): + ''' Ausgabe Richtung Telegram''' + + if len(tonnen) > 0: + tonnen = ' und '.join(tonnen) + # Jetzt den Telegram Bot ansprechen + try: + os.system(f'echo "Morgen {wird} {tonnen} abgeholt" | {config["telegram"]["pathBot"]} -u {config["telegram"]["tgReceiver"]}') + except Exception as e: + os.system(f'logger "Fehler Muellbot Telegram: {e}"') + + def ausgabe_signal(self, tonnen, wird): + ''' Ausgabe Richtung Signal''' + + if len(tonnen) > 0: + + tonnen = ' und '.join(tonnen) + + # Signal Bot + try: + os.system(f'{config["signal"]["path"]} -a {config["signal"]["account"]} send -g {config["signal"]["group_id"]} -m "Morgen {wird} {tonnen} abgeholt"') + except Exception as e: + os.system(f'logger "Fehler Muellbot Signal: {e}"') + + def ausgabe_text(self, tonnen, wird): + ''' Ausgabe als Text''' + + if len(tonnen) > 0: + tonnen = ' und '.join(tonnen) + + print(f'Morgen {wird} {tonnen} abgeholt') diff --git a/python/muell.py b/python/muell.py index ec84b32..6c1f9d8 100644 --- a/python/muell.py +++ b/python/muell.py @@ -1,184 +1,9 @@ #!/usr/bin/python3 -''' -Um die Werte für kommune und strasse zu erfahren, rufe https://www.egst.de/de/abfallabholung/ per -Browser auf und ermittle die gesendeten Daten in der Entwicklerkonsole deines Browsers. - -Die Angaben hier im Skript stehen für Hörstel als Kommune und Im Wiesengrund als Strasse - -Um die Daten an Telegram zu senden, musst du dir ein Skript/Programm schreiben, dass die API -von Telegram verwendet und Daten an dich oder deine Gruppe senden kann! - -Beispiel für eine Konfigurationsdatei: -kommune: 2601 -strasse: 2146 -telegram: - tgReceiver: - pathBot: '~/bin/YMBot' -signal: - path: "~/bin/signal-cli" - account: "+49" - group_id: "" -''' - -import csv -import os -import re -import requests -import httplib2 -import urllib -import yaml - -from datetime import datetime, timedelta -from pathlib import Path - -# Funktionen -def get_index_positions(list_of_elems, element): - ''' Returns the indexes of all occurrences of given element in - the list- listOfElements ''' - index_pos_list = [] - for i in range(len(list_of_elems)): - if list_of_elems[i] == element: - index_pos_list.append(i) - - return index_pos_list - -def readConfig(): - ''' Liest ~/.muell.yaml und speichert Daten in Dictionary ''' - - home = Path.home() # home ist ohne / am Ende! - config = {} - with open(f'{home}/.muell.yaml', 'r') as config_file: - config = yaml.safe_load(config_file) - - return config - -def init(key, modus, host, headers): - ''' Initialisiert das System und sucht entsprechende Daten heraus ''' - - url = f'https://{host}/?key={key}&modus={modus}&waction=init' - http = httplib2.Http() - (resp, content) = http.request(url, "POST", headers = headers, body = urllib.parse.urlencode(postdata)) - - result = re.findall(r"", str(content)) - - datum = result[1].split(' ') - - name = None - value = None - for i in datum: - i = i.split('=') - if i[0] == 'name': - name = i[1][1:-1] - if i[0] == 'value': - value = i[1][1:-1] - - return name, value - -def read_file(jahr): - ''' Liest Daten aus muell.csv und liefert ggf. Ergebnis zurück''' +from Muell import Muell - antwort_liste = list() - headline = list() - try: - with open(f'muell{jahr}.csv', 'r', encoding="latin1") as f: - csv_reader = csv.reader(f, delimiter=';') - lines = 0 - antwort_liste = [] - for row in csv_reader: - if lines == 0: - # print(f'Column names are {", ".join(row)}') - headline = row - lines += 1 - else: - if isinstance(row, list): - antwort_liste.append(row) - except FileNotFoundError: - pass - - return (antwort_liste, headline) - -# Variablen -key = 'e21758b9c711463552fb9c70ac7d4273' -modus = 'd6c5855a62cf32a4dadbc2831f0f295f' - -host = 'api.abfall.io' - -url = f'https://{host}/?key={key}&modus={modus}&waction=export_csv' -export_als = f"{{'action':'{url}','target':''}}" - -current_year = datetime.today().year -zeitraum = f'{current_year}0101-{current_year}1231' - -postdata = { - 'f_id_abfalltyp_0': '50', - 'f_id_abfalltyp_1': '161', - 'f_id_abfalltyp_2': '53', - 'f_id_abfalltyp_3': '187', - 'f_id_abfalltyp_4': '169', - 'f_abfallarten_index_max': '5', - 'f_abfallarten': '50,161,53,187', - 'f_zeitraum': zeitraum, - 'f_export_als': export_als, -} - -headers = { - 'User-Agent': 'Mozilla/5.0 (Linux; x68_64; x64; rv:88.0) Gecko/20100101 Firefox/88.0', - 'Content-Type': 'application/x-www-form-urlencoded', -} - if __name__ == '__main__': - (antwort_liste, headline) = read_file(current_year) - config = readConfig() - - if not antwort_liste: - # Keine vernünftigen Daten, ergo mal gucken, was die Webseite ergibt - postdata['f_id_kommune'] = config['kommune'] - postdata['f_id_strasse'] = config['strasse'] - - (name, value) = init(key, modus, host, headers) - if name != None and value != None: - postdata[name] = value - - http = httplib2.Http() - (resp, content) = http.request(url, "POST", headers = headers, body = urllib.parse.urlencode(postdata)) - - antwort = str(content) - f = open(f'muell{current_year}.csv', 'wb') - f.write(content) - f.close() - (antwort_liste, headline) = read_file(current_year) - - tomorrow = (datetime.now() + timedelta(1)).strftime('%d.%m.%Y') - index = set(); - for row in antwort_liste: - try: - if (row.count(tomorrow) > 1): - for i in range(len(row)): - if (row[i] == tomorrow and i < 4): - index.add(i) - else: - pos = row.index(tomorrow) - if pos < 4: - index.add(pos) - except ValueError: - pass - - wird = 'wird' if len(index) < 2 else 'werden' - tonnen = [] - for pos in index: - tonnen.append(headline[pos]) - - if len(tonnen) > 0: - tonnen = ' und '.join(tonnen) - # Jetzt den Telegram Bot ansprechen - try: - os.system(f'echo "Morgen {wird} {tonnen} abgeholt" | {config["telegram"]["pathBot"]} -u {config["telegram"]["tgReceiver"]}') - except Exception as e: - os.system(f'logger "Fehler Muellbot Telegram: {e}"') - - # Signal Bot - try: - os.system(f'{config["signal"]["path"]} -a {config["signal"]["account"]} send -g {config["signal"]["group_id"]} -m "Morgen {wird} {tonnen} abgeholt"') - except Exception as e: - os.system(f'logger "Fehler Muellbot Signal: {e}"') + muell = Muell() + tonnen, wird = muell.get_data() + muell.ausgabe_telegram(tonnen, wird) + muell.ausgabe_signal(tonnen, wird)