- Go API server with PostgreSQL + Redis - AI floor replacement (OpenRouter Gemini) - Product database (10 brands, 3539 products) - Recommendation engine, calculator, articles - Redis async queue + worker pool - Hot product caching, brand view tracking - JWT auth, favorites, projects - Docker deployment ready Co-Authored-By: Claude <noreply@anthropic.com>
191 lines
6.2 KiB
Python
191 lines
6.2 KiB
Python
"""
|
|
Scrape brand logos from the web.
|
|
Tries Clearbit Logo API first, falls back to scraping the homepage.
|
|
Outputs brand_logos.json for Go to read.
|
|
"""
|
|
import json
|
|
import os
|
|
import sys
|
|
import urllib.request
|
|
import urllib.error
|
|
import ssl
|
|
import re
|
|
|
|
# ── Brand → domain mapping ──────────────────────────────────
|
|
BRANDS = {
|
|
"Armstrong": "armstrongflooring.com",
|
|
"Bruce": "bruce.com",
|
|
"COREtec": "coretecfloors.com",
|
|
"Daltile": "daltile.com",
|
|
"Karndean": "karndean.com",
|
|
"LifeProof": "lifeproof.com",
|
|
"Mohawk Flooring": "mohawkflooring.com",
|
|
"MSI Surfaces": "msisurfaces.com",
|
|
"Pergo": "pergo.com",
|
|
"Shaw Floors": "shawfloors.com",
|
|
}
|
|
|
|
OUTPUT_DIR = "static/brand_logos"
|
|
OUTPUT_JSON = "brand_logos.json"
|
|
|
|
ssl_ctx = ssl.create_default_context()
|
|
ssl_ctx.check_hostname = False
|
|
ssl_ctx.verify_mode = ssl.CERT_NONE
|
|
|
|
|
|
def fetch_url(url, timeout=10):
|
|
"""Fetch URL and return response data."""
|
|
try:
|
|
req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"})
|
|
resp = urllib.request.urlopen(req, timeout=timeout, context=ssl_ctx)
|
|
return resp.read(), resp.getcode()
|
|
except Exception as e:
|
|
print(f" [!] Failed: {e}")
|
|
return None, 0
|
|
|
|
|
|
def try_clearbit(domain):
|
|
"""Try Clearbit Logo API: https://logo.clearbit.com/{domain}"""
|
|
url = f"https://logo.clearbit.com/{domain}"
|
|
print(f" Trying Clearbit: {url}")
|
|
data, code = fetch_url(url)
|
|
if data and code == 200:
|
|
return data, "png"
|
|
return None, None
|
|
|
|
|
|
def try_homepage_favicon(domain):
|
|
"""Try to find logo from homepage HTML (og:image or favicon)."""
|
|
url = f"https://{domain}"
|
|
print(f" Fetching homepage: {url}")
|
|
html_bytes, code = fetch_url(url)
|
|
if not html_bytes:
|
|
return None, None
|
|
html = html_bytes.decode("utf-8", errors="ignore")
|
|
|
|
# Try og:image first
|
|
m = re.search(r'<meta[^>]+property=["\']og:image["\'][^>]+content=["\']([^"\']+)', html)
|
|
if m:
|
|
img_url = m.group(1)
|
|
if img_url.startswith("//"):
|
|
img_url = "https:" + img_url
|
|
elif img_url.startswith("/"):
|
|
img_url = url + img_url
|
|
print(f" Found og:image: {img_url}")
|
|
data, _ = fetch_url(img_url)
|
|
if data:
|
|
ext = img_url.split(".")[-1].split("?")[0] or "png"
|
|
return data, ext
|
|
|
|
# Try favicon
|
|
m = re.search(r'<link[^>]+rel=["\'](?:icon|shortcut icon)["\'][^>]+href=["\']([^"\']+)', html)
|
|
if m:
|
|
favicon_url = m.group(1)
|
|
if favicon_url.startswith("//"):
|
|
favicon_url = "https:" + favicon_url
|
|
elif favicon_url.startswith("/"):
|
|
favicon_url = url + favicon_url
|
|
print(f" Found favicon: {favicon_url}")
|
|
data, _ = fetch_url(favicon_url)
|
|
if data:
|
|
return data, "ico"
|
|
|
|
# Try /favicon.ico
|
|
favicon_url = f"https://{domain}/favicon.ico"
|
|
print(f" Trying: {favicon_url}")
|
|
data, _ = fetch_url(favicon_url)
|
|
if data:
|
|
return data, "ico"
|
|
|
|
return None, None
|
|
|
|
|
|
def try_wikipedia(name):
|
|
"""Try to get logo from Wikipedia."""
|
|
search_name = name.replace(" ", "_")
|
|
# Try the Wikipedia REST API
|
|
api_url = f"https://en.wikipedia.org/api/rest_v1/page/summary/{search_name}"
|
|
print(f" Trying Wikipedia: {api_url}")
|
|
data, code = fetch_url(api_url)
|
|
if data and code == 200:
|
|
try:
|
|
info = json.loads(data)
|
|
thumb = info.get("thumbnail", {}).get("source")
|
|
if thumb:
|
|
print(f" Found Wikipedia thumbnail: {thumb}")
|
|
img_data, _ = fetch_url(thumb)
|
|
if img_data:
|
|
return img_data, "png"
|
|
except Exception:
|
|
pass
|
|
|
|
# Try infobox image via action=query
|
|
query_url = f"https://en.wikipedia.org/w/api.php?action=query&titles={search_name}&prop=pageimages&format=json&pithumbsize=200"
|
|
print(f" Trying Wikipedia query: {query_url}")
|
|
data, code = fetch_url(query_url)
|
|
if data and code == 200:
|
|
try:
|
|
result = json.loads(data)
|
|
pages = result.get("query", {}).get("pages", {})
|
|
for page in pages.values():
|
|
thumb = page.get("thumbnail", {}).get("source")
|
|
if thumb:
|
|
print(f" Found: {thumb}")
|
|
img_data, _ = fetch_url(thumb)
|
|
if img_data:
|
|
return img_data, "png"
|
|
except Exception:
|
|
pass
|
|
|
|
return None, None
|
|
|
|
|
|
def main():
|
|
os.makedirs(OUTPUT_DIR, exist_ok=True)
|
|
results = {}
|
|
|
|
for brand, domain in BRANDS.items():
|
|
print(f"\n{'='*60}")
|
|
print(f"Brand: {brand} ({domain})")
|
|
|
|
img_data = None
|
|
ext = "png"
|
|
|
|
# Strategy 1: Clearbit
|
|
img_data, ext = try_clearbit(domain)
|
|
|
|
# Strategy 2: Homepage og:image / favicon
|
|
if not img_data:
|
|
img_data, ext = try_homepage_favicon(domain)
|
|
|
|
# Strategy 3: Wikipedia
|
|
if not img_data:
|
|
img_data, ext = try_wikipedia(brand)
|
|
|
|
if img_data and len(img_data) > 100:
|
|
filename = f"{brand.lower().replace(' ', '_').replace('.', '')}.{ext}"
|
|
filepath = os.path.join(OUTPUT_DIR, filename)
|
|
with open(filepath, "wb") as f:
|
|
f.write(img_data)
|
|
local_url = f"/static/brand_logos/{filename}"
|
|
print(f" [OK] Saved: {filepath} ({len(img_data)} bytes)")
|
|
results[brand] = local_url
|
|
else:
|
|
print(f" [FAIL] No logo found for {brand}")
|
|
results[brand] = ""
|
|
|
|
# Write JSON mapping
|
|
json_path = os.path.join(os.path.dirname(__file__) or ".", "..", "..", OUTPUT_JSON)
|
|
# Go binary runs from project root, output to root
|
|
root_json_path = OUTPUT_JSON
|
|
with open(root_json_path, "w", encoding="utf-8") as f:
|
|
json.dump(results, f, indent=2, ensure_ascii=False)
|
|
print(f"\n{'='*60}")
|
|
print(f"Logo mapping saved to {root_json_path}")
|
|
print(f"Logos saved to {OUTPUT_DIR}/")
|
|
print(json.dumps(results, indent=2, ensure_ascii=False))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|