Test CI

main.py

@@ -0,0 +1,131 @@
+import requests
+import yaml
+import os
+
+# from dotenv import load_dotenv
+
+# load_dotenv()
+
+url = os.getenv("ADGUARD_URL")
+username = os.getenv("ADGUARD_USER")
+password = os.getenv("ADGUARD_PASSWORD")
+yaml_file = os.getenv("YAML_FILE")
+
+print(f"AdGuard URL: {url}")
+print(f"Username: {username}")
+
+session = requests.Session()
+
+class IndentDumper(yaml.Dumper):
+    def increase_indent(self, flow=False, indentless=False):
+        return super().increase_indent(flow, False)
+
+def login():
+    r = session.post(f"{url}/control/login", json={"name": username, "password": password})
+    if r.status_code != 200:
+        raise Exception(f"Login fehlgeschlagen: {r.status_code} {r.text}")
+    print("Login erfolgreich.")
+
+def load_yaml():
+    with open(yaml_file, "r") as f:
+        return yaml.safe_load(f)["rewrites"]
+
+def save_yaml(data):
+    with open(yaml_file, "w") as f:
+        yaml.dump(
+            {"rewrites": data},
+            f,
+            Dumper=IndentDumper,
+            default_flow_style=False,
+            sort_keys=False
+        )
+
+def get_rewrites():
+    r = session.get(f"{url}/control/rewrite/list")
+    r.raise_for_status()
+    print(r.json())
+    return r.json()  # AdGuard gibt direkt eine Liste zurück
+
+def add_rewrite(entry):
+    r = session.post(f"{url}/control/rewrite/add", json=entry)
+    r.raise_for_status()
+    print(f"Hinzugefügt: {entry}")
+
+def delete_rewrite(entry):
+    r = session.post(f"{url}/control/rewrite/delete", json=entry)
+    r.raise_for_status()
+    print(f"Gelöscht: {entry}")
+
+def update_rewrite(old_entry, new_entry):
+    r = session.put(f"{url}/control/rewrite/update", json={"target": old_entry, "update": new_entry})
+    r.raise_for_status()
+    print(f"Aktualisiert: {old_entry} -> {new_entry}")
+
+def merge_rewrites(existing, desired):
+    # Mapping: domain -> IP (bestehend)
+    existing_map = {r["domain"]: r["answer"] for r in existing}
+    desired_map = {entry["domain"]: entry["answer"] for entry in desired}
+
+    updated = existing.copy()
+
+    # Hinzufügen und Ändern
+    for domain, ip in desired_map.items():
+        if domain in existing_map:
+            # Ändern, wenn die IP-Adresse unterschiedlich ist
+            if existing_map[domain] != ip:
+                print(f"Ändere {domain}: {existing_map[domain]} -> {ip}")
+                for r in updated:
+                    if r["domain"] == domain:
+                        r["answer"] = ip
+        else:
+            # Hinzufügen, wenn der Eintrag noch nicht existiert
+            print(f"Füge hinzu: {domain} -> {ip}")
+            updated.append({"domain": domain, "answer": ip})
+
+    # Entfernen von Einträgen, die nicht mehr gewünscht sind
+    for domain in existing_map.keys():
+        if domain not in desired_map:
+            updated = [r for r in updated if r["domain"] != domain]
+            print(f"Lösche: {domain} : {ip}")
+
+    return updated
+
+def apply_rewrites(desired):
+    existing = get_rewrites()
+
+    existing_map = {r["domain"]: r["answer"] for r in existing}
+    desired_map = {r["domain"]: r["answer"] for r in desired}
+
+    # Änderungen oder Neueinträge
+    for domain, ip in desired_map.items():
+        if domain in existing_map:
+            if existing_map[domain] != ip:
+                update_rewrite(
+                    {"domain": domain, "answer": existing_map[domain]},
+                    {"domain": domain, "answer": ip}
+                )
+        else:
+            add_rewrite({"domain": domain, "answer": ip})
+
+    # Entferne veraltete
+    for domain, ip in existing_map.items():
+        if domain not in desired_map:
+            delete_rewrite({"domain": domain, "answer": ip})
+
+def main():
+    login()
+    current = get_rewrites()
+    desired = load_yaml()
+
+    # Merge existing and desired rewrites
+    merged = merge_rewrites(current, desired)
+
+    # Save merged rewrites back to YAML
+    save_yaml(merged)
+
+    # Apply the merged rewrites to AdGuard
+    apply_rewrites(merged)
+    print("Alle Änderungen wurden erfolgreich angewendet.")
+
+if __name__ == "__main__":
+    main()