from copy import deepcopy
import hashlib
import json

import pytest

from extensions.main_fan.history_compaction import RecordReader, agent_view, compact_evidence, summarize_point


def point(system='safety', role='fan_state'):
    return {'node_id': 'node1', 'system': system, 'role': role, 'fan_id': '1号', 'label': '主通1一级', 'unit': '', 'mapping_confirmed': False}


class Reader:
    def __init__(self, samples):
        self.samples = samples
    def data(self, record):
        return {'samples': self.samples[record['key']]}


def segment(key='a', start='2026-10-01 00:00:00', end='2026-10-03 00:00:00'):
    return {'key': key, 'status': 'completed', 'node_id': 'node1', 'start_time': start, 'end_time': end}


def sample(at, value='开', **extras):
    return {'time': at, 'value_label': value, 'status_code': '0', 'status_label': '正常', **extras}


@pytest.mark.parametrize('stopped_label', ['停', '关'])
def test_sort_merge_cross_day_and_deduplicate_boundary_without_extending_edges(stopped_label):
    reader = Reader({'a': [sample('2026-10-01 23:59:00'), sample('2026-10-02 00:00:00')],
        'b': [sample('2026-10-02 00:02:00', stopped_label), sample('2026-10-02 00:00:00'),
              sample('2026-10-02 00:01:00', stopped_label)]})
    result = summarize_point(point(), [segment('b'), segment('a')], reader)
    intervals = result['state_intervals']
    assert len(intervals) == 2 and intervals[0]['state'] == '开' and intervals[1]['state'] == '停'
    assert intervals[0]['observed_from'] == '2026-10-01T23:59:00+08:00'
    assert intervals[-1]['observed_until'] == '2026-10-02T00:02:00+08:00'
    assert result['quality']['duplicate_time_count'] == 1
    assert result['state_changes'][0]['uncertainty_seconds'] == 60
    assert result['state_changes'][0]['machine_switch_confirmed'] is False
    assert result['state_changes'][0]['to_state'] == '停'
    assert reader.samples['b'][0]['value_label'] == stopped_label


def test_fault_conflicting_timestamp_and_long_gap_break_observed_intervals():
    samples = [sample('2026-10-01 00:00:00'), sample('2026-10-01 00:01:00'),
        sample('2026-10-01 00:02:00', status_label='故障', status_code='2'),
        sample('2026-10-01 00:03:00'), sample('2026-10-01 00:04:00'),
        sample('2026-10-01 00:04:00', '停'), sample('2026-10-01 00:05:00'), sample('2026-10-01 00:30:00')]
    result = summarize_point(point(), [segment()], Reader({'a': samples}))
    assert result['quality']['conflicting_timestamp_count'] == 1 and result['quality']['invalid_value_count'] == 1
    assert result['quality']['gap_count'] == 1 and result['state_changes'] == []
    assert sum(row['state'] == '未知' for row in result['state_intervals']) == 2
    view = agent_view({'points': [result]}, '/workspace/details.json')['points'][0]
    assert len(view['state_periods']) == 1 and view['state_periods'][0]['observed_parts'] == 4
    assert view['unknown_interval_count'] == 2


def test_state_change_across_unknown_preserves_sampling_uncertainty():
    result = summarize_point(point(), [segment()], Reader({'a': [sample('2026-10-01 01:00:00'),
        sample('2026-10-01 01:01:00', status_label='故障'), sample('2026-10-01 01:05:00', '停')]}))
    change = result['state_changes'][0]
    assert change['after'] == '2026-10-01T01:00:00+08:00' and change['by'] == '2026-10-01T01:05:00+08:00'
    assert change['uncertainty_seconds'] == 300 and change['intervening_unknown_or_gap']


@pytest.mark.parametrize('role', ['fan_state', 'flow'])
def test_equipment_zeros_remain_data_and_state_codes_are_not_assumed(role):
    class EquipmentReader:
        def data(self, record):
            return {'payload': {'code': 200, 'data': [{'nodeId': 'node1', 'sbNodeCurve': [
                {'dataTime': '2026-10-01 01:00:00', 'realData': '0'},
                {'dataTime': '2026-10-01 01:01:00', 'realData': '0'}]}]}}
    result = summarize_point(point('equipment', role), [segment()], EquipmentReader())
    assert result['quality']['raw_sample_count'] == 2
    if role == 'fan_state':
        assert len(result['state_intervals']) == 1 and result['state_intervals'][0]['state'] == '原始值0'
    else:
        assert result['values']['zero_count'] == 2 and result['values']['numeric_sample_count'] == 2


def test_zip_evidence_roundtrip_retry_and_tampered_source(tmp_path):
    resolve = lambda virtual: tmp_path / virtual.removeprefix('/workspace/')
    record = segment(); record['files'] = {}
    for name, value in [('request', {'nodeId': 'node1'}), ('response', {'code': 200, 'data': []}), ('data', {'samples': []})]:
        virtual = '/workspace/daily/a/' + name + '.json'
        local = resolve(virtual); local.parent.mkdir(parents=True, exist_ok=True)
        local.write_text(json.dumps(value)); record['files'][virtual] = hashlib.sha256(local.read_bytes()).hexdigest()
    bad = deepcopy(record)
    raw = resolve('/workspace/daily/a/data.json'); raw.write_text('tampered')
    with pytest.raises(ValueError, match='指纹'):
        compact_evidence([bad], '/workspace/failed.zip', resolve)
    raw.write_text(json.dumps({'samples': []}))
    records, files, cleanup, _ = compact_evidence([record], '/workspace/packed.zip', resolve)
    assert len(files) == 1 and len(cleanup) == 3
    with RecordReader(resolve) as reader:
        assert reader.data(records[0]) == {'samples': []}
    stamp = resolve('/workspace/packed.zip').stat().st_mtime_ns
    again, _, cleanup, _ = compact_evidence(records, '/workspace/packed.zip', resolve)
    assert not cleanup and again == records and resolve('/workspace/packed.zip').stat().st_mtime_ns == stamp


def test_independent_verifier_reads_compacted_evidence_without_daily_files(tmp_path):
    from scripts.verify_main_fan_batch import read_segment
    resolve = lambda virtual: tmp_path / virtual.lstrip('/')
    record = segment(); record['files'] = {}
    for kind, value in [('request', {}), ('response', {'code': 200}), ('data', {'samples': []})]:
        virtual = '/workspace/staging/' + kind + '.json'
        path = resolve(virtual); path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(json.dumps(value)); record['files'][virtual] = hashlib.sha256(path.read_bytes()).hexdigest()
    records, _, cleanup, _ = compact_evidence([record], '/workspace/packed.zip', resolve)
    for virtual in cleanup:
        resolve(virtual).unlink()
    assert read_segment(tmp_path, records[0]) == {'samples': []}


def test_noisy_state_overview_is_bounded_and_points_to_complete_details():
    samples = [sample(f'2026-10-01 01:{minute:02d}:00', '开' if minute % 2 else '停') for minute in range(60)]
    result = summarize_point(point(), [segment()], Reader({'a': samples}))
    view = agent_view({'points': [result]}, '/workspace/details.json')['points'][0]
    assert len(result['state_intervals']) == 60 and len(view['state_periods']) == 20
    assert view['state_periods_truncated'] and view['state_change_count'] == 59