import os
import json
import time
import smtplib
import logging
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from pathlib import Path
import requests
from dotenv import load_dotenv
# Optional: for scheduling within Python
import schedule
# Set up basic logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s')
# Load environment variables
load_dotenv()
SMTP_SERVER = os.getenv("SMTP_SERVER")
SMTP_PORT = int(os.getenv("SMTP_PORT", 587))
EMAIL_USER = os.getenv("EMAIL_USER")
EMAIL_PASS = os.getenv("EMAIL_PASS")
RECIPIENT_EMAIL = os.getenv("RECIPIENT_EMAIL")
KEYWORDS = [kw.strip().lower() for kw in os.getenv("KEYWORDS", "").split(",") if kw.strip()]
# File to store IDs of jobs already notified
SEEN_FILE = Path("seen_jobs.json")
def load_seen_jobs():
if SEEN_FILE.exists():
try:
with open(SEEN_FILE, "r") as f:
data = json.load(f)
if isinstance(data, list):
return set(data)
except Exception as e:
logging.warning(f"Could not read seen jobs file: {e}")
return set()
def save_seen_jobs(seen_set):
try:
with open(SEEN_FILE, "w") as f:
json.dump(list(seen_set), f)
except Exception as e:
logging.error(f"Error saving seen jobs file: {e}")
def fetch_jobs():
"""
Fetch job listings from RemoteOK JSON API.
RemoteOK API endpoint: https://remoteok.com/api
"""
url = "https://remoteok.com/api" # public endpoint
headers = {
"User-Agent": "Mozilla/5.0 (compatible; JobNotifier/1.0)"
}
try:
resp = requests.get(url, headers=headers, timeout=10)
resp.raise_for_status()
data = resp.json()
# The first element may be metadata; filter typical job dicts by presence of 'id' and 'position'
jobs = []
for item in data:
if isinstance(item, dict) and item.get("id") and item.get("position"):
jobs.append(item)
logging.info(f"Fetched {len(jobs)} job postings")
return jobs
except Exception as e:
logging.error(f"Error fetching jobs: {e}")
return []
def filter_jobs(jobs, keywords, seen_ids):
"""
Return list of job dicts matching any keyword and not in seen_ids.
"""
new_matches = []
for job in jobs:
job_id = str(job.get("id"))
if job_id in seen_ids:
continue
text = (job.get("position","") + " " + job.get("description","") + " " + job.get("company","")).lower()
if any(kw in text for kw in keywords):
new_matches.append(job)
logging.info(f"Found {len(new_matches)} new matching jobs")
return new_matches
def format_email(jobs):
"""
Create an HTML/plain text summary of the jobs.
"""
if not jobs:
return None, None
subject = f"{len(jobs)} new remote job(s) matching your keywords"
plain_lines = []
html_lines = ['']
html_lines.append(f"
{len(jobs)} new remote job(s) found
")
for job in jobs:
title = job.get("position")
company = job.get("company")
url = job.get("url") or job.get("apply_url") or job.get("url") # fields may vary
date = job.get("date") or job.get("date_posted", "")
snippet = job.get("description","")[:200].replace("\n", " ").strip()
plain_lines.append(f"- {title} at {company} ({date}): {url}")
html_lines.append(f"- {title} at {company} ({date})
"
f"{snippet}...
View Job ")
html_lines.append("
")
plain_text = "\n".join(plain_lines)
html_text = "\n".join(html_lines)
return subject, (plain_text, html_text)
def send_email(subject, body_plain, body_html):
"""
Send an email via SMTP.
"""
msg = MIMEMultipart("alternative")
msg["Subject"] = subject
msg["From"] = EMAIL_USER
msg["To"] = RECIPIENT_EMAIL
part1 = MIMEText(body_plain, "plain")
part2 = MIMEText(body_html, "html")
msg.attach(part1)
msg.attach(part2)
try:
server = smtplib.SMTP(SMTP_SERVER, SMTP_PORT)
server.ehlo()
server.starttls()
server.login(EMAIL_USER, EMAIL_PASS)
server.sendmail(EMAIL_USER, RECIPIENT_EMAIL, msg.as_string())
server.quit()
logging.info("Email sent successfully")
except Exception as e:
logging.error(f"Failed to send email: {e}")
def job_notifier():
seen = load_seen_jobs()
jobs = fetch_jobs()
new_jobs = filter_jobs(jobs, KEYWORDS, seen)
if new_jobs:
subject, bodies = format_email(new_jobs)
if subject:
send_email(subject, bodies[0], bodies[1])
# Mark these as seen
for job in new_jobs:
seen.add(str(job.get("id")))
save_seen_jobs(seen)
else:
logging.info("No new matching jobs; nothing to email.")
def main():
# Run once immediately
job_notifier()
# If you prefer to schedule within this script (e.g. every day at 09:00):
# schedule.every().day.at("09:00").do(job_notifier)
# while True:
# schedule.run_pending()
# time.sleep(60)
if __name__ == "__main__":
main()
8