upstream-frappe-crm/crm/domain_enrichment/pipeline.py

# Copyright (c) 2026, Frappe Technologies Pvt. Ltd. and contributors
# For license information, please see license.txt

"""Orchestration: load config, crawl, run extractors, run enabled sources, assemble
an ``EnrichmentResult``.

``run()`` accepts an optional ``progress`` callback ``fn(step_index, message)`` so
Phase 4's ``tasks.py`` can surface the progress steps over realtime without the
engine depending on Frappe. The pipeline never writes to the DB or CRM docs (that
is Phase 4) and never raises for fetch/source failures -- they are captured as
errors/notes on the result.
"""

from __future__ import annotations

from urllib.parse import urlparse

from . import extractors
from .config import EnrichmentConfig, get_config
from .crawler import crawl, probe_about_pages
from .http import build_session
from .result import EnrichmentResult, Field, Method

PROGRESS_STEPS = [
	"Discovering pages",
	"Crawling website",
	"Extracting company information",
	"Extracting contacts",
	"Classifying industry",
	"Saving results",
	"Completed",
]


def _normalize_website(url: str) -> str:
	url = (url or "").strip()
	if not url:
		raise ValueError("website URL is required")
	if not urlparse(url).scheme:
		url = "https://" + url
	parsed = urlparse(url)
	if parsed.scheme not in ("http", "https"):
		raise ValueError(f"unsupported URL scheme: {parsed.scheme}")
	if not parsed.netloc:
		raise ValueError(f"invalid URL: {url}")
	return url


def run(website: str, cfg: EnrichmentConfig = None, progress=None) -> EnrichmentResult:
	"""Run the full enrichment engine for ``website`` and return the result.

	:param website: the site to enrich (scheme optional; https assumed).
	:param cfg: an :class:`EnrichmentConfig`; loaded via ``get_config()`` if omitted.
	:param progress: optional ``fn(step_index, message)`` callback (0-based index
	    into ``PROGRESS_STEPS``); exceptions in it are swallowed.
	"""
	website = _normalize_website(website)
	cfg = cfg or get_config()
	result = EnrichmentResult(website=website)

	def emit(step_index, message=""):
		if progress:
			try:
				progress(step_index, message or PROGRESS_STEPS[step_index])
			except Exception:
				pass  # progress reporting must never break enrichment

	session = build_session(cfg)
	try:
		# 1 + 2. Discover + crawl (BFS does both; reported together).
		emit(0)
		crawled = crawl(website, cfg, session=session, progress=lambda msg: emit(1, msg))
		pages = [page for page, _soup in crawled]
		soups_by_url = {page.url: soup for page, soup in crawled}
		result.pages_crawled = [{"url": p.url, "status": p.status_code, "error": p.error} for p in pages]
		result.errors = [{"url": p.url, "error": p.error} for p in pages if p.error]

		# Short-circuit on a blocked / unreachable homepage with a clear reason.
		reason = extractors.diagnose_readability(pages)
		if reason in ("blocked", "unreachable"):
			result.notes.append(extractors.READABILITY_MESSAGES[reason])
			emit(6)
			return result
		if reason == "empty":
			# JS-rendered body: <head> metadata may still be usable -- fall through.
			result.notes.append(extractors.READABILITY_MESSAGES["empty"])

		homepage = pages[0]
		home_soup = soups_by_url.get(homepage.url)

		# If the BFS didn't surface an About page, probe the common paths -- it's the
		# best source of a company description (and helps industry/contacts too). Only
		# when the page budget allows more fetches.
		max_pages = int(cfg.setting("max_pages", 10) or 10)
		if home_soup and len(pages) < max_pages and not any(extractors._is_about_page(p.url) for p in pages):
			for page, soup in probe_about_pages(
				homepage.url, cfg, session=session, skip_urls=[p.url for p in pages]
			):
				pages.append(page)
				soups_by_url[page.url] = soup
				result.pages_crawled.append(
					{"url": page.url, "status": page.status_code, "error": page.error}
				)

		# 3. Company information (each field carries provenance).
		emit(2)
		company = extractors.extract_company_info(homepage, home_soup) if home_soup else {}
		result.company_name = company.get("company_name") or Field()
		result.logo = company.get("logo") or Field()
		result.image = company.get("image") or Field()
		result.description = (
			extractors.select_description(
				pages,
				soups_by_url,
				industry_rules=cfg.rules("Industry"),
				company_name=result.company_name.value,
			)
			or company.get("description")
			or Field()
		)

		# 4. Contacts: emails, phones, social.
		emit(3)
		result.emails = extractors.extract_emails(pages)
		result.phones = extractors.extract_phones(pages)
		result.social_profiles = extractors.extract_social_profiles(
			pages,
			soups_by_url,
			cfg.rules("Social"),
			extra_links=company.get("social_links"),
		)

		# 5. Industry.
		emit(4)
		industry, confidence = extractors.classify_industry(pages, company, cfg.rules("Industry"))
		result.industry = Field(industry, website, Method.KEYWORD_CLASSIFIER) if industry else Field()
		result.industry_confidence = confidence

		# 6 + 7. Done (saving handled by the caller in Phase 4).
		emit(5)
		emit(6)
		return result
	finally:
		session.close()