saksham
1a4c80958f
FastAPI + Celery + Next.js + Postgres/Redis app with company monitoring, source collection, LLM-based change analysis, enrichment, and account security (Turnstile, escalating lockout, email verification).
77 lines
2.5 KiB
Python
77 lines
2.5 KiB
Python
"""Shared content extraction/normalization helpers used by every collector.
|
|
|
|
Centralizing this (rather than letting each collector roll its own) is what
|
|
makes cross-collector dedup and hashing behave consistently.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import re
|
|
from urllib.parse import urlsplit, urlunsplit
|
|
|
|
import trafilatura
|
|
from bs4 import BeautifulSoup
|
|
|
|
_WHITESPACE_RE = re.compile(r"[ \t\f\v]+")
|
|
_BLANK_LINES_RE = re.compile(r"\n{3,}")
|
|
|
|
# Query params that vary per-request/session but don't change page meaning -
|
|
# stripped so the same logical page always canonicalizes identically.
|
|
_NOISE_QUERY_PREFIXES = ("utm_", "fbclid", "gclid", "mc_", "_hs")
|
|
|
|
|
|
def normalize_whitespace(text: str) -> str:
|
|
text = text.replace("\r\n", "\n").replace("\r", "\n")
|
|
text = _WHITESPACE_RE.sub(" ", text)
|
|
lines = [line.strip() for line in text.split("\n")]
|
|
text = "\n".join(lines)
|
|
return _BLANK_LINES_RE.sub("\n\n", text).strip()
|
|
|
|
|
|
def compute_content_hash(text: str) -> str:
|
|
return hashlib.sha256(text.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def canonicalize_url(url: str) -> str:
|
|
parts = urlsplit(url)
|
|
query_pairs = [
|
|
pair
|
|
for pair in parts.query.split("&")
|
|
if pair and not pair.split("=")[0].startswith(_NOISE_QUERY_PREFIXES)
|
|
]
|
|
path = parts.path.rstrip("/") or "/"
|
|
return urlunsplit((parts.scheme.lower(), parts.netloc.lower(), path, "&".join(query_pairs), ""))
|
|
|
|
|
|
def extract_readable_text(html: str, url: str) -> tuple[str, str]:
|
|
"""Returns (text, extraction_method). Prefers trafilatura (boilerplate
|
|
removal tuned for articles/press releases); falls back to a plain
|
|
BeautifulSoup text extraction for pages trafilatura can't parse (e.g.
|
|
thin job listing pages)."""
|
|
extracted = trafilatura.extract(
|
|
html,
|
|
url=url,
|
|
include_comments=False,
|
|
include_tables=True,
|
|
favor_precision=True,
|
|
)
|
|
if extracted and extracted.strip():
|
|
return normalize_whitespace(extracted), "trafilatura"
|
|
|
|
soup = BeautifulSoup(html, "lxml")
|
|
for tag in soup(["script", "style", "nav", "footer", "header", "noscript"]):
|
|
tag.decompose()
|
|
text = soup.get_text(separator="\n")
|
|
return normalize_whitespace(text), "beautifulsoup_fallback"
|
|
|
|
|
|
def extract_title(html: str) -> str | None:
|
|
soup = BeautifulSoup(html, "lxml")
|
|
if soup.title and soup.title.string:
|
|
return soup.title.string.strip()
|
|
h1 = soup.find("h1")
|
|
if h1:
|
|
return h1.get_text(strip=True)
|
|
return None
|