"""Run orchestrator: coordinates all sites, browser, extraction, DB, and notification.""" from __future__ import annotations import random import time import uuid from datetime import date, datetime from pathlib import Path from typing import Optional from loguru import logger from gimme_job.config import GlobalConfig, list_enabled_sites, load_site_manifest, merge_query from gimme_job.constants import FailureClassification, RunStatus from gimme_job.models.dto import SearchQuery from gimme_job.models.manifest import SiteManifest from gimme_job.models.runtime import RunSummary, SiteRunResult class RunOrchestrator: def __init__(self, global_config: GlobalConfig, session_factory): self.cfg = global_config self.session_factory = session_factory def run_all( self, site_filter: Optional[str] = None, dry_run: bool = False, skip_notify: bool = False, limit_sites: Optional[int] = None, ) -> RunSummary: from gimme_job.adapters.registry import _load_all_adapters _load_all_adapters() run_date = date.today() summary = RunSummary(run_date=datetime.utcnow()) # Determine sites to run if site_filter: site_ids = [site_filter] else: site_ids = list_enabled_sites() if limit_sites: site_ids = site_ids[:limit_sites] if not site_ids: logger.warning("No enabled sites found") return summary logger.info(f"Running {len(site_ids)} sites: {site_ids}") for i, site_id in enumerate(site_ids): result = self._run_one_site(site_id, run_date, dry_run) summary.add_result(result) # Record to DB (unless dry run) if not dry_run: from gimme_job.db.engine import db_session from gimme_job.db.repo import SiteRunRepo from gimme_job.runtime.health import update_site_status with db_session() as session: SiteRunRepo().record_run(session, result) update_site_status(site_id, result, session) # Inter-site delay (skip after last site) if i < len(site_ids) - 1: delay = random.randint( self.cfg.runtime.min_delay_ms, self.cfg.runtime.max_delay_ms, ) / 1000.0 logger.debug(f"Sleeping {delay:.1f}s before next site") time.sleep(delay) # Summarize and notify if not dry_run and not skip_notify and summary.total_new > 0: self._summarize_and_notify(summary, run_date) return summary def _run_one_site( self, site_id: str, run_date: date, dry_run: bool ) -> SiteRunResult: result = SiteRunResult(site_id=site_id, started_at=datetime.utcnow()) try: manifest = load_site_manifest(site_id) except FileNotFoundError: logger.error(f"[{site_id}] Manifest not found") result.status = RunStatus.FAILED result.error_summary = "Manifest file not found" result.mark_done() return result query = merge_query(self.cfg, manifest) run_id = str(uuid.uuid4())[:8] from gimme_job.utils.paths import screenshot_path, dom_snapshot_path, trace_path try: if manifest.search_override.multi_keyword_mode == "separate" and query.keywords: # Run one scrape per keyword and merge results from gimme_job.runtime.dedupe import deduplicate_in_batch all_candidates = [] for kw in query.keywords: from gimme_job.models.dto import SearchQuery kw_query = query.model_copy(update={"keywords": [kw]}) kw_run_id = f"{run_id}-{kw[:8]}" logger.info(f"[{site_id}] Keyword sweep: '{kw}'") kw_candidates = self._scrape_site(site_id, manifest, kw_query, kw_run_id, result, dry_run=dry_run) all_candidates.extend(kw_candidates) # Deduplicate across keyword sweeps by fingerprint candidates = deduplicate_in_batch(all_candidates) else: candidates = self._scrape_site(site_id, manifest, query, run_id, result, dry_run=dry_run) except Exception as e: logger.error(f"[{site_id}] Scrape error: {e}") result.status = RunStatus.FAILED result.error_summary = str(e) result.mark_done() return result result.items_found = len(candidates) if not dry_run and candidates: from gimme_job.db.engine import db_session from gimme_job.db.repo import JobPostingRepo with db_session() as session: total, new_count = JobPostingRepo().upsert_candidates(session, candidates, run_date) result.new_items = new_count elif dry_run: result.new_items = len(candidates) # treat all as new in dry run if result.status != RunStatus.FAILED: result.status = RunStatus.SUCCESS result.mark_done() logger.info( f"[{site_id}] Done: {result.items_found} found, {result.new_items} new" ) return result def _scrape_site( self, site_id: str, manifest: SiteManifest, query: SearchQuery, run_id: str, result: SiteRunResult, dry_run: bool = False, ): from gimme_job.adapters.registry import get_adapter from gimme_job.runtime.browser import BrowserManager from gimme_job.runtime.dedupe import deduplicate_in_batch, ensure_fingerprints from gimme_job.runtime.extractor import apply_post_filters from gimme_job.utils.paths import dom_snapshot_path, screenshot_path, trace_path headless = manifest.browser.headless_on_run profile = manifest.browser.profile_name or self.cfg.runtime.profile_name bm = BrowserManager( profile_name=profile, headless=headless, slow_mo=self.cfg.runtime.slow_mo_ms, ) all_cards = [] try: context = bm.open_context() bm.start_tracing() page = bm.new_page() # Set timeouts page.set_default_timeout(self.cfg.runtime.default_timeout_ms) page.set_default_navigation_timeout(self.cfg.runtime.navigation_timeout_ms) adapter = get_adapter(site_id, manifest) # Navigate url = manifest.build_start_url(query) logger.info(f"[{site_id}] Navigating to {url}") page.goto(url, timeout=self.cfg.runtime.navigation_timeout_ms) # Apply search (for input-field based search) adapter.apply_search(page, query) # Apply filters (date filter etc.) adapter.apply_filters(page, query) # Wait for content if manifest.search.result_list_wait_selector: try: page.wait_for_selector( manifest.search.result_list_wait_selector, timeout=self.cfg.runtime.default_timeout_ms, state="attached", ) except Exception: logger.debug(f"[{site_id}] Wait selector timed out — continuing anyway") # Check if site has existing data (for early stop on update runs) site_has_data = False if not dry_run: try: from gimme_job.db.engine import db_session from gimme_job.db.repo import JobPostingRepo with db_session() as _s: site_has_data = JobPostingRepo().count_by_site(_s, site_id) > 0 except Exception: pass # Collect cards across pages max_pages = min( manifest.pagination.max_pages, self.cfg.runtime.max_pages_per_site, ) for page_idx in range(max_pages): page_cards = adapter.collect_cards(page, manifest) all_cards.extend(page_cards) logger.debug(f"[{site_id}] Page {page_idx+1}: {len(page_cards)} cards") if not page_cards: break # no results on this page — stop paginating # Early stop: if site has existing data and this page's cards are all known, # we've caught up to previously stored content — no need to paginate further. if site_has_data and page_idx >= 1: try: from gimme_job.db.engine import db_session from gimme_job.db.repo import JobPostingRepo page_fps = [] for card in page_cards: try: c = adapter.normalize(card) if c.fingerprint: page_fps.append(c.fingerprint) except Exception: pass if page_fps: with db_session() as _s: known = JobPostingRepo().get_existing_fingerprints(_s, page_fps) if len(known) >= len(page_fps): logger.info( f"[{site_id}] Page {page_idx+1}: all {len(page_fps)} items" " already in DB — stopping early" ) break except Exception as e: logger.debug(f"[{site_id}] Early stop check failed: {e}") if page_idx < max_pages - 1: if not adapter.paginate(page, page_idx, manifest): break time.sleep(0.5) # Take screenshot try: ss_path = screenshot_path(site_id, run_id) page.screenshot(path=str(ss_path), full_page=False) result.screenshot_path = ss_path except Exception as e: logger.debug(f"[{site_id}] Screenshot failed: {e}") # Save DOM snapshot try: dom_path = dom_snapshot_path(site_id, run_id) dom_path.parent.mkdir(parents=True, exist_ok=True) dom_path.write_text(page.content(), encoding="utf-8") result.dom_snapshot_path = dom_path except Exception as e: logger.debug(f"[{site_id}] DOM snapshot failed: {e}") finally: try: t_path = trace_path(site_id, run_id) bm.save_trace(t_path) result.trace_path = t_path except Exception: pass bm.close() # Normalize cards adapter = get_adapter(site_id, manifest) # re-instantiate (stateless) candidates = [] for card in all_cards: try: candidate = adapter.normalize(card) if candidate.is_valid(): candidates.append(candidate) except Exception as e: logger.debug(f"[{site_id}] Normalize error: {e}") # Apply post-filters if manifest.post_filters.include_posted_text: include_lower = [t.lower() for t in manifest.post_filters.include_posted_text] candidates = [ c for c in candidates if not c.posted_text or any(t in (c.posted_text or "").lower() for t in include_lower) ] if manifest.post_filters.include_title_keywords: kws = [t.lower() for t in manifest.post_filters.include_title_keywords] before = len(candidates) candidates = [ c for c in candidates if any(kw in (c.title or "").lower() for kw in kws) ] logger.info( f"[{site_id}] title keyword filter: {len(candidates)}/{before} kept " f"(keywords: {manifest.post_filters.include_title_keywords})" ) # Ensure fingerprints and deduplicate within batch candidates = ensure_fingerprints(candidates) candidates = deduplicate_in_batch(candidates) return candidates def _summarize_and_notify(self, summary: RunSummary, run_date: date) -> None: try: from gimme_job.db.engine import db_session from gimme_job.db.repo import JobPostingRepo from gimme_job.runtime.notifier import NotificationDispatcher, build_listing_messages from gimme_job.runtime.telegram import TelegramClient with db_session() as session: postings = JobPostingRepo().get_today_new(session, run_date) if not postings: return chunks = build_listing_messages(postings) client = TelegramClient() if client.is_configured(): for chunk in chunks: client.send_message(chunk) dispatcher = NotificationDispatcher( global_config=self.cfg, session_factory=self.session_factory ) full_text = "\n\n".join(chunks) dispatcher.send_markdown_fallback(full_text, run_date) except Exception as e: logger.error(f"Summarize/notify failed: {e}")