gimme-job/gimme_job/runtime/orchestrator.py

"""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, run_started_at=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,
            )
            consecutive_all_known = 0
            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,
                # increment counter. Stop only after 2 consecutive all-known pages to avoid
                # missing new items that appear after a block of already-seen pages.
                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):
                                consecutive_all_known += 1
                                logger.info(
                                    f"[{site_id}] Page {page_idx+1}: all {len(page_fps)} items"
                                    f" already in DB (consecutive: {consecutive_all_known})"
                                )
                                if consecutive_all_known >= 2:
                                    logger.info(f"[{site_id}] 2 consecutive all-known pages — stopping early")
                                    break
                            else:
                                consecutive_all_known = 0
                    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})"
            )

        if manifest.post_filters.exclude_title_keywords:
            exc_kws = [t.lower() for t in manifest.post_filters.exclude_title_keywords]
            before = len(candidates)
            candidates = [
                c for c in candidates
                if not any(kw in (c.title or "").lower() for kw in exc_kws)
            ]
            logger.info(
                f"[{site_id}] title exclude filter: {len(candidates)}/{before} kept "
                f"(excluded: {manifest.post_filters.exclude_title_keywords})"
            )

        if manifest.post_filters.exclude_company_keywords:
            exc_co = [t.lower() for t in manifest.post_filters.exclude_company_keywords]
            before = len(candidates)
            candidates = [
                c for c in candidates
                if not any(kw in (c.company or "").lower() for kw in exc_co)
            ]
            logger.info(
                f"[{site_id}] company exclude filter: {len(candidates)}/{before} kept "
                f"(excluded: {manifest.post_filters.exclude_company_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, run_started_at=None) -> 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, since=run_started_at)
                if not postings:
                    return
                chunks = build_listing_messages(postings, run_started_at=run_started_at)

            client = TelegramClient()
            if client.is_configured():
                import time as _time
                for i, chunk in enumerate(chunks):
                    client.send_message(chunk)
                    if i < len(chunks) - 1:
                        _time.sleep(1)  # avoid Telegram rate limiting between chunks

            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}")