#!/usr/bin/env python3
"""Local exact audit of weighted rectangular facility assignments and proposals.

CSV inputs are preserved; outputs must be fresh. This ordinary planning model
does not claim eligibility for a strict-metric approximation theorem.
"""
import argparse
import csv
from datetime import datetime, timezone
from fractions import Fraction as Q
import hashlib
import html
import io
import json
import math
from pathlib import Path
import re
import sys

SCHEMA = 'facility-assignment-review-v1'
FILES = ('study.json', 'clients.csv', 'sites.csv', 'costs.csv', 'selected.csv', 'assignments.csv')
MAX_BYTES = 8 * 1024 * 1024
MAX_TOTAL_BYTES = 16 * 1024 * 1024
MAX_CLIENTS, MAX_SITES = 512, 128
ID = re.compile(r'[A-Za-z0-9][A-Za-z0-9_.:-]{0,63}\Z')
NUMBER = re.compile(r'-?(?:[0-9]+(?:\.[0-9]+)?|[0-9]+/[0-9]+)\Z')


def sha(data):
    return hashlib.sha256(data).hexdigest()


def number(value):
    if not isinstance(value, str) or len(value) > 160 or not NUMBER.fullmatch(value):
        raise ValueError('Use bounded decimal or rational strings; no exponents or rounding')
    try:
        q = Q(value)
    except (ValueError, ZeroDivisionError):
        raise ValueError('Invalid rational cell') from None
    if q < 0 or max(q.numerator.bit_length(), q.denominator.bit_length()) > 128:
        raise ValueError('Require nonnegative rational with at most 128-bit components')
    return q


def unique_object(pairs):
    out = {}
    for k, v in pairs:
        if k in out:
            raise ValueError('Duplicate JSON key: ' + k)
        out[k] = v
    return out


def reject_constant(value):
    raise ValueError('Non-finite JSON value: ' + value)


def text(value, name, limit=160):
    if not isinstance(value, str) or not value or len(value) > limit or any(ord(c) < 32 for c in value):
        raise ValueError('Invalid ' + name)
    return value


def identifier(value):
    if not isinstance(value, str) or not ID.fullmatch(value):
        raise ValueError('Invalid stable ID')
    return value


def rows(raw, columns):
    reader = csv.DictReader(io.StringIO(raw.decode('utf-8-sig'), newline=''), strict=True)
    if reader.fieldnames is None or len(reader.fieldnames) != len(columns) or set(reader.fieldnames) != set(columns):
        raise ValueError('CSV header must contain exactly: ' + ', '.join(columns))
    out = []
    for row in reader:
        if None in row or any(v is None for v in row.values()):
            raise ValueError('CSV row has wrong field count')
        out.append(row)
    return out


def load_inputs(directory):
    if set(p.name for p in directory.iterdir()) != set(FILES):
        raise ValueError('Input directory must contain exactly the six recognized files')
    raw = {}
    for name in FILES:
        with (directory / name).open('rb') as f:
            raw[name] = f.read(MAX_BYTES + 1)
    check_raw(raw)
    return raw


def check_raw(raw):
    if not isinstance(raw, dict) or set(raw) != set(FILES):
        raise ValueError('Require exactly the six recognized input filenames')
    if any(not isinstance(v, bytes) or len(v) > MAX_BYTES for v in raw.values()):
        raise ValueError('Inputs must be byte strings, at most eight MiB each')
    if sum(map(len, raw.values())) > MAX_TOTAL_BYTES:
        raise ValueError('Combined input limit is sixteen MiB')


def normalize(raw):
    study = json.loads(raw['study.json'].decode('utf-8-sig'), object_pairs_hook=unique_object, parse_constant=reject_constant)
    fields = {'schema', 'title', 'cost_units', 'demand_units', 'provenance', 'k', 'selection_rule', 'assignment_rule', 'objective', 'additional_constraints'}
    if not isinstance(study, dict) or set(study) != fields or study['schema'] != SCHEMA:
        raise ValueError('Invalid study schema/fields')
    for key in ('title', 'cost_units', 'demand_units', 'provenance'):
        text(study[key], key, 2000 if key == 'provenance' else 160)
    unsupported = []
    if study['selection_rule'] not in ('exactly_k', 'at_most_k'):
        unsupported.append('Unsupported selection rule')
    if study['assignment_rule'] != 'whole_client_to_one_open_site':
        unsupported.append('Only whole-client assignment to one open site is implemented')
    if study['objective'] != 'weighted_sum_assignment_cost':
        unsupported.append('Only weighted assignment cost is implemented')
    if not isinstance(study['additional_constraints'], list):
        raise ValueError('Additional constraints must be a list')
    for value in study['additional_constraints']:
        text(value, 'additional constraint', 500)
    if study['additional_constraints']:
        unsupported.append('Additional constraints are not implemented: ' + '; '.join(study['additional_constraints']))
    clients = {}
    for row in rows(raw['clients.csv'], ('client_id', 'demand')):
        c = identifier(row['client_id'])
        if c in clients:
            raise ValueError('Duplicate client ID: ' + c)
        clients[c] = str(number(row['demand']))
    sites = {}
    for row in rows(raw['sites.csv'], ('site_id', 'capacity', 'required')):
        s = identifier(row['site_id'])
        if s in sites:
            raise ValueError('Duplicate site ID: ' + s)
        if row['required'] not in ('true', 'false'):
            raise ValueError('Required flag must be exactly true or false')
        sites[s] = dict(capacity=None if row['capacity'] == '' else str(number(row['capacity'])), required=row['required'] == 'true')
    if not 1 <= len(clients) <= MAX_CLIENTS or not 1 <= len(sites) <= MAX_SITES:
        raise ValueError('Require 1..512 clients and 1..128 sites')
    if type(study['k']) is not int or not 1 <= study['k'] <= len(sites):
        raise ValueError('k must be an integer from one through site count')
    costs = {c: {} for c in sorted(clients)}
    entries = rows(raw['costs.csv'], ('client_id', 'site_id', 'unit_cost'))
    if len(entries) != len(clients) * len(sites):
        raise ValueError('Require one unit cost for every client-site pair')
    for row in entries:
        c, s = row['client_id'], row['site_id']
        if c not in clients or s not in sites or s in costs[c]:
            raise ValueError('Duplicate or unknown client-site cost pair')
        costs[c][s] = str(number(row['unit_cost']))
    selected = [identifier(x['site_id']) for x in rows(raw['selected.csv'], ('site_id',))]
    assignments = [dict(client_id=identifier(x['client_id']), site_id=identifier(x['site_id'])) for x in rows(raw['assignments.csv'], ('client_id', 'site_id'))]
    return dict(study=study, clients=dict(sorted(clients.items())), sites=dict(sorted(sites.items())), costs={c: dict(sorted(row.items())) for c, row in costs.items()}, selected=sorted(selected), assignments=sorted(assignments, key=lambda x: (x['client_id'], x['site_id'])), unsupported=unsupported)


def model(normalized):
    return dict(study=normalized['study'], clients={c: Q(v) for c, v in normalized['clients'].items()}, sites={s: dict(capacity=None if v['capacity'] is None else Q(v['capacity']), required=v['required']) for s, v in normalized['sites'].items()}, costs={c: {s: Q(v) for s, v in row.items()} for c, row in normalized['costs'].items()})


def audit_plan(data, selected, assignments):
    study, clients, sites, costs = (data[k] for k in ('study', 'clients', 'sites', 'costs'))
    issues = []
    chosen = set(selected)
    if len(chosen) != len(selected):
        issues.append('Duplicate selected site IDs')
    if not chosen <= sites.keys():
        issues.append('Unknown selected site IDs: ' + ', '.join(sorted(chosen - sites.keys())))
    if not chosen or len(chosen) > study['k'] or (study['selection_rule'] == 'exactly_k' and len(chosen) != study['k']):
        issues.append('Selected site count violates ' + study['selection_rule'] + '=' + str(study['k']))
    required = {s for s, v in sites.items() if v['required']}
    if not required <= chosen:
        issues.append('Required sites missing: ' + ', '.join(sorted(required - chosen)))
    seen = set()
    for row in assignments:
        c, s = row['client_id'], row['site_id']
        if c not in clients or c in seen:
            issues.append('Unknown or duplicate assignment client: ' + c)
        seen.add(c)
        if s not in chosen or s not in sites:
            issues.append('Assignment uses an unknown or closed site: ' + s)
    if clients.keys() - seen:
        issues.append('Unassigned clients: ' + ', '.join(sorted(clients.keys() - seen)))
    result = dict(valid=False, selected_ids=sorted(chosen), issues=issues, exact_cost=None, assignments=assignments, loads=None)
    if issues:
        return result
    loads = {s: Q(0) for s in sorted(chosen)}
    cost = Q(0)
    for row in assignments:
        c, s = row['client_id'], row['site_id']
        loads[s] += clients[c]
        cost += clients[c] * costs[c][s]
    for s, load in loads.items():
        if sites[s]['capacity'] is not None and load > sites[s]['capacity']:
            issues.append('Capacity exceeded at ' + s + ': load ' + str(load) + ', capacity ' + str(sites[s]['capacity']))
    result['loads'] = {s: str(v) for s, v in loads.items()}
    result['valid'] = not issues
    if result['valid']:
        result['exact_cost'] = str(cost)
    return result


def nearest_plan(data, selected):
    chosen = sorted(set(selected))
    if not chosen or not set(chosen) <= data['sites'].keys():
        return None
    assignments = [dict(client_id=c, site_id=min(chosen, key=lambda s: (data['costs'][c][s], s))) for c in data['clients']]
    return audit_plan(data, selected, assignments)


def relaxation_bound(data):
    # Dropping opening, required-site and capacity constraints permits each client
    # its cheapest candidate. Every feasible whole-client plan costs at least this.
    return sum((w * min(data['costs'][c].values()) for c, w in data['clients'].items()), Q(0))


def integer_scale(values):
    scale = 1
    for value in values:
        scale = math.lcm(scale, value.denominator)
        if scale > 1_000_000:
            raise ValueError('Exact solver scale exceeds one million; input has not been rounded')
    return scale


def solver_proposal(data):
    import ortools
    from ortools.sat.python import cp_model
    clients, sites = list(data['clients']), list(data['sites'])
    demand_scale = integer_scale(list(data['clients'].values()) + [v['capacity'] for v in data['sites'].values() if v['capacity'] is not None])
    weighted = [[data['clients'][c] * data['costs'][c][s] for s in sites] for c in clients]
    objective_scale = integer_scale(v for row in weighted for v in row)
    coefficients = [[int(v * objective_scale) for v in row] for row in weighted]
    demands = [int(data['clients'][c] * demand_scale) for c in clients]
    capacities = [None if data['sites'][s]['capacity'] is None else int(data['sites'][s]['capacity'] * demand_scale) for s in sites]
    if sum(map(sum, coefficients)) > 2**60 or sum(demands) > 2**60 or any(v is not None and v > 2**60 for v in capacities):
        raise ValueError('Exact solver integer range exceeds the conservative limit; input has not been rounded')
    cp = cp_model.CpModel()
    ys = [cp.new_bool_var('open_' + str(j)) for j in range(len(sites))]
    cp.add(sum(ys) >= 1)
    if data['study']['selection_rule'] == 'exactly_k':
        cp.add(sum(ys) == data['study']['k'])
    else:
        cp.add(sum(ys) <= data['study']['k'])
    for j, s in enumerate(sites):
        if data['sites'][s]['required']:
            cp.add(ys[j] == 1)
    xs = [[cp.new_bool_var(f'assign_{i}_{j}') for j in range(len(sites))] for i in range(len(clients))]
    for i, row in enumerate(xs):
        cp.add(sum(row) == 1)
        for j, x in enumerate(row):
            cp.add(x <= ys[j])
    for j, cap in enumerate(capacities):
        if cap is not None:
            cp.add(sum(demands[i] * xs[i][j] for i in range(len(clients))) <= cap * ys[j])
    cp.minimize(sum(coefficients[i][j] * xs[i][j] for i in range(len(clients)) for j in range(len(sites))))
    error = cp.validate()
    if error:
        raise ValueError(error)
    solver = cp_model.CpSolver()
    solver.parameters.max_time_in_seconds = 2.0
    solver.parameters.num_search_workers = 1
    solver.parameters.random_seed = 125
    status = solver.solve(cp)
    witness = None
    if status in (cp_model.OPTIMAL, cp_model.FEASIBLE):
        selected = [s for s, y in zip(sites, ys) if solver.value(y)]
        assignments = [dict(client_id=c, site_id=sites[j]) for i, c in enumerate(clients) for j in range(len(sites)) if solver.value(xs[i][j])]
        witness = audit_plan(data, selected, assignments)
    return dict(status=solver.status_name(status), checked_plan=witness, version=ortools.__version__, time_limit_seconds=2, workers=1, seed=125, demand_scale=demand_scale, objective_scale=objective_scale, reported_scaled_objective=float(solver.objective_value) if witness else None, reported_scaled_best_bound=float(solver.best_objective_bound), reported_wall_seconds=float(solver.wall_time), model_sha256=sha(str(cp.proto).encode()), scope='Solver status and floating numerical bounds are reported only; independent quality uses an exact checked plan and the separately justified relaxation bound.')


def review(raw, solver='none'):
    if solver not in ('none', 'ortools'):
        raise ValueError('Unsupported solver')
    check_raw(raw)
    result = dict(schema=SCHEMA, generated_at_utc=datetime.now(timezone.utc).isoformat(), status='input_invalid', errors=[], input_files=[dict(name=k, bytes=len(v), sha256=sha(v)) for k, v in sorted(raw.items())], submitted=None, nearest_reassignment=None, solver=None, comparison=None, scope='Exact frozen weighted assignment model only. No strict-metric/source-theorem eligibility, geographic validation or customer benefit is established.')
    try:
        normalized = normalize(raw)
    except (ValueError, UnicodeDecodeError, csv.Error) as e:
        result['errors'].append(str(e))
        return result, None
    result['study'] = normalized['study']
    result['normalized_input_sha256'] = sha(json.dumps(normalized, sort_keys=True, separators=(',', ':')).encode())
    result['model_support'] = dict(supported=not normalized['unsupported'], issues=normalized['unsupported'])
    if normalized['unsupported']:
        result['status'] = 'unsupported_model'
        return result, normalized
    data = model(normalized)
    result['submitted'] = audit_plan(data, normalized['selected'], normalized['assignments'])
    result['nearest_reassignment'] = nearest_plan(data, normalized['selected'])
    if solver == 'ortools':
        try:
            result['solver'] = solver_proposal(data)
        except ImportError as e:
            result['solver'] = dict(status='unavailable', reason=str(e))
        except ValueError as e:
            result['solver'] = dict(status='refused', reason=str(e))
    else:
        result['solver'] = dict(status='not_requested')
    candidates = [('submitted', result['submitted']), ('nearest_reassignment', result['nearest_reassignment']), ('solver_proposal', result['solver'].get('checked_plan'))]
    valid = [(name, p) for name, p in candidates if p and p['valid']]
    origin, best = min(valid, key=lambda x: Q(x[1]['exact_cost'])) if valid else (None, None)
    lower = relaxation_bound(data)
    upper = Q(best['exact_cost']) if best else None
    if upper is not None and upper < lower:
        raise ArithmeticError('Checked objective contradicts independently derived lower bound')
    result['comparison'] = dict(best_origin=origin, best_checked_plan=best, exact_lower_bound=str(lower), lower_bound_scope='Each client may use its cheapest candidate; opening count, required sites and capacities relaxed', exact_gap=str(upper-lower) if upper is not None else None, independently_known_optimum=str(upper) if upper is not None and upper == lower else None, improvement_over_submitted=str(Q(result['submitted']['exact_cost'])-upper) if result['submitted']['valid'] and upper is not None else None)
    result['status'] = 'review_complete' if result['submitted']['valid'] else 'submitted_plan_infeasible'
    return result, normalized


def report_html(result):
    esc = lambda v: html.escape(str(v), quote=True)
    study = result.get('study', {})
    submitted, comparison = result['submitted'], result['comparison']
    best = comparison['best_checked_plan'] if comparison else None
    status = {'review_complete': 'Review complete', 'submitted_plan_infeasible': 'Submitted plan fails the declared constraints', 'input_invalid': 'Input needs correction', 'unsupported_model': 'Model needs an additional checker'}[result['status']]
    cards = [('Submitted cost', submitted['exact_cost'] if submitted and submitted['valid'] else 'Not accepted'), ('Best checked cost', best['exact_cost'] if best else 'Unknown'), ('Relaxed lower bound', comparison['exact_lower_bound'] if comparison else 'Not assessed'), ('Best plan proven optimal?', 'Yes, for this model' if comparison and comparison['independently_known_optimum'] is not None else 'Unknown')]
    body = '<div class="cards">' + ''.join('<div><span>'+esc(k)+'</span><strong>'+esc(v)+'</strong></div>' for k,v in cards) + '</div><h2>Review findings</h2><ul>'
    issues = result['errors'] + result.get('model_support', {}).get('issues', []) + (submitted['issues'] if submitted else [])
    if not issues:
        issues = ['Selected sites, required IDs, complete whole-client assignments and capacity loads passed exact checks.']
    body += ''.join('<li>'+esc(x)+'</li>' for x in issues) + '</ul>'
    if comparison:
        body += '<p>Lower bound: '+esc(comparison['lower_bound_scope'])+'.</p>'
        if comparison['exact_gap'] is not None:
            body += '<p>The best checked cost is <strong>at most '+esc(comparison['exact_gap'])+' above the optimum</strong> for this model. This is a bound on suboptimality, not a measured difference from a known optimum.</p>'
        if comparison['improvement_over_submitted'] is not None:
            body += '<p>Best checked plan reduces submitted cost by <strong>'+esc(comparison['improvement_over_submitted'])+'</strong>.</p>'
    nearest = result['nearest_reassignment']
    if nearest and not nearest['valid']:
        body += '<p>The nearest-site reassignment was rejected: '+esc('; '.join(nearest['issues']))+'.</p>'
    for label, plan in [('Submitted plan', submitted), ('Best checked plan', best)]:
        if not plan or not plan['valid']:
            continue
        body += '<h2>'+label+'</h2><p>Open sites: '+esc(', '.join(plan['selected_ids']))+'</p><div class="table-wrap"><table><tr><th>Site ID</th><th>Assigned demand</th></tr>'
        body += ''.join('<tr><td>'+esc(s)+'</td><td>'+esc(v)+'</td></tr>' for s,v in plan['loads'].items())+'</table></div>'
        body += '<details><summary>Client assignments</summary><div class="table-wrap"><table><tr><th>Client ID</th><th>Site ID</th></tr>' + ''.join('<tr><td>'+esc(x['client_id'])+'</td><td>'+esc(x['site_id'])+'</td></tr>' for x in plan['assignments'])+'</table></div></details>'
    if result['solver']:
        names = {'not_requested':'Not run','unavailable':'Unavailable in this environment','refused':'Input exceeds adapter limits','OPTIMAL':'Optimal, as reported by the solver','FEASIBLE':'Feasible, as reported by the solver','INFEASIBLE':'Infeasible, as reported by the solver','UNKNOWN':'No completed solver conclusion','MODEL_INVALID':'Solver model rejected'}
        body += '<h2>Conventional solver</h2><p>'+esc(names.get(result['solver']['status'],result['solver']['status']))+'. '+esc(result['solver'].get('reason','Solver status and numerical bounds remain separate from independent optimality.'))+'</p>'
    body += '<h2>Input provenance</h2><p>'+esc(study.get('provenance','No accepted study declaration.'))+'</p><p>Supplied provenance is not independently verified.</p><ul>'
    body += ''.join('<li><a download href="inputs/'+esc(x['name'])+'">'+esc(x['name'])+'</a> · SHA-256 <code>'+esc(x['sha256'])+'</code></li>' for x in result['input_files'])+'</ul>'
    body += '<details><summary>Evidence and scope</summary><p>'+esc(result['scope'])+'</p><p>Limits: 512 clients, 128 sites, eight MiB per file, sixteen MiB combined and 128-bit rational input components. Optional solver: two seconds, one worker; exact scales capped at one million and integer range capped conservatively.</p><a href="report.json" download>Review JSON</a> · <a href="manifest.json" download>File manifest</a></details>'
    css='body{font:16px/1.55 system-ui,sans-serif;color:#173b39;background:#f2f5f0;margin:0}main{max-width:1080px;margin:40px auto;padding:32px;background:white;border:1px solid #d6e1db;border-radius:12px}h1{font-size:34px;line-height:1.15}h2{font-size:22px;margin-top:30px}.eyebrow{letter-spacing:.12em;text-transform:uppercase;font-size:12px}.status{padding:12px 18px;background:#e8f3ed;border-left:4px solid #357b68}.cards{display:grid;grid-template-columns:repeat(4,1fr);gap:12px;margin:24px 0}.cards>div{border:1px solid #ccdcd2;border-radius:8px;padding:16px}.cards span{display:block;font-size:13px}.cards strong{display:block;font-size:23px;margin-top:8px;overflow-wrap:anywhere}a{color:#286653}code{font-size:12px;overflow-wrap:anywhere}table{border-collapse:collapse;width:100%}th,td{text-align:left;padding:10px 14px;border:1px solid #d9e2dd}th{background:#e8f3ed}.table-wrap{overflow-x:auto}details{padding:14px;border:1px solid #d9e2dd;margin-top:20px}summary{cursor:pointer;font-weight:600}@media(max-width:700px){main{margin:12px;padding:20px}.cards{grid-template-columns:repeat(2,1fr)}h1{font-size:28px}}'
    timestamp = datetime.fromisoformat(result['generated_at_utc']).strftime('%d %B %Y, %H:%M UTC')
    return '<!doctype html><html lang="en"><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1"><title>'+esc(study.get('title','Facility assignment review'))+'</title><style>'+css+'</style><main><div class="eyebrow">Optimization Assurance · Assignment review</div><h1>'+esc(study.get('title','Facility assignment review'))+'</h1><div class="status">'+esc(status)+'</div><p>Objective units: '+esc(study.get('demand_units','unspecified'))+' × '+esc(study.get('cost_units','unspecified'))+' · '+esc(timestamp)+'</p>'+body+'<footer><p>Review sign-off: not supplied. This report records computational checks and unresolved items.</p></footer></main></html>\n'


def export_review(raw, output, solver='none'):
    if output.exists():
        raise FileExistsError('Choose a fresh output directory; existing evidence is preserved')
    result, normalized = review(raw, solver)
    output.mkdir(parents=True, exist_ok=False)
    (output/'inputs').mkdir()
    for name, data in raw.items():
        with (output/'inputs'/name).open('xb') as f:
            f.write(data)
    def write(name, value):
        with (output/name).open('x') as f:
            json.dump(value,f,indent=2,allow_nan=False);f.write('\n')
    write('report.json',result)
    if normalized:
        write('normalized.json',normalized)
    with (output/'report.html').open('x') as f:
        f.write(report_html(result))
    files=[dict(path=str(p.relative_to(output)),bytes=p.stat().st_size,sha256=sha(p.read_bytes())) for p in sorted(output.rglob('*')) if p.is_file()]
    write('manifest.json',dict(schema=SCHEMA,files=files,tools=[dict(name=Path(__file__).name,sha256=sha(Path(__file__).read_bytes()))],python=sys.version,scope='Checksums bind local bytes; no independent authenticity, theorem eligibility, geographic validity or buyer validation.'))
    return result


def main():
    parser=argparse.ArgumentParser(description=__doc__)
    parser.add_argument('input_directory',type=Path)
    parser.add_argument('--output',required=True,type=Path)
    parser.add_argument('--solver',choices=('none','ortools'),default='none')
    args=parser.parse_args()
    result=export_review(load_inputs(args.input_directory),args.output,args.solver)
    print(json.dumps(dict(status=result['status'],report=str((args.output/'report.html').resolve()),comparison=result['comparison'])))


if __name__=='__main__':
    main()