kunthawat/mitsu-chaiyaporn
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
eea7c4780e0f9b5d04273d22bdbbfb4d0317deef
mitsu-chaiyaporn/mirror.py
kunthawat e04ea1dd27 Mirror www.mitsuthailand.com as static site with dynamic URLs 
- 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
2026-08-06 13:50:48 +07:00

260 lines
9.0 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""Crawl www.mitsuthailand.com -> site/ as a static mirror with dynamic (relative) URLs."""
import os, re, queue, threading, urllib.parse, html
import httpx
from common import (BASE_HOST, BASE_URL, MEDIA_HOST, YP_HOST, OUT_DIR,
assets_rel_for, safe_asset_path, enqueue_asset,
claim_asset_urls, mark_asset_result, is_downloaded,
canonical_page_path)
client = httpx.Client(
follow_redirects=True, timeout=40.0,
headers={"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
"AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0 Safari/537.36",
"Accept-Language": "th,en;q=0.9"},
)
# ---------------- URL classification ----------------
def resolve_url(raw, base):
raw = html.unescape(raw).strip()
if not raw or raw.startswith(("data:", "mailto:", "tel:", "javascript:", "#")):
return None
if raw.startswith(("//", "http")):
parsed = urllib.parse.urlparse(raw)
host = (parsed.hostname or "").lower()
if host not in (BASE_HOST, MEDIA_HOST, YP_HOST):
return None
if "/lp/" in parsed.path or parsed.path.startswith("/lp"):
return None
return urllib.parse.urlunparse(parsed._replace(scheme="https")) if parsed.scheme in ("", "http") else raw
absurl = urllib.parse.urljoin(base, raw)
parsed = urllib.parse.urlparse(absurl)
host = (parsed.hostname or "").lower()
if host and host not in (BASE_HOST, MEDIA_HOST, YP_HOST):
return None
if "/lp/" in parsed.path or parsed.path.startswith("/lp"):
return None
return absurl
def is_page_url(url):
parsed = urllib.parse.urlparse(url)
host = (parsed.hostname or "").lower()
# The www.yellowpages.co.th/lp/... pages are a YellowPages mirror of the same content
# and will die too; skip them. Only mirror the main mitsuthailand.com pages.
if host == YP_HOST:
return False
path = parsed.path.rstrip("/")
# Skip /lp/... "live preview" duplicate copies of the same content (very long paths).
if "/lp/" in path or path.startswith("/lp"):
return False
# skip Drupal file assets (they have extensions or are /sites /core /themes /system paths)
if path.startswith(("/sites/", "/core/", "/themes/", "/system/", "/sites")):
return False
ext = os.path.splitext(path)[1].lower()
# Drupal content routes (no file extension) are pages
return ext in (".html", "")
# ---------------- asset workers ----------------
_jobs = queue.Queue()
_shutdown = threading.Event()
_submitted = set() # URLs already pushed to _jobs (guarantees no dup put)
_sub_lock = threading.Lock()
def start_workers(n=12):
for _ in range(n):
t = threading.Thread(target=_worker, daemon=True)
t.start()
def stop_workers():
_shutdown.set()
def submit_assets(urls):
"""Push each URL to _jobs exactly once; returns count of new submissions."""
new = []
with _sub_lock:
for u in urls:
if u not in _submitted and not is_downloaded(u):
_submitted.add(u)
new.append(u)
for u in new:
_jobs.put(u)
return len(new)
def _worker():
while not _shutdown.is_set():
try:
url = _jobs.get(timeout=0.5)
except queue.Empty:
continue
try:
r = client.get(url)
if r.status_code != 200:
print(f" [asset {r.status_code}] {url}")
mark_asset_result(url, None)
continue
ctype = r.headers.get("content-type", "")
data = r.content
rel = safe_asset_path(assets_rel_for(url), url)
dest = os.path.join(OUT_DIR, rel)
os.makedirs(os.path.dirname(dest), exist_ok=True)
if "css" in ctype or url.rstrip("?;").endswith(".css"):
try:
text = data.decode("utf-8", "replace")
text = rewrite_css(url, text)
data = text.encode("utf-8")
except Exception as e:
print(" [css err]", e)
with open(dest, "wb") as f:
f.write(data)
mark_asset_result(url, rel)
except Exception as e:
print(f" [asset ERR] {url}: {e}")
mark_asset_result(url, None)
# ---------------- CSS rewriting ----------------
def rewrite_css(css_url, css_text):
def repl(m):
u = m.group(1).strip()
if u.startswith(("data:", "#", "http", "//")) or u.startswith("url("):
return m.group(0)
abs = urllib.parse.urljoin(css_url, u)
parsed = urllib.parse.urlparse(abs)
host = (parsed.hostname or "").lower()
if host and host not in (BASE_HOST, MEDIA_HOST, YP_HOST):
return m.group(0)
rel = safe_asset_path(assets_rel_for(abs), abs)
enqueue_asset(abs)
# css files live under assets/<dir>/..., reference is relative to that file's dir
css_dir = os.path.dirname(assets_rel_for(css_url)) # e.g. assets/sites/storage/files/css
from_dir = os.path.join(OUT_DIR, css_dir)
rel_from = os.path.relpath(os.path.join(OUT_DIR, rel), from_dir)
return 'url("{}")'.format(rel_from)
return re.sub(r'url\(\s*["\']?([^"\'()]+)["\']?\s*\)', repl, css_text)
# ---------------- crawl + render ----------------
def fetch_sitemap_urls():
urls = {BASE_URL, BASE_URL + "/"}
try:
r = client.get(f"{BASE_URL}/sitemap.xml")
if r.status_code == 200:
for m in re.finditer(r"<loc>([^<]+)</loc>", r.text):
u = m.group(1).strip().replace("http://", "https://")
if BASE_HOST in u:
urls.add(u)
except Exception as e:
print("sitemap err", e)
return urls
def crawl_pages(seed):
crawled = {} # canonical path -> representative URL
to_crawl = queue.Queue()
for u in seed:
to_crawl.put(u)
for _ in range(5):
found = {}
while not to_crawl.empty():
try:
url = to_crawl.get_nowait()
except queue.Empty:
break
c_key = canonical_page_path(url)
if c_key in crawled:
continue
crawled[c_key] = url
try:
r = client.get(url)
except Exception as e:
print(" [page ERR]", url, e)
continue
if r.status_code != 200:
print(" [page", r.status_code, "]", url)
continue
text = r.text
for l in re.findall(r'(?:href|src|action)="([^"]+)"', text):
a = resolve_url(l, url)
if a is None:
continue
if is_page_url(a):
ca = canonical_page_path(a)
if ca not in crawled and ca not in found:
found[ca] = a
else:
enqueue_asset(a)
for ck, u in found.items():
to_crawl.put(u)
print(f" pass: {len(crawled)} pages, +{len(found)} new")
if not found:
break
return list(crawled.values())
def render_pages(pages):
from rewrite import rewrite_html
seen = set()
for url in sorted(pages):
try:
canon = canonical_page_path(url)
if canon in seen:
continue
seen.add(canon)
r = client.get(url)
if r.status_code != 200:
print(" [render", r.status_code, "]", url)
continue
text = rewrite_html(r.text, url)
rel_path = canon.lstrip("/")
dest = os.path.join(OUT_DIR, rel_path, "index.html")
os.makedirs(os.path.dirname(dest), exist_ok=True)
with open(dest, "w", encoding="utf-8") as f:
f.write(text)
print(f" [page OK] {canon}")
except Exception as e:
print(" [render ERR]", url, e)
def drain_assets():
"""Pull newly-queued assets (common._queue) into _jobs until nothing is pending
or in-flight. Returns when all submitted assets have a result AND no new items
were enqueued in the meantime."""
import time as _time
idle_rounds = 0
while True:
batch = claim_asset_urls()
if batch:
submit_assets(batch)
idle_rounds = 0
if not _submitted:
return
if all(is_downloaded(u) for u in list(_submitted)):
# confirm nothing new got queued before returning
if not claim_asset_urls():
return
_time.sleep(0.3)
def main():
print("Mirroring", BASE_URL, "->", OUT_DIR)
seed = fetch_sitemap_urls()
print(f"Seeded {len(seed)} URLs.")
pages = crawl_pages(seed)
print(f"Total pages: {len(pages)}")
start_workers()
drain_assets()
render_pages(pages)
drain_assets()
stop_workers()
print(f"Pages rendered: {len(pages)}")
print("Done. Site in", OUT_DIR)
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink