File size: 7,442 Bytes
0584798 ed7ac51 0584798 a8990ec 0584798 ed7ac51 0584798 ed7ac51 0584798 ed7ac51 0584798 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 | from __future__ import annotations
import csv
import json
import os
from dataclasses import dataclass
from functools import lru_cache
from pathlib import Path
try:
from .config import IAB_TAXONOMY_GRAPH_PATH, IAB_TAXONOMY_PATH, IAB_TAXONOMY_VERSION # type: ignore
except ImportError:
from config import IAB_TAXONOMY_GRAPH_PATH, IAB_TAXONOMY_PATH, IAB_TAXONOMY_VERSION
_DEFAULT_MODEL_REPO_ID = "admesh/agentic-intent-classifier"
@dataclass(frozen=True)
class IabNode:
unique_id: str
parent_id: str | None
label: str
path: tuple[str, ...]
@property
def level(self) -> int:
return len(self.path)
@property
def path_label(self) -> str:
return path_to_label(self.path)
class IabTaxonomy:
def __init__(self, nodes: list[IabNode]):
self.nodes = nodes
self._path_index = {node.path: node for node in nodes}
self._children_index: dict[tuple[str, ...], list[IabNode]] = {}
self._level_index: dict[int, list[IabNode]] = {}
for node in nodes:
self._children_index.setdefault(node.path[:-1], []).append(node)
self._level_index.setdefault(node.level, []).append(node)
for children in self._children_index.values():
children.sort(key=lambda item: item.path)
for level_nodes in self._level_index.values():
level_nodes.sort(key=lambda item: item.path)
def get_node(self, path: tuple[str, ...]) -> IabNode:
if path not in self._path_index:
raise KeyError(f"Unknown IAB path: {path}")
return self._path_index[path]
def build_level(self, path: tuple[str, ...]) -> dict:
node = self.get_node(path)
return {"id": node.unique_id, "label": node.label}
def has_path(self, path: tuple[str, ...]) -> bool:
return path in self._path_index
def immediate_children(self, prefix: tuple[str, ...]) -> list[IabNode]:
return list(self._children_index.get(prefix, []))
def siblings(self, path: tuple[str, ...]) -> list[IabNode]:
node = self.get_node(path)
return [candidate for candidate in self._children_index.get(path[:-1], []) if candidate.path != node.path]
def level_nodes(self, level: int) -> list[IabNode]:
return list(self._level_index.get(level, []))
def to_training_graph(self) -> dict:
nodes = []
for node in self.nodes:
child_nodes = self.immediate_children(node.path)
sibling_nodes = self.siblings(node.path)
nodes.append(
{
"node_id": node.unique_id,
"parent_id": node.parent_id,
"level": node.level,
"label": node.label,
"path": list(node.path),
"path_label": node.path_label,
"child_ids": [child.unique_id for child in child_nodes],
"child_paths": [child.path_label for child in child_nodes],
"sibling_ids": [sibling.unique_id for sibling in sibling_nodes],
"sibling_paths": [sibling.path_label for sibling in sibling_nodes],
"canonical_surface_name": node.label,
}
)
return {
"taxonomy": "IAB Content Taxonomy",
"taxonomy_version": IAB_TAXONOMY_VERSION,
"node_count": len(nodes),
"level_counts": {
f"tier{level}": len(self.level_nodes(level))
for level in range(1, 5)
},
"nodes": nodes,
}
def build_content_object(self, path: tuple[str, ...], mapping_mode: str, mapping_confidence: float) -> dict:
if not path:
raise ValueError("IAB path must not be empty")
payload = {
"taxonomy": "IAB Content Taxonomy",
"taxonomy_version": IAB_TAXONOMY_VERSION,
"tier1": self.build_level(path[:1]),
"mapping_mode": mapping_mode,
"mapping_confidence": round(float(mapping_confidence), 4),
}
if len(path) >= 2:
payload["tier2"] = self.build_level(path[:2])
if len(path) >= 3:
payload["tier3"] = self.build_level(path[:3])
if len(path) >= 4:
payload["tier4"] = self.build_level(path[:4])
return payload
def build_content_object_from_label(
self,
path_label: str,
mapping_mode: str,
mapping_confidence: float,
) -> dict:
return self.build_content_object(
path=parse_path_label(path_label),
mapping_mode=mapping_mode,
mapping_confidence=mapping_confidence,
)
def parse_path_label(path_label: str) -> tuple[str, ...]:
path = tuple(part.strip() for part in path_label.split(">") if part.strip())
if not path:
raise ValueError("IAB path label must not be empty")
return path
def path_to_label(path: tuple[str, ...]) -> str:
if not path:
raise ValueError("IAB path must not be empty")
return " > ".join(path)
def _load_rows(path: Path) -> list[dict]:
with path.open("r", encoding="utf-8") as handle:
reader = csv.reader(handle, delimiter="\t")
rows = list(reader)
header = rows[1]
data_rows = rows[2:]
parsed = []
for row in data_rows:
padded = row + [""] * (len(header) - len(row))
parsed.append(dict(zip(header, padded)))
return parsed
def _resolve_taxonomy_path() -> Path:
"""Resolve taxonomy TSV path for local and HF trust_remote_code environments."""
if IAB_TAXONOMY_PATH.exists():
return IAB_TAXONOMY_PATH
# HF dynamic modules often do not contain non-Python data files.
# Fetch the taxonomy TSV directly from the model repo as a fallback.
repo_id = os.environ.get("ADMESH_MODEL_REPO_ID", _DEFAULT_MODEL_REPO_ID).strip() or _DEFAULT_MODEL_REPO_ID
revision = os.environ.get("ADMESH_MODEL_REVISION", "").strip() or None
filename = f"data/iab-content/Content Taxonomy {IAB_TAXONOMY_VERSION}.tsv"
try:
from huggingface_hub import hf_hub_download
except ModuleNotFoundError as exc:
raise FileNotFoundError(
f"Taxonomy TSV missing at {IAB_TAXONOMY_PATH}; install huggingface_hub or provide local taxonomy file."
) from exc
downloaded = hf_hub_download(
repo_id=repo_id,
repo_type="model",
filename=filename,
revision=revision,
)
return Path(downloaded)
@lru_cache(maxsize=1)
def get_iab_taxonomy() -> IabTaxonomy:
nodes = []
for row in _load_rows(_resolve_taxonomy_path()):
path = tuple(
value.strip()
for key in ("Tier 1", "Tier 2", "Tier 3", "Tier 4")
if (value := row.get(key, "").strip())
)
if not path:
continue
nodes.append(
IabNode(
unique_id=row["Unique ID"].strip(),
parent_id=row["Parent"].strip() or None,
label=row["Name"].strip(),
path=path,
)
)
return IabTaxonomy(nodes)
def write_training_graph(path: Path = IAB_TAXONOMY_GRAPH_PATH) -> Path:
taxonomy = get_iab_taxonomy()
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(taxonomy.to_training_graph(), indent=2, sort_keys=True) + "\n", encoding="utf-8")
return path
|