a44b9ef852
- Mutare și reorganizare foldere proiecte - Actualizare path-uri în TOOLS.md - Sincronizare configurații agenți - 79 fișiere actualizate
112 lines
2.9 KiB
Python
Executable File
112 lines
2.9 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
ANAF Page Monitor - Simple hash-based change detection
|
|
Checks configured pages and reports changes via stdout
|
|
"""
|
|
|
|
import json
|
|
import hashlib
|
|
import urllib.request
|
|
import ssl
|
|
import os
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
SCRIPT_DIR = Path(__file__).parent
|
|
CONFIG_FILE = SCRIPT_DIR / "config.json"
|
|
HASHES_FILE = SCRIPT_DIR / "hashes.json"
|
|
LOG_FILE = SCRIPT_DIR / "monitor.log"
|
|
|
|
# SSL context that doesn't verify (some ANAF pages have cert issues)
|
|
SSL_CTX = ssl.create_default_context()
|
|
SSL_CTX.check_hostname = False
|
|
SSL_CTX.verify_mode = ssl.CERT_NONE
|
|
|
|
def log(msg):
|
|
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
with open(LOG_FILE, "a") as f:
|
|
f.write(f"[{timestamp}] {msg}\n")
|
|
|
|
def load_json(path, default=None):
|
|
try:
|
|
with open(path) as f:
|
|
return json.load(f)
|
|
except:
|
|
return default if default is not None else {}
|
|
|
|
def save_json(path, data):
|
|
with open(path, "w") as f:
|
|
json.dump(data, f, indent=2)
|
|
|
|
def fetch_page(url, timeout=30):
|
|
"""Fetch page content"""
|
|
try:
|
|
req = urllib.request.Request(url, headers={
|
|
'User-Agent': 'Mozilla/5.0 (compatible; ANAF-Monitor/1.0)'
|
|
})
|
|
with urllib.request.urlopen(req, timeout=timeout, context=SSL_CTX) as resp:
|
|
return resp.read()
|
|
except Exception as e:
|
|
log(f"ERROR fetching {url}: {e}")
|
|
return None
|
|
|
|
def compute_hash(content):
|
|
"""Compute SHA256 hash of content"""
|
|
return hashlib.sha256(content).hexdigest()
|
|
|
|
def check_page(page, hashes):
|
|
"""Check a single page for changes. Returns change info or None."""
|
|
page_id = page["id"]
|
|
name = page["name"]
|
|
url = page["url"]
|
|
|
|
content = fetch_page(url)
|
|
if content is None:
|
|
return None
|
|
|
|
new_hash = compute_hash(content)
|
|
old_hash = hashes.get(page_id)
|
|
|
|
if old_hash is None:
|
|
log(f"INIT: {page_id} - storing initial hash")
|
|
hashes[page_id] = new_hash
|
|
return None
|
|
|
|
if new_hash != old_hash:
|
|
log(f"CHANGE DETECTED: {page_id} - {name}")
|
|
log(f" URL: {url}")
|
|
log(f" Old hash: {old_hash}")
|
|
log(f" New hash: {new_hash}")
|
|
hashes[page_id] = new_hash
|
|
return {"id": page_id, "name": name, "url": url}
|
|
|
|
log(f"OK: {page_id} - no changes")
|
|
return None
|
|
|
|
def main():
|
|
log("=== Starting ANAF monitor check ===")
|
|
|
|
config = load_json(CONFIG_FILE, {"pages": []})
|
|
hashes = load_json(HASHES_FILE, {})
|
|
|
|
changes = []
|
|
for page in config["pages"]:
|
|
change = check_page(page, hashes)
|
|
if change:
|
|
changes.append(change)
|
|
|
|
save_json(HASHES_FILE, hashes)
|
|
|
|
log("=== Monitor check complete ===")
|
|
|
|
# Output changes as JSON for the caller
|
|
if changes:
|
|
print(json.dumps({"changes": changes}))
|
|
else:
|
|
print(json.dumps({"changes": []}))
|
|
|
|
return len(changes)
|
|
|
|
if __name__ == "__main__":
|
|
exit(main())
|