"""Shared parts of the tracking plan audit scripts: constants, helpers, file readers, Plan and Inventory.

Imported by audit_plan.py, plan_quality.py and the drift modules; never run on its own.
Standard library only, Python 3.8 or later.
"""
import csv
import os
import re
import sys
from collections import OrderedDict, defaultdict
from datetime import date

# ---------- constants ----------

HERE = os.path.dirname(os.path.realpath(__file__))  # realpath, so a symlinked install resolves to its target
SKILL_DIR = os.path.dirname(HERE)
ASSETS = os.path.join(SKILL_DIR, "assets")

TOOL_COLUMNS = OrderedDict([
    ("ga4", "ga4_name"), ("mixpanel", "mixpanel_name"), ("amplitude", "amplitude_name"),
    ("posthog", "posthog_name"), ("segment", "segment_name"), ("rudderstack", "rudderstack_name"),
])
ROUTE_TOOLS = {"segment", "rudderstack"}  # usually a route in delivery_paths, not only a destination
NOT_MAPPED = {"", "none", "not sent", "not_sent", "n/a", "na", "-", "no", "false"}
STATUSES = ("proposed", "approved", "implemented", "verified", "deprecated", "retired")
TYPES = ("string", "integer", "number", "boolean", "timestamp", "array", "object")
TYPE_ALIASES = {
    "str": "string", "text": "string", "varchar": "string", "enum": "string",
    "int": "integer", "int64": "integer", "long": "integer", "bigint": "integer",
    "float": "number", "float64": "number", "double": "number", "decimal": "number", "numeric": "number",
    "bool": "boolean", "datetime": "timestamp", "date": "timestamp",
    "list": "array", "dict": "object", "map": "object", "json": "object",
}
COMPATIBLE = {
    "string": {"string", "numeric_string", "timestamp"}, "integer": {"integer"},
    "number": {"integer", "number"}, "boolean": {"boolean"}, "timestamp": {"timestamp"},
    "array": {"array"}, "object": {"object"},
}
FAMILY = {"string": "text", "numeric_string": "text", "timestamp": "text", "integer": "numeric",
          "number": "numeric", "boolean": "boolean", "array": "array", "object": "object"}
SOURCE_ALIASES = {
    "browser": "web", "javascript": "web", "js": "web", "analytics.js": "web", "analytics-next": "web",
    "posthog-js": "web", "mixpanel-browser": "web", "amplitude-ts": "web", "gtag": "web", "gtm": "web",
    "iphone": "ios", "ipad": "ios", "swift": "ios", "analytics-ios": "ios", "analytics-swift": "ios",
    "posthog-ios": "ios", "analytics-android": "android", "analytics-kotlin": "android",
    "posthog-android": "android", "backend": "server", "api": "server", "http": "server",
    "http-api": "server", "node": "server", "node.js": "server", "python": "server", "ruby": "server",
    "go": "server", "java": "server", "php": "server", "analytics-node": "server",
    "analytics-python": "server", "analytics-ruby": "server", "analytics-go": "server",
    "analytics-java": "server", "posthog-node": "server", "posthog-python": "server",
    "billing_webhook": "server", "crm": "server", "measurement_protocol": "server",
}
OBS_SYNONYMS = {
    "row_count": "rows", "occurrences": "rows", "count": "rows", "n": "rows",
    "event_name": "event", "event_type": "event", "key": "property", "property_name": "property",
    "param": "property", "parameter": "property", "type": "value_type", "data_type": "value_type",
    "platform": "source", "lib": "source", "library": "source",
}
DEFINITION_SYNONYMS = {"parametername": "parameter_name", "parameter": "parameter_name",
                       "event_parameter": "parameter_name", "type": "kind"}
SENTINEL_NOTE = "sentinel and empty strings count as missing"
PII_EXACT = {"email", "e_mail", "email_address", "phone", "phone_number", "mobile_number", "first_name",
             "last_name", "full_name", "firstname", "lastname", "fullname", "address", "street_address",
             "home_address", "ip", "ip_address", "password", "passwd", "date_of_birth", "dob",
             "birthdate", "ssn", "credit_card", "card_number"}
PII_LAST_TOKEN = {"email", "phone", "password", "ssn", "address"}
MONEY_TOKENS = {"revenue", "price", "amount", "tax", "shipping", "discount", "cost", "mrr", "arr"}
MONEY_EXACT = {"value", "total", "subtotal", "order_total", "cart_total"}
QUANTITY_TOKENS = {"duration", "size", "weight", "distance", "bytes", "seconds", "ms", "quantity", "qty"}
TEST_TOKENS = {"test", "tmp", "temp", "debug", "qa", "staging"}
# Identifiers whose null share D14 checks per event when the plan names them without a null rule.
ID_RULE_KEYS = {"user_id", "account_id", "workspace_id", "organization_id", "org_id", "company_id", "group_id"}
NULL_RULE_RE = re.compile(r"null|optional|not applicable|n/a|absent|missing|empty|\b(when|if|unless|before|after)\b")
# Event parameters that fill GA4 predefined dimensions or metrics, so they need no custom definition
# (Data API schema and ecommerce guide, checked 2026-11-02; refresh from the live pages).
GA4_BUILTIN_PARAMS = {
    "achievement_id", "character", "content_group", "content_id", "content_type", "coupon", "currency",
    "file_extension", "file_name", "group_id", "items", "level", "link_classes", "link_domain", "link_id",
    "link_text", "link_url", "method", "outbound", "page_location", "page_referrer", "page_title",
    "percent_scrolled", "search_term", "shipping", "shipping_tier", "tax", "transaction_id", "value",
    "video_provider", "video_title", "video_url", "virtual_currency_name", "visible",
}
RULE_NAMES = OrderedDict([
    ("D01", "planned-unseen"), ("D02", "unplanned-event"), ("D03", "name-variant"),
    ("D04", "dynamic-names"), ("D05", "lifecycle"), ("D06", "source-mismatch"),
    ("D07", "required-incomplete"), ("D08", "optional-unseen"), ("D09", "unplanned-property"),
    ("D10", "type-mismatch"), ("D11", "enum-drift"), ("D12", "volume-and-duplicates"),
    ("D13", "pii"), ("D14", "limits-and-identifiers"), ("D15", "ga4-registration"),
])
# Print order: privacy, double counting and wrong values, missing data, lifecycle, hygiene.
PRINT_ORDER = ["D13", "D12", "D10", "D11", "D14", "D06", "D03", "D07", "D01", "D08", "D15", "D05", "D04",
               "D02", "D09"]
SEVERITY_ORDER = {"FAIL": 0, "WARN": 1, "INFO": 2, "PASS": 3, "NOT CHECKED": 4}
CLASSES = ("planned", "variant", "vendor", "generated", "dynamic", "test", "orphan")
LATE_DAYS = 3  # GA4 keeps filling daily tables for up to 3 days after the table date
SETTLED_DAYS = LATE_DAYS + 1  # so a table is final on the fourth day after its date (answers 7029846, 9358801)
UUID_RE = re.compile(r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}")
HEX_RE = re.compile(r"(?<![0-9A-Za-z])(?=[0-9a-fA-F]*[0-9])(?=[0-9a-fA-F]*[a-fA-F])[0-9a-fA-F]{8,}(?![0-9A-Za-z])")
DIGITS_RE = re.compile(r"\d{2,}")
EMAIL_RE = re.compile(r"[^@\s]+@[^@\s]+\.[A-Za-z]{2,}")
PHONE_RE = re.compile(r"^(\+\d[\d\s().-]{8,18}\d|\(?\d{3}\)?[\s.-]\d{3}[\s.-]\d{4})$")
PHONE_IN_TEXT_RE = re.compile(r"(?<![\w+])\+\d[\d\s().-]{8,18}\d(?!\d)")
URL_QUERY_RE = re.compile(r"[?&](email|token|password|code|reset|key)=", re.I)
GENERATED_RE = re.compile(r"^(Viewed|Loaded) .+ (Page|Screen)$")
VERSION_RE = re.compile(r"^v?\d+(\.\d+)+([-+ (].*)?$")
# GA4 event naming rule (event naming rules help page, checked 2026-11-06): a letter first, then only
# letters, numbers and underscores; non-English letters count as letters. Reserved names and prefixes come
# from the @ga4-reserved sections of vendor-events.txt (event names and prefixes) and vendor-params.txt
# (parameter prefixes), so they can be refreshed.
GA4_EVENT_NAME_RE = re.compile(r"^[^\W\d_]\w*$")
RESERVED_SCOPE = "ga4-reserved"
# Reference count routes (D12). A server route loses nothing to consent or blockers; a browser route
# (a web tag or an app SDK) does, so it is judged against the band its contract documents.
ROUTE_ALIASES = {
    "server": "server", "backend": "server", "api": "server", "http": "server", "http-api": "server",
    "measurement_protocol": "server", "measurement protocol": "server", "server-side": "server",
    "server_side": "server", "sgtm": "server",
    "browser": "browser", "web": "browser", "client": "browser", "client-side": "browser",
    "client_side": "browser", "gtm": "browser", "gtag": "browser", "ios": "browser", "android": "browser",
}
REDACTION_TOKENS = {"<email>", "<phone>", "<url_query>", "<long_text>", "<other>", "<rare>"}
TAIL_TOKENS = ("<other>", "<rare>")
TRANSPORT_NOTES = {
    "mixpanel": ("Mixpanel deduplicates at query time on event, distinct_id, time and $insert_id, so its "
                 "reports drop them, while raw exports and warehouse copies keep them"),
    "amplitude": ("Amplitude ignores a repeat of the same device_id and insert_id within 7 days at "
                  "ingestion, so pairs kept in its export arrived more than 7 days apart"),
}
DEFAULT_TRANSPORT_NOTE = ("check whether this destination deduplicates on that key; raw exports and "
                          "warehouse copies keep both")
# Fallback for library names that SOURCE_ALIASES does not list, tried in this order.
SOURCE_GUESSES = [
    ("web", re.compile(r"javascript|browser|(^|[^a-z])(js|web)([^a-z]|$)")),
    ("android", re.compile(r"android|kotlin")),
    ("ios", re.compile(r"(^|[^a-z])(ios|iphone|ipad|swift)([^a-z]|$)")),
    ("server", re.compile(r"(^|[^a-z])(node|python|ruby|go|java|php|dotnet|net|csharp|elixir|rust|server)"
                          r"([^a-z]|$)")),
]


def fail(message):
    print(f"error: {message}", file=sys.stderr)
    sys.exit(2)


# ---------- helpers ----------

def num(n):
    return f"{n:,}"


def pct(share):
    value = share * 100
    if 0 < value < 0.01:
        return "<0.01%"
    text = f"{value:.2f}" if value < 10 else f"{value:.1f}"
    if value < 100 and float(text) >= 100:
        text = "99.9"
    return text + "%"


def plural(n, word):
    return f"{num(n)} {word}" + ("" if n == 1 else "s")


def where(base, plan_name=None, source=None):
    """Label such as sign_up.method (plan signup_method, source web)."""
    bits = []
    if plan_name:
        bits.append(f"plan {plan_name}")
    if source is not None:
        bits.append(f"source {source or 'all'}")
    return f"{base} ({', '.join(bits)})" if bits else base


def first_items(items, limit=8):
    items = list(items)
    if len(items) <= limit:
        return ", ".join(items)
    return ", ".join(items[:limit]) + f" and {len(items) - limit} more"


def src_list(sources):
    return ", ".join(sorted(s or "no source label" for s in sources))


def join_and(items):
    items = [str(x) for x in items]
    return items[0] if len(items) == 1 else ", ".join(items[:-1]) + " and " + items[-1]


def lines_text(lines):
    return ("line " if len(lines) == 1 else "lines ") + join_and(lines)


def norm_key(name):
    """Lowercase, split camelCase, turn spaces, hyphens, dots, colons and slashes into underscores."""
    text = re.sub(r"([a-z0-9])([A-Z])", r"\1_\2", (name or "").strip())
    text = re.sub(r"[\s\-./:\\]+", "_", text.lower())
    return re.sub(r"_+", "_", text).strip("_")


def mask(name):
    text = UUID_RE.sub("{uuid}", name)
    text = HEX_RE.sub("{hex}", text)
    return DIGITS_RE.sub("{n}", text)


def masked_example(name):
    """Keep the last two digits of each digit run so examples differ without printing whole IDs."""
    return re.sub(r"\d{3,}", lambda m: "*" * (len(m.group(0)) - 2) + m.group(0)[-2:], UUID_RE.sub("{uuid}", name))


def ratio_text(ratio):
    return f"{ratio:.3f}" if 0.99 <= ratio <= 1.01 and ratio != 1 else f"{ratio:.2f}"


def norm_type(text):
    t = (text or "").strip().lower()
    t = TYPE_ALIASES.get(t, t)
    return t if t in TYPES else None


def norm_source(text):
    """Map a platform or library name to web, ios, android or server; unknown names pass through."""
    parts = (text or "").strip().lower().split("/")
    s = (parts[1] if parts[0].startswith("@") and len(parts) > 1 else parts[0]).strip()  # "@scope/lib" or "lib/1.2"
    if s in SOURCE_ALIASES:
        return SOURCE_ALIASES[s]
    for platform, pattern in SOURCE_GUESSES:
        if pattern.search(s):
            return platform
    return s


def plan_sources(text):
    out = set()
    for part in re.split(r"[;,|]", text or ""):
        s = part.strip().lower()
        if not s:
            continue
        if s == "mobile_app":
            out.update({"ios", "android"})
        else:
            out.add(norm_source(s))
    return out


def plan_scope(text):
    t = (text or "").strip().lower()
    if t in ("user", "person", "profile", "user property", "user_property"):
        return "USER"
    if t.startswith("item"):
        return "ITEM"
    return "EVENT"


def parse_bool(text):
    t = (text or "").strip().lower()
    if t in ("true", "yes", "y", "required", "1"):
        return True
    if t in ("false", "no", "n", "optional", "0"):
        return False
    return None


def parse_date(text):
    t = (text or "").strip()
    m = re.match(r"(\d{4})-?(\d{2})-?(\d{2})", t)
    if not m:
        return None
    try:
        return date(int(m.group(1)), int(m.group(2)), int(m.group(3)))
    except ValueError:
        return None


def is_identifier(name):
    n = norm_key(name)
    return n == "id" or n.endswith("_id") or n in {"distinct_id", "anonymous_id", "device_id"}


def is_money(name, unit=""):
    n = norm_key(name)
    u = (unit or "").lower()
    if re.search(r"currency|cents|minor unit|major unit|\b[a-z]{3}\b minor", u) or re.fullmatch(r"[A-Z]{3}", unit or ""):
        return True
    tokens = set(n.split("_"))
    return n in MONEY_EXACT or bool(tokens & MONEY_TOKENS)


def is_quantity(name):
    return bool(set(norm_key(name).split("_")) & QUANTITY_TOKENS)


def is_pii_name(name):
    n = norm_key(name)
    if n in PII_EXACT:
        return True
    return n.split("_")[-1] in PII_LAST_TOKEN


def is_template_text(text):
    t = (text or "").strip()
    return ((not t) or bool(re.fullmatch(r"(?i)(tbd|todo|tba|\?+|-+|x{3,}|n/?a)", t))
            or bool(re.search(r"<[^<>]{1,40}>", t)))


def list_items(text):
    items = []
    for part in re.split(r"[;\n]", text or ""):
        part = re.sub(r"^\s*\d+\s*[.)]?\s*", "", part).strip()
        if part:
            items.append(part)
    return items


def t2_count(text):
    numbers = set()
    for a, b in re.findall(r"T2\.(\d+)\s*-\s*T2\.(\d+)", text or ""):
        numbers.update(range(int(a), int(b) + 1))
    for a in re.findall(r"T2\.(\d+)", text or ""):
        numbers.add(int(a))
    return len(numbers) if numbers else None


def id_rule(ids_text, key):
    """'required', 'nullable' or None (not named) for one identifier in the plan's identifiers cell."""
    for item in re.split(r"[;\n]", ids_text or ""):
        text = item.strip().lower()
        if re.search(r"(?<![a-z_])" + re.escape(key) + r"(?![a-z_])", text):
            return "nullable" if NULL_RULE_RE.search(text) else "required"
    return None


def redact(value):
    if EMAIL_RE.search(value):
        return "<email>"
    if PHONE_RE.match(value) or PHONE_IN_TEXT_RE.search(value):
        return "<phone>"
    if URL_QUERY_RE.search(value):
        return "<url_query>"
    if len(value) > 64:
        return "<long_text>"
    return value


def norm_route(text):
    """'server' or 'browser' for a reference row's route, or None when the value is neither."""
    return ROUTE_ALIASES.get((text or "").strip().lower())


def reserved_lists(vendor):
    """The @ga4-reserved names and prefixes of a vendor list, as loaded by load_vendor."""
    names, prefixes = vendor
    return set(names.get(RESERVED_SCOPE, ())), list(prefixes.get(RESERVED_SCOPE, ()))


def ga4_event_name_problem(name, reserved):
    """Why GA4 cannot take a planned event name (the 40-character limit is checked apart), or None."""
    if not GA4_EVENT_NAME_RE.match(name):
        return "breaks GA4's event naming rule (a letter first, then only letters, numbers and underscores)"
    names, prefixes = reserved
    if name in names:
        return "is an event name GA4 reserves"
    hit = next((p for p in prefixes if name.startswith(p)), None)
    return f"starts with {hit}, a prefix GA4 reserves for event names" if hit else None


def ga4_param_name_problem(name, reserved):
    """Prefixes only: the exact names Google lists (currency, user_id) are valid event parameters that
    just cannot become custom definitions, so they are no plan defect."""
    hit = next((p for p in reserved[1] if name.startswith(p)), None)
    return f"starts with {hit}, a prefix GA4 reserves for parameter names" if hit else None


# ---------- reading files and the plan ----------

def read_csv(path, required, label, synonyms=None):
    try:
        with open(path, newline="", encoding="utf-8-sig") as handle:
            reader = csv.reader(handle)
            header = next(reader, None)
            if header is None:
                fail(f"{path} is empty ({label})")
            keys = [h.strip().lower() for h in header]
            if synonyms:
                present = set(keys)
                keys = [synonyms[k] if k in synonyms and synonyms[k] not in present else k for k in keys]
            rows = []
            for number, record in enumerate(reader, start=2):
                if not any(cell.strip() for cell in record):
                    continue
                row = {k: (record[i].strip() if i < len(record) else "") for i, k in enumerate(keys)}
                row["_line"] = number
                rows.append(row)
    except OSError as exc:
        fail(f"cannot read {path}: {exc.strerror}")
    except csv.Error as exc:
        fail(f"{path} is not a readable CSV: {exc}")
    missing = [c for c in required if c not in keys]
    if missing:
        fail(f"{path} is missing required column {', '.join(missing)} ({label})")
    return set(keys), rows


def to_int(text, path, row, column):
    t = (text or "").replace(",", "").replace("_", "").replace(" ", "")
    if t == "":
        return None
    try:
        return int(float(t))
    except ValueError:
        fail(f"{path} line {row['_line']}: column {column} should be a number, got '{text}'")


def load_vendor(path):
    names, prefixes, scope = defaultdict(set), defaultdict(list), "all"
    try:
        with open(path, encoding="utf-8") as handle:
            for raw in handle:
                line = raw.strip()
                if not line or line.startswith("#"):
                    continue
                if line.startswith("@"):
                    scope = line[1:].strip().lower() or "all"
                elif line.startswith("prefix:"):
                    prefixes[scope].append(line[len("prefix:"):])
                else:
                    names[scope].add(line)
    except OSError as exc:
        fail(f"cannot read vendor list {path}: {exc.strerror}")
    return names, prefixes


def is_vendor(name, tool, vendor):
    names, prefixes = vendor
    for scope in ("all", tool):
        if name in names.get(scope, ()):
            return True
        if any(name.startswith(p) for p in prefixes.get(scope, ())):
            return True
    return False


class Plan(object):
    def __init__(self, events_path, props_path):
        self.event_columns, rows = read_csv(events_path, ["event"], "plan events")
        self.prop_columns, prop_rows = read_csv(props_path, ["event", "property"], "plan properties")
        self.events = OrderedDict()
        # Every line that defines each event or (event, property); the first row is the definition and
        # Q07 reports the others, because a repeated row would otherwise replace it without a word.
        self.event_lines, self.prop_lines = defaultdict(list), defaultdict(list)
        for row in rows:
            name = row["event"]
            if not name:
                continue
            self.event_lines[name].append(row["_line"])
            if name in self.events:
                continue
            sources, overrides, derived = set(), defaultdict(set), []
            text = row.get("sources", "")
            if not text and row.get("source_system"):
                text = row["source_system"]
            for part in re.split(r"[;,]", text):
                part = part.strip()
                if not part:
                    continue
                if ":" in part:
                    tool, src = part.split(":", 1)
                    overrides[tool.strip().lower()].update(plan_sources(src))
                else:
                    sources.update(plan_sources(part))
            # A browser companion recorded only in delivery_paths (tracking-plan-builder sheets) is a web source.
            for part in row.get("delivery_paths", "").split(";"):
                path = part.strip().lower()
                tool = path.split(" via ")[0].split(" ")[0] if path else ""
                if tool not in TOOL_COLUMNS or tool in overrides:
                    continue
                browser = "companion" in path or " via gtm" in path or " via gtag" in path
                if browser and not re.search(r"server|sgtm|measurement.protocol", path):
                    overrides[tool].add("web")
                    derived.append(tool)
            tier_match = re.search(r"\d", row.get("risk_tier", ""))
            self.events[name] = {
                "name": name, "row": row, "status": row.get("status", "").lower(),
                "tier": int(tier_match.group(0)) if tier_match else None,
                "sources": sources, "overrides": overrides, "derived": derived,
                "introduced_at": parse_date(row.get("introduced_at", "")),
            }
        self.props = []
        for row in prop_rows:
            if not row["property"]:
                continue
            key = (row["event"] or "*", row["property"])
            self.prop_lines[key].append(row["_line"])
            if len(self.prop_lines[key]) > 1:
                continue
            allowed_text = row.get("allowed_values_or_format", row.get("allowed_values", ""))
            allowed = [v.strip() for v in allowed_text.split("|") if v.strip()] if "|" in allowed_text else None
            self.props.append({
                "event": row["event"] or "*", "name": row["property"], "row": row,
                "raw_type": row.get("type", ""), "type": norm_type(row.get("type", "")),
                "required": parse_bool(row.get("required", "")), "allowed": allowed, "format": allowed_text,
                "unit": row.get("unit", ""), "policy": row.get("unknown_value_policy", "").lower(),
                "null_means": row.get("null_means", ""), "description": row.get("description", ""),
                "excluded": {t for t in re.split(r"[;,|\s]+", row.get("excluded_destinations", "").lower()) if t},
                "ga4": row.get("ga4_parameter", ""), "scope": row.get("scope", ""),
            })

    def repeated_events(self):
        """Event names defined on more than one row, with their lines."""
        return OrderedDict((n, lines) for n, lines in self.event_lines.items() if len(lines) > 1)

    def repeated_props(self):
        """(event, property) pairs defined on more than one row, with their lines."""
        return OrderedDict((k, lines) for k, lines in self.prop_lines.items() if len(lines) > 1)

    def event_collisions(self):
        """Distinct event names that become one after norm_key, such as report_generated and Report Generated."""
        groups = defaultdict(list)
        for name in self.events:
            groups[norm_key(name)].append(name)
        return [names for names in groups.values() if len(names) > 1]

    def unknown_prop_events(self):
        """Events named by property rows that are neither * nor in the events tab, with the rows' lines."""
        out = OrderedDict()
        for (event, _), lines in self.prop_lines.items():
            if event != "*" and event not in self.events:
                out.setdefault(event, []).extend(lines)
        return out

    def final_name(self, event, tool):
        column = TOOL_COLUMNS.get(tool)
        if column is None or column not in self.event_columns:
            return event["name"], False
        value = event["row"].get(column, "")
        return (None if value.lower() in NOT_MAPPED else value), True

    def props_for(self, event_name, tool=None):
        """Event-specific and global properties for one event, minus those excluded for the tool."""
        specific = [p for p in self.props if p["event"] == event_name]
        names = {p["name"] for p in specific}
        merged = specific + [p for p in self.props if p["event"] == "*" and p["name"] not in names]
        return [p for p in merged if not tool or tool not in p["excluded"]]

    @staticmethod
    def tool_prop_name(prop, tool):
        return prop["ga4"] if tool == "ga4" and prop["ga4"] else prop["name"]


# ---------- observed inventory ----------

class Inventory(object):
    def __init__(self, paths):
        self.events = defaultdict(lambda: defaultdict(list))
        self.props = defaultdict(lambda: defaultdict(lambda: defaultdict(list)))
        self.identity = defaultdict(list)
        self.files = paths
        first_read = {}  # (tool, raw source label, event, property, type) -> (file, line)
        for path in paths:
            _, rows = read_csv(path, ["tool", "event"], "observed inventory", OBS_SYNONYMS)
            if not rows:
                fail(f"{path} has a header but no rows (observed inventory): a run on it would check nothing")
            for row in rows:
                source = norm_source(row.get("source", ""))
                if VERSION_RE.match(source):
                    fail(f"{path} line {row['_line']}: source '{row.get('source', '')}' looks like an app version. "
                         "A version split is read by hand for the version tail rule and is never passed to "
                         "drift; profile again without --source-key version_name.")
                o = {
                    "tool": row.get("tool", "").lower(), "source": source,
                    "event": row["event"], "property": row.get("property", ""),
                    "type": (row.get("value_type", "") or "").strip().lower(),
                    "rows": to_int(row.get("rows"), path, row, "rows"),
                    "event_rows": to_int(row.get("event_rows"), path, row, "event_rows"),
                    "first": parse_date(row.get("first_seen", "")), "last": parse_date(row.get("last_seen", "")),
                    "values": self.parse_values(row.get("values", "")),
                }
                for column in ("pii_hits", "sentinel_hits", "empty_hits", "duplicate_keys", "transport_duplicates",
                               "max_params"):
                    o[column] = to_int(row.get(column), path, row, column)
                if not o["tool"] or not o["event"]:
                    continue
                # Rows whose raw source labels differ (analytics.js and analytics-next) still sum into one
                # source; the same row read twice would double every count, so it stops the run.
                raw_source = row.get("source", "").strip().lower()
                key = (o["tool"], raw_source, o["event"], o["property"], o["type"])
                if key in first_read:
                    where_first = "{} line {}".format(*first_read[key])
                    what = f"{o['event']}.{o['property']}" if o["property"] else o["event"]
                    fail(f"{path} line {row['_line']} repeats the {o['tool']} row for {what} (source "
                         f"'{raw_source or 'none'}'{', type ' + o['type'] if o['type'] else ''}) first read at "
                         f"{where_first}. Rows given twice are summed and double every count: pass each query "
                         "result once, and never a stacked file together with its parts.")
                first_read[key] = (path, row["_line"])
                if o["type"] == "identifier":
                    self.identity[o["tool"]].append(o)
                elif o["property"]:
                    self.props[o["tool"]][o["event"]][o["property"]].append(o)
                else:
                    self.events[o["tool"]][o["event"]].append(o)

    @staticmethod
    def parse_values(text):
        out = OrderedDict()
        for part in (text or "").split("|"):
            if "=" not in part:
                continue
            value, count = part.rsplit("=", 1)
            value = value.strip()
            if value[:1] == "'" and value[1:2] in ("=", "+", "-", "@"):
                value = value[1:]
            try:
                out[value] = out.get(value, 0) + int(count.replace(",", "").strip())
            except ValueError:
                continue
        return out or None

    def tools(self):
        return sorted(set(self.events) | set(self.props) | set(self.identity))

    def names(self, tool):
        return set(self.events[tool]) | set(n for n in self.props[tool] if n != "*")

    def has_counts(self, tool, name):
        """True when an event-level row carries a rows number (0 counts; an empty cell means names only)."""
        return any(o["rows"] is not None for o in self.events[tool].get(name, []))

    def listed(self, tool, name):
        return bool(self.events[tool].get(name) or self.props[tool].get(name))

    def total(self, tool, names):
        return sum(o["rows"] or 0 for n in names for o in self.events[tool].get(n, []))

    def tool_total(self, tool):
        return sum(o["rows"] or 0 for rows in self.events[tool].values() for o in rows)

    def by_source(self, tool, names):
        out = defaultdict(int)
        for n in names:
            for o in self.events[tool].get(n, []):
                if o["rows"]:
                    out[o["source"]] += o["rows"]
        if not out:  # fall back to event_rows on property rows (Part B only, for example)
            for n in names:
                for rows in self.props[tool].get(n, {}).values():
                    for o in rows:
                        if o["event_rows"]:
                            out[o["source"]] = max(out[o["source"]], o["event_rows"])
        return out

    def last_by_source(self, tool, names):
        out = {}
        for n in names:
            for o in self.events[tool].get(n, []):
                if o["rows"] and o["last"]:
                    out[o["source"]] = max(out.get(o["source"], o["last"]), o["last"])
        return out

    def last_seen(self, tool, names):
        dates = [o["last"] for n in names for o in self.events[tool].get(n, []) if o["last"]]
        return max(dates) if dates else None


def summary_counts(items):
    def count(s):
        return sum(1 for x in items if x["severity"] == s)
    return (f"{count('FAIL')} FAIL, {count('WARN')} WARN, {count('PASS')} PASS, {count('NOT CHECKED')} not checked. "
            "Nothing was changed.")
