import csv, io
from collections import Counter, defaultdict
from datetime import datetime, timezone
from decimal import Decimal
from zoneinfo import ZoneInfo

ORDERS = '''id,version,utc,status,currency,amount,customer
O1,1,2027-05-31T16:30:00Z,paid,CNY,100,C1
O1,2,2027-05-31T16:30:00Z,paid,CNY,120,C1
O1,2,2027-05-31T16:30:00Z,paid,CNY,120,C1
O2,1,2027-06-12T00:00:00Z,paid,CNY,200,C2
O3,1,2027-06-15T00:00:00Z,cancelled,CNY,80,C1
O4,1,2027-06-30T16:30:00Z,paid,CNY,300,C1
O5,1,2027-06-20T00:00:00Z,paid,USD,50,C1
O6,2,2027-06-21T00:00:00Z,paid,CNY,40,C1
O6,2,2027-06-21T00:00:00Z,paid,CNY,60,C1'''
REFUNDS = [
    {'refund_id':'R1','order_id':'O1','currency':'CNY','amount':Decimal('20')},
    {'refund_id':'R1','order_id':'O1','currency':'CNY','amount':Decimal('20')},
    {'refund_id':'R2','order_id':'O2','currency':'CNY','amount':Decimal('30')},
    {'refund_id':'R3','order_id':'O9','currency':'CNY','amount':Decimal('10')},
]
DIMS = [
    {'customer':'C1','effective':'2027-05','segment':'SMB'},
    {'customer':'C1','effective':'2027-07','segment':'Enterprise'},
]
TZ = ZoneInfo('Asia/Shanghai')
START = datetime(2027,6,1,tzinfo=TZ)
END = datetime(2027,7,1,tzinfo=TZ)
CUTOFF = END.astimezone(timezone.utc)
def parse(s): return datetime.fromisoformat(s.replace('Z','+00:00'))
def rows(text): return list(csv.DictReader(io.StringIO(text)))
def money(x): return Decimal(x)
def fmt(d): return ', '.join(f'{k}={v:.2f}' for k,v in sorted(d.items())) or '(none)'
def sums(rs):
    out=defaultdict(Decimal)
    for r in rs: out[r['currency']] += r['amount']
    return dict(out)
raw=rows(ORDERS)
for i,r in enumerate(raw,1):
    r['source_row']=f'orders.csv:{i+1}'
    r['version']=int(r['version']); r['instant']=parse(r['utc']); r['amount']=money(r['amount'])

# Exact export repeats versus same-precedence disagreements.
seen=set(); dedup=[]; dup=Counter()
for r in raw:
    signature=tuple(r[k] for k in ('id','version','utc','status','currency','amount','customer'))
    if signature in seen: dup[r['id']]+=1
    else: seen.add(signature); dedup.append(r)
print('EXACT_DUPLICATE_EXTRA_ROWS',dict(dup))

# Literal legacy SQL behavior: UTC text prefix, no status/refund filter, inner join all dimension rows.
legacy=[r for r in raw if r['utc'].startswith('2027-06')]
joined=[]
for r in legacy:
    matches=[d for d in DIMS if d['customer']==r['customer']]
    for d in matches: joined.append((r,d))
legacy_sums=defaultdict(Decimal)
for r,d in joined: legacy_sums[r['currency']]+=r['amount']
print('LEGACY_STAGE_utc_prefix rows=%d keys=%d gross=%s' %
      (len(legacy),len({r['id'] for r in legacy}),fmt(sums(legacy))))
print('LEGACY_STAGE_inner_customer_join rows=%d keys=%d gross=%s' %
      (len(joined),len({r['id'] for r,d in joined}),fmt(legacy_sums)))

# Anti-join checks are scoped to populations and identify actual missing keys.
known_customers={d['customer'] for d in DIMS}
print('CUSTOMER_ANTI_JOIN_legacy_population',sorted({r['customer'] for r in legacy if r['customer'] not in known_customers}))

# Contract: cutoff first, then choose max version; inspect all tied rows, never break value conflicts.
within_cutoff=[r for r in dedup if r['instant']<=CUTOFF]
by_order=defaultdict(list)
for r in within_cutoff: by_order[r['id']].append(r)
selected=[]; conflicts=[]
for oid,group in sorted(by_order.items()):
    maxrank=max((r['version'],r['instant']) for r in group)
    versions=[r for r in group if (r['version'],r['instant'])==maxrank]
    # Version is precedence; tied rows must agree on all economic/status/timestamp fields.
    signatures={(r['instant'],r['status'],r['currency'],r['amount'],r['customer']) for r in versions}
    if len(signatures)>1:
        conflicts.append((oid,versions)); continue
    selected.append(versions[0])
month=[r for r in selected if START<=r['instant'].astimezone(TZ)<END]
eligible=[r for r in month if r['status']=='paid']
conflict_month=[(oid,grp) for oid,grp in conflicts if START<=grp[0]['instant'].astimezone(TZ)<END]
print('CONTRACT cutoff=%s dedup_in_cutoff=%d unambiguous_selected=%d unambiguous_month=%d confirmed_paid=%d unresolved_month_keys=%d' %
      (CUTOFF.isoformat(),len(within_cutoff),len(selected),len(month),len(eligible),len(conflict_month)))
print('TIED_PRECEDENCE_CONFLICTS',[(oid,[(r['source_row'],str(r['amount'])) for r in grp]) for oid,grp in conflicts])
print('ELIGIBLE_ORDER_KEYS',[(r['id'],r['source_row'],r['currency'],str(r['amount'])) for r in eligible])

# Correctly enrich at order instant: latest customer dimension effective on/before order local month.
enriched=[]
for r in eligible:
    ym=r['instant'].astimezone(TZ).strftime('%Y-%m')
    matches=[d for d in DIMS if d['customer']==r['customer'] and d['effective']<=ym]
    segment=max(matches,key=lambda d:d['effective'])['segment'] if matches else 'UNKNOWN'
    enriched.append((r,segment))
print('ASOF_ENRICHED',[(r['id'],segment) for r,segment in enriched])

# Deduplicate refunds by refund_id; conflicting reuse is unresolved, not silently chosen.
refund_by_id=defaultdict(list)
for r in REFUNDS: refund_by_id[r['refund_id']].append(r)
refunds=[]; refund_conflicts=[]
for rid,grp in sorted(refund_by_id.items()):
    sig={(r['order_id'],r['currency'],r['amount']) for r in grp}
    if len(sig)>1: refund_conflicts.append(rid)
    else: refunds.append(grp[0])
paid_keys={r['id']:r for r in eligible}
orphans=[r for r in refunds if r['order_id'] not in paid_keys]
linked=[r for r in refunds if r['order_id'] in paid_keys and paid_keys[r['order_id']]['currency']==r['currency']]
refund_totals=sums(linked)
gross=sums(eligible)
net={c:gross.get(c,Decimal(0))-refund_totals.get(c,Decimal(0)) for c in set(gross)|set(refund_totals)}
print('GROSS_verified_paid',fmt(gross))
print('LINKED_REFUNDS',[(r['refund_id'],r['order_id'],str(r['amount'])) for r in linked],fmt(refund_totals))
print('NET_verified_subset',fmt(net))
print('ORPHAN_REFUNDS',[(r['refund_id'],r['order_id'],r['currency'],str(r['amount'])) for r in orphans])

assert dict(legacy_sums)=={'CNY':Decimal('960'),'USD':Decimal('100')}
assert net=={'CNY':Decimal('270'),'USD':Decimal('50')}
assert [r['id'] for r in eligible]==['O1','O2','O5']
assert [(oid,[r['amount'] for r in grp]) for oid,grp in conflicts]==[('O6',[Decimal('40'),Decimal('60')])]
assert [(r['refund_id'],r['order_id']) for r in orphans]==[('R3','O9')]