#!/usr/bin/env python3 import json, re, sys, time, html, os from pathlib import Path from urllib.parse import urljoin, urlparse, urldefrag, parse_qsl, urlencode, urlunparse from urllib.request import Request, urlopen from urllib.error import HTTPError, URLError from collections import deque, defaultdict ROOT = Path(__file__).resolve().parents[1] BASE = os.environ.get('MPC_AUDIT_BASE', 'http://127.0.0.1:43104').rstrip('/') _LOCAL_NETLOC = urlparse(BASE).netloc.lower() HOSTS = {'marlothparkcentral.com','www.marlothparkcentral.com',_LOCAL_NETLOC} BAD_CODES = {402,403,404,405} MAX = 2000 # Basic URL extraction patterns for HTML/CSS/JS asset surfaces ATTR_RE = re.compile(r'''(?:href|src|action|poster)\s*=\s*["']([^"'#][^"']*)["']''', re.I) SRCSET_RE = re.compile(r'''(?:srcset)\s*=\s*["']([^"']+)["']''', re.I) URL_CSS_RE = re.compile(r'''url\(([^)]+)\)''', re.I) A_RE = re.compile(r''']*?href\s*=\s*["']([^"']+)["'][^>]*>''', re.I) META_REFRESH_RE = re.compile(r'''http-equiv=["']refresh["'][^>]*content=["'][^"']*url=([^"']+)''', re.I) ROUTE_BAD = [] VISITED = {} DISCOVERED_BY = {} SKIPPED_EXTERNAL = set() ignore_prefixes = ('mailto:', 'tel:', 'sms:', 'whatsapp:', 'javascript:', 'data:', 'blob:', '#') ignore_paths_prefix = ('/_next/data/',) # The API form endpoints are not public links/pages; do not GET arbitrary POST APIs from form action. api_skip_prefixes = ('/api/mpc/contact-message','/api/mpc/accommodation-enquiry','/api/mpc/stay-enquiry','/api/mpc/order-pack','/api/mpc/free-guide-lead','/api/mpc/gated-delivery/request-code','/api/mpc/gated-delivery/verify','/api/mpc/guide-delivery','/api/mpc/alerts-subscribe','/api/mpc/preferences','/api/mpc/business-listing-application','/api/admin/') def normalise(raw, current='/'): if not raw: return None raw = html.unescape(raw.strip()) if not raw or raw.startswith(ignore_prefixes): return None # srcset: caller handles pieces raw = raw.replace('&', '&') u = urljoin(BASE + current, raw) u, _ = urldefrag(u) p = urlparse(u) if p.scheme not in ('http','https'): return None host = p.netloc.lower() if host not in HOSTS: SKIPPED_EXTERNAL.add(u) return None # Convert canonical domain to local path = p.path or '/' if any(path.startswith(x) for x in ignore_paths_prefix): return None # Normalize trailing slash except root and files; keep query for assets but sort it query = urlencode(sorted(parse_qsl(p.query, keep_blank_values=True))) if p.query else '' # Avoid GETing API form endpoints that are intentionally POST-only, unless image/download static APIs if any(path.startswith(x) for x in api_skip_prefixes): return None base_parts = urlparse(BASE) return urlunparse((base_parts.scheme,base_parts.netloc,path,'',query,'')) def fetch(url): req = Request(url, headers={'User-Agent':'MPC-launch-crawler/1.0','Accept':'text/html,application/xhtml+xml,application/xml,image/*,*/*;q=0.8'}) try: with urlopen(req, timeout=15) as r: body = r.read(2_500_000) ctype = r.headers.get('content-type','').split(';')[0].lower() return r.status, ctype, body, str(r.url) except HTTPError as e: try: body = e.read(1_000_000) except Exception: body=b'' return e.code, e.headers.get('content-type','').split(';')[0].lower(), body, url except URLError as e: return 'ERR', '', str(e).encode(), url except Exception as e: return 'ERR', '', repr(e).encode(), url def extract_links(body, ctype, path): text = '' links=[] if ctype in ('text/html','application/xhtml+xml','text/css','application/javascript','text/javascript') or path.endswith(('.js','.css','.html')): try: text = body.decode('utf-8', errors='ignore') except Exception: text='' if ctype in ('text/html','application/xhtml+xml') or path.endswith('.html') or path == '/': for m in ATTR_RE.finditer(text): links.append(m.group(1)) for m in A_RE.finditer(text): links.append(m.group(1)) for m in META_REFRESH_RE.finditer(text): links.append(m.group(1)) for m in SRCSET_RE.finditer(text): for part in m.group(1).split(','): bits = part.strip().split() if bits: links.append(bits[0]) if ctype in ('text/css','text/html','application/javascript','text/javascript') or path.endswith(('.css','.js','.html')): for m in URL_CSS_RE.finditer(text): links.append(m.group(1).strip(' "\'')) return links def initial_urls(): urls = ['/'] rm = ROOT/'.next/routes-manifest.json' if rm.exists(): data=json.load(open(rm)) for r in data.get('staticRoutes',[]): page=r.get('page') if page and not page.startswith('/api') and page not in ('/_not-found',): urls.append(page) for r in data.get('dynamicRoutes',[]): page=r.get('page') # include dynamic pages only for known data slugs below # Known content slugs from data files and sitemap sitemap_url = BASE + '/sitemap.xml' try: st,ct,body,final=fetch(sitemap_url) if st==200: for loc in re.findall(rb'(.*?)', body): u=normalise(loc.decode('utf-8','ignore'), '/') if u: urls.append(urlparse(u).path) except Exception: pass # Add common dynamic assets/pages discovered from source strings if present for pattern_file in ['data/stays.ts','data/suppliers.ts','data/product-catalog.ts','data/route-packs.ts','data/mpc-content.ts']: p=ROOT/pattern_file if p.exists(): txt=p.read_text(errors='ignore') for slug in re.findall(r'''slug\s*:\s*["']([^"']+)["']''', txt): for prefix in ['/accommodation/','/local-options/','/products/','/route-packs/']: urls.append(prefix+slug) for href in re.findall(r'''href\s*:\s*["']([^"']+)["']''', txt): if href.startswith('/'): urls.append(href) out=[] seen=set() for u in urls: n=normalise(u, '/') if n and n not in seen: out.append(n); seen.add(n) return out q=deque(initial_urls()) for u in list(q): DISCOVERED_BY[u]='seed' while q and len(VISITED)=500)) or status=='ERR' if bad: ROUTE_BAD.append((url,status,ctype,DISCOVERED_BY.get(url,''),body[:300].decode('utf-8','ignore'))) continue if isinstance(status,int) and status in (200,301,302,303,307,308): for raw in extract_links(body, ctype, path): # srcset may have descriptors; normalise handles bad tokens poorly; skip pure descriptors if raw in ('1x','2x') or re.match(r'^\d+w$', raw): continue nu=normalise(raw, path) if nu and nu not in VISITED and nu not in q: DISCOVERED_BY[nu]=url q.append(nu) # Also inspect server/source for raw hrefs not discovered because client components may render later for p in list(ROOT.glob('**/*')): if p.is_file() and p.suffix in ('.tsx','.ts','.js','.css','.html') and 'node_modules' not in p.parts and '.next/cache' not in str(p): try: txt=p.read_text(errors='ignore') except Exception: continue for raw in re.findall(r'''(?:href|src|action)\s*=\s*[{]?["'`]([^"'`{}]+)["'`]''', txt): nu=normalise(raw, '/') if nu and urlparse(nu).path.startswith('/api') is False and nu not in VISITED and len(VISITED)=500)) or status=='ERR' if bad: ROUTE_BAD.append((nu,status,ctype,DISCOVERED_BY[nu],body[:300].decode('utf-8','ignore'))) summary={ 'checked_count': len(VISITED), 'bad_count': len(ROUTE_BAD), 'bad': ROUTE_BAD, 'visited': VISITED, 'external_skipped_count': len(SKIPPED_EXTERNAL), 'external_skipped_sample': sorted(SKIPPED_EXTERNAL)[:50] } out=ROOT/'operations/reports/v6.82.105-deep-launch-audit-crawl.json' out.parent.mkdir(parents=True, exist_ok=True) out.write_text(json.dumps(summary, indent=2), encoding='utf-8') print(f"checked_count={len(VISITED)} bad_count={len(ROUTE_BAD)} external_skipped={len(SKIPPED_EXTERNAL)}") if ROUTE_BAD: for row in ROUTE_BAD[:100]: print('BAD', row[:4]) else: print('NO_BAD_ROUTE_LINK_STATUS')