You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
350 lines
14 KiB
Python
350 lines
14 KiB
Python
"""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, SummaryRepo
|
|
|
|
with db_session() as session:
|
|
postings = JobPostingRepo().get_today_new(session, run_date)
|
|
|
|
if not postings:
|
|
return
|
|
|
|
from gimme_job.runtime.summarizer import OllamaSummarizer
|
|
summarizer = OllamaSummarizer(
|
|
base_url=self.cfg.summarization.ollama_base_url,
|
|
model=self.cfg.summarization.model,
|
|
temperature=self.cfg.summarization.temperature,
|
|
)
|
|
text = summarizer.summarize(postings)
|
|
summary.summary_text = text
|
|
|
|
with db_session() as session:
|
|
SummaryRepo().save_summary(session, run_date, text, self.cfg.summarization.model)
|
|
|
|
from gimme_job.runtime.notifier import NotificationDispatcher
|
|
dispatcher = NotificationDispatcher(
|
|
global_config=self.cfg, session_factory=self.session_factory
|
|
)
|
|
dispatcher.send(text, run_date)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Summarize/notify failed: {e}")
|