Harwinder
e608d367e2
Add remote_only config option that appends LinkedIn's f_WT=2 filter param to search URLs. Change location from Toronto to Canada for broader remote job coverage. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
525 lines
20 KiB
Python
525 lines
20 KiB
Python
"""
|
|
daemon.py
|
|
---------
|
|
Enhanced job scraper daemon with:
|
|
1. Job card scraping
|
|
2. Detail page fetching
|
|
3. Resume-based match scoring
|
|
4. Smart alerts - only 70%+ matches with breakdown
|
|
|
|
Usage:
|
|
python daemon.py # single run
|
|
python daemon.py --loop # run every 2 hours
|
|
python daemon.py --loop --interval 3600 # run every hour
|
|
"""
|
|
|
|
import requests
|
|
from bs4 import BeautifulSoup
|
|
import time
|
|
import random
|
|
import re
|
|
import json
|
|
import yaml
|
|
import argparse
|
|
import os
|
|
import sys
|
|
from datetime import datetime
|
|
from urllib.parse import quote
|
|
import logging
|
|
import mysql.connector
|
|
from mysql.connector import Error
|
|
|
|
from resume_parser import parse_resume_text, parse_resume_file
|
|
from matcher import score_job
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
|
|
datefmt="%Y-%m-%d %H:%M:%S",
|
|
)
|
|
logger = logging.getLogger("job-matcher")
|
|
|
|
|
|
def load_config(config_path="config.yaml"):
|
|
with open(config_path, "r") as f:
|
|
return yaml.safe_load(f)
|
|
|
|
|
|
def load_profile(profile_path="profile.json", resume_path=None):
|
|
if os.path.exists(profile_path):
|
|
with open(profile_path, "r") as f:
|
|
profile = json.load(f)
|
|
if "full_text" not in profile and resume_path and os.path.exists(resume_path):
|
|
full_profile = parse_resume_file(resume_path)
|
|
profile["full_text"] = full_profile.get("full_text", "")
|
|
return profile
|
|
elif resume_path and os.path.exists(resume_path):
|
|
logger.info(f"No profile.json found. Parsing resume: {resume_path}")
|
|
profile = parse_resume_file(resume_path)
|
|
from resume_parser import save_profile
|
|
save_profile(profile, profile_path)
|
|
return profile
|
|
else:
|
|
logger.error("No profile.json or resume file found. Run resume_parser.py first.")
|
|
sys.exit(1)
|
|
|
|
|
|
class LinkedInJobScraper:
|
|
def __init__(self, config):
|
|
self.config = config.get("scraper", {})
|
|
self.session = requests.Session()
|
|
self.session.headers.update({
|
|
"User-Agent": (
|
|
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
|
|
"(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
|
|
),
|
|
"Accept": "text/html,application/xhtml+xml",
|
|
"Accept-Language": "en-US,en;q=0.9",
|
|
})
|
|
self.detail_delay = self.config.get("delay_between_details", [4, 8])
|
|
|
|
def scrape_jobs(self, keyword, location, pages=3):
|
|
jobs = []
|
|
seen_links = set()
|
|
encoded_keyword = quote(keyword)
|
|
encoded_location = quote(location)
|
|
remote_only = self.config.get("remote_only", False)
|
|
delay_range = self.config.get("delay_between_pages", [3, 6])
|
|
|
|
for page in range(pages):
|
|
logger.info(f" Page {page + 1}/{pages} for '{keyword}'")
|
|
start_index = page * 25
|
|
url = (
|
|
f"https://www.linkedin.com/jobs/search?"
|
|
f"keywords={encoded_keyword}&location={encoded_location}"
|
|
f"&start={start_index}"
|
|
)
|
|
if remote_only:
|
|
url += "&f_WT=2"
|
|
|
|
try:
|
|
response = self.session.get(url, timeout=15)
|
|
if response.status_code == 429:
|
|
wait = 60 * (page + 1)
|
|
logger.warning(f"Rate limited. Sleeping {wait}s...")
|
|
time.sleep(wait)
|
|
continue
|
|
if response.status_code != 200:
|
|
logger.error(f"Failed page {page + 1}: HTTP {response.status_code}")
|
|
continue
|
|
|
|
soup = BeautifulSoup(response.content, "html.parser")
|
|
job_cards = soup.find_all("div", class_="base-search-card")
|
|
|
|
if not job_cards:
|
|
logger.info("No job cards found (possibly blocked).")
|
|
break
|
|
|
|
for card in job_cards:
|
|
job = self._extract_card(card)
|
|
if not job or job["link"] in seen_links:
|
|
continue
|
|
seen_links.add(job["link"])
|
|
jobs.append(job)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error scraping page {page + 1}: {e}")
|
|
|
|
time.sleep(random.uniform(*delay_range))
|
|
return jobs
|
|
|
|
def _extract_card(self, card):
|
|
try:
|
|
title_elem = card.find("h3", class_="base-search-card__title")
|
|
if not title_elem:
|
|
return None
|
|
title = title_elem.get_text(strip=True)
|
|
|
|
link_elem = card.find("a", class_="base-card__full-link")
|
|
if not link_elem or not link_elem.get("href"):
|
|
return None
|
|
link = link_elem["href"].split("?")[0]
|
|
|
|
company_elem = card.find("h4", class_="base-search-card__subtitle")
|
|
company = company_elem.get_text(strip=True) if company_elem else "Unknown"
|
|
|
|
location_elem = card.find("span", class_="job-search-card__location")
|
|
location = location_elem.get_text(strip=True) if location_elem else ""
|
|
|
|
date_elem = card.find("time")
|
|
posted_date = date_elem.get_text(strip=True) if date_elem else "Just posted"
|
|
|
|
benefit_elem = card.find("div", class_="job-posting-benefits")
|
|
actively_hiring = False
|
|
if benefit_elem:
|
|
actively_hiring = "actively hiring" in benefit_elem.get_text().lower()
|
|
|
|
return {
|
|
"title": title, "company": company, "location": location,
|
|
"posted_date": posted_date, "link": link,
|
|
"actively_hiring": actively_hiring, "description": None,
|
|
}
|
|
except Exception as e:
|
|
logger.debug(f"Card extraction error: {e}")
|
|
return None
|
|
|
|
def fetch_job_description(self, job_url):
|
|
try:
|
|
response = self.session.get(job_url, timeout=15)
|
|
if response.status_code != 200:
|
|
logger.debug(f"Detail fetch failed ({response.status_code}): {job_url}")
|
|
return None
|
|
|
|
soup = BeautifulSoup(response.content, "html.parser")
|
|
|
|
desc_div = soup.find("div", class_="show-more-less-html__markup")
|
|
if desc_div:
|
|
return desc_div.get_text(separator="\n", strip=True)
|
|
|
|
for selector in [
|
|
{"class_": "description__text"},
|
|
{"class_": "job-description"},
|
|
{"class_": "details"},
|
|
]:
|
|
fallback = soup.find("div", **selector)
|
|
if fallback:
|
|
text = fallback.get_text(separator="\n", strip=True)
|
|
if len(text) > 50:
|
|
return text
|
|
|
|
sections = soup.find_all("section", class_="show-more-less-html")
|
|
if sections:
|
|
return "\n".join(s.get_text(separator="\n", strip=True) for s in sections)
|
|
|
|
logger.debug(f"No description found on detail page: {job_url}")
|
|
return None
|
|
except Exception as e:
|
|
logger.debug(f"Detail fetch exception: {e}")
|
|
return None
|
|
|
|
def fetch_descriptions_batch(self, jobs, max_fetches=50):
|
|
fetched_count = 0
|
|
for job in jobs:
|
|
if job.get("description"):
|
|
continue
|
|
if fetched_count >= max_fetches:
|
|
logger.info(f"Hit max detail fetches ({max_fetches}).")
|
|
break
|
|
|
|
logger.info(f" Fetching details: {job['title']} @ {job['company']}")
|
|
description = self.fetch_job_description(job["link"])
|
|
if description:
|
|
job["description"] = description
|
|
fetched_count += 1
|
|
else:
|
|
job["description"] = ""
|
|
|
|
time.sleep(random.uniform(*self.detail_delay))
|
|
|
|
logger.info(f"Fetched {fetched_count} job descriptions.")
|
|
return jobs
|
|
|
|
|
|
def pre_filter_jobs(jobs, config):
|
|
title_blacklist = [t.lower() for t in config.get("title_blacklist", [])]
|
|
company_blacklist = [c.lower() for c in config.get("company_blacklist", [])]
|
|
filtered = []
|
|
for job in jobs:
|
|
title_lower = job["title"].lower()
|
|
company_lower = job["company"].lower()
|
|
if any(bl in title_lower for bl in title_blacklist):
|
|
continue
|
|
if any(bl in company_lower for bl in company_blacklist):
|
|
continue
|
|
filtered.append(job)
|
|
skipped = len(jobs) - len(filtered)
|
|
if skipped:
|
|
logger.info(f"Pre-filter: {skipped} jobs skipped, {len(filtered)} remaining.")
|
|
return filtered
|
|
|
|
|
|
class JobAlertSystem:
|
|
def __init__(self, db_config):
|
|
self.db_config = db_config
|
|
|
|
def is_job_fresh(self, posted_date):
|
|
if not posted_date:
|
|
return False
|
|
text = posted_date.lower()
|
|
if any(x in text for x in ["just", "minute", "hour", "now", "today"]):
|
|
return True
|
|
match = re.search(r"(\d+)\s+day", text)
|
|
if match:
|
|
return int(match.group(1)) <= 4
|
|
return False
|
|
|
|
def send_telegram_alert(self, token, chat_id, job, match_result):
|
|
url = f"https://api.telegram.org/bot{token}/sendMessage"
|
|
score = match_result["score"]
|
|
breakdown = match_result["breakdown"]
|
|
matched = match_result.get("matched_skills", [])
|
|
missing = match_result.get("missing_skills", [])
|
|
|
|
filled = score // 10
|
|
bar = "█" * filled + "░" * (10 - filled)
|
|
matched_str = ", ".join(matched[:8]) if matched else "—"
|
|
missing_str = ", ".join(missing[:5]) if missing else "None"
|
|
|
|
message = (
|
|
f"🎯 *{score}% Match — {match_result['verdict']}*\n"
|
|
f"{bar}\n\n"
|
|
f"*{job['title']}*\n"
|
|
f"🏢 {job['company']}\n"
|
|
f"📍 {job['location']}\n\n"
|
|
f"📊 *Breakdown:*\n"
|
|
f" Skills: {breakdown['skills']['score']}% | "
|
|
f"Title: {breakdown['title']['score']}% | "
|
|
f"Exp: {breakdown['experience']['score']}%\n"
|
|
f" Keywords: {breakdown['keywords']['score']}% | "
|
|
f"Certs: {breakdown['certifications']['score']}%\n\n"
|
|
f"✅ *Match:* {matched_str}\n"
|
|
f"❌ *Gap:* {missing_str}\n\n"
|
|
f"[View Job]({job['link']})"
|
|
)
|
|
|
|
try:
|
|
r = requests.post(url, json={
|
|
"chat_id": chat_id, "text": message,
|
|
"parse_mode": "Markdown", "disable_web_page_preview": True,
|
|
}, timeout=10)
|
|
if r.status_code != 200:
|
|
logger.error(f"Telegram error: {r.text}")
|
|
return False
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"Telegram exception: {e}")
|
|
return False
|
|
|
|
def send_summary_alert(self, token, chat_id, stats):
|
|
url = f"https://api.telegram.org/bot{token}/sendMessage"
|
|
message = (
|
|
f"✅ *Job Matcher Run Complete*\n"
|
|
f"⏰ {datetime.now().strftime('%H:%M %d-%b-%Y')}\n\n"
|
|
f"📊 *Stats:*\n"
|
|
f" Scanned: {stats['total_scanned']}\n"
|
|
f" Passed pre-filter: {stats['after_prefilter']}\n"
|
|
f" Descriptions fetched: {stats['descriptions_fetched']}\n"
|
|
f" Matches (70%+): {stats['matches']}\n"
|
|
f" Alerts sent: {stats['alerts_sent']}\n"
|
|
f" Top score: {stats.get('top_score', 'N/A')}%"
|
|
)
|
|
try:
|
|
requests.post(url, json={
|
|
"chat_id": chat_id, "text": message, "parse_mode": "Markdown",
|
|
}, timeout=10)
|
|
except Exception as e:
|
|
logger.error(f"Summary send error: {e}")
|
|
|
|
def process_jobs(self, jobs, profile, config):
|
|
conn = None
|
|
cursor = None
|
|
min_score = config.get("min_match_score", 70)
|
|
weights = config.get("weights", None)
|
|
aspirational = config.get("aspirational_skills", [])
|
|
|
|
stats = {
|
|
"total_scanned": len(jobs), "after_prefilter": len(jobs),
|
|
"descriptions_fetched": sum(1 for j in jobs if j.get("description")),
|
|
"matches": 0, "alerts_sent": 0, "top_score": 0,
|
|
}
|
|
|
|
try:
|
|
conn = mysql.connector.connect(**self.db_config)
|
|
cursor = conn.cursor(dictionary=True)
|
|
|
|
cursor.execute("SELECT * FROM users")
|
|
users = cursor.fetchall()
|
|
|
|
cursor.execute("SELECT user_id, job_id FROM alert_logs")
|
|
alerted = {(r["user_id"], r["job_id"]) for r in cursor.fetchall()}
|
|
|
|
job_sql = """
|
|
INSERT IGNORE INTO jobs
|
|
(title, company, location, posted_date_raw, job_link,
|
|
is_actively_hiring, description, match_score, match_breakdown,
|
|
matched_skills, missing_skills, verdict, detail_fetched_at, scored_at)
|
|
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
|
|
"""
|
|
|
|
for job in jobs:
|
|
if not self.is_job_fresh(job["posted_date"]):
|
|
continue
|
|
|
|
job_text = job.get("description") or ""
|
|
full_text = f"{job['title']}\n{job_text}"
|
|
|
|
match_result = score_job(
|
|
profile=profile, job_title=job["title"],
|
|
job_text=full_text, weights=weights,
|
|
aspirational_skills=aspirational,
|
|
)
|
|
|
|
score = match_result["score"]
|
|
stats["top_score"] = max(stats["top_score"], score)
|
|
if score >= min_score:
|
|
stats["matches"] += 1
|
|
|
|
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
|
|
cursor.execute(job_sql, (
|
|
job["title"], job["company"], job["location"],
|
|
job["posted_date"], job["link"], job["actively_hiring"],
|
|
job.get("description"), score,
|
|
json.dumps(match_result["breakdown"], default=str),
|
|
json.dumps(match_result["matched_skills"]),
|
|
json.dumps(match_result["missing_skills"]),
|
|
match_result["verdict"],
|
|
now if job.get("description") else None, now,
|
|
))
|
|
|
|
if cursor.rowcount == 1:
|
|
job_id = cursor.lastrowid
|
|
else:
|
|
cursor.execute(
|
|
"SELECT job_id FROM jobs WHERE job_link = %s", (job["link"],)
|
|
)
|
|
result = cursor.fetchone()
|
|
if not result:
|
|
continue
|
|
job_id = result["job_id"]
|
|
cursor.execute("""
|
|
UPDATE jobs SET match_score=%s, match_breakdown=%s,
|
|
matched_skills=%s, missing_skills=%s, verdict=%s,
|
|
description=%s, scored_at=%s WHERE job_id=%s
|
|
""", (
|
|
score,
|
|
json.dumps(match_result["breakdown"], default=str),
|
|
json.dumps(match_result["matched_skills"]),
|
|
json.dumps(match_result["missing_skills"]),
|
|
match_result["verdict"],
|
|
job.get("description"), now, job_id,
|
|
))
|
|
|
|
if score < min_score:
|
|
continue
|
|
|
|
for user in users:
|
|
user_min = user.get("min_match_score", min_score)
|
|
if score < user_min:
|
|
continue
|
|
key = (user["user_id"], job_id)
|
|
if key in alerted:
|
|
continue
|
|
if self.send_telegram_alert(
|
|
user["telegram_token"], user["telegram_chat_id"],
|
|
job, match_result,
|
|
):
|
|
cursor.execute(
|
|
"INSERT INTO alert_logs (user_id, job_id) VALUES (%s, %s)",
|
|
(user["user_id"], job_id)
|
|
)
|
|
alerted.add(key)
|
|
stats["alerts_sent"] += 1
|
|
time.sleep(1)
|
|
|
|
logger.info(
|
|
f" ✅ {score}% | {job['title']} @ {job['company']} "
|
|
f"[{', '.join(match_result['matched_skills'][:5])}]"
|
|
)
|
|
|
|
conn.commit()
|
|
|
|
for user in users:
|
|
self.send_summary_alert(
|
|
user["telegram_token"], user["telegram_chat_id"], stats,
|
|
)
|
|
|
|
except Error as e:
|
|
logger.error(f"Database error: {e}")
|
|
finally:
|
|
if cursor:
|
|
cursor.close()
|
|
if conn and conn.is_connected():
|
|
conn.close()
|
|
|
|
return stats
|
|
|
|
|
|
def run_once(config, profile):
|
|
scraper_cfg = config.get("scraper", {})
|
|
db_config = config.get("database", {})
|
|
scraper = LinkedInJobScraper(config)
|
|
alert_system = JobAlertSystem(db_config)
|
|
|
|
keywords = scraper_cfg.get("keywords", ["Network Engineer"])
|
|
location = scraper_cfg.get("location", "Toronto, ON")
|
|
pages = scraper_cfg.get("pages_per_keyword", 3)
|
|
keyword_delay = scraper_cfg.get("delay_between_keywords", [5, 10])
|
|
max_details = config.get("max_detail_fetches_per_run", 50)
|
|
|
|
all_jobs = []
|
|
for keyword in keywords:
|
|
logger.info(f"🔍 Scraping: {keyword}")
|
|
jobs = scraper.scrape_jobs(keyword=keyword, location=location, pages=pages)
|
|
all_jobs.extend(jobs)
|
|
time.sleep(random.uniform(*keyword_delay))
|
|
|
|
unique = {}
|
|
for job in all_jobs:
|
|
unique[job["link"]] = job
|
|
jobs = list(unique.values())
|
|
logger.info(f"📋 Total unique jobs: {len(jobs)}")
|
|
|
|
jobs = pre_filter_jobs(jobs, config)
|
|
|
|
logger.info("📄 Fetching job descriptions...")
|
|
jobs = scraper.fetch_descriptions_batch(jobs, max_fetches=max_details)
|
|
|
|
desc_count = sum(1 for j in jobs if j.get("description"))
|
|
logger.info(f"📄 Jobs with descriptions: {desc_count}/{len(jobs)}")
|
|
|
|
logger.info("🎯 Scoring jobs against your profile...")
|
|
stats = alert_system.process_jobs(jobs, profile, config)
|
|
|
|
logger.info("=" * 50)
|
|
logger.info(f"📊 RUN COMPLETE")
|
|
logger.info(f" Scanned: {stats['total_scanned']}")
|
|
logger.info(f" Matches (70%+): {stats['matches']}")
|
|
logger.info(f" Alerts sent: {stats['alerts_sent']}")
|
|
logger.info(f" Top score: {stats['top_score']}%")
|
|
logger.info("=" * 50)
|
|
return stats
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Job Matcher Daemon")
|
|
parser.add_argument("--config", default="config.yaml", help="Config file path")
|
|
parser.add_argument("--profile", default="profile.json", help="Resume profile path")
|
|
parser.add_argument("--resume", default=None, help="Resume file (txt/pdf) to parse")
|
|
parser.add_argument("--loop", action="store_true", help="Run continuously")
|
|
parser.add_argument("--interval", type=int, default=7200, help="Loop interval in seconds (default: 2hrs)")
|
|
args = parser.parse_args()
|
|
|
|
config = load_config(args.config)
|
|
profile = load_profile(args.profile, args.resume)
|
|
|
|
logger.info(f"Profile loaded: {len(profile.get('skills', {}))} skills, "
|
|
f"{profile.get('years_experience', '?')} years exp, "
|
|
f"certs: {profile.get('certifications', [])}")
|
|
|
|
if args.loop:
|
|
logger.info(f"🔁 Daemon mode: running every {args.interval}s")
|
|
while True:
|
|
try:
|
|
run_once(config, profile)
|
|
except Exception as e:
|
|
logger.error(f"Run failed: {e}")
|
|
logger.info(f"💤 Sleeping {args.interval}s until next run...")
|
|
time.sleep(args.interval)
|
|
else:
|
|
run_once(config, profile)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|