first commit
This commit is contained in:
+18
@@ -0,0 +1,18 @@
|
||||
{
|
||||
"domains": [
|
||||
"example.com",
|
||||
"example.net"
|
||||
],
|
||||
"check_interval_minutes": 60,
|
||||
"smtp": {
|
||||
"host": "smtp.gmail.com",
|
||||
"port": 587,
|
||||
"use_tls": true,
|
||||
"username": "your_account@gmail.com",
|
||||
"password": "app_password_here",
|
||||
"from": "your_account@gmail.com",
|
||||
"to": [
|
||||
"your_account@gmail.com"
|
||||
]
|
||||
}
|
||||
}
|
||||
+151
@@ -0,0 +1,151 @@
|
||||
#!/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
|
||||
Reference in New Issue
Block a user