"""Rules F: BigQuery export (G17, G18), plus the expiration checks G04 reads."""
import re
from datetime import timedelta
from .common import DAYS_PER_MONTH, daterange, get, n0, num, pct, rid_of, todate

def expiry_parts(e):
    """(table date, expiry date) from earliest_expiry: {table, expires}, BQ2b's own table_name and
    option_value columns, the BQ2b rows as a list (the first row counts), or a plain string."""
    if isinstance(e, list): e = e[0] if e else None
    t, v = (e.get('table') or e.get('table_name'), e.get('expires') or e.get('option_value')) if isinstance(e, dict) else (e, e)
    table_date = re.search(r'events_(\d{8})', str(t or ''))
    expiry = re.search(r'\d{4}-\d{2}-\d{2}', str(v or ''))
    return (todate(table_date.group(1)) if table_date else None, todate(expiry.group(0)) if expiry else None)

def table_keep_days(c):
    """(days the earliest-expiring daily table is kept, description) from BQ2b, or None."""
    d, x = expiry_parts((c.bq or {}).get('earliest_expiry'))
    return ((x - d).days, f'events_{d:%Y%m%d} expires on {x}') if d and x else None

def daily_table_dates(bq):
    """The dates of the BQ1 daily tables and the newest one. Table names or BQ1 rows both read."""
    have = {todate(str(t.get('table_name') if isinstance(t, dict) else t)[-8:]) for t in bq.get('daily_tables') or []} - {None}
    return have, todate(str(bq.get('newest_daily_table') or '')[-8:]) or max(have, default=None)

def bq_history_gap(c, need):
    """Why the BigQuery export cannot hold `need` months of history, or '' when the results show it can."""
    if c.bq is None: return 'no BigQuery results supplied, so table expiration and freshness are unread'
    days_needed = need * DAYS_PER_MONTH
    exp = c.bq.get('default_table_expiration_days')
    if exp and num(exp) < days_needed: return f'the dataset default table expiration is {num(exp):g} days'
    kd = table_keep_days(c)
    if kd and kd[0] < days_needed: return f'daily tables still expire ({kd[1]}, {kd[0]} days after its date)'
    if num(c.bq.get('tables_with_expiry')) and not kd:
        how = 'earliest_expiry not readable: write {table, expires}' if c.bq.get('earliest_expiry') else 'run BQ2b'
        return f'{int(num(c.bq["tables_with_expiry"]))} daily tables carry an expiration ({how})'
    newest = daily_table_dates(c.bq)[1]
    if not newest: return 'the newest daily table is unknown'
    if (c.asof - newest).days > c.th['table_max_age_days']: return f'the newest daily table is {newest}, so the export is stale'
    return ''

def g17(c, f):
    if c.bq_links is None: return f.nc('BigQuery links not read')
    req = c.cfg.get('bigquery_required')
    if not c.bq_links:
        return f.fail('BigQuery export is required but there is no link') if req else \
            f.warn('no BigQuery link: raw event history starts only from the day a link is made')
    x = c.bq_links[0]
    daily, streaming = get(x, 'dailyExportEnabled'), get(x, 'streamingExportEnabled')
    if not daily and not streaming: f.fail('daily and streaming export are both off')
    elif not daily and c.cfg.get('warehouse_uses_revenue'):
        f.warn('streaming export only, for revenue reporting: streaming is best effort and can have gaps')
    exported = {rid_of(s) for s in get(x, 'exportStreams', [])}
    prod = [str(s) for s in c.cfg.get('production_stream_ids') or [rid_of(get(s, 'name')) for s in c.web]]
    miss = [s for s in prod if exported and s not in exported]
    if miss: f.fail(f'production stream {", ".join(miss)} is not in exportStreams')
    bad = [e for e in get(x, 'excludedEvents', []) if e == 'purchase' or e in (c.ke_names or set())]
    if bad and c.cfg.get('warehouse_uses_revenue'):
        f.fail(f'excludedEvents drops {", ".join(bad)} while the warehouse reports revenue or conversions')
    f.ok(f'daily export to {get(x, "project")} ({get(x, "datasetLocation")}) of every production stream, nothing needed excluded')

def g18(c, f):
    links, req = c.bq_links, c.cfg.get('bigquery_required')
    if links == []: return f.na('no BigQuery link, so no export to rate; G17 rates the missing link' if req else 'no BigQuery link and none required')
    if c.rows('R02') is None and c.bq is None: return f.nc('R02 daily volume and BigQuery results not read')
    x = (links or [{}])[0]
    daily_on = links is None or bool(get(x, 'dailyExportEnabled'))  # answer 9823238: the limit is on daily export only
    if not daily_on:
        f.partial.append(('streaming-only export' if get(x, 'streamingExportEnabled') else 'daily export off') + ': no daily event limit applies')
    exported = {rid_of(s) for s in get(x, 'exportStreams', [])} or None
    excl, evs = set(get(x, 'excludedEvents', [])), c.agg('R01', 'eventName', 'eventCount')
    kept = 1 - (sum(evs.get(e, 0) for e in excl) / sum(evs.values()) if excl and evs else 0)
    days = {d: v * kept for d, v in c.daily(exported, c.start, c.end).items()}
    src = dict.fromkeys(days, 'R02')
    for r in (c.bq or {}).get('daily_counts') or []:  # raw exported events, consent-denied pings included
        d, v = todate(r.get('day')), num(r.get('events'))
        if d and c.start <= d <= c.end and v > days.get(d, 0): days[d], src[d] = v, 'BigQuery'
    need = int(c.cfg.get('needs_history_months') or 14)
    bq = c.bq or {}
    have, newest = daily_table_dates(bq)
    bq1 = bool(have or newest)
    stale = bool(daily_on and newest and (c.asof - newest).days > c.th['table_max_age_days'])
    stale_told = False
    limit_read = bool(days) and daily_on and not c.is360
    if limit_read:
        peak = max(days, key=days.get)
        over = sorted(d for d, v in days.items() if v > c.th['export_limit'])
        what = f'peak {n0(days[peak])} on {peak}' + (' in BigQuery, consent-denied pings included' if src[peak] == 'BigQuery' else '')
        if over:
            text = (f'{len(over)} day{"s" if len(over) != 1 else ""} above the 1,000,000 standard daily export limit ({what}); '
                    'Google pauses the daily export and does not reprocess paused days')
            gaps = [d for d in daterange(over[0], c.asof - timedelta(days=2)) if have and d not in have]
            if c.bq is None: f.fail(text + '; at risk: BigQuery results not supplied, so a pause is not ruled out', risk=True)
            elif stale or gaps:
                f.fail(text + '; the export looks paused: ' + (f'the newest daily table is {newest}' if stale else f'daily table {gaps[0]} is missing'), loss=True)
                stale_told = stale
            elif not bq1: f.fail(text + '; at risk: BQ1 daily table list not supplied, so a pause is not ruled out', risk=True)
            else: f.fail(text + '; at risk: daily tables still complete', risk=True)
        elif days[peak] >= c.th['export_warn']:
            f.warn(f'exported events peaked at {n0(days[peak])} on {peak}, near the 1,000,000 daily limit'
                   + (' (BigQuery count, consent-denied pings included)' if src[peak] == 'BigQuery' else ''))
    if c.bq is None:
        f.partial.append('no BigQuery results: tables, freshness and expiration not checked')
        return f.ok('daily exported events (R02) under the limit' if limit_read else 'BigQuery link read')
    if daily_on and not bq1: f.partial.append('BQ1 not supplied: freshness and missing tables not checked')
    days_needed = need * DAYS_PER_MONTH
    exp_days = bq.get('default_table_expiration_days')
    if exp_days and num(exp_days) < days_needed:
        f.fail(f'dataset default table expiration is {num(exp_days):g} days while {need} months of history are needed '
               '(BigQuery sandbox tables expire after 60 days)', loss=True)
    kd, n_exp = table_keep_days(c), int(num(bq.get('tables_with_expiry')))
    if kd and kd[0] < days_needed:
        f.fail(f'{n_exp or "some"} daily tables still carry an expiration ({kd[1]}, {kd[0]} days after its date) while {need} months are needed; '
               'a table keeps the expiration it was created with, so a sandbox upgrade or a new dataset default does not remove it', loss=True)
    elif n_exp and bq.get('earliest_expiry'):  # BQ2b ran: say so, and still no PASS on an expiration nobody could read
        f.warn(f'{n_exp} daily tables carry an expiration that could not be read')
        f.partial.append('earliest_expiry not readable: write {table, expires}')
    elif n_exp:
        f.warn(f'{n_exp} daily tables carry an expiration; run BQ2b to see when the first one expires')
    if stale and not stale_told:
        f.fail(f'the newest daily table is {newest}, {(c.asof - newest).days} days old', loss=True)
    gaps = [c.asof - timedelta(days=i) for i in range(2, 32) if c.asof - timedelta(days=i) not in have]
    if daily_on and have and gaps: f.warn(f'{len(gaps)} missing daily table{"s" if len(gaps) != 1 else ""} in the last 30 days, first {min(gaps)}')
    api, worst = c.daily(exported), (0, None)
    for r in c.bq.get('daily_counts') or []:
        d = todate(r.get('day'))
        bqv = r.get('events_not_denied', r.get('events'))
        bqv = None if bqv is None else num(bqv)  # BigQuery returns INT64 values as JSON strings
        settled = d is not None and (c.asof - d).days > c.th['table_update_days']  # answers 7029846, 9358801: a table can change for up to 3 days after its date
        if settled and d in api and api[d] and bqv is not None and abs(bqv - api[d]) / api[d] > worst[0]:
            worst = (abs(bqv - api[d]) / api[d], d)
    if worst[0] > c.th['export_tolerance']:
        f.warn(f'export and Data API daily event counts differ by {pct(worst[0])} on {worst[1]} (after consent adjustment)')
    f.ok('; '.join(x for x, ok in (('exported volume under the limit', limit_read), ('no daily event limit applies', not daily_on),
                                    ('tables fresh, complete and kept long enough', daily_on and bq1),
                                    ('tables kept long enough', not (daily_on and bq1))) if ok))