#!/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): /
with "logo" in class/id/src/alt or inside a logo/brand container, the first image in
/, 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 .png with dest_dir/.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 "= 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"|", " ", 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" 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:
.png plus .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()