"""Rules G and H: custom definitions (G19, G20), attribution and channels (G21, G22)."""
import json
import re
from collections import defaultdict
from .common import ID_NAMES, PREDEFINED, QUOTA, get, lookback_days, n0, num, pct, up

def custom_defs(c):
    cds = get(c.snap, 'customDimensions')
    if cds is not None:
        return [(up(get(d, 'scope') or 'EVENT'), get(d, 'parameterName', ''), get(d, 'displayName', '')) for d in cds]
    api = get(c.snap, 'metadataApi')  # getMetadata keeps dimensions; the MCP tool returns custom_dimensions
    meta = next((v for v in (get(api, 'dimensions'), get(api, 'customDimensions')) if v is not None), None)
    if meta is None: return None
    scope = {'customEvent': 'EVENT', 'customUser': 'USER', 'customItem': 'ITEM'}
    return [(scope[a.split(':')[0]], a.split(':', 1)[1], get(d, 'uiName', '')) for d in meta
            for a in [get(d, 'apiName', '')] if a.split(':')[0] in scope]

def custom_metric_count(c):
    cms = get(c.snap, 'customMetrics')
    if cms is not None: return len(cms)
    api = get(c.snap, 'metadataApi')
    cms = get(api, 'customMetrics')
    return len(cms) if cms is not None else sum(str(get(m, 'apiName', '')).startswith('custom') for m in get(api, 'metrics') or [])

def g19(c, f):
    defs = custom_defs(c)
    if defs is None: return f.nc('custom definitions not read')
    q = QUOTA[c.is360]
    used = defaultdict(int)
    for scope, _, _ in defs: used[scope] += 1
    used['metrics'] = custom_metric_count(c)
    used['calculated'] = len(get(c.snap, 'calculatedMetrics') or [])
    for k in ('EVENT', 'USER', 'ITEM', 'metrics', 'calculated'):
        if used[k] >= c.th['quota_share'] * q[k]:
            f.warn(f'{k.lower()} custom definitions: {used[k]} of {q[k]} ({pct(used[k] / q[k])})')
    pre = sorted({p for _, p, _ in defs if p.lower() in PREDEFINED})
    ids = sorted({p for _, p, _ in defs if p.lower() in ID_NAMES and p.lower() not in PREDEFINED})
    legacy = sorted({dn for _, _, dn in defs if re.search(r'\[[a-z0-9_]+\]\s*$', dn or '')})
    if pre: f.warn(f'definitions that duplicate predefined dimensions: {", ".join(pre)}')
    if ids: f.warn(f'registered IDs or timestamps: {", ".join(ids)} (use User-ID for users; IDs create (other) rows)')
    if legacy: f.warn(f'legacy display names ending in [event_name]: {", ".join(legacy[:5])}')
    f.ok('quotas under 80%, no IDs or duplicates')

def g20(c, f):
    r14 = c.reports.get('R14')
    if custom_defs(c) is not None and not [d for d in custom_defs(c) if d[0] != 'ITEM']:
        return f.na('no event or user custom dimensions registered')
    if not r14: return f.nc('R14 custom dimension probes not read')
    dead = sorted(k for k, v in r14.items() if num(v.get('not_set_share_28d')) >= 0.999)
    high = sorted((k, int(num(v.get('distinct_1d')))) for k, v in r14.items()
                  if num(v.get('distinct_1d')) > c.th['cardinality_per_day'])
    if dead: f.warn(f'only (not set) for 28 days: {", ".join(dead)} (wrong parameter name or scope, or a dead slot)')
    if high: f.warn('more than 500 distinct values in one day: ' + ', '.join(f'{k} ({n0(n)})' for k, n in high))
    blind = sorted(k for k, v in r14.items() if v.get('distinct_1d') is None)
    if blind: f.partial.append(f'one-day cardinality not read for {len(blind)} count-only probes')
    if c.bq and c.bq.get('unregistered_key_event_params'):  # key names, or the BQ6 rows as returned
        params = [str(x.get('key') if isinstance(x, dict) else x) for x in c.bq['unregistered_key_event_params']]
        f.warn(f'parameters on key events in BigQuery but not registered: {", ".join(params)}')
    defs = custom_defs(c) or []
    if len([d for d in defs if d[0] != 'ITEM']) > len(r14):
        f.partial.append(f'probed {len(r14)} of {len([d for d in defs if d[0] != "ITEM"])} event and user dimensions')
    f.ok(f'{len(r14)} probed definitions populated, under 500 values a day')

def g21(c, f):
    att = get(c.snap, 'attributionSettings')
    if att is None: return f.nc('attribution settings not read')
    model = up(get(att, 'reportingAttributionModel'))
    acq = lookback_days(get(att, 'acquisitionConversionEventLookbackWindow'))
    oth = lookback_days(get(att, 'otherConversionEventLookbackWindow'))
    cycle = int(c.cfg.get('sales_cycle_days') or 0)
    if 'GOOGLE_PAID' in model and not c.documented('attribution_model'):
        f.warn(f'{model} without a decision: only Google paid channels receive key event credit in reports')
    if cycle and oth and oth < cycle:
        f.warn(f'other key events lookback is {oth} days, shorter than the {cycle}-day sales cycle (it also shortens how long direct sessions keep an earlier source)')
    if acq == 7 and not c.documented('acquisition_lookback'):
        f.warn('acquisition lookback is 7 days without a decision')
    for key, d, text, _ in c.changes({'attribution'}):
        if not c.annotated(d):
            f.warn(f'{text} on {d} has no annotation' + c.ev(key))
            c.propose(d, text, 'Attribution setting changed in Admin. Model changes rewrite history; lookback changes apply forward. Compare periods with care.')
    f.ok(f'model {model}, lookbacks {acq} and {oth} days cover the sales cycle')

def g22(c, f):
    r04, cgs = c.rows('R04'), get(c.snap, 'channelGroups')
    if r04 is None and cgs is None: return f.nc('R04 channels and channel groups not read')
    prim = next((g for g in cgs or [] if get(g, 'primary') and not get(g, 'systemDefined')), None)
    custom_primary = prim is not None or (cgs is None and any(
        r.get('sessionPrimaryChannelGroup') not in (None, r.get('sessionDefaultChannelGroup')) for r in r04 or []))
    if r04:
        ch, tot = c.agg('R04', 'sessionDefaultChannelGroup', 'sessions'), sum(x.get('sessions', 0) for x in r04)
        shares = [('the default channel group', ch.get('Unassigned', 0) / tot if tot else 0)]
        if custom_primary and any('sessionPrimaryChannelGroup' in x for x in r04):
            pch = c.agg('R04', 'sessionPrimaryChannelGroup', 'sessions')
            shares.append((f'the primary group "{get(prim, "displayName")}"' if prim else 'the primary group',
                           pch.get('Unassigned', 0) / tot if tot else 0))
        un = max(v for _, v in shares)
        said = f'Unassigned is {pct(un)} of sessions' if len(shares) == 1 else \
            f'Unassigned holds {pct(shares[0][1])} of sessions in {shares[0][0]} and {pct(shares[1][1])} in {shares[1][0]}'
        if un >= c.th['unassigned_fail']: f.gated(said, c.low(tot, 'min_sessions'))
        elif un >= c.th['unassigned_warn']:
            f.warn(f'{said} (diagnose with ga4-unassigned-not-set-diagnosis)')
        prev = c.agg('R04P', 'sessionDefaultChannelGroup', 'sessions')
        if prev and tot and sum(prev.values()):
            rise = ch.get('Direct', 0) / tot - prev.get('Direct', 0) / sum(prev.values())
            if rise >= c.th['direct_rise']:
                f.warn(f'Direct share rose {100 * rise:.1f} percentage points on the previous 28 days')
        elif not prev: f.partial.append('R04P previous window not read: Direct trend not checked')
    src = ('R06', 'defaultChannelGroup') if c.rows('R06') is not None else ('R04', 'sessionDefaultChannelGroup')
    kes = c.agg(src[0], src[1], 'keyEvents')
    tke = sum(kes.values())
    if tke and kes.get('Unassigned', 0) / tke >= c.th['unassigned_key_event_fail']:
        f.gated(f'{pct(kes["Unassigned"] / tke)} of key events are credited to Unassigned', c.low(tke, 'min_key_events'))
    for g in [prim] if prim else []:
        rules = get(g, 'groupingRule', [])
        names, exprs = [get(r, 'displayName', '') for r in rules], [get(r, 'expression') for r in rules]
        ai = next((i for i, n in enumerate(names) if re.search(r'\bai\b|assistant', n, re.I)), None)
        ref = next((i for i, n in enumerate(names) if n.lower() == 'referral'), None)
        medium = any('ai-assistant' in json.dumps(e).lower() for e in exprs if e)
        if ai is not None and ref is not None and ai > ref:
            f.warn(f'primary group "{get(g, "displayName")}": {names[ai]} sits after Referral, so assistant visits that arrive '
                   'with medium referral land in Referral (first match wins)')
        if ai is None and not medium:
            f.warn(f'primary group "{get(g, "displayName")}" has no channel for medium ai-assistant (used since 13 May 2026), '
                   'so assistant referrals can land in Unassigned in this group; verify the rules at run time')
        elif ai is not None and not medium and not exprs[ai]:
            f.partial.append(f'rule expression of "{names[ai]}" not read: confirm it matches medium ai-assistant')
        if '(not set)' in json.dumps(rules):
            f.warn(f'primary group "{get(g, "displayName")}" has a rule that matches (not set)')
    for key, d, text, _ in c.changes({'channel_groups'}, c.th['primary_group_days']):
        if not c.annotated(d):
            f.warn(f'{text} on {d} has no annotation' + c.ev(key))
            c.propose(d, text, 'Channel group change. Custom groups apply to history; a new primary group applies forward only.')
    f.ok('; '.join(x for x, ok in (('Unassigned under 2% of sessions (R04)', bool(r04)),
                                    ('primary channel group order sound', prim is not None),
                                    ('channel groups read, none custom and primary', cgs is not None and prim is None)) if ok))