Compare commits
No commits in common. "e1e0d7b230f1a342593604f166850723f9682016" and "1d98a136c72df8fbc6d7dae26ddf1f38c0d8eb3a" have entirely different histories.
e1e0d7b230
...
1d98a136c7
24
main.py
24
main.py
@ -1,24 +0,0 @@
|
||||
# _ __________________
|
||||
# | | / / _/ ____/ _/ /
|
||||
# | | / // // / __ / // /
|
||||
# | |/ // // /_/ // // /___
|
||||
# |___/___/\____/___/_____/
|
||||
# © Uthmn 2025 under MIT license
|
||||
|
||||
import time
|
||||
|
||||
import services.apt
|
||||
import services.mail
|
||||
|
||||
from dotenv import load_dotenv
|
||||
from os import getenv
|
||||
from os.path import dirname, join
|
||||
|
||||
# Load environment variables from .env file
|
||||
dotenv_path = join(dirname(__file__), ".env")
|
||||
load_dotenv(dotenv_path)
|
||||
|
||||
# TODO: Flag --now to run temporarily and just execute hourly tasks instantly then exit
|
||||
# TODO: Logging
|
||||
# TODO: Email alerts + structure
|
||||
# TODO: Use typer for CLI
|
||||
@ -1,10 +0,0 @@
|
||||
click==8.3.0
|
||||
dotenv==0.9.9
|
||||
markdown-it-py==4.0.0
|
||||
mdurl==0.1.2
|
||||
Pygments==2.19.2
|
||||
python-dotenv==1.1.1
|
||||
rich==14.2.0
|
||||
shellingham==1.5.4
|
||||
typer==0.19.2
|
||||
typing_extensions==4.15.0
|
||||
@ -1,76 +0,0 @@
|
||||
# _ __________________
|
||||
# | | / / _/ ____/ _/ /
|
||||
# | | / // // / __ / // /
|
||||
# | |/ // // /_/ // // /___
|
||||
# |___/___/\____/___/_____/
|
||||
# © Uthmn 2025 under MIT license
|
||||
|
||||
import subprocess
|
||||
import os
|
||||
import sys
|
||||
import re
|
||||
|
||||
def require_root():
|
||||
if os.geteuid() != 0:
|
||||
print("This script requires root privileges. Please run with sudo.")
|
||||
sys.exit(1)
|
||||
|
||||
def detect_apt():
|
||||
try:
|
||||
subprocess.run(["apt", "--version"], capture_output=True, text=True, check=True)
|
||||
return True
|
||||
except FileNotFoundError:
|
||||
return False
|
||||
except subprocess.CalledProcessError as e:
|
||||
raise RuntimeError(f"Error running apt: {e.stderr or e}")
|
||||
|
||||
def check_updates():
|
||||
# Update package list
|
||||
try:
|
||||
subprocess.run(["apt", "update"], capture_output=True, text=True, check=True)
|
||||
except FileNotFoundError:
|
||||
raise FileNotFoundError("apt not found on this system.")
|
||||
except subprocess.CalledProcessError as e:
|
||||
raise RuntimeError(f"Error running 'apt update': {e.stderr or e.stdout or e}")
|
||||
|
||||
# List upgradable packages
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["apt", "list", "--upgradable"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=True
|
||||
)
|
||||
upgrades = result.stdout.strip()
|
||||
lines = [line for line in upgrades.splitlines() if line.strip()]
|
||||
|
||||
if len(lines) <= 1: # Only the header
|
||||
return {} # No upgrades
|
||||
|
||||
package_dict = {}
|
||||
for line in lines[1:]: # skip header
|
||||
# Example line format:
|
||||
# docker-ce/focal-security 5:24.0.1~3-0~ubuntu-focal amd64 [upgradable from: 5:23.0.0~3-0~ubuntu-focal]
|
||||
match = re.match(
|
||||
r"([^/]+)/([^\s]+)\s+([^\s]+)\s+[^\s]+\s+\[upgradable from: ([^\]]+)\]", line
|
||||
)
|
||||
if match:
|
||||
name, repo, latest_version, installed_version = match.groups()
|
||||
# Detect security update based on repo name containing "security"
|
||||
is_security = "security" in repo.lower()
|
||||
package_dict[name] = {
|
||||
"installed_version": installed_version,
|
||||
"latest_version": latest_version,
|
||||
"repo": repo,
|
||||
"security": is_security
|
||||
}
|
||||
return package_dict
|
||||
|
||||
except subprocess.CalledProcessError as e:
|
||||
raise RuntimeError(f"Error running 'apt list --upgradable': {e.stderr or e.stdout or e}")
|
||||
|
||||
if __name__ == "__main__":
|
||||
require_root()
|
||||
if detect_apt():
|
||||
updates = check_updates()
|
||||
print(updates)
|
||||
121
services/mail.py
121
services/mail.py
@ -1,121 +0,0 @@
|
||||
# _ __________________
|
||||
# | | / / _/ ____/ _/ /
|
||||
# | | / // // / __ / // /
|
||||
# | |/ // // /_/ // // /___
|
||||
# |___/___/\____/___/_____/
|
||||
# © Uthmn 2025 under MIT license
|
||||
|
||||
import smtplib
|
||||
from email.mime.text import MIMEText
|
||||
|
||||
from imaplib import IMAP4_SSL
|
||||
from email import message_from_bytes
|
||||
|
||||
from dotenv import load_dotenv
|
||||
from os import getenv
|
||||
from os.path import dirname, join
|
||||
|
||||
# Load environment variables from .env file
|
||||
dotenv_path = join(dirname(__file__), "../.env")
|
||||
load_dotenv(dotenv_path)
|
||||
|
||||
# SMTP server settings
|
||||
SMTP_SERVER = getenv("SMTP_SERVER")
|
||||
SMTP_PORT = int(getenv("SMTP_PORT"))
|
||||
SMTP_USERNAME = getenv("SMTP_USERNAME")
|
||||
SMTP_PASSWORD = getenv("SMTP_PASSWORD")
|
||||
|
||||
# IMAP server settings
|
||||
IMAP_SERVER = getenv("IMAP_SERVER")
|
||||
IMAP_PORT = int(getenv("IMAP_PORT"))
|
||||
IMAP_USERNAME = getenv("IMAP_USERNAME")
|
||||
IMAP_PASSWORD = getenv("IMAP_PASSWORD")
|
||||
|
||||
def send_email(receiver_email, subject, body):
|
||||
sender_email = SMTP_USERNAME
|
||||
|
||||
# Create the email
|
||||
message = MIMEText(body, "html")
|
||||
message["Subject"] = subject
|
||||
message["From"] = sender_email
|
||||
message["To"] = receiver_email
|
||||
|
||||
# Connect using SSL
|
||||
try:
|
||||
with smtplib.SMTP_SSL(SMTP_SERVER, SMTP_PORT) as server:
|
||||
server.login(SMTP_USERNAME, SMTP_PASSWORD)
|
||||
server.sendmail(sender_email, receiver_email, message.as_string())
|
||||
print("HTML email sent successfully!")
|
||||
except Exception as e:
|
||||
print(f"Failed to send email: {e}")
|
||||
|
||||
def receive_emails(folder="INBOX", limit=50):
|
||||
emails = {}
|
||||
|
||||
try:
|
||||
M = IMAP4_SSL(host=IMAP_SERVER, port=IMAP_PORT)
|
||||
M.login(IMAP_USERNAME, IMAP_PASSWORD)
|
||||
M.select(folder)
|
||||
|
||||
# Get all message IDs
|
||||
typ, data = M.search(None, 'ALL')
|
||||
if typ != 'OK' or not data[0]:
|
||||
print("No messages found.")
|
||||
return emails
|
||||
|
||||
all_ids = data[0].split()
|
||||
last_ids = all_ids[-limit:] # Take only the last 'limit' messages
|
||||
|
||||
for num in last_ids:
|
||||
typ, msg_data = M.fetch(num, '(RFC822)')
|
||||
if typ != 'OK':
|
||||
print(f"Failed to fetch message {num}")
|
||||
continue
|
||||
|
||||
# Parse the email
|
||||
msg = message_from_bytes(msg_data[0][1])
|
||||
msg_id = msg.get('Message-ID') or num.decode()
|
||||
|
||||
# Extract headers
|
||||
subject = msg.get('Subject', '')
|
||||
sender = msg.get('From', '')
|
||||
date = msg.get('Date', '')
|
||||
|
||||
# Extract plain text and HTML
|
||||
body_text = ""
|
||||
body_html = ""
|
||||
if msg.is_multipart():
|
||||
for part in msg.walk():
|
||||
content_type = part.get_content_type()
|
||||
content_disposition = str(part.get("Content-Disposition"))
|
||||
if "attachment" in content_disposition:
|
||||
continue
|
||||
payload = part.get_payload(decode=True)
|
||||
if payload:
|
||||
payload = payload.decode(errors="ignore")
|
||||
if content_type == "text/plain":
|
||||
body_text = payload
|
||||
elif content_type == "text/html":
|
||||
body_html = payload
|
||||
else:
|
||||
payload = msg.get_payload(decode=True)
|
||||
if payload:
|
||||
body_text = payload.decode(errors="ignore")
|
||||
|
||||
emails[msg_id] = {
|
||||
"subject": subject,
|
||||
"from": sender,
|
||||
"date": date,
|
||||
"body_text": body_text,
|
||||
"body_html": body_html
|
||||
}
|
||||
|
||||
M.close()
|
||||
M.logout()
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error receiving emails: {e}")
|
||||
|
||||
return emails
|
||||
|
||||
print(receive_emails())
|
||||
Loading…
Reference in New Issue
Block a user