134 lines
4.8 KiB
Python
134 lines
4.8 KiB
Python
"""LeadScopingEngine – orchestrates scraping + AI extraction for lead generation."""
|
||
from __future__ import annotations
|
||
import json
|
||
import logging
|
||
import os
|
||
import time
|
||
from pathlib import Path
|
||
from typing import Optional
|
||
|
||
from scraper import AsyncScraper
|
||
from extractor import DeepSeekExtractor
|
||
from models import CompanyTechProfile, ScrapeResult, LeadGenResponse
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
CONFIG_PATH = Path(os.getenv("LEADGEN_CONFIG", "/app/data/leadgen_config.json"))
|
||
|
||
CAREER_PATHS = ["/careers", "/jobs", "/about/careers", "/careers/openings", "/company/careers", "/en/careers"]
|
||
|
||
|
||
def _load_config() -> dict:
|
||
if CONFIG_PATH.exists():
|
||
try:
|
||
return json.loads(CONFIG_PATH.read_text())
|
||
except Exception:
|
||
pass
|
||
return {}
|
||
|
||
|
||
class LeadScopingEngine:
|
||
"""Orchestrates async scraping and AI-powered tech signal extraction."""
|
||
|
||
def __init__(
|
||
self,
|
||
api_key: Optional[str] = None,
|
||
api_base: Optional[str] = None,
|
||
scrape_timeout_ms: int = 30_000,
|
||
):
|
||
self.api_key = api_key
|
||
self.api_base = api_base
|
||
self.scraper = AsyncScraper(timeout_ms=scrape_timeout_ms)
|
||
self._extractor: Optional[DeepSeekExtractor] = None
|
||
|
||
@property
|
||
def extractor(self) -> DeepSeekExtractor:
|
||
if self._extractor is None:
|
||
cfg = _load_config()
|
||
key = self.api_key or cfg.get("api_key", "")
|
||
base = self.api_base or cfg.get("base_url", "https://openrouter.ai/api/v1")
|
||
self._extractor = DeepSeekExtractor(api_key=key, base_url=base)
|
||
return self._extractor
|
||
|
||
async def analyze_url(self, url: str) -> LeadGenResponse:
|
||
"""Analyze a company URL: scrape main page + careers → extract → return structured profile."""
|
||
t0 = time.monotonic()
|
||
|
||
# Step 1: Scrape main page
|
||
logger.info("Scraping %s ...", url)
|
||
main_result: ScrapeResult = await self.scraper.scrape(url)
|
||
|
||
if main_result.error or not main_result.text_content:
|
||
logger.warning("Scrape failed for %s: %s", url, main_result.error)
|
||
return LeadGenResponse(
|
||
success=False,
|
||
error=main_result.error or "No content extracted",
|
||
elapsed_ms=round((time.monotonic() - t0) * 1000, 1),
|
||
)
|
||
|
||
# Step 2: Also scrape careers page for hiring data
|
||
combined_text = main_result.text_content
|
||
career_sections = []
|
||
base = url.rstrip("/")
|
||
|
||
for path in CAREER_PATHS:
|
||
career_url = base + path
|
||
try:
|
||
cr = await self.scraper.scrape(career_url)
|
||
if cr.text_content and len(cr.text_content) > 100:
|
||
logger.info("Careers page found: %s (%d chars)", career_url, len(cr.text_content))
|
||
career_sections.append(f"\n\n=== JOB LISTINGS FROM {career_url} ===\n\n" + cr.text_content[:20000])
|
||
break # Stop after first match
|
||
except Exception:
|
||
continue
|
||
|
||
if career_sections:
|
||
combined_text += "\n".join(career_sections)
|
||
|
||
logger.info(
|
||
"Scraped %s — main: %d chars, careers: %d chars — %.1fs",
|
||
url, len(main_result.text_content),
|
||
sum(len(c) for c in career_sections),
|
||
time.monotonic() - t0,
|
||
)
|
||
|
||
# Step 3: AI Extraction
|
||
try:
|
||
cfg = _load_config()
|
||
model = cfg.get("model", "qwen/qwen3-coder:free")
|
||
fallbacks = cfg.get("fallback_models", [])
|
||
profile: CompanyTechProfile = await self.extractor.extract(
|
||
combined_text, url, model=model, fallback_models=fallbacks
|
||
)
|
||
except Exception as exc:
|
||
logger.exception("AI extraction failed for %s", url)
|
||
return LeadGenResponse(
|
||
success=False,
|
||
error=f"AI extraction error: {str(exc)}",
|
||
elapsed_ms=round((time.monotonic() - t0) * 1000, 1),
|
||
)
|
||
|
||
elapsed = round((time.monotonic() - t0) * 1000, 1)
|
||
logger.info(
|
||
"Analysis complete — %s | Tech: %d | Pains: %d | Roles: %d | Skills: %d | Maturity: %s | %.0fms",
|
||
profile.company_name,
|
||
len(profile.detected_technologies),
|
||
len(profile.current_pain_points),
|
||
len(profile.hiring_roles),
|
||
len(profile.desired_skills),
|
||
profile.estimated_data_maturity,
|
||
elapsed,
|
||
)
|
||
|
||
return LeadGenResponse(
|
||
success=True,
|
||
profile=profile,
|
||
elapsed_ms=elapsed,
|
||
)
|
||
|
||
async def batch_analyze(self, urls: list[str]) -> list[LeadGenResponse]:
|
||
"""Analyze multiple URLs concurrently."""
|
||
import asyncio
|
||
tasks = [self.analyze_url(url) for url in urls]
|
||
return await asyncio.gather(*tasks, return_exceptions=False)
|