152 lines
4.3 KiB
Python
152 lines
4.3 KiB
Python
#!/usr/bin/env python3
|
|
import argparse
|
|
import json
|
|
import logging
|
|
import re
|
|
import smtplib
|
|
import subprocess
|
|
import time
|
|
from datetime import datetime
|
|
from email.mime.text import MIMEText
|
|
|
|
AVAILABLE_MARKERS = [
|
|
"no match for",
|
|
"not found",
|
|
"no data found",
|
|
"no entries found",
|
|
"domain not found",
|
|
"status: available",
|
|
"status: free",
|
|
"is available for registration",
|
|
"no matching record",
|
|
"object does not exist",
|
|
"nothing found",
|
|
]
|
|
|
|
EXPIRY_PATTERNS = [
|
|
r"registry expiry date:\s*(.+)",
|
|
r"expiration date:\s*(.+)",
|
|
r"expiry date:\s*(.+)",
|
|
r"expire date:\s*(.+)",
|
|
r"paid-till:\s*(.+)",
|
|
r"renewal date:\s*(.+)",
|
|
]
|
|
|
|
DATE_FORMATS = (
|
|
"%Y-%m-%dT%H:%M:%SZ",
|
|
"%Y-%m-%dT%H:%M:%S%z",
|
|
"%Y-%m-%d",
|
|
"%d-%b-%Y",
|
|
"%d.%m.%Y",
|
|
"%Y.%m.%d",
|
|
)
|
|
|
|
|
|
def whois_lookup(domain, timeout=15):
|
|
result = subprocess.run(
|
|
["whois", domain], capture_output=True, text=True, timeout=timeout
|
|
)
|
|
return result.stdout
|
|
|
|
|
|
def format_date(raw):
|
|
raw = raw.strip()
|
|
for fmt in DATE_FORMATS:
|
|
try:
|
|
return datetime.strptime(raw, fmt).strftime("%Y-%m-%d")
|
|
except ValueError:
|
|
continue
|
|
return raw
|
|
|
|
|
|
def check_domain(domain):
|
|
try:
|
|
text = whois_lookup(domain)
|
|
except (subprocess.TimeoutExpired, OSError) as e:
|
|
logging.warning("whois для %s не выполнен: %s", domain, e)
|
|
return "unknown", None
|
|
|
|
lower = text.lower()
|
|
for marker in AVAILABLE_MARKERS:
|
|
if marker in lower:
|
|
return "free", None
|
|
|
|
for pattern in EXPIRY_PATTERNS:
|
|
m = re.search(pattern, text, re.IGNORECASE)
|
|
if m:
|
|
return "busy", format_date(m.group(1))
|
|
|
|
return "busy", None
|
|
|
|
|
|
def build_report(domains):
|
|
lines = []
|
|
any_free = False
|
|
for domain in domains:
|
|
status, expiry = check_domain(domain)
|
|
if status == "free":
|
|
lines.append(f"{domain} - свободен")
|
|
any_free = True
|
|
elif status == "busy":
|
|
lines.append(f"{domain} - занят до {expiry}" if expiry else f"{domain} - занят")
|
|
else:
|
|
lines.append(f"{domain} - не удалось проверить")
|
|
return "\n".join(lines), any_free
|
|
|
|
|
|
def send_email(smtp_cfg, subject, body):
|
|
msg = MIMEText(body, "plain", "utf-8")
|
|
msg["Subject"] = subject
|
|
msg["From"] = smtp_cfg["from"]
|
|
msg["To"] = ", ".join(smtp_cfg["to"])
|
|
|
|
if smtp_cfg.get("use_tls", True):
|
|
with smtplib.SMTP(smtp_cfg["host"], smtp_cfg["port"], timeout=30) as server:
|
|
server.starttls()
|
|
server.login(smtp_cfg["username"], smtp_cfg["password"])
|
|
server.sendmail(smtp_cfg["from"], smtp_cfg["to"], msg.as_string())
|
|
else:
|
|
with smtplib.SMTP_SSL(smtp_cfg["host"], smtp_cfg["port"], timeout=30) as server:
|
|
server.login(smtp_cfg["username"], smtp_cfg["password"])
|
|
server.sendmail(smtp_cfg["from"], smtp_cfg["to"], msg.as_string())
|
|
|
|
|
|
def run_check(config):
|
|
body, any_free = build_report(config["domains"])
|
|
subject = "Проверка доменов" + (" (есть свободные)" if any_free else "")
|
|
logging.info("Результат проверки:\n%s", body)
|
|
send_email(config["smtp"], subject, body)
|
|
logging.info("Письмо отправлено (тема: %s)", subject)
|
|
|
|
|
|
def load_config(path):
|
|
with open(path, encoding="utf-8") as f:
|
|
return json.load(f)
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Отслеживание освобождения доменов")
|
|
parser.add_argument("--config", default="config.json", help="путь к config.json")
|
|
parser.add_argument("--once", action="store_true", help="одна проверка без цикла")
|
|
args = parser.parse_args()
|
|
|
|
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
|
|
|
|
while True:
|
|
config = load_config(args.config)
|
|
try:
|
|
run_check(config)
|
|
except Exception:
|
|
logging.exception("Ошибка при выполнении проверки")
|
|
|
|
if args.once:
|
|
break
|
|
time.sleep(config.get("check_interval_minutes", 60) * 60)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main()
|
|
except KeyboardInterrupt:
|
|
pass
|