| 1 | from __future__ import annotations |
| 2 | |
| 3 | import hashlib |
| 4 | import re |
| 5 | import unicodedata |
| 6 | from datetime import datetime, timezone |
| 7 | from decimal import Decimal |
| 8 | from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit |
| 9 | |
| 10 | from bs4 import BeautifulSoup |
| 11 | |
| 12 | from app.models import DocumentType, NormalizedNumber |
| 13 | |
| 14 | |
| 15 | TRACKING_PARAMS = {"utm_source", "utm_medium", "utm_campaign", "utm_term", "utm_content", "fbclid", "gclid"} |
| 16 | |
| 17 | |
| 18 | def canonicalize_url(url: str) -> str: |
| 19 | parts = urlsplit(url.strip()) |
| 20 | scheme = parts.scheme.lower() |
| 21 | host = (parts.hostname or "").lower() |
| 22 | port = f":{parts.port}" if parts.port and parts.port not in {80, 443} else "" |
| 23 | path = re.sub(r"/+", "/", parts.path or "/") |
| 24 | if path != "/" and path.endswith("/"): |
| 25 | path = path[:-1] |
| 26 | query = urlencode(sorted((k, v) for k, v in parse_qsl(parts.query, keep_blank_values=True) if k not in TRACKING_PARAMS)) |
| 27 | return urlunsplit((scheme, host + port, path, query, "")) |
| 28 | |
| 29 | |
| 30 | def normalize_text(text: str) -> str: |
| 31 | text = text.replace("\x00", " ") |
| 32 | text = repair_text_encoding(text) |
| 33 | text = _normalize_typography(text) |
| 34 | return re.sub(r"\s+", " ", text).strip() |
| 35 | |
| 36 | |
| 37 | def repair_text_encoding(text: str) -> str: |
| 38 | text = _repair_known_mojibake(text) |
| 39 | if any(marker in text for marker in ("â", "Ã", "�")): |
| 40 | try: |
| 41 | repaired = text.encode("latin1").decode("utf-8") |
| 42 | except UnicodeError: |
| 43 | repaired = text |
| 44 | else: |
| 45 | if repaired.count("�") <= text.count("�"): |
| 46 | text = repaired |
| 47 | return unicodedata.normalize("NFKC", text) |
| 48 | |
| 49 | |
| 50 | def _normalize_typography(text: str) -> str: |
| 51 | return ( |
| 52 | text.replace("\u2010", "-") |
| 53 | .replace("\u2011", "-") |
| 54 | .replace("\u2012", "-") |
| 55 | .replace("\u2013", "-") |
| 56 | .replace("\u2014", "-") |
| 57 | .replace("\u2018", "'") |
| 58 | .replace("\u2019", "'") |
| 59 | .replace("\u201c", '"') |
| 60 | .replace("\u201d", '"') |
| 61 | .replace("\u00a0", " ") |
| 62 | ) |
| 63 | |
| 64 | |
| 65 | def _repair_known_mojibake(text: str) -> str: |
| 66 | replacements = { |
| 67 | "\u00c3\u00a2\u00c2\u0080\u00c2\u0091": "-", |
| 68 | "\u00c3\u00a2\u00c2\u0080\u00c2\u0099": "'", |
| 69 | "\u00c3\u00a2\u00c2\u0080\u00c2\u009c": '"', |
| 70 | "\u00c3\u00a2\u00c2\u0080\u00c2\u009d": '"', |
| 71 | "\u00c3\u00a2\u00c2\u0080\u00c2\u0093": "-", |
| 72 | "\u00c3\u00a2\u00c2\u0080\u00c2\u0094": "-", |
| 73 | "\u00e2\u201a\u00ac": "\u20ac", |
| 74 | "\u00e2\u201a\u00b9": "\u20b9", |
| 75 | "\u00c3\u00a2\u00e2\u20ac\u0161\u00c2\u00ac": "\u20ac", |
| 76 | "\u00c3\u00a2\u00e2\u20ac\u0161\u00c2\u00b9": "\u20b9", |
| 77 | } |
| 78 | for bad, good in replacements.items(): |
| 79 | text = text.replace(bad, good) |
| 80 | return text |
| 81 | |
| 82 | |
| 83 | def content_hash(text: str) -> str: |
| 84 | return hashlib.sha256(normalize_text(text).lower().encode("utf-8")).hexdigest() |
| 85 | |
| 86 | |
| 87 | def detect_document_type(content_type: str, url: str = "") -> DocumentType: |
| 88 | lower = content_type.lower() |
| 89 | if "pdf" in lower or url.lower().endswith(".pdf"): |
| 90 | return DocumentType.PDF_REFERENCE |
| 91 | if "html" in lower: |
| 92 | return DocumentType.HTML |
| 93 | if "xml" in lower or "rss" in lower: |
| 94 | return DocumentType.RSS_XML |
| 95 | if "text/plain" in lower: |
| 96 | return DocumentType.TEXT |
| 97 | return DocumentType.UNKNOWN |
| 98 | |
| 99 | |
| 100 | def extract_text(content: str, content_type: str) -> tuple[str | None, str | None]: |
| 101 | doc_type = detect_document_type(content_type) |
| 102 | if doc_type == DocumentType.PDF_REFERENCE: |
| 103 | return None, normalize_text(content) |
| 104 | if doc_type == DocumentType.HTML: |
| 105 | soup = BeautifulSoup(content, "html.parser") |
| 106 | title = normalize_text(soup.title.get_text(" ")) if soup.title else None |
| 107 | for tag in soup(["script", "style", "noscript", "svg", "nav", "header", "footer", "form", "button"]): |
| 108 | tag.decompose() |
| 109 | for tag in soup.select("[role='navigation'], .navigation, .navbar, .breadcrumb, .language, .social"): |
| 110 | tag.decompose() |
| 111 | article = _best_article_node(soup) |
| 112 | return title, clean_article_text(normalize_text(article.get_text(" "))) |
| 113 | return None, normalize_text(content) |
| 114 | |
| 115 | |
| 116 | def clean_article_text(text: str) -> str: |
| 117 | text = normalize_text(text) |
| 118 | start_patterns = [ |
| 119 | r"\b[A-Z][A-Za-z.\- ]+,\s+(?:Germany,\s+)?(?:January|February|March|April|May|June|July|August|September|October|November|December)\s+\d{1,2},\s+20\d{2}\b", |
| 120 | r"\b(?:January|February|March|April|May|June|July|August|September|October|November|December)\s+\d{1,2},\s+20\d{2}\b", |
| 121 | r"\b\d{1,2}\.\d{1,2}\.20\d{2}\b", |
| 122 | ] |
| 123 | first = min((match.start() for pattern in start_patterns if (match := re.search(pattern, text, re.IGNORECASE))), default=None) |
| 124 | if first is not None and first > 0: |
| 125 | heading_start = text.rfind(". ", 0, first) |
| 126 | start = heading_start + 2 if heading_start >= 0 else 0 |
| 127 | text = text[start:] |
| 128 | boilerplate = [ |
| 129 | "Navigation", |
| 130 | "Suche", |
| 131 | "German English", |
| 132 | "Facebook", |
| 133 | "Instagram", |
| 134 | "linkedIn", |
| 135 | "Xing", |
| 136 | "Close search", |
| 137 | "Search / HOME", |
| 138 | ] |
| 139 | for phrase in boilerplate: |
| 140 | text = text.replace(phrase, " ") |
| 141 | return normalize_text(text) |
| 142 | |
| 143 | |
| 144 | def _best_article_node(soup: BeautifulSoup): |
| 145 | candidates = soup.select("article, main, [class*='press'], [class*='news'], [class*='content']") |
| 146 | if not candidates: |
| 147 | return soup |
| 148 | return max(candidates, key=lambda tag: len(tag.get_text(" "))) |
| 149 | |
| 150 | |
| 151 | def extract_published_at(text: str) -> datetime | None: |
| 152 | european = re.search(r"\b(?P<day>\d{1,2})\.(?P<month>\d{1,2})\.(?P<year>20\d{2})\b", text) |
| 153 | if european: |
| 154 | return datetime( |
| 155 | int(european.group("year")), |
| 156 | int(european.group("month")), |
| 157 | int(european.group("day")), |
| 158 | tzinfo=timezone.utc, |
| 159 | ) |
| 160 | named = re.search( |
| 161 | r"\b(?P<month>January|February|March|April|May|June|July|August|September|October|November|December)\s+" |
| 162 | r"(?P<day>\d{1,2}),\s+(?P<year>20\d{2})\b", |
| 163 | text, |
| 164 | re.IGNORECASE, |
| 165 | ) |
| 166 | if named: |
| 167 | month = { |
| 168 | "january": 1, |
| 169 | "february": 2, |
| 170 | "march": 3, |
| 171 | "april": 4, |
| 172 | "may": 5, |
| 173 | "june": 6, |
| 174 | "july": 7, |
| 175 | "august": 8, |
| 176 | "september": 9, |
| 177 | "october": 10, |
| 178 | "november": 11, |
| 179 | "december": 12, |
| 180 | }[named.group("month").lower()] |
| 181 | return datetime(int(named.group("year")), month, int(named.group("day")), tzinfo=timezone.utc) |
| 182 | return None |
| 183 | |
| 184 | |
| 185 | _MONEY_PATTERN = re.compile( |
| 186 | r"(?P<prefix>₹|€|\$|INR|EUR|USD)?\s*(?P<number>[+-]?\d+(?:,\d{2,3})*(?:\.\d+)?)\s*(?P<scale>crore|lakh|million|billion|mn|bn)?", |
| 187 | re.IGNORECASE, |
| 188 | ) |
| 189 | _CAPACITY_PATTERN = re.compile(r"(?P<number>\d+(?:\.\d+)?)\s*(?P<unit>MW|GW|units?)", re.IGNORECASE) |
| 190 | _PERCENT_PATTERN = re.compile(r"(?P<number>[+-]?\d+(?:\.\d+)?)\s*%") |
| 191 | |
| 192 | _MONEY_PATTERN = re.compile( |
| 193 | r"(?P<prefix>₹|€|₹|€|\$|INR|EUR|USD)?\s*(?P<number>[+-]?\d+(?:,\d{2,3})*(?:\.\d+)?)\s*(?P<scale>crore|lakh|million|billion|mn|bn)?", |
| 194 | re.IGNORECASE, |
| 195 | ) |
| 196 | |
| 197 | |
| 198 | _MONEY_PATTERN = re.compile( |
| 199 | r"(?P<prefix>\u20b9|\u20ac|\$|INR|EUR|USD)?\s*(?P<number>[+-]?\d+(?:,\d{2,3})*(?:\.\d+)?)\s*(?P<scale>crore|lakh|million|billion|mn|bn)?", |
| 200 | re.IGNORECASE, |
| 201 | ) |
| 202 | |
| 203 | |
| 204 | def normalize_numbers(text: str) -> list[NormalizedNumber]: |
| 205 | values: list[NormalizedNumber] = [] |
| 206 | for match in _MONEY_PATTERN.finditer(text): |
| 207 | original = match.group(0).strip() |
| 208 | prefix = (match.group("prefix") or "").upper() |
| 209 | scale = (match.group("scale") or "").lower() |
| 210 | if not prefix and scale not in {"crore", "lakh", "million", "billion", "mn", "bn"}: |
| 211 | continue |
| 212 | number = Decimal(match.group("number").replace(",", "")) |
| 213 | multiplier = { |
| 214 | "lakh": Decimal("100000"), |
| 215 | "crore": Decimal("10000000"), |
| 216 | "million": Decimal("1000000"), |
| 217 | "mn": Decimal("1000000"), |
| 218 | "billion": Decimal("1000000000"), |
| 219 | "bn": Decimal("1000000000"), |
| 220 | "": Decimal("1"), |
| 221 | }[scale] |
| 222 | currency = {"₹": "INR", "€": "EUR", "$": "USD"}.get(prefix, prefix or None) |
| 223 | if prefix in {"₹", "€", "$"}: |
| 224 | currency = {"₹": "INR", "€": "EUR", "$": "USD"}[prefix] |
| 225 | if prefix == "€": |
| 226 | currency = "EUR" |
| 227 | elif prefix == "₹": |
| 228 | currency = "INR" |
| 229 | values.append(NormalizedNumber(original=original, value=number * multiplier, currency=currency)) |
| 230 | for match in _CAPACITY_PATTERN.finditer(text): |
| 231 | values.append(NormalizedNumber(original=match.group(0), value=Decimal(match.group("number")), unit=match.group("unit").upper())) |
| 232 | for match in _PERCENT_PATTERN.finditer(text): |
| 233 | values.append(NormalizedNumber(original=match.group(0), value=Decimal(match.group("number")), unit="PERCENT")) |
| 234 | return values |