initial commit

wowhead_scraper/parsers/common.py

@@ -0,0 +1,142 @@
+"""Extractors for the data blobs Wowhead embeds in its pages.
+
+Almost everything worth having is in inline JSON / JS object literals rather than in the HTML,
+so these helpers find a blob by its marker and hand it to chompjs (which accepts unquoted keys
+and bare identifiers that plain json.loads would reject).
+"""
+import json
+import re
+from typing import Any, Dict, List, Optional, Tuple
+
+import chompjs
+from parsel import Selector
+
+from config import LIST_ROW_CAP
+
+SIDES = {1: "Alliance", 2: "Horde", 3: "Both"}
+
+_GATHERER_TYPES = {1: "npc", 2: "object", 3: "item", 5: "quest"}
+_ENTITY_LINK = re.compile(r"\[url=/\w+/(npc|object|item|quest|zone)=(\d+)[^\]]*\](.*?)\[/url\]")
+_MARKUP_TAG = re.compile(r"\[/?[a-z]+[^\]]*\]")
+_SMALL_WORDS = {"a", "an", "and", "at", "for", "in", "of", "on", "the", "to"}
+
+
+def js_object_after(html: str, marker: str) -> Optional[Any]:
+    """Parse the JS object/array literal that starts right after `marker`."""
+    i = html.find(marker)
+    if i < 0:
+        return None
+    return chompjs.parse_js_object(html[i + len(marker):])
+
+
+def json_script(html: str, script_id: str) -> Optional[Any]:
+    m = re.search(rf'<script type="application/json" id="{re.escape(script_id)}">(.*?)</script>', html, re.S)
+    return json.loads(m.group(1)) if m else None
+
+
+def listview(html: str, listview_id: str) -> Optional[Dict[str, Any]]:
+    """Return a `new Listview({...})` config by id. List pages may use either embedding style."""
+    for m in re.finditer(r"new Listview\(\{", html):
+        head = html[m.end():m.end() + 300]
+        if re.search(rf"\bid:\s*'{re.escape(listview_id)}'", head):
+            return chompjs.parse_js_object(html[m.end() - 1:])
+    for lv in json_script(html, "data.page.listPage.listviews") or []:
+        if lv.get("id") == listview_id:
+            return lv
+    return None
+
+
+def filter_categories(html: str) -> Dict[int, Tuple[str, Dict[int, str]]]:
+    """Category tree from the list page's Filter.init config: {id: (name, {sub_id: sub_name})}.
+
+    Top-level and sub-category ids overlap (Kalimdor and Dun Morogh are both 1), hence the nesting.
+    Names come from url slugs, so "elwynn-forest" becomes "Elwynn Forest".
+    """
+    config = js_object_after(html, "Filter.init(") or {}
+
+    def name(c):
+        words = c["url"].split("-")
+        return " ".join(w if i and w in _SMALL_WORDS else w.capitalize() for i, w in enumerate(words))
+
+    return {int(c["id"]): (name(c), {int(s["id"]): name(s) for s in c.get("categories") or []})
+            for c in config.get("categories") or []}
+
+
+def list_page(html: str, listview_id: str) -> Tuple[List[Dict[str, Any]], bool, Optional[int]]:
+    """Rows of a list page, whether Wowhead truncated them at its row cap, and its total match count."""
+    lv = listview(html, listview_id) or {}
+    rows = lv.get("data") or []
+    truncated = bool(lv.get("_truncated")) or len(rows) >= LIST_ROW_CAP
+    # "5,442 quests found (1,000 displayed)" only appears when truncated
+    m = re.match(r"([\d,]+) \w+ found", lv.get("note") or "")
+    total = lv.get("totalCount") or (int(m.group(1).replace(",", "")) if m else None)
+    return rows, truncated, total if truncated else len(rows)
+
+
+def entity_blob(html: str, kind: str, entity_id: int) -> Optional[Dict[str, Any]]:
+    """The `$.extend(g_quests[123], {...})` record a detail page carries for itself."""
+    return js_object_after(html, f"$.extend(g_{kind}s[{entity_id}], ")
+
+
+def gatherer(html: str) -> Dict[str, Dict[int, str]]:
+    """Names of every entity the page references, keyed by type then id."""
+    names: Dict[str, Dict[int, str]] = {}
+    for m in re.finditer(r"WH\.Gatherer\.addData\((\d+), \d+, ", html):
+        kind = _GATHERER_TYPES.get(int(m.group(1)))
+        if kind:
+            for eid, data in chompjs.parse_js_object(html[m.end():]).items():
+                names.setdefault(kind, {})[int(eid)] = data.get("name_enus")
+    return names
+
+
+def quick_facts(html: str) -> List[str]:
+    """The "Quick Facts" infobox as a list of raw markup lines ([li] items)."""
+    m = re.search(r'WH\.markup\.printHtml\(("(?:[^"\\]|\\.)*"), "infobox-contents-0"', html)
+    if not m:
+        return []
+    markup = json.loads(m.group(1))
+    return re.findall(r"\[li\](.*?)\[/li\]", markup)
+
+
+def markup_text(markup: str) -> str:
+    return " ".join(_MARKUP_TAG.sub("", markup).replace("&nbsp;", " ").split())
+
+
+def markup_links(markup: str) -> List[Dict[str, Any]]:
+    return [{"type": t, "id": int(i), "name": markup_text(n)} for t, i, n in _ENTITY_LINK.findall(markup)]
+
+
+def mapper(html: str) -> Optional[Dict[str, Any]]:
+    return js_object_after(html, "new Mapper(")
+
+
+def selector(html: str) -> Selector:
+    # Keep <br> line breaks in extracted text.
+    return Selector(text=re.sub(r"<br\s*/?>", "\n", html))
+
+
+def text_after(sel: Selector, xpath: str, stop_tags=("h2",)) -> Optional[str]:
+    """Text following the first node matching `xpath`, up to the first sibling in `stop_tags`."""
+    found = sel.xpath(xpath)
+    if not found:
+        return None
+    node = found[0].root
+    parts = [node.tail or ""]
+    for sibling in node.itersiblings():
+        if sibling.tag in stop_tags:
+            break
+        if sibling.tag not in ("script", "table"):
+            parts.append("".join(sibling.itertext()))
+        parts.append(sibling.tail or "")
+    return clean_text("".join(parts))
+
+
+def section_text(sel: Selector, heading: str) -> Optional[str]:
+    """Text of an <h2> section such as Description, Progress or Completion."""
+    return text_after(sel, f"//h2[normalize-space(.)='{heading}']")
+
+
+def clean_text(text: str) -> Optional[str]:
+    lines = [" ".join(line.split()) for line in text.splitlines()]
+    text = "\n".join(lines).strip()
+    return re.sub(r"\n{3,}", "\n\n", text) or None

wowhead_scraper/parsers/object.py

@@ -0,0 +1,68 @@
+from typing import Any, Dict, List
+
+from parsers.common import SIDES, entity_blob, js_object_after, listview, markup_text, quick_facts
+
+LISTVIEW_ID = "objects"
+CSV_FIELDS = ["id", "name", "type", "type_id", "skill", "zone_ids", "url"]
+
+
+def list_row(raw: Dict[str, Any], categories: Dict[int, tuple]) -> Dict[str, Any]:
+    type_name, _ = categories.get(raw.get("type"), (None, {}))
+    return {
+        "id": raw["id"],
+        "name": raw.get("name"),
+        "type": type_name,
+        "type_id": raw.get("type"),
+        "skill": raw.get("skill"),
+        # -1 entries are placeholders for "unknown location"
+        "zone_ids": ";".join(str(z) for z in raw.get("location", []) if z >= 0),
+    }
+
+
+def parse_page(html: str, row: Dict[str, Any]) -> Dict[str, Any]:
+    object_id = int(row["id"])
+    blob = entity_blob(html, "object", object_id) or {}
+    return {
+        "id": object_id,
+        "name": blob.get("name", row.get("name")),
+        "type": row.get("type"),
+        "type_id": blob.get("type"),
+        "skill": int(row["skill"]) if row.get("skill") not in (None, "") else None,
+        "locations": _locations(js_object_after(html, "var g_mapperData = ")),
+        "starts": _quests(html, "starts"),
+        "ends": _quests(html, "ends"),
+        "contains": _loot(html),
+        "quick_facts": [markup_text(f) for f in quick_facts(html)],
+        "url": row.get("url"),
+    }
+
+
+def _locations(mapper_data) -> List[Dict[str, Any]]:
+    locations = []
+    for zone_id, floors in (mapper_data or {}).items():
+        for floor in floors:
+            locations.append({
+                "zone_id": int(zone_id),
+                "zone": floor.get("uiMapName"),
+                "ui_map_id": floor.get("uiMapId"),
+                "count": floor.get("count"),
+                "coords": floor.get("coords"),
+            })
+    return locations
+
+
+def _quests(html: str, listview_id: str) -> List[Dict[str, Any]]:
+    lv = listview(html, listview_id) or {}
+    return [{"id": q["id"], "name": q.get("name"), "level": q.get("level"),
+             "reqlevel": q.get("reqlevel"), "side": SIDES.get(q.get("side"))}
+            for q in lv.get("data") or []]
+
+
+def _loot(html: str) -> List[Dict[str, Any]]:
+    lv = listview(html, "contains") or {}
+    opened = lv.get("_totalCount")
+    return [{"id": i["id"], "name": i.get("name"), "count": i.get("count"),
+             # same figure Wowhead shows in its "%" column
+             "percent": round(100 * i["count"] / opened, 2) if opened and i.get("count", -1) > 0 else None,
+             "stack": i.get("stack")}
+            for i in lv.get("data") or []]

wowhead_scraper/parsers/quest.py

@@ -0,0 +1,131 @@
+import re
+from typing import Any, Dict, List
+
+from parsers.common import (SIDES, entity_blob, gatherer, markup_links, markup_text, mapper, quick_facts,
+                            section_text, selector, text_after)
+
+LISTVIEW_ID = "quests"
+CSV_FIELDS = ["id", "name", "level", "reqlevel", "side", "category", "category_id",
+              "category2", "category2_id", "exp", "money", "url"]
+
+_MAPPER_TYPES = {1: "npc", 2: "object", 3: "item"}
+
+
+def list_row(raw: Dict[str, Any], categories: Dict[int, tuple]) -> Dict[str, Any]:
+    top_name, sub_names = categories.get(raw.get("category2"), (None, {}))
+    return {
+        "id": raw["id"],
+        "name": raw.get("name"),
+        "level": raw.get("level"),
+        "reqlevel": raw.get("reqlevel"),
+        "side": SIDES.get(raw.get("side")),
+        "category": sub_names.get(raw.get("category")),
+        "category_id": raw.get("category"),
+        "category2": top_name,
+        "category2_id": raw.get("category2"),
+        "exp": raw.get("xp"),
+        "money": raw.get("money"),
+    }
+
+
+def parse_page(html: str, row: Dict[str, Any]) -> Dict[str, Any]:
+    quest_id = int(row["id"])
+    sel = selector(html)
+    blob = entity_blob(html, "quest", quest_id) or {}
+    names = gatherer(html)
+    facts = quick_facts(html)
+
+    return {
+        "id": quest_id,
+        "name": blob.get("name", row.get("name")),
+        "level": blob.get("level"),
+        "reqlevel": blob.get("reqlevel"),
+        "side": SIDES.get(blob.get("side")),
+        "category": row.get("category"),
+        "category2": row.get("category2"),
+        "objectives": text_after(sel, "//h1[@class='heading-size-1']", stop_tags=("table", "div", "h2", "script")),
+        "objective_items": _objective_items(sel),
+        "description": section_text(sel, "Description"),
+        "progress": section_text(sel, "Progress"),
+        "completion": section_text(sel, "Completion"),
+        "start": _fact_links(facts, "Start:"),
+        "end": _fact_links(facts, "End:"),
+        "locations": _locations(mapper(html)),
+        "series": _series(sel, quest_id),
+        "rewards": {
+            "xp": blob.get("xp"),
+            # the page's record omits money when it is 0
+            "money": blob.get("money", 0),
+            "item_choices": _items(blob.get("itemchoices"), names),
+            "items": _items(blob.get("itemrewards"), names),
+            "reputation": _reputation(sel, blob.get("reprewards")),
+        },
+        "quick_facts": [markup_text(f) for f in facts],
+        "url": row.get("url"),
+    }
+
+
+def _fact_links(facts: List[str], label: str) -> List[Dict[str, Any]]:
+    for fact in facts:
+        if markup_text(fact).startswith(label):
+            return markup_links(fact)
+    return []
+
+
+def _locations(mapper_config) -> List[Dict[str, Any]]:
+    """Flatten mapper points: who/what is where, and its role (start, end, objective)."""
+    points = []
+    for zone_id, zone in ((mapper_config or {}).get("objectives") or {}).items():
+        for level in zone.get("levels", []):
+            for p in level:
+                points.append({
+                    "role": p.get("point"),
+                    "type": _MAPPER_TYPES.get(p.get("type"), p.get("type")),
+                    "id": p.get("id"),
+                    "name": p.get("name"),
+                    "zone_id": int(zone_id),
+                    "zone": zone.get("zone"),
+                    "coord": p.get("coord"),
+                    "coords": p.get("coords"),
+                })
+    return points
+
+
+def _objective_items(sel) -> List[Dict[str, Any]]:
+    result = []
+    for tr in sel.xpath("(//table[@class='icon-list'])[1]//tr"):
+        href = tr.xpath(".//a/@href").get() or ""
+        m = re.search(r"/(\w+)=(\d+)", href)
+        # A count for items/kills; a standing such as "Friendly" for faction objectives.
+        quantity = tr.attrib.get("data-icon-list-quantity", "1")
+        result.append({
+            "type": m.group(1) if m else None,
+            "id": int(m.group(2)) if m else None,
+            "name": " ".join(tr.xpath(".//a//text()").getall()).strip() or None,
+            "quantity": int(quantity) if quantity.isdigit() else None,
+            "standing": None if quantity.isdigit() else quantity,
+            "provided": "(Provided)" in "".join(tr.xpath(".//td//text()").getall()),
+        })
+    return result
+
+
+def _series(sel, quest_id: int) -> List[Dict[str, Any]]:
+    series = []
+    for td in sel.xpath("//table[@class='series']//tr/td"):
+        href = td.xpath(".//a/@href").get()
+        m = re.search(r"quest=(\d+)", href or "")
+        series.append({"id": int(m.group(1)) if m else quest_id,
+                       "name": " ".join(td.xpath(".//text()").getall()).strip()})
+    return series
+
+
+def _items(pairs, names) -> List[Dict[str, Any]]:
+    return [{"id": i, "name": names.get("item", {}).get(i), "count": c} for i, c in pairs or []]
+
+
+def _reputation(sel, pairs) -> List[Dict[str, Any]]:
+    factions = {}
+    for a in sel.xpath("//a[contains(@href, '/faction=')]"):
+        m = re.search(r"faction=(\d+)", a.attrib["href"])
+        factions[int(m.group(1))] = a.xpath("string()").get().strip()
+    return [{"faction_id": f, "faction": factions.get(f), "value": v} for f, v in pairs or []]