"""The audit context (every input, with lookups) and the Finding each rule fills.""" import re from collections import defaultdict from datetime import date, timedelta from .common import DEFAULTS, PRIORITY, RULES, get, kv, row_count, table, todate, up from .inputs import norm_reduced class Ctx: def __init__(s, snap, cfg, ui, bq=None, ads=None, exports=None, tagcov=None, asof=None): s.snap, s.cfg, s.ui, s.bq, s.ads, s.tagcov = snap or {}, cfg or {}, ui or {}, bq, ads, tagcov s.th = dict(DEFAULTS, **(s.cfg.get('thresholds') or {})) s.asof = todate(asof or s.cfg.get('as_of')) or date.today() s.reports = {k: norm_reduced(k, v) for k, v in dict(get(s.snap, 'reports', {}), **(exports or {})).items()} s.prop = get(s.snap, 'property') or {} s.is360 = '360' in up(get(s.prop, 'serviceLevel') or s.cfg.get('service_level')) s.streams = get(s.snap, 'dataStreams') s.web = [x for x in s.streams or [] if 'WEB' in up(get(x, 'type'))] s.app_only = bool(s.streams) and not s.web s.kes = get(s.snap, 'keyEvents') s.ke_names = None if s.kes is None else {get(k, 'eventName') for k in s.kes} s.ads_links, s.bq_links = get(s.snap, 'googleAdsLinks'), get(s.snap, 'bigqueryLinks') s.dec = s.cfg.get('documented_decisions') or {} s.intended = s.cfg.get('key_events_intended') or {} w = get(s.snap, 'window') or {} s.end = todate(w.get('end')) or s.asof - timedelta(days=2) s.start = todate(w.get('start')) or s.end - timedelta(days=27) s.filters = [kv(v) for _, v in s.uirows('filters.') if v.lower() != 'none'] s.filters_supplied = bool(s.uirows('filters.')) s.proposed = [] r04, r03 = s.rows('R04'), s.rows('R03') s.sessions_note = '' if r04: s._sessions = sum(x.get('sessions', 0) for x in r04) elif r03 and any('sessions' in x for x in r03): s._sessions, s.sessions_note = sum(x.get('sessions', 0) for x in r03), 'sessions summed from R03, approximate' else: s._sessions = None def uival(s, key): return (s.ui.get(key) or {}).get('value') or None def uirows(s, prefix): return [(k, r['value']) for k, r in s.ui.items() if k.startswith(prefix) and r.get('value')] def ev(s, key): if not s.uival(key): return '' r = s.ui[key] return f' [intake {key}' + (f', evidence {r["evidence"]}' if r.get('evidence') else '') + ']' def rows(s, rid): return table(s.reports.get(rid)) def agg(s, rid, dim, met): out = defaultdict(float) for r in s.rows(rid) or []: out[str(r.get(dim, ''))] += r.get(met, 0) return out def total(s, rid, met): r = s.rows(rid) return sum(x.get(met, 0) for x in r) if r else None def sessions(s): return s._sessions def trunc(s, rid): """A note when a report returned fewer rows than its rowCount, else ''.""" rows, rc = s.rows(rid), row_count(s.reports.get(rid)) if rows is None or rc is None or len(rows) >= rc: return '' return f'{rid} truncated ({len(rows)} of {rc} rows), shares approximate' def key_events_total(s): for rid in ('R06', 'R04', 'R01'): t = s.total(rid, 'keyEvents') if t: return t return None def events_total(s): for rid in ('R11', 'R03', 'R01'): t = s.total(rid, 'eventCount') if t: return t return None def daily(s, streams=None, lo=None, hi=None): out = defaultdict(float) for x in s.rows('R02') or []: d = todate(x.get('date')) if d and (not lo or d >= lo) and (not hi or d <= hi) and (not streams or str(x.get('streamId')) in streams): out[d] += x.get('eventCount', 0) return out def documented(s, key): return key in s.dec def annotated(s, d): for a in get(s.snap, 'annotations') or []: ad, rng = get(a, 'annotationDate'), get(a, 'annotationDateRange') spans = [(ad, ad)] if ad else [(get(rng, 'startDate'), get(rng, 'endDate'))] if rng else [] for lo, hi in spans: lo = date(int(get(lo, 'year')), int(get(lo, 'month')), int(get(lo, 'day'))) hi = date(int(get(hi, 'year')), int(get(hi, 'month')), int(get(hi, 'day'))) if lo - timedelta(days=s.th['annotation_days']) <= d <= hi + timedelta(days=s.th['annotation_days']): return True return False def changes(s, areas, days=None): out = [] for k, v in s.uirows('change_history.'): c = kv(v) d, text = todate(c.get('date')), c.get('change', '') area = c.get('area') or next((a for a, pat in (('attribution', 'lookback|attribution|model'), ('channel_groups', 'channel'), ('filters', 'filter'), ('key_events', 'key event|counting|conversion'), ('retention', 'retention'), ('consent', 'consent')) if re.search(pat, text, re.I)), 'other') if d and (s.asof - d).days <= (days or s.th['change_window_days']) and area in areas: out.append((k, d, text, area)) return out def propose(s, d, title, desc): s.proposed.append({'date': d.isoformat(), 'title': title[:60], 'description': desc[:150]}) def low(s, denom, minimum, label='low volume'): """The cap label when the window is below a low-volume minimum, else ''.""" return label if denom is not None and denom < s.th[minimum] else '' def low_sessions(s): return s.low(s.sessions(), 'min_sessions', f'low volume ({s.sessions_note})' if s.sessions_note else 'low volume') class Finding: def __init__(s, rid): s.id, (s.slug, s.area) = rid, RULES[rid] s.subs, s.partial, s.loss, s.risk = [], [], [], [] s.state, s.reason, s.ok_text, s.prio = None, '', '', PRIORITY[rid] def add(s, status, text, prio=None, loss=False, risk=False): s.subs.append((status, text)) if prio and prio < s.prio: s.prio = prio if loss and status == 'FAIL': s.loss.append(text) if risk and status == 'FAIL': s.risk.append(text) def fail(s, text, **k): s.add('FAIL', text, **k) def warn(s, text, **k): s.add('WARN', text, **k) def gated(s, text, low, ev='', **k): """FAIL unless a cap applies (low volume, truncated report), then WARN with the cap's label.""" label = low if isinstance(low, str) else 'low volume' s.add('WARN' if low else 'FAIL', text + (f'; capped at WARN: {label}' if low else '') + ev, **k) def nc(s, reason): s.state, s.reason = 'NOT CHECKED', reason def na(s, reason): s.state, s.reason = 'N/A', reason def ok(s, text): s.ok_text = text @property def status(s): if s.state: return s.state st = [x for x, _ in s.subs] return 'FAIL' if 'FAIL' in st else 'WARN' if 'WARN' in st else 'PASS'