File: examplev6.py

package info (click to toggle)
python-hole 0.9.0-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 128 kB
  • sloc: python: 680; makefile: 3
file content (149 lines) | stat: -rw-r--r-- 5,430 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
"""Example for the usage of the hole module."""

import asyncio
import json
from datetime import datetime

import aiohttp

from hole import Hole

PASSWORD = "your_password_here"
HOST = "your_host_here"
PROTOCOL = "https"
PORT = 443
VERIFY_TLS = False


async def main():
    """Get the data from a Pi-hole instance."""
    async with aiohttp.ClientSession() as session:
        async with Hole(
            host=HOST,
            session=session,
            password=PASSWORD,
            protocol=PROTOCOL,
            port=PORT,
            verify_tls=VERIFY_TLS,
        ) as pihole:
            await pihole.get_data()

            print("\n=== Version Information ===")
            print(
                f"Core Version: {pihole.core_current} (Latest: {pihole.core_latest}, Update Available: {pihole.core_update})"
            )
            print(
                f"Web Version: {pihole.web_current} (Latest: {pihole.web_latest}, Update Available: {pihole.web_update})"
            )
            print(
                f"FTL Version: {pihole.ftl_current} (Latest: {pihole.ftl_latest}, Update Available: {pihole.ftl_update})"
            )

            print("\n=== Basic Statistics ===")
            print(f"Status: {pihole.status}")
            print(f"Domains being blocked: {pihole.domains_being_blocked}")
            print(f"Total queries today: {pihole.dns_queries_today}")
            print(f"Queries blocked today: {pihole.ads_blocked_today}")
            print(f"Percentage blocked: {pihole.ads_percentage_today}%")

            print("\n=== Client Statistics ===")
            print(f"Total clients ever seen: {pihole.clients_ever_seen}")
            print(f"Active clients: {pihole.unique_clients}")

            print("\n=== Query Statistics ===")
            print(f"Queries forwarded: {pihole.queries_forwarded}")
            print(f"Queries cached: {pihole.queries_cached}")
            print(f"Unique domains: {pihole.unique_domains}")

            print("\n=== Top Permitted Domains ===")
            for domain in pihole.top_queries:
                print(f"{domain['domain']}: {domain['count']} queries")

            print("\n=== Top Blocked Domains (Ads) ===")
            for domain in pihole.top_ads:
                print(f"{domain['domain']}: {domain['count']} queries")

            print("\n=== Forward Destinations ===")
            for upstream in pihole.forward_destinations:
                print(
                    f"Name: {upstream['name']}, IP: {upstream['ip']}, Count: {upstream['count']}"
                )

            print("\n=== Reply Types ===")
            for reply_type, count in pihole.reply_types.items():
                print(f"{reply_type}: {count}")

            print("\n=== Raw Data ===")
            print(
                json.dumps(
                    {
                        "data": pihole.data,
                        "blocked_domains": pihole.blocked_domains,
                        "permitted_domains": pihole.permitted_domains,
                        "clients": pihole.clients,
                        "upstreams": pihole.upstreams,
                        "blocking_status": pihole.blocking_status,
                        "versions": pihole.versions,
                    },
                    indent=4,
                    sort_keys=True,
                )
            )


async def toggle_blocking():
    """Example of enabling and disabling Pi-hole blocking."""
    async with aiohttp.ClientSession() as session:
        async with Hole(
            host=HOST,
            session=session,
            password=PASSWORD,
            protocol=PROTOCOL,
            port=PORT,
            verify_tls=VERIFY_TLS,
        ) as pihole:
            await pihole.get_data()
            initial_status = pihole.status
            print(f"\nInitial Pi-hole status: {initial_status}")

            print("\nDisabling Pi-hole blocking for 60 seconds...")
            disable_result = await pihole.disable(duration=60)

            await pihole.get_data()
            if pihole.status != "disabled":
                print(
                    f"ERROR: Failed to disable Pi-hole! Status is still: {pihole.status}"
                )
                return
            print(f"Successfully disabled Pi-hole. Status: {pihole.status}")
            print(f"Disable operation response: {disable_result}")

            print("\nWaiting 5 seconds...")
            await asyncio.sleep(5)

            print("\nEnabling Pi-hole blocking...")
            enable_result = await pihole.enable()

            await pihole.get_data()
            if pihole.status != "enabled":
                print(
                    f"ERROR: Failed to enable Pi-hole! Status is still: {pihole.status}"
                )
                return
            print(f"Successfully enabled Pi-hole. Status: {pihole.status}")
            print(f"Enable operation response: {enable_result}")

            if pihole.status == initial_status:
                print(
                    "\nToggle test completed successfully! Pi-hole returned to initial state."
                )
            else:
                print(
                    f"\nWARNING: Final status ({pihole.status}) differs from initial status ({initial_status})"
                )


if __name__ == "__main__":
    print(f"=== Pi-hole Statistics as of {datetime.now()} ===")
    asyncio.run(main())
    asyncio.run(toggle_blocking())