kunthawat
e04ea1dd27
- Full static mirror of the YellowPages/Drupal site (39 pages + all assets) - All URLs rewritten to relative/domain-agnostic form (works on any domain) - RFQ + request-quotation + contact forms converted to email (FormSubmit.co) - Google Maps embed replaced with static Google Maps iframe - nginx Dockerfile + config for EasyPanel deployment - mirror.py/rewrite.py/cleanup.py tooling for re-crawling updates
90 lines
3.0 KiB
Python
90 lines
3.0 KiB
Python
#!/usr/bin/env python3
|
|
"""Shared low-level helpers for the mirror (no circular deps)."""
|
|
import os, hashlib, urllib.parse, threading, queue
|
|
|
|
BASE_HOST = "www.mitsuthailand.com"
|
|
BASE_URL = f"https://{BASE_HOST}"
|
|
MEDIA_HOST = "media.yellowpages.co.th"
|
|
YP_HOST = "www.yellowpages.co.th"
|
|
OUT_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "site")
|
|
|
|
# ---------------- asset path mapping ----------------
|
|
def url_path_to_rel(parsed):
|
|
"""Same-host (mitsuthailand.com) asset -> site-relative path mirroring the
|
|
original site structure (e.g. /sites/storage/... -> sites/storage/...).
|
|
Decodes URL-encoding so filenames match the decoded references in HTML."""
|
|
path = urllib.parse.unquote(parsed.path).lstrip("/")
|
|
parts = [seg for seg in path.split("/") if seg not in ("", "..", ".")]
|
|
if not parts:
|
|
parts = ["index"]
|
|
name = parts[-1]
|
|
if parsed.query:
|
|
q = hashlib.md5(parsed.query.encode()).hexdigest()[:8]
|
|
stem, dot, ext = name.rpartition(".")
|
|
name = f"{stem}_{q}{dot}{ext}" if dot else f"{name}_{q}"
|
|
return os.path.join(*parts)
|
|
|
|
def external_url_to_rel(parsed):
|
|
path = urllib.parse.unquote(parsed.path).lstrip("/")
|
|
parts = [seg for seg in path.split("/") if seg not in ("", "..", ".")]
|
|
if not parts:
|
|
parts = ["index"]
|
|
name = parts[-1]
|
|
if parsed.query:
|
|
q = hashlib.md5(parsed.query.encode()).hexdigest()[:8]
|
|
stem, dot, ext = name.rpartition(".")
|
|
name = f"{stem}_{q}{dot}{ext}" if dot else f"{name}_{q}"
|
|
return os.path.join("assets", "external", *parts[:-1], name)
|
|
|
|
def assets_rel_for(url):
|
|
parsed = urllib.parse.urlparse(url)
|
|
host = (parsed.hostname or "").lower()
|
|
if host == BASE_HOST or parsed.netloc == "" or parsed.netloc.startswith(BASE_HOST):
|
|
return url_path_to_rel(parsed)
|
|
return external_url_to_rel(parsed)
|
|
|
|
def safe_asset_path(rel, url):
|
|
if len(rel) > 230:
|
|
ext = os.path.splitext(rel)[1]
|
|
rel = os.path.join("assets", "misc", hashlib.sha1(url.encode()).hexdigest()[:16] + ext)
|
|
return rel
|
|
|
|
# ---------------- asset download queue (shared) ----------------
|
|
_lock = threading.Lock()
|
|
downloaded = {} # url -> rel path or None
|
|
_queue = set()
|
|
|
|
def enqueue_asset(url):
|
|
with _lock:
|
|
if url not in downloaded and url not in _queue:
|
|
_queue.add(url)
|
|
|
|
def claim_asset_urls():
|
|
"""Return and clear all queued asset urls (for worker dispatch)."""
|
|
with _lock:
|
|
items = list(_queue)
|
|
_queue.clear()
|
|
return items
|
|
|
|
def mark_asset_result(url, rel):
|
|
with _lock:
|
|
downloaded[url] = rel
|
|
|
|
def is_downloaded(url):
|
|
with _lock:
|
|
return url in downloaded
|
|
|
|
def pending_count():
|
|
with _lock:
|
|
return len(_queue)
|
|
|
|
|
|
def canonical_page_path(url):
|
|
"""Return the canonical decoded site-root-relative path for a page URL.
|
|
Collapses encoded (%20 etc.) and decoded variants into one path."""
|
|
parsed = urllib.parse.urlparse(url)
|
|
path = urllib.parse.unquote(parsed.path).rstrip("/")
|
|
if path == "":
|
|
return "/"
|
|
return path
|