Job_Aggregation--Tracking_Platform/main.py at main · Primavat/Job_Aggregation--Tracking_Platform · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
#!/usr/bin/env python3
"""
Global Tech Jobs & Internships Pipeline
========================================
Fetches jobs from public APIs → Claude AI filters/classifies → Google Sheets

Usage:
    python main.py                  # Run once
    python main.py --schedule       # Run on cron schedule
    python main.py --filter remote  # Remote-only roles
    python main.py --filter intern  # Internships only
    python main.py --notify         # Enable Telegram notifications
"""

import argparse
import logging
import sys
import time
import os

import schedule

from config import Config
from collectors import JobCollector
from claude_processor import ClaudeProcessor
from sheets_manager import SheetsManager
from notifier import Notifier
from utils import setup_logging, load_seen_ids, save_seen_ids

logger = logging.getLogger(__name__)


def run_pipeline(cfg: Config, job_filter: str | None = None, notify: bool = False):
    """Full pipeline: collect → process → upload."""

    logger.info("=" * 60)
    logger.info("Pipeline run started")

    seen_ids = load_seen_ids(cfg.SEEN_IDS_FILE)
    notifier  = Notifier(cfg) if notify else None

    # ── 1. Collect ────────────────────────────────────────────────────────
    logger.info("Phase 1: Collecting jobs from APIs …")
    collector = JobCollector(cfg)
    raw_jobs  = collector.collect_all()
    logger.info(f"  Collected {len(raw_jobs)} raw listings")

    if not raw_jobs:
        logger.warning("No jobs collected. Check API keys / network.")
        return

    # ── 2. Deduplicate against history ───────────────────────────────────
    new_jobs = [j for j in raw_jobs if j["id"] not in seen_ids]
    logger.info(f"  {len(new_jobs)} new (after dedup against {len(seen_ids)} seen)")

    if not new_jobs:
        logger.info("No new jobs to process. All done.")
        return

    # ── 3. Claude: filter, classify, clean ───────────────────────────────
    logger.info("Phase 2: Claude AI processing …")
    processor = ClaudeProcessor(cfg)
    processed = processor.process_batch(new_jobs)
    logger.info(f"  {len(processed)} passed domain filter")

    # ── 4. Optional user-side filter ────────────────────────────────────
    if job_filter == "remote":
        processed = [j for j in processed
                     if "remote" in (j.get("location") or "").lower()]
        logger.info(f"  → {len(processed)} after remote filter")
    elif job_filter == "intern":
        processed = [j for j in processed
                     if j.get("job_type", "").lower() == "internship"]
        logger.info(f"  → {len(processed)} after internship filter")

    if not processed:
        logger.info("No qualifying jobs after filtering.")
        return

    # ── 5. Google Sheets upload ───────────────────────────────────────────
    logger.info("Phase 3: Uploading to Google Sheets …")
    sheets = SheetsManager(cfg)
    added  = sheets.upload(processed)
    logger.info(f"  Uploaded {added} new rows")

    # ── 6. Persist seen IDs ───────────────────────────────────────────────
    for j in processed:
        seen_ids.add(j["id"])
    save_seen_ids(cfg.SEEN_IDS_FILE, seen_ids)

    # ── 7. Notify ────────────────────────────────────────────────────────
    if notifier and added > 0:
        notifier.send(processed[:5], added)   # preview of top 5

    logger.info(f"Pipeline complete — {added} jobs added to sheet.")
    logger.info("=" * 60)


def main():
    parser = argparse.ArgumentParser(description="Global Tech Jobs Pipeline")
    parser.add_argument("--schedule", action="store_true",
                        help="Run on a recurring schedule (default: every 6 hours)")
    parser.add_argument("--filter", choices=["remote", "intern"],
                        help="Optional post-filter: remote or internship only")
    parser.add_argument("--notify", action="store_true",
                        help="Send Telegram notification after each run")
    parser.add_argument("--interval", type=int, default=6,
                        help="Schedule interval in hours (default: 6)")
    args = parser.parse_args()

    setup_logging()
    cfg = Config()

    if args.schedule:
        run_pipeline(cfg, args.filter, args.notify)
    else:
        run_pipeline(cfg, args.filter, args.notify)


if __name__ == "__main__":
    main()