"""Read-only exact-code conflict audit of manifest-selected TRAIN/VALID rows."""
import argparse
import hashlib
import json
from collections import Counter, defaultdict
from pathlib import Path

import pandas as pd

from src.contracts.prompt import code_for_sample


def digest(value):
    return hashlib.sha256(value.encode()).hexdigest()


def audit_selection(selection, frame):
    groups, normalized, pairs = defaultdict(list), defaultdict(list), defaultdict(list)
    seen = set()
    for ref in selection:
        ident = ref['prompt_id']
        if ident in seen:
            raise ValueError('duplicate selected ID')
        seen.add(ident)
        row = frame.loc[ident].to_dict()
        if int(row['target']) != ref['target'] or row['pair_id'] != ref['pair_id']:
            raise ValueError('manifest label/pair mismatch')
        code = code_for_sample(row)
        item = dict(sample_id=ident, pair_id=ref['pair_id'], target=ref['target'])
        fingerprint = digest(code)
        groups[fingerprint].append(item)
        normalized[digest(''.join(code.split()))].append(item)
        pairs[ref['pair_id']].append((ref['target'], fingerprint))
    if not pairs or any(sorted(label for label, _ in rows) != [0, 1] for rows in pairs.values()):
        raise ValueError('empty or incomplete pair population')
    conflicts = {h: rows for h, rows in groups.items() if len({r['target'] for r in rows}) > 1}
    unavoidable = sum(min(Counter(r['target'] for r in rows).values()) for rows in conflicts.values())
    return dict(rows=len(seen), pairs=len(pairs), unique_exact_code=len(groups),
                exact_opposite_label_groups=len(conflicts),
                rows_in_exact_conflicts=sum(map(len, conflicts.values())),
                exact_identical_within_pair=sum(rows[0][1] == rows[1][1] for rows in pairs.values()),
                deterministic_code_only_accuracy_ceiling=1-unavoidable/len(seen),
                whitespace_removed_opposite_label_groups_screen_only=sum(
                    len({r['target'] for r in rows}) > 1 for rows in normalized.values()),
                conflict_examples=list(conflicts.values())[:10],
                exact_conflict_groups=conflicts), groups


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--manifest', type=Path, required=True)
    parser.add_argument('--corpus', type=Path, default=Path('data/processed/corpus.parquet'))
    parser.add_argument('--output', type=Path, required=True)
    args = parser.parse_args()
    if args.output.resolve() in {args.manifest.resolve(), args.corpus.resolve()}:
        raise ValueError('output must not overwrite evidence')
    manifest = json.loads(args.manifest.read_text())
    corpus_hash = hashlib.sha256(args.corpus.read_bytes()).hexdigest()
    if corpus_hash != manifest['input_sha256']['data/processed/corpus.parquet']:
        raise ValueError('corpus differs from frozen snapshot')
    frame = pd.read_parquet(args.corpus).set_index('sample_id', verify_integrity=True)
    result = dict(scope='Frozen TRAIN and VALID selections only; no TEST selection or evaluation',
                  corpus_sha256=corpus_hash,
                  manifest_sha256=hashlib.sha256(args.manifest.read_bytes()).hexdigest(),
                  dataset_snapshot_hash=manifest['dataset_snapshot_hash'],
                  limitations=[
                      'Exact source-code equality only; whitespace screening is not semantic equivalence.',
                      'Does not inspect bounded/tokenized prompts or missing repository context.',
                      'Consistency does not establish that labels are semantically correct.'])
    groups = {}
    for split in ('train', 'valid'):
        result[split], groups[split] = audit_selection(manifest[split+'_selection'], frame)
    shared = set(groups['train']) & set(groups['valid'])
    result['cross_split_exact_code_overlap'] = dict(
        hash_groups=len(shared), train_rows=sum(len(groups['train'][h]) for h in shared),
        valid_rows=sum(len(groups['valid'][h]) for h in shared),
        opposite_label_groups=sum(len({r['target'] for r in groups['train'][h]+groups['valid'][h]}) > 1 for h in shared))
    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(json.dumps(result, indent=2)+'\n')
    print(args.output)


if __name__ == '__main__':
    main()