=
9e3e7c3e94
- logo_fetch.py: find a company's logo from its domain (homepage logo candidates, staging URLs, touch icons, favicon services), cached by domain - generate_sprite.py: --domain / --logo-cache, domain column in --batch, logos as PIL images, generate(..., write_files=False) for use as a library
679 lines
27 KiB
Python
Executable File
679 lines
27 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Find a business's logo from its domain.
|
|
|
|
from logo_fetch import fetch_logo
|
|
path = fetch_logo("example.com", "logos/") # Path to logos/example.com.png, or None
|
|
|
|
Strategy, best source first. The first image with a brand hue in it wins; an image that is
|
|
only white/black/gray is kept as a fallback in case nothing colourful turns up.
|
|
|
|
1. Logos on the homepage (https then http, without then with www): <img>/<svg>/<picture>
|
|
with "logo" in class/id/src/alt or inside a logo/brand container, the first image in
|
|
<header>/<nav>, JSON-LD "logo" and og:image URLs that say "logo". Images whose file
|
|
name/alt names something other than the domain, or that sit in partner/client/carousel
|
|
sections, rank lower; only candidates close to the best one's score are tried, so a
|
|
wall of client logos can't beat the site's own black-and-white logo.
|
|
2. The same logos on each page in `extra_urls` (e.g. a staging copy of the site).
|
|
3. Site icons from those pages: apple-touch-icon, large <link rel=icon>, msapplication-TileImage.
|
|
4. Favicon services: Google s2 (sz=128), then DuckDuckGo ip3. Their default globe icons
|
|
(served with HTTP 404) are rejected.
|
|
og:image is only used when its URL says "logo": otherwise it is nearly always a photo.
|
|
|
|
Parked or for-sale domains and domains that redirect to social/review sites are skipped
|
|
(their icons belong to someone else). SVG is rasterized with cairosvg if it is installed,
|
|
else with the `rsvg-convert` binary if it is on PATH, else skipped. Images are stored as PNG
|
|
(at most 256 px) in dest_dir/<domain>.png with dest_dir/<domain>.json holding the method that
|
|
found it; a miss is cached too, so reruns make no requests (pass refresh=True to retry).
|
|
|
|
CLI: python logo_fetch.py example.com [--out logos] [--extra https://staging.example.net/] [--refresh]
|
|
"""
|
|
import argparse
|
|
import base64
|
|
import io
|
|
import json
|
|
import re
|
|
import shutil
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
from html.parser import HTMLParser
|
|
from pathlib import Path
|
|
from urllib.parse import unquote, urljoin, urlparse
|
|
|
|
import numpy as np
|
|
import requests
|
|
from PIL import Image
|
|
|
|
UA = ("Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) "
|
|
"Chrome/128.0.0.0 Safari/537.36")
|
|
HEADERS = {"User-Agent": UA, "Accept-Language": "en-US,en;q=0.9",
|
|
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8"}
|
|
TIMEOUT = (4, 8) # (connect, read) seconds per request
|
|
DNS_SECONDS = 5
|
|
DOWNLOAD_SECONDS = 12 # wall-clock cap for one download
|
|
HOMEPAGE_SECONDS = 20 # stop trying more homepage variants after this
|
|
SEARCH_SECONDS = 75 # stop trying more candidates after this (keeps what was found)
|
|
MAX_PAGE_BYTES = 3_000_000
|
|
MAX_IMAGE_BYTES = 6_000_000
|
|
MAX_LOGO_TRIES = 6 # downloads per page and tier
|
|
MAX_ICON_TRIES = 3
|
|
CACHE_VERSION = 2
|
|
STORE_SIZE = 256
|
|
|
|
LOGO_SCORE = 50 # candidates scoring at least this are "logos", below are "icons"
|
|
|
|
# Third-party badges and widgets that often say "logo" but aren't the business's own.
|
|
FOREIGN = re.compile(r"bbb|better.?business|google|facebook|instagram|twitter|linkedin|youtube|tiktok|"
|
|
r"pinterest|yelp|angi|homeadvisor|thumbtack|houzz|nextdoor|porch|visa|mastercard|"
|
|
r"amex|discover|paypal|venmo|badge|award|review|trustpilot|guild|certif|seal|"
|
|
r"partner|payment|financ|wix|squarespace|godaddy|wordpress|elementor|gravatar|"
|
|
r"emoji|spinner|loader|loading|placeholder|blank|pixel|spacer|arrow|close|menu", re.I)
|
|
# Theme / template stand-ins that a site shows before it gets its real logo.
|
|
PLACEHOLDER = re.compile(r"\bmock|placeholder|your.?logo|logo.?here|my.?company|sample.?logo|dummy|"
|
|
r"default.?logo|logo.?default|lorem|demo.?logo|logoipsum|your.?company", re.I)
|
|
PARKING_HOSTS = ("sedoparking.com", "parkingcrew.net", "bodis.com", "hugedomains.com", "dan.com",
|
|
"afternic.com", "godaddy.com", "namecheap.com", "above.com", "undeveloped.com",
|
|
"sedo.com", "domainmarket.com", "brandbucket.com", "squadhelp.com", "atom.com",
|
|
"parklogic.com", "domainnamesales.com", "uniregistry.com", "porkbun.com",
|
|
"expireddomains.com", "expireddomains.net")
|
|
PLATFORM_HOSTS = ("facebook.com", "instagram.com", "yelp.com", "google.com", "goo.gl", "linktr.ee",
|
|
"linkedin.com", "twitter.com", "x.com", "nextdoor.com", "angi.com", "thumbtack.com",
|
|
"homeadvisor.com", "houzz.com", "bbb.org", "youtube.com", "tiktok.com")
|
|
PARKED_TEXT = re.compile(r"domain (?:name )?(?:is|may be) for sale|buy this domain|this domain is parked|"
|
|
r"parked free|domain parking|parkingcrew|sedoparking|is available for purchase|"
|
|
r"the domain .{0,40} is for sale|future home of|website coming soon|expired domains?|"
|
|
r"domain (?:has )?expired|this site can.t be reached|account suspended|404 not found", re.I)
|
|
|
|
|
|
def normalize_domain(value):
|
|
"""'HTTPS://www.Example.com/about' -> 'example.com'. None if it isn't a plausible domain."""
|
|
if not value:
|
|
return None
|
|
v = value.strip().lower()
|
|
v = re.sub(r"^[a-z][a-z0-9+.-]*://", "", v)
|
|
v = re.split(r"[/?#\s]", v, maxsplit=1)[0]
|
|
v = v.split("@")[-1].split(":")[0].strip(".")
|
|
if v.startswith("www."):
|
|
v = v[4:]
|
|
if "." not in v or not re.fullmatch(r"[a-z0-9.-]+", v) or ".." in v:
|
|
return None
|
|
if not re.search(r"\.[a-z]{2,}$", v):
|
|
return None
|
|
return v
|
|
|
|
|
|
def _host(url):
|
|
h = (urlparse(url).hostname or "").lower()
|
|
return h[4:] if h.startswith("www.") else h
|
|
|
|
|
|
def _host_in(url, hosts):
|
|
h = _host(url)
|
|
return any(h == d or h.endswith("." + d) for d in hosts)
|
|
|
|
|
|
# ============================================================ page parsing
|
|
|
|
class _Candidate:
|
|
__slots__ = ("url", "data", "score", "kind")
|
|
|
|
def __init__(self, url=None, data=None, score=0, kind="img"):
|
|
self.url, self.data, self.score, self.kind = url, data, score, kind
|
|
|
|
|
|
def _srcset_best(srcset):
|
|
best, best_v = None, -1.0
|
|
for part in srcset.split(","):
|
|
bits = part.strip().split()
|
|
if not bits:
|
|
continue
|
|
v = 1.0
|
|
if len(bits) > 1:
|
|
m = re.match(r"([\d.]+)([wx])", bits[1])
|
|
if m:
|
|
v = float(m.group(1))
|
|
if v > best_v:
|
|
best, best_v = bits[0], v
|
|
return best
|
|
|
|
|
|
def _is_hint(text):
|
|
return "logo" in text or bool(re.search(r"navbar-brand|site-brand|brand-logo|branding|site-title|brand\b", text))
|
|
|
|
|
|
def _is_bad_container(text):
|
|
"""Sections that show other companies' logos (partners, clients, brands carried)."""
|
|
return bool(re.search(r"partner|client|customer|sponsor|carousel|slider|swiper|marquee|logos|brands|"
|
|
r"featured|testimonial|review|award|certif|affiliat|manufactur|as-seen", text))
|
|
|
|
|
|
GENERIC_WORDS = set("""logo logos image images icon icons favicon header footer site main white black dark
|
|
light color colour colors final retina mobile small large transparent horizontal vertical sticky scaled
|
|
cropped copy untitled design designs default new web website primary secondary full alt nav navbar brand
|
|
branding home top wordpress upload uploads content screenshot img png jpg jpeg svg webp gif edited
|
|
version original mark text title company business group the and with removebg background bg trans
|
|
frame artboard layer asset assets file photo picture custom resized min hires lores rgb cmyk vector
|
|
square round circle wide long stacked inverse reverse reversed blue green red yellow orange purple
|
|
gray grey""".split())
|
|
|
|
|
|
def _words(*texts):
|
|
out = []
|
|
for t in texts:
|
|
t = re.sub(r"\.(png|jpe?g|svg|webp|gif|ico|avif)$", "", t)
|
|
t = re.sub(r"([a-z])([A-Z])", r"\1 \2", t)
|
|
out += [w for w in re.findall(r"[a-z]{4,}", t.lower()) if w not in GENERIC_WORDS]
|
|
return out
|
|
|
|
|
|
def brand_label(domain):
|
|
"""'shop.miriellopainting.co.uk' -> 'miriellopainting': the name part of a domain."""
|
|
if not domain:
|
|
return None
|
|
parts = domain.split(".")
|
|
if len(parts) >= 3 and len(parts[-1]) == 2 and parts[-2] in ("co", "com", "org", "net", "ac", "gov"):
|
|
parts = parts[:-2]
|
|
else:
|
|
parts = parts[:-1]
|
|
return parts[-1] if parts else None
|
|
|
|
|
|
def relevance(words, brand):
|
|
"""+1 if the words name the brand, -1 if they name something else, 0 if they say nothing."""
|
|
if not words or not brand:
|
|
return 0
|
|
return 1 if any(w in brand or brand in w for w in words) else -1
|
|
|
|
|
|
class _PageParser(HTMLParser):
|
|
CONTAINERS = {"header", "nav", "a", "div", "span", "figure", "picture", "footer", "section", "li",
|
|
"p", "h1", "h2", "button", "main", "aside", "ul"}
|
|
|
|
def __init__(self, base_url, html, brand=None):
|
|
super().__init__(convert_charrefs=True)
|
|
self.base = base_url
|
|
self.html = html
|
|
self.brand = brand
|
|
self.line_starts = [0] + [m.end() for m in re.finditer("\n", html)]
|
|
self.stack = [] # (tag, logo_hint, bad_container)
|
|
self.cands = []
|
|
self.header_imgs = 0
|
|
self.svg_start = None
|
|
self.svg_depth = 0
|
|
self.svg_score = 0
|
|
|
|
def _offset(self):
|
|
line, col = self.getpos()
|
|
return self.line_starts[line - 1] + col
|
|
|
|
def _ctx(self):
|
|
tags = {t for t, _, _ in self.stack}
|
|
return (any(h for _, h, _ in self.stack), bool(tags & {"header", "nav"}), "footer" in tags,
|
|
any(b for _, _, b in self.stack))
|
|
|
|
def _add(self, url, score, kind="img"):
|
|
if not url:
|
|
return
|
|
url = url.strip()
|
|
if url.startswith("data:"):
|
|
self.cands.append(_Candidate(data=url, score=score, kind=kind))
|
|
elif not url.startswith(("javascript:", "#", "about:")):
|
|
self.cands.append(_Candidate(url=urljoin(self.base, url), score=score, kind=kind))
|
|
|
|
def _adjust(self, score, words, in_header, in_footer, in_bad):
|
|
rel = relevance(words, self.brand)
|
|
score += 30 * rel if rel > 0 else 40 * rel
|
|
if in_header:
|
|
score += 15
|
|
if in_footer:
|
|
score -= 15
|
|
if in_bad:
|
|
score -= 40
|
|
return score
|
|
|
|
def handle_starttag(self, tag, attrs):
|
|
a = {k: (v or "") for k, v in attrs}
|
|
ident = f"{a.get('class', '')} {a.get('id', '')}".lower()
|
|
if tag in ("img", "source"):
|
|
self._img(tag, a, ident)
|
|
elif tag == "svg":
|
|
if self.svg_start is None:
|
|
in_hint, in_header, in_footer, in_bad = self._ctx()
|
|
if _is_hint(ident) or in_hint:
|
|
self.svg_start = self._offset()
|
|
self.svg_depth = 0
|
|
self.svg_score = self._adjust(80, _words(a.get("aria-label", ""), a.get("title", "")),
|
|
in_header, in_footer, in_bad)
|
|
if self.svg_start is not None:
|
|
self.svg_depth += 1
|
|
elif tag == "link":
|
|
self._link(a)
|
|
elif tag == "meta":
|
|
prop = (a.get("property") or a.get("name") or "").lower()
|
|
content = a.get("content", "")
|
|
if prop in ("og:image", "og:image:url", "twitter:image") and "logo" in content.lower():
|
|
self._add(content, 55, "og")
|
|
elif prop == "msapplication-tileimage" and content:
|
|
self._add(content, 20, "icon")
|
|
if tag in self.CONTAINERS:
|
|
self.stack.append((tag, _is_hint(ident), _is_bad_container(ident)))
|
|
|
|
def handle_startendtag(self, tag, attrs):
|
|
self.handle_starttag(tag, attrs)
|
|
if tag in self.CONTAINERS and self.stack and self.stack[-1][0] == tag:
|
|
self.stack.pop()
|
|
|
|
def handle_endtag(self, tag):
|
|
if tag == "svg" and self.svg_start is not None:
|
|
self.svg_depth -= 1
|
|
if self.svg_depth <= 0:
|
|
end = self.html.find(">", self._offset()) + 1
|
|
raw = self.html[self.svg_start:end]
|
|
if "<use" not in raw and len(raw) < 400_000:
|
|
if "xmlns=" not in raw[:300]:
|
|
raw = raw.replace("<svg", '<svg xmlns="http://www.w3.org/2000/svg"', 1)
|
|
self.cands.append(_Candidate(data="svg:" + raw, score=self.svg_score, kind="svg"))
|
|
self.svg_start = None
|
|
return
|
|
if tag not in self.CONTAINERS:
|
|
return
|
|
for i in range(len(self.stack) - 1, -1, -1):
|
|
if self.stack[i][0] == tag:
|
|
del self.stack[i:]
|
|
return
|
|
|
|
def _img(self, tag, a, ident):
|
|
src = (a.get("data-src") or a.get("data-lazy-src") or a.get("data-original") or "")
|
|
srcset = a.get("data-srcset") or a.get("data-lazy-srcset") or a.get("srcset") or ""
|
|
plain = a.get("src", "")
|
|
if not src:
|
|
src = plain if plain and not (plain.startswith("data:") and len(plain) < 300) else ""
|
|
if srcset:
|
|
best = _srcset_best(srcset)
|
|
if best and not best.startswith("data:"):
|
|
src = best
|
|
if not src:
|
|
return
|
|
try:
|
|
if 0 < int(a.get("width", "0") or 0) <= 10 and 0 < int(a.get("height", "0") or 0) <= 10:
|
|
return
|
|
except ValueError:
|
|
pass
|
|
fname = unquote(urlparse(src).path.rsplit("/", 1)[-1]) if not src.startswith("data:") else ""
|
|
alt = a.get("alt", "")
|
|
low = f"{fname} {alt}".lower()
|
|
in_hint, in_header, in_footer, in_bad = self._ctx()
|
|
score = 0
|
|
if "custom-logo" in ident:
|
|
score = 110
|
|
elif "logo" in ident:
|
|
score = 100
|
|
elif "logo" in fname.lower():
|
|
score = 90
|
|
elif in_hint:
|
|
score = 85
|
|
elif "logo" in alt.lower():
|
|
score = 70
|
|
elif "brand" in ident:
|
|
score = 60
|
|
elif in_header and self.header_imgs == 0:
|
|
score = 52
|
|
if in_header:
|
|
self.header_imgs += 1
|
|
if not score or tag == "source" and not in_hint:
|
|
return
|
|
words = _words(fname, alt, a.get("title", ""))
|
|
if PLACEHOLDER.search(low) or relevance(words, self.brand) <= 0 and FOREIGN.search(low):
|
|
return
|
|
self._add(src, self._adjust(score, words, in_header, in_footer, in_bad))
|
|
|
|
def _link(self, a):
|
|
rel = a.get("rel", "").lower()
|
|
href = a.get("href", "")
|
|
if not href or "icon" not in rel or "mask-icon" in rel or PLACEHOLDER.search(href.rsplit("/", 1)[-1]):
|
|
return
|
|
size = 0
|
|
m = re.match(r"(\d+)x(\d+)", a.get("sizes", ""))
|
|
if m:
|
|
size = int(m.group(1))
|
|
if "apple-touch-icon" in rel:
|
|
score = 35
|
|
elif size >= 96 or href.lower().endswith(".svg"):
|
|
score = 30
|
|
else:
|
|
score = 10 + min(size, 64) // 8
|
|
self._add(href, score, "icon")
|
|
|
|
|
|
JSONLD_LOGO = re.compile(r'"logo"\s*:\s*(?:\{[^{}]*?"(?:url|contentUrl)"\s*:\s*)?"(https?:[^"]+)"')
|
|
|
|
|
|
def parse_page(base_url, html, brand=None):
|
|
"""Logo candidates on a page, best first: [_Candidate]. Each URL appears once. `brand` is
|
|
the domain's name part (brand_label): images named after something else rank lower."""
|
|
p = _PageParser(base_url, html, brand)
|
|
try:
|
|
p.feed(html)
|
|
p.close()
|
|
except Exception: # malformed markup: keep what was found so far
|
|
pass
|
|
cands = p.cands
|
|
for m in JSONLD_LOGO.finditer(html):
|
|
url = m.group(1).replace("\\/", "/")
|
|
words = _words(unquote(url.rsplit("/", 1)[-1]))
|
|
if relevance(words, brand) > 0 or not FOREIGN.search(url.rsplit("/", 1)[-1]):
|
|
cands.append(_Candidate(url=url, score=75 + (30 * relevance(words, brand) if words else 0)))
|
|
seen, out = set(), []
|
|
for c in sorted(cands, key=lambda c: -c.score):
|
|
key = c.url or c.data
|
|
if key not in seen:
|
|
seen.add(key)
|
|
out.append(c)
|
|
return out
|
|
|
|
|
|
# ============================================================ network + images
|
|
|
|
def _get(session, url, limit, accept=None, deadline=DOWNLOAD_SECONDS):
|
|
"""GET with a size cap and a wall-clock deadline (a server trickling bytes can't stall
|
|
the search). Returns (response, body bytes) or (None, None). A body cut off by the
|
|
deadline is returned as is (fine for HTML; a cut-off image fails to decode)."""
|
|
start = time.monotonic()
|
|
try:
|
|
r = session.get(url, timeout=TIMEOUT, stream=True, allow_redirects=True,
|
|
headers={"Accept": accept} if accept else None)
|
|
except (requests.RequestException, ValueError):
|
|
return None, None
|
|
try:
|
|
body = bytearray()
|
|
for chunk in r.iter_content(65536):
|
|
body += chunk
|
|
if len(body) > limit or time.monotonic() - start > deadline:
|
|
break
|
|
return r, bytes(body)
|
|
except (requests.RequestException, ValueError):
|
|
return None, None
|
|
finally:
|
|
r.close()
|
|
|
|
|
|
def _resolves(host):
|
|
"""True if the host resolves within DNS_SECONDS. getaddrinfo has no timeout of its own
|
|
(requests would hang on a stuck lookup too), so it runs in a daemon thread that is
|
|
abandoned if it takes too long."""
|
|
done = threading.Event()
|
|
ok = []
|
|
|
|
def lookup():
|
|
try:
|
|
socket.getaddrinfo(host, 443, proto=socket.IPPROTO_TCP)
|
|
ok.append(True)
|
|
except (OSError, UnicodeError):
|
|
pass
|
|
done.set()
|
|
|
|
threading.Thread(target=lookup, daemon=True).start()
|
|
return done.wait(DNS_SECONDS) and bool(ok)
|
|
|
|
|
|
def fetch_homepage(session, domain):
|
|
"""(final_url, html) for the first of https/http x bare/www that answers with HTML. Hosts
|
|
that don't resolve are skipped, and the attempts stop after HOMEPAGE_SECONDS."""
|
|
start = time.monotonic()
|
|
hosts = [h for h in (domain, f"www.{domain}") if _resolves(h)]
|
|
for url in [f"{scheme}://{h}/" for scheme in ("https", "http") for h in hosts]:
|
|
if time.monotonic() - start > HOMEPAGE_SECONDS:
|
|
break
|
|
r, body = _get(session, url, MAX_PAGE_BYTES)
|
|
if r is None:
|
|
continue
|
|
if r.status_code >= 400 or "html" not in r.headers.get("content-type", "html").lower():
|
|
if r.status_code in (403, 404, 410, 500, 502, 503):
|
|
continue
|
|
return None, None
|
|
return r.url, body.decode(r.encoding or "utf-8", errors="replace")
|
|
return None, None
|
|
|
|
|
|
def is_parked(final_url, html, domain=None):
|
|
"""True if the page is a parked / for-sale / placeholder page or a social-site redirect."""
|
|
if _host_in(final_url, PARKING_HOSTS + PLATFORM_HOSTS) and not (domain and _host_in(final_url, (domain,))):
|
|
return True
|
|
text = re.sub(r"<script.*?</script>|<style.*?</style>", " ", html[:400_000], flags=re.S | re.I)
|
|
text = re.sub(r"<[^>]+>", " ", text)
|
|
return len(text.split()) < 400 and bool(PARKED_TEXT.search(text))
|
|
|
|
|
|
def _rasterize_svg(data):
|
|
try:
|
|
import cairosvg
|
|
return cairosvg.svg2png(bytestring=data, output_width=STORE_SIZE)
|
|
except ImportError:
|
|
pass
|
|
except Exception:
|
|
return None
|
|
exe = shutil.which("rsvg-convert")
|
|
if not exe:
|
|
return None
|
|
try:
|
|
out = subprocess.run([exe, "-w", str(STORE_SIZE), "--keep-aspect-ratio", "-f", "png"],
|
|
input=data, capture_output=True, timeout=15)
|
|
except (OSError, subprocess.SubprocessError):
|
|
return None
|
|
return out.stdout if out.returncode == 0 and out.stdout else None
|
|
|
|
|
|
def _decode_data_uri(uri):
|
|
if uri.startswith("svg:"):
|
|
return uri[4:].encode("utf-8")
|
|
head, _, payload = uri.partition(",")
|
|
try:
|
|
return base64.b64decode(payload) if ";base64" in head else unquote(payload).encode("utf-8")
|
|
except (ValueError, TypeError):
|
|
return None
|
|
|
|
|
|
def _looks_svg(data, ctype=""):
|
|
head = data[:512].lstrip().lower()
|
|
return "svg" in ctype or head.startswith(b"<svg") or (head.startswith(b"<?xml") and b"<svg" in data[:4096])
|
|
|
|
|
|
def load_image(data, ctype=""):
|
|
"""Bytes -> RGBA PIL image (largest frame/size), or None if it can't be read."""
|
|
if not data:
|
|
return None
|
|
if _looks_svg(data, ctype):
|
|
data = _rasterize_svg(data)
|
|
if not data:
|
|
return None
|
|
try:
|
|
img = Image.open(io.BytesIO(data))
|
|
img.seek(0)
|
|
img.load()
|
|
img = img.convert("RGBA")
|
|
except Exception:
|
|
return None
|
|
if max(img.size) < 16:
|
|
return None
|
|
img.thumbnail((STORE_SIZE, STORE_SIZE), Image.LANCZOS)
|
|
return img
|
|
|
|
|
|
def judge(img):
|
|
"""'hued' if the image has a brand colour, 'neutral' if only white/black/gray, None if it has
|
|
no usable colour (blank, transparent, a single flat block)."""
|
|
from generate_sprite import extract_logo_colors, is_neutral # lazily: keeps import light
|
|
arr = np.array(img)
|
|
opaque = arr[:, :, 3] >= 200
|
|
if opaque.sum() < 50 or opaque.mean() < 0.02:
|
|
return None
|
|
px = arr[opaque][:, :3].astype(float)
|
|
if px.std(axis=0).max() < 3 and opaque.all():
|
|
return None # one flat colour filling the whole canvas
|
|
colors = extract_logo_colors(img)
|
|
if not colors:
|
|
return None
|
|
return "neutral" if all(is_neutral(c) for c, _ in colors) else "hued"
|
|
|
|
|
|
# ============================================================ fetch
|
|
|
|
class _Search:
|
|
def __init__(self, session):
|
|
self.session = session
|
|
self.fallback = None # (img, method): first neutral-only image
|
|
self.tried = set()
|
|
self.start = time.monotonic()
|
|
|
|
def try_candidate(self, c, method):
|
|
"""Returns (img, method) if c is a usable hued image; remembers neutral ones."""
|
|
key = c.url or c.data
|
|
if key in self.tried or time.monotonic() - self.start > SEARCH_SECONDS:
|
|
return None
|
|
self.tried.add(key)
|
|
if c.data:
|
|
data, ctype = _decode_data_uri(c.data), ""
|
|
else:
|
|
r, data = _get(self.session, c.url, MAX_IMAGE_BYTES, accept="image/avif,image/webp,image/*,*/*;q=0.8")
|
|
if r is None or r.status_code != 200 or len(data or b"") > MAX_IMAGE_BYTES:
|
|
return None
|
|
ctype = r.headers.get("content-type", "").lower()
|
|
if "html" in ctype:
|
|
return None
|
|
img = load_image(data, ctype)
|
|
if img is None:
|
|
return None
|
|
verdict = judge(img)
|
|
if verdict == "hued":
|
|
return img, method
|
|
if verdict == "neutral" and self.fallback is None:
|
|
self.fallback = (img, method)
|
|
return None
|
|
|
|
def try_list(self, cands, method, limit):
|
|
for c in cands[:limit]:
|
|
found = self.try_candidate(c, method)
|
|
if found:
|
|
return found
|
|
return None
|
|
|
|
|
|
def _page_candidates(session, url, domain=None, brand=None):
|
|
"""(candidates, parked) for the domain's homepage (url=None) or for one page url."""
|
|
if url is None:
|
|
final, html = fetch_homepage(session, domain)
|
|
else:
|
|
r, body = _get(session, url, MAX_PAGE_BYTES)
|
|
if r is None or r.status_code >= 400 or "html" not in r.headers.get("content-type", "html").lower():
|
|
return [], False
|
|
final, html = r.url, body.decode(r.encoding or "utf-8", errors="replace")
|
|
if not html:
|
|
return [], False
|
|
if is_parked(final, html, domain if url is None else None):
|
|
return [], True
|
|
return parse_page(final, html, brand), False
|
|
|
|
|
|
def find_logo(domain=None, extra_urls=(), session=None):
|
|
"""Search without caching. Returns (PIL image, method) or (None, None). method is one of
|
|
site-logo, extra-logo, site-icon, extra-icon, google-favicon, ddg-favicon,
|
|
or any of those plus '-mono' when only a white/black/gray image was found."""
|
|
session = session or requests.Session()
|
|
session.headers.update(HEADERS)
|
|
s = _Search(session)
|
|
brand = brand_label(domain)
|
|
site, parked = (_page_candidates(session, None, domain, brand) if domain else ([], False))
|
|
extras = None
|
|
|
|
def extra_pages():
|
|
nonlocal extras
|
|
if extras is None:
|
|
extras = [_page_candidates(session, url, brand=brand)[0] for url in extra_urls]
|
|
return extras
|
|
|
|
def logos(cands):
|
|
"""Logo-tier candidates close to the best one: a strong own logo shuts out weak guesses."""
|
|
tier = [c for c in cands if c.score >= LOGO_SCORE]
|
|
best = max((c.score for c in tier), default=0)
|
|
return [c for c in tier if c.score >= best - 30]
|
|
|
|
def icons(cands):
|
|
return [c for c in cands if c.score < LOGO_SCORE]
|
|
|
|
found = s.try_list(logos(site), "site-logo", MAX_LOGO_TRIES)
|
|
for cands in ([] if found else extra_pages()):
|
|
found = found or s.try_list(logos(cands), "extra-logo", MAX_LOGO_TRIES)
|
|
found = found or s.try_list(icons(site), "site-icon", MAX_ICON_TRIES)
|
|
for cands in ([] if found else extra_pages()):
|
|
found = found or s.try_list(icons(cands), "extra-icon", MAX_ICON_TRIES)
|
|
if domain and not parked and not found:
|
|
for url, method in ((f"https://www.google.com/s2/favicons?domain={domain}&sz=128", "google-favicon"),
|
|
(f"https://icons.duckduckgo.com/ip3/{domain}.ico", "ddg-favicon")):
|
|
found = s.try_candidate(_Candidate(url=url), method)
|
|
if found:
|
|
break
|
|
if found:
|
|
return found
|
|
if s.fallback:
|
|
return s.fallback[0], s.fallback[1] + "-mono"
|
|
return None, None
|
|
|
|
|
|
def _cache_key(domain, key):
|
|
k = key or domain
|
|
if not k:
|
|
raise ValueError("fetch_logo needs a domain or a key")
|
|
return re.sub(r"[^a-z0-9._-]+", "_", k.lower())
|
|
|
|
|
|
def fetch_logo_info(domain, dest_dir, extra_urls=(), key=None, refresh=False, session=None):
|
|
"""Cached find_logo. Returns (Path or None, method or None). The cache lives in dest_dir:
|
|
<key>.png plus <key>.json ({"method": ...}); key defaults to the normalized domain."""
|
|
domain = normalize_domain(domain) if domain else None
|
|
dest = Path(dest_dir)
|
|
k = _cache_key(domain, key)
|
|
png, meta = dest / f"{k}.png", dest / f"{k}.json"
|
|
if not refresh and meta.exists():
|
|
try:
|
|
info = json.loads(meta.read_text())
|
|
if info.get("version") == CACHE_VERSION:
|
|
if info.get("method") and png.exists():
|
|
return png, info["method"]
|
|
if not info.get("method"):
|
|
return None, None
|
|
except (ValueError, OSError):
|
|
pass
|
|
img, method = find_logo(domain, extra_urls, session)
|
|
dest.mkdir(parents=True, exist_ok=True)
|
|
if img is not None:
|
|
img.save(png)
|
|
elif png.exists():
|
|
png.unlink()
|
|
meta.write_text(json.dumps({"version": CACHE_VERSION, "domain": domain, "method": method}))
|
|
return (png if img is not None else None), method
|
|
|
|
|
|
def fetch_logo(domain, dest_dir, extra_urls=(), key=None, refresh=False):
|
|
"""Path to a cached PNG of the domain's logo in dest_dir, or None if none was found."""
|
|
return fetch_logo_info(domain, dest_dir, extra_urls, key, refresh)[0]
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser(description="Find a business logo from its domain")
|
|
ap.add_argument("domain")
|
|
ap.add_argument("--out", default="logos", help="Cache folder (default logos/)")
|
|
ap.add_argument("--extra", action="append", default=[], help="Extra page to search (repeatable)")
|
|
ap.add_argument("--refresh", action="store_true", help="Ignore the cache")
|
|
a = ap.parse_args()
|
|
path, method = fetch_logo_info(a.domain, a.out, a.extra, refresh=a.refresh)
|
|
if not path:
|
|
sys.exit(f"No logo found for {a.domain}")
|
|
print(f"{path} ({method})")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|