wmj
a1c5fecd83
- Switch Prisma schema from SQLite to PostgreSQL - Add netlify.toml for Netlify deployment config - Add @netlify/plugin-nextjs for serverless Next.js - Remove Docker build files (Dockerfile, docker-compose, start.sh) - Remove auto-healer and scripts directories - Update next.config.ts (remove standalone output) - Database: falah_mobile_v1 on Contabo Postgres
295 lines
10 KiB
Python
Executable File
295 lines
10 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Falah OS — Broken Link Checker
|
|
===============================
|
|
Crawls all known Falah OS pages, extracts every internal link,
|
|
and verifies each returns a valid HTTP response.
|
|
|
|
Usage:
|
|
python3 falah-broken-link-checker.py # Full crawl
|
|
python3 falah-broken-link-checker.py --ci-exit # Exit 1 on broken links
|
|
python3 falah-broken-link-checker.py --domain mobile # Only mobile app domain
|
|
python3 falah-broken-link-checker.py --quick # Only top-level pages
|
|
|
|
Output: /root/.hermes/reports/broken-links/broken-links-YYYYMMDD.json
|
|
"""
|
|
|
|
import argparse, json, os, re, sys, time, urllib.request, urllib.error
|
|
from urllib.parse import urljoin, urlparse
|
|
from html.parser import HTMLParser
|
|
from datetime import datetime
|
|
|
|
# ── Config ─────────────────────────────────────────────────────
|
|
DOMAINS = {
|
|
"mobile": {
|
|
"base": "https://falahos.my/mobile",
|
|
"seed_pages": [
|
|
"/", "/prayer", "/dhikr", "/qibla", "/halal-monitor",
|
|
"/nur", "/forum", "/souq", "/groups", "/learn",
|
|
"/wallet", "/profile",
|
|
],
|
|
"exclude_patterns": [
|
|
r"tel:", r"mailto:", r"javascript:", r"#.*",
|
|
r"api/", r"\.(pdf|zip|doc|jpg|png|svg|ico)$",
|
|
r"https?://(www\.)?(facebook|twitter|x|instagram|linkedin|tiktok|youtube)\.com",
|
|
],
|
|
},
|
|
"istore": {
|
|
"base": "https://store.falah-os.com",
|
|
"seed_pages": ["/", "/apps", "/marketplace", "/developer", "/account"],
|
|
"exclude_patterns": [
|
|
r"tel:", r"mailto:", r"javascript:", r"#.*",
|
|
r"api/", r"\.(pdf|zip|doc)$",
|
|
r"https?://(www\.)?(facebook|twitter|x|instagram|linkedin|tiktok|youtube)\.com",
|
|
],
|
|
},
|
|
}
|
|
|
|
REPORT_DIR = "/root/.hermes/reports/broken-links"
|
|
TIMEOUT = 15
|
|
MAX_LINKS = 300 # Max links to check per run (avoid rate limiting)
|
|
USER_AGENT = "Falah-Broken-Link-Checker/1.0"
|
|
|
|
|
|
class LinkExtractor(HTMLParser):
|
|
"""Extract all href attributes from <a> tags."""
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.links = []
|
|
|
|
def handle_starttag(self, tag, attrs):
|
|
ad = dict(attrs)
|
|
if tag == "a" and "href" in ad:
|
|
self.links.append(ad["href"])
|
|
# Also check sources and imports
|
|
if tag == "link" and "href" in ad:
|
|
self.links.append(ad["href"])
|
|
if tag == "script" and "src" in ad:
|
|
self.links.append(ad["src"])
|
|
if tag == "img" and "src" in ad:
|
|
self.links.append(ad["src"])
|
|
|
|
|
|
def fetch_page(url, timeout=TIMEOUT):
|
|
"""Fetch a page and return its text content."""
|
|
req = urllib.request.Request(
|
|
url, headers={"User-Agent": USER_AGENT}, method="GET"
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
|
raw = resp.read()
|
|
return {
|
|
"status": resp.status,
|
|
"text": raw.decode("utf-8", errors="replace"),
|
|
"error": None,
|
|
}
|
|
except urllib.error.HTTPError as e:
|
|
raw = e.read()
|
|
return {
|
|
"status": e.code,
|
|
"text": raw.decode("utf-8", errors="replace"),
|
|
"error": None,
|
|
}
|
|
except Exception as e:
|
|
return {"status": 0, "text": "", "error": str(e)}
|
|
|
|
|
|
def check_link(url, timeout=TIMEOUT):
|
|
"""Verify a single link returns a valid response.
|
|
Returns (status, error_string)."""
|
|
# Skip non-HTTP links
|
|
if not url.startswith("http"):
|
|
return None, "non-http"
|
|
|
|
# HEAD request to check without downloading body
|
|
req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT}, method="HEAD")
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
|
if resp.status in (200, 301, 302, 307, 308):
|
|
return resp.status, None
|
|
return resp.status, f"HTTP {resp.status}"
|
|
except urllib.error.HTTPError as e:
|
|
if e.code in (301, 302, 307, 308):
|
|
return e.code, None # Redirects are fine
|
|
return e.code, f"HTTP {e.code}"
|
|
except urllib.error.URLError as e:
|
|
return 0, f"DNS/timeout: {str(e.reason)[:60]}"
|
|
except Exception as e:
|
|
return 0, str(e)[:60]
|
|
|
|
|
|
def is_internal(url, base_domain):
|
|
"""Check if URL belongs to the same domain."""
|
|
parsed = urlparse(url)
|
|
return parsed.netloc == "" or base_domain in parsed.netloc
|
|
|
|
|
|
def should_exclude(url, patterns):
|
|
"""Check if URL matches any exclude pattern."""
|
|
for pat in patterns:
|
|
if re.search(pat, url):
|
|
return True
|
|
return False
|
|
|
|
|
|
def normalize_url(href, base_url):
|
|
"""Convert a relative href to absolute URL."""
|
|
if href.startswith("//"):
|
|
return "https:" + href
|
|
return urljoin(base_url, href)
|
|
|
|
|
|
def crawl_and_check(domain_config, quick=False):
|
|
"""Crawl all pages in a domain and check every internal link."""
|
|
base = domain_config["base"]
|
|
seed_pages = domain_config["seed_pages"]
|
|
exclude_patterns = domain_config["exclude_patterns"]
|
|
base_domain = urlparse(base).netloc
|
|
|
|
visited_pages = set()
|
|
all_links = set()
|
|
broken_links = []
|
|
checked = 0
|
|
skipped = 0
|
|
|
|
print(f"\n{'='*60}")
|
|
print(f" Crawling: {base}")
|
|
print(f" Seeds: {len(seed_pages)} pages")
|
|
print(f"{'='*60}")
|
|
|
|
# Phase 1: Crawl pages and extract links
|
|
for seed in seed_pages:
|
|
url = base.rstrip("/") + ("/" + seed.lstrip("/") if seed != "/" else "")
|
|
if url in visited_pages:
|
|
continue
|
|
visited_pages.add(url)
|
|
|
|
print(f" 📄 Fetching: {url}")
|
|
page = fetch_page(url)
|
|
if page["status"] != 200:
|
|
print(f" ⚠️ Page returned HTTP {page['status']} — skipping extraction")
|
|
broken_links.append({
|
|
"page": url,
|
|
"link": url,
|
|
"status": page["status"],
|
|
"error": page["error"] or f"HTTP {page['status']}",
|
|
"type": "page_fetch_failed",
|
|
})
|
|
continue
|
|
|
|
# Extract links from HTML
|
|
extractor = LinkExtractor()
|
|
try:
|
|
extractor.feed(page["text"])
|
|
except Exception as e:
|
|
print(f" ⚠️ HTML parse error: {e}")
|
|
continue
|
|
|
|
for href in extractor.links:
|
|
abs_url = normalize_url(href, url)
|
|
if not is_internal(abs_url, base_domain):
|
|
skipped += 1
|
|
continue
|
|
if should_exclude(abs_url, exclude_patterns):
|
|
skipped += 1
|
|
continue
|
|
all_links.add(abs_url)
|
|
|
|
print(f" Found {len(extractor.links)} links, {len(set(normalize_url(h, url) for h in extractor.links if is_internal(normalize_url(h, url), base_domain) and not should_exclude(normalize_url(h, url), exclude_patterns)))} internal to check")
|
|
|
|
print(f"\n Total unique internal links to check: {len(all_links)}")
|
|
|
|
# Phase 2: Verify each link
|
|
limit = min(MAX_LINKS, len(all_links)) if not quick else 50
|
|
link_list = list(all_links)[:limit]
|
|
|
|
print(f" Checking {len(link_list)} links (MAX_LINKS={limit})...\n")
|
|
|
|
for i, url in enumerate(link_list, 1):
|
|
checked += 1
|
|
status, error = check_link(url)
|
|
if error:
|
|
broken_links.append({
|
|
"page": visited_pages.copy().pop() if visited_pages else base,
|
|
"link": url,
|
|
"status": status,
|
|
"error": error,
|
|
"type": "broken" if status in (0, 404, 410) else "warning",
|
|
})
|
|
print(f" {i:3d}. ❌ {status} {url[:90]}")
|
|
print(f" Error: {error}")
|
|
else:
|
|
if i <= 5 or i % 20 == 0: # Show progress periodically
|
|
print(f" {i:3d}. ✅ {status} {url[:80]}")
|
|
|
|
return {
|
|
"domain": base,
|
|
"checked": checked,
|
|
"skipped": skipped,
|
|
"total_extracted": len(all_links),
|
|
"limit_applied": limit < len(all_links),
|
|
"broken": broken_links,
|
|
"broken_count": len(broken_links),
|
|
"success_count": checked - len(broken_links),
|
|
}
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Falah OS Broken Link Checker")
|
|
parser.add_argument("--ci-exit", action="store_true", help="Exit 1 on broken links")
|
|
parser.add_argument("--domain", choices=["mobile", "istore", "all"], default="all",
|
|
help="Domain to check (default: all)")
|
|
parser.add_argument("--quick", action="store_true", help="Quick check (fewer links)")
|
|
args = parser.parse_args()
|
|
|
|
os.makedirs(REPORT_DIR, exist_ok=True)
|
|
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
report = {
|
|
"timestamp": datetime.now().isoformat(),
|
|
"domains_checked": [],
|
|
"total_broken": 0,
|
|
"total_checked": 0,
|
|
"passed": True,
|
|
}
|
|
|
|
domains_to_check = []
|
|
if args.domain in ("mobile", "all"):
|
|
domains_to_check.append(("mobile", DOMAINS["mobile"]))
|
|
if args.domain in ("istore", "all"):
|
|
domains_to_check.append(("istore", DOMAINS["istore"]))
|
|
|
|
for name, config in domains_to_check:
|
|
result = crawl_and_check(config, quick=args.quick)
|
|
report["domains_checked"].append(result)
|
|
report["total_broken"] += result["broken_count"]
|
|
report["total_checked"] += result["checked"]
|
|
|
|
report["passed"] = report["total_broken"] == 0
|
|
|
|
# Save report
|
|
report_file = os.path.join(REPORT_DIR, f"broken-links-{timestamp}.json")
|
|
with open(report_file, "w") as f:
|
|
json.dump(report, f, indent=2, default=str)
|
|
|
|
print(f"\n{'='*60}")
|
|
print(f" BROKEN LINK CHECK COMPLETE")
|
|
print(f" Total checked: {report['total_checked']}")
|
|
print(f" Total broken: {report['total_broken']}")
|
|
print(f" Status: {'✅ PASS' if report['passed'] else '❌ FAIL'}")
|
|
print(f" Report: {report_file}")
|
|
print(f"{'='*60}")
|
|
|
|
# Print broken links summary
|
|
if report["total_broken"] > 0:
|
|
print(f"\n Broken Links:")
|
|
for domain in report["domains_checked"]:
|
|
for bl in domain["broken"]:
|
|
print(f" [{bl['status']}] {bl['link']}")
|
|
print(f" ({bl['error']})")
|
|
|
|
sys.exit(0 if report["passed"] else (1 if args.ci_exit else 0))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|