410 lines
13 KiB
Python
Executable File
410 lines
13 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""giac_lookup.py – GIAC Certified Holder Directory lookup tool (static SSR parser, no browser).
|
||
|
||
Usage:
|
||
giac_lookup John Doe
|
||
giac_lookup Doe
|
||
giac_lookup 123456
|
||
giac_lookup Smith --json
|
||
|
||
Network:
|
||
Respects standard environment variables (http_proxy, https_proxy, no_proxy)
|
||
or explicit proxy via `--proxy <url>`. Requires direct or proxied internet egress.
|
||
|
||
Dependencies:
|
||
Python standard library only.
|
||
"""
|
||
import argparse
|
||
from datetime import date
|
||
import html as htmlmod
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
import time
|
||
import urllib.error
|
||
import urllib.parse
|
||
import urllib.request
|
||
|
||
BASE = "https://www.giac.org"
|
||
UA = ("Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
|
||
"(KHTML, like Gecko) Chrome/149.0.0.0 Safari/537.36")
|
||
|
||
HEADERS = {
|
||
"User-Agent": UA,
|
||
"Accept": ("text/html,application/xhtml+xml,application/xml;q=0.9,"
|
||
"image/avif,image/webp,*/*;q=0.8"),
|
||
"Accept-Language": "en-US,en;q=0.9",
|
||
"Accept-Encoding": "identity",
|
||
"Sec-Fetch-Dest": "document",
|
||
"Sec-Fetch-Mode": "navigate",
|
||
"Sec-Fetch-Site": "none",
|
||
"Upgrade-Insecure-Requests": "1",
|
||
}
|
||
|
||
|
||
def http_get(url, proxy=None, timeout=40, retries=3):
|
||
"""Execute HTTP GET request using configured headers, retries, and proxy resolution."""
|
||
def _open(px):
|
||
handlers = [urllib.request.ProxyHandler(
|
||
{k: px for k in ("http", "https")})] if px else []
|
||
opener = urllib.request.build_opener(*handlers)
|
||
req = urllib.request.Request(url, headers=HEADERS)
|
||
with opener.open(req, timeout=timeout) as resp:
|
||
return resp.read().decode("utf-8", "ignore")
|
||
|
||
proxy = proxy if proxy is not None else (
|
||
os.environ.get("https_proxy") or
|
||
os.environ.get("HTTPS_PROXY") or
|
||
os.environ.get("http_proxy") or
|
||
os.environ.get("HTTP_PROXY"))
|
||
|
||
last_err = None
|
||
for attempt in range(retries):
|
||
try:
|
||
return _open(proxy)
|
||
except urllib.error.HTTPError as e:
|
||
last_err = e
|
||
if e.code in (403, 407, 429):
|
||
break
|
||
time.sleep(1.5 * (attempt + 1))
|
||
except Exception as e: # noqa: BLE001
|
||
last_err = e
|
||
time.sleep(1.5 * (attempt + 1))
|
||
raise RuntimeError(
|
||
f"GET request failed for {url} "
|
||
f"(proxy={'configured' if proxy else 'direct'}): {last_err}")
|
||
|
||
|
||
def is_blocked(html_text):
|
||
"""Detect WAF/Incapsula challenge or block responses."""
|
||
markers = (
|
||
"Request unsuccessful. Incapsula incident ID",
|
||
"You are browsing this site with a prohibited",
|
||
"blocked your request",
|
||
)
|
||
low = html_text.lower()
|
||
return any(m.lower() in low for m in markers)
|
||
|
||
|
||
# ---------- Progress Logging ----------
|
||
|
||
_QUIET = False
|
||
|
||
|
||
def _info(msg):
|
||
"""Write informational log to stderr so stdout remains clean for piping."""
|
||
if not _QUIET:
|
||
print(msg, file=sys.stderr, flush=True)
|
||
|
||
|
||
def set_quiet(q):
|
||
global _QUIET
|
||
_QUIET = q
|
||
|
||
|
||
# ---------- Parsing ----------
|
||
|
||
TAG_TEXT = re.compile(r"<[^>]+>")
|
||
|
||
|
||
def _clean(s):
|
||
s = TAG_TEXT.sub(" ", s)
|
||
s = htmlmod.unescape(s)
|
||
return re.sub(r"\s+", " ", s).strip()
|
||
|
||
|
||
CARD_SPLIT = re.compile(r'data-testid="certification-holder-card"[^>]*>')
|
||
|
||
|
||
def parse_card(card_html):
|
||
title_m = re.search(r"<h3[^>]*>(.*?)</h3>", card_html, re.S)
|
||
title = _clean(title_m.group(1)) if title_m else None
|
||
code = None
|
||
if title:
|
||
cm = re.search(r"\(([A-Z0-9]{2,6})\)\s*$", title)
|
||
if cm:
|
||
code = cm.group(1)
|
||
flat = _clean(card_html)
|
||
|
||
def field(label):
|
||
m = re.search(label + r"\s*:\s*([0-9]{4}-[0-9]{2}-[0-9]{2})", flat)
|
||
return m.group(1) if m else None
|
||
|
||
num_m = re.search(r"Analyst Number\s*:\s*(\d+)", flat)
|
||
return {
|
||
"title": title,
|
||
"code": code,
|
||
"analyst_number": num_m.group(1) if num_m else None,
|
||
"obtained": field(r"Obtained"),
|
||
"expiry": field(r"Expiry"),
|
||
}
|
||
|
||
|
||
def _section(html_text, heading):
|
||
start = html_text.find(heading)
|
||
if start < 0:
|
||
return ""
|
||
nxt = [html_text.find(h, start + len(heading))
|
||
for h in ("Inactive Certifications", "Subscribe to GIAC")
|
||
if html_text.find(h, start + len(heading)) > start]
|
||
end = min(nxt) if nxt else len(html_text)
|
||
return html_text[start:end]
|
||
|
||
|
||
def parse_detail(html_text):
|
||
name_m = re.search(r'<h1[^>]*>(.*?)</h1>', html_text, re.S)
|
||
name = _clean(name_m.group(1)) if name_m else None
|
||
flat = _clean(html_text)
|
||
id_m = re.search(r"Analyst ID\s*:\s*(\d+)", flat)
|
||
since_m = re.search(r"Certified since\s*:\s*([A-Za-z]+\s+\d{4})", flat)
|
||
active = _cards_in(html_text, "Active Certifications")
|
||
inactive = _cards_in(html_text, "Inactive Certifications")
|
||
return {
|
||
"name": name,
|
||
"analyst_id": id_m.group(1) if id_m else None,
|
||
"certified_since": since_m.group(1) if since_m else None,
|
||
"active": active,
|
||
"inactive": inactive,
|
||
}
|
||
|
||
|
||
def _cards_in(html_text, heading):
|
||
section = _section(html_text, heading)
|
||
out = []
|
||
for seg in CARD_SPLIT.split(section)[1:]:
|
||
card = parse_card(seg)
|
||
if card["title"]:
|
||
out.append(card)
|
||
return out
|
||
|
||
|
||
def parse_search(html_text):
|
||
"""Parse directory search results into list of {analyst_id, name, path}."""
|
||
rows = re.findall(
|
||
r'<a[^>]*href="(/certified-professional/[^"]+)"[^>]*>(.*?)</a>',
|
||
html_text, re.S)
|
||
out = []
|
||
seen = set()
|
||
for href, inner in rows:
|
||
m = re.match(r"/certified-professional/(.+)/(\d+)$", href)
|
||
if not m:
|
||
continue
|
||
id_num = m.group(2)
|
||
name = _clean(inner)
|
||
name = re.sub(r"^\d+\s+", "", name).strip()
|
||
key = (id_num, name)
|
||
if key in seen:
|
||
continue
|
||
seen.add(key)
|
||
out.append({"analyst_id": id_num, "name": name, "path": href})
|
||
return out
|
||
|
||
|
||
# ---------- API Calls ----------
|
||
|
||
def search(query, proxy=None):
|
||
_info(f"→ Searching for '{query}' ...")
|
||
q = urllib.parse.quote(query, safe="")
|
||
url = f"{BASE}/certified-professional?certified_holder%5Bquery%5D={q}"
|
||
html_text = http_get(url, proxy)
|
||
if is_blocked(html_text):
|
||
raise RuntimeError("Incapsula/WAF block encountered. Please try again later.")
|
||
if "certified-professional/" not in html_text:
|
||
_info("→ No matches found.")
|
||
return []
|
||
results = parse_search(html_text)
|
||
_info(f"→ {len(results)} match(es) found.")
|
||
return results
|
||
|
||
|
||
def person(slug_id, proxy=None):
|
||
_info(f"→ Loading profile {slug_id} ...")
|
||
url = f"{BASE}/certified-professional/{slug_id}"
|
||
html_text = http_get(url, proxy)
|
||
if is_blocked(html_text):
|
||
raise RuntimeError("Incapsula/WAF block encountered on profile page.")
|
||
p = parse_detail(html_text)
|
||
_info(f"→ {len(p['active'])} active / {len(p['inactive'])} inactive certifications.")
|
||
return p
|
||
|
||
|
||
# ---------- Cost Estimation ----------
|
||
|
||
DEFAULT_FIRST_EUR = float(os.environ.get("GIAC_FIRST_PRICE_EUR", "8090"))
|
||
DEFAULT_RENEWAL_USD = 499.0
|
||
DEFAULT_DISCOUNT_USD = 249.0
|
||
DEFAULT_FX = float(os.environ.get("GIAC_USD_TO_EUR", "0.88"))
|
||
NORMAL_YEARS = 4.0
|
||
COTERM_THRESHOLD = 9
|
||
|
||
|
||
def _iso_date(s):
|
||
y, m, d = (int(x) for x in s.split("-"))
|
||
return date(y, m, d)
|
||
|
||
|
||
def estimate_costs(p, first_eur=None, renewal_usd=DEFAULT_RENEWAL_USD,
|
||
discount_usd=DEFAULT_DISCOUNT_USD, fx=None):
|
||
first_eur = DEFAULT_FIRST_EUR if first_eur is None else first_eur
|
||
fx = DEFAULT_FX if fx is None else fx
|
||
renewal_eur = renewal_usd * fx
|
||
discount_eur = discount_usd * fx
|
||
all_certs = p["active"] + p["inactive"]
|
||
first_total_eur = len(all_certs) * first_eur
|
||
|
||
recerts = []
|
||
for c in p["active"]:
|
||
if c["obtained"] and c["expiry"]:
|
||
try:
|
||
years = (_iso_date(c["expiry"]) - _iso_date(c["obtained"])).days / 365.25
|
||
except ValueError:
|
||
continue
|
||
cycles = max(0, int(round(years / NORMAL_YEARS)) - 1) if years > NORMAL_YEARS + 0.15 else 0
|
||
if cycles:
|
||
recerts.append({"title": c["title"], "code": c["code"], "cycles": cycles})
|
||
|
||
recert_events = sum(r["cycles"] for r in recerts)
|
||
low_eur = first_total_eur + recert_events * discount_eur
|
||
high_eur = first_total_eur + recert_events * renewal_eur
|
||
return {
|
||
"first_count": len(all_certs),
|
||
"first_unit_eur": first_eur,
|
||
"first_total_eur": first_total_eur,
|
||
"recerts": recerts,
|
||
"recert_events": recert_events,
|
||
"low_eur": low_eur,
|
||
"high_eur": high_eur,
|
||
"fx": fx,
|
||
"coterm_note": len(p["active"]) >= COTERM_THRESHOLD,
|
||
}
|
||
|
||
|
||
def print_estimate(est):
|
||
"""Print one-line summary estimated expenditure."""
|
||
if not est:
|
||
return
|
||
if est["recert_events"]:
|
||
tag = (f"{est['low_eur']:,.0f} - {est['high_eur']:,.0f} EUR "
|
||
f"(initial certifications: {est['first_count']}x {est['first_unit_eur']:,.0f} EUR + "
|
||
f"{est['recert_events']} renewals, multi-discount vs standard rate)")
|
||
else:
|
||
tag = f"{est['low_eur']:,.0f} EUR (initial certifications: {est['first_count']}x {est['first_unit_eur']:,.0f} EUR)"
|
||
if est["coterm_note"]:
|
||
tag += " · 9+ active certifications: co-termination applies"
|
||
print()
|
||
print(f"Estimated historical investment: ~{tag}")
|
||
|
||
|
||
# ---------- Output Formatting ----------
|
||
|
||
def fmt_date(iso):
|
||
if not iso:
|
||
return "-"
|
||
y, m, d = iso.split("-")
|
||
return f"{d}.{m}.{y}"
|
||
|
||
|
||
def print_search(results):
|
||
if not rows_print(results):
|
||
return
|
||
print(f"{'ID':>8} Name")
|
||
print("-" * 60)
|
||
for r in results:
|
||
print(f"{r['analyst_id']:>8} {r['name']}")
|
||
print(f"\n{len(results)} match(es)")
|
||
|
||
|
||
def rows_print(results):
|
||
if not results:
|
||
print("No matches found.")
|
||
return False
|
||
return True
|
||
|
||
|
||
def print_report(p):
|
||
print(f"GIAC Certification Status: {p['name'] or '?'}")
|
||
print(f"Analyst ID: {p['analyst_id'] or '?'} · "
|
||
f"Certified since: {p['certified_since'] or '?'}")
|
||
print("Source: giac.org Certified Professional Directory "
|
||
f"(retrieved {time.strftime('%Y-%m-%d')})")
|
||
for heading, label in (("active", "Active Certifications"),
|
||
("inactive", "Inactive Certifications (Expired)")):
|
||
certs = p[heading]
|
||
print(f"\n### {label} ({len(certs)})\n")
|
||
if not certs:
|
||
print("- None")
|
||
continue
|
||
w = max(len(c["title"] or "") for c in certs)
|
||
print(f"{'Certification':<{w}} Analyst No. Obtained Expires")
|
||
print("-" * (w + 42))
|
||
for c in certs:
|
||
print(f"{c['title'] or '?':<{w}} {c['analyst_number'] or '-':>11} "
|
||
f"{fmt_date(c['obtained'])} {fmt_date(c['expiry'])}")
|
||
|
||
print_estimate(estimate_costs(p))
|
||
|
||
|
||
def resolve_and_report(query, proxy):
|
||
results = search(query, proxy)
|
||
if not results:
|
||
print(f"No matches found for '{query}'.")
|
||
return None
|
||
if len(results) == 1:
|
||
r = results[0]
|
||
else:
|
||
print(f"{len(results)} matches for '{query}':\n")
|
||
for i, r in enumerate(results, 1):
|
||
print(f" [{i}] {r['analyst_id']:>7} {r['name']}")
|
||
print()
|
||
try:
|
||
sel = input("Select number (Enter=1, 'q'=quit): ").strip()
|
||
except EOFError:
|
||
sel = "1"
|
||
if sel.lower() == "q":
|
||
return None
|
||
idx = int(sel) - 1 if sel else 0
|
||
r = results[idx]
|
||
_info(f"→ Selected: {r['name']} (ID {r['analyst_id']})")
|
||
slug = r["path"].split("/certified-professional/", 1)[1]
|
||
p = person(slug, proxy)
|
||
p["name"] = p["name"] or r["name"]
|
||
p["analyst_id"] = p["analyst_id"] or r["analyst_id"]
|
||
print_report(p)
|
||
return p
|
||
|
||
|
||
# ---------- CLI Entrypoint ----------
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser(
|
||
prog="giac_lookup",
|
||
description="Lookup GIAC certifications by name, surname, or analyst ID.")
|
||
ap.add_argument("query", nargs="+",
|
||
help="Name, surname, or Analyst ID (e.g. 'John Doe', 'Smith', '123456')")
|
||
ap.add_argument("--json", action="store_true",
|
||
help="Output raw search results as JSON")
|
||
ap.add_argument("--search-only", "-s", action="store_true",
|
||
help="Only list matching holders without fetching full certification details")
|
||
ap.add_argument("--proxy", default=None,
|
||
help="Optional proxy override (defaults to environment variables).")
|
||
ap.add_argument("--quiet", "-q", action="store_true",
|
||
help="Suppress progress output on stderr (for scripting/piping).")
|
||
args = ap.parse_args()
|
||
set_quiet(args.quiet)
|
||
|
||
query = " ".join(args.query).strip()
|
||
proxy = args.proxy
|
||
|
||
if args.json:
|
||
results = search(query, proxy)
|
||
print(json.dumps(results, indent=2, ensure_ascii=False))
|
||
elif args.search_only:
|
||
print_search(search(query, proxy))
|
||
else:
|
||
resolve_and_report(query, proxy)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|