"""Independent row parser (spec 18.1). Re-implements the CSV interpretation,
the dotted timestamp parse, decimal price scaling and inherent row checks with
code paths that differ from engine/parse.py. Must agree with the producer on
every canonical row, every malformed class and every counter.
"""

import re

from .. import versions

# regex-based line splitter (independent of the byte-scanner in parse.py)
_TERM_RE = re.compile(rb"\r\n|\n|\r")


def v_split_lines(data):
    """Yield (line_bytes, terminator_len) using a regex matcher."""
    pos = 0
    for m in _TERM_RE.finditer(data):
        yield data[pos:m.start()], len(m.group(0))
        pos = m.end()
    if pos < len(data):
        yield data[pos:], 0


_MONTH_BEFORE = [0, 31, 59, 90, 120, 151, 181, 212, 243, 273, 304, 334]
_TS_RE = re.compile(
    r"^(\d{4})\.(\d{2})\.(\d{2}) (\d{2}):(\d{2}):(\d{2})(?:\.(\d{3}))?$")


def v_civil_days(y, m, d):
    """Independent civil-days implementation (month-offset table). Valid in
    the canonical timestamp window 2000..certify+7d."""
    leap = (y % 4 == 0 and y % 100 != 0) or (y % 400 == 0)
    doy = _MONTH_BEFORE[m - 1] + d + (1 if (leap and m > 2) else 0)
    y = y - 1
    n = ((y * 365 + y // 4 - y // 100 + y // 400)
         - (1969 * 365 + 1969 // 4 - 1969 // 100 + 1969 // 400))
    return n + doy - 1


def v_timestamp_dotted(text):
    """Independent dotted timestamp parser -> (ts_ms, None) or (None, err)."""
    m = _TS_RE.match(text)
    if not m:
        return None, "vparse: timestamp shape not recognized"
    y, mo, d, hh, mi, ss = (int(g) for g in m.group(1, 2, 3, 4, 5, 6))
    ms = int(m.group(7)) if m.group(7) else 0
    if not (1 <= mo <= 12 and 1 <= d <= 31 and hh <= 23 and mi <= 59
            and ss <= 59 and ms <= 999):
        return None, "vparse: timestamp field out of range"
    days = v_civil_days(y, mo, d)
    ts = (days * 86400 + hh * 3600 + mi * 60 + ss) * 1000 + ms
    return ts, None


def v_price_micro(text, scale):
    """Independent decimal->micro parser. Returns (value, None) or
    (None, 'MALFORMED_PRICE_PARSE'|'MALFORMED_PRICE_PRECISION')."""
    if not re.match(r"^[0-9]+(?:\.[0-9]+)?$", text):
        return None, "MALFORMED_PRICE_PARSE"
    if "." in text:
        whole, frac = text.split(".", 1)
        if len(frac) > 6:
            return None, "MALFORMED_PRICE_PRECISION"
        value = int(whole) * scale + int((frac + "000000")[:6])
        return value, None
    return int(text) * scale, None


def v_volume(text):
    """Independent volume token parse."""
    if not re.match(r"^[0-9]+$", text):
        return None, "MALFORMED_VOLUME"
    return int(text), None


_HEADER_WORDS = ("datetime", "date", "time")


def v_is_header(first_field):
    low = first_field.lower()
    for w in _HEADER_WORDS:
        if low.startswith(w) and (len(low) == len(w) or not low[len(w)].isdigit()):
            return True
    return False


def v_parse_chunk(data, *, chunk_index, byte_start, global_line_start, cfg,
                  expect_header, certify_time_ms):
    """Independent producer-parser counterpart. Same return contract as
    engine.parse.parse_chunk."""
    grammar = cfg.get("source_grammar", "dotted_v1")
    if grammar == "tickstory_mt5":
        return v_parse_chunk_tickstory(
            data, chunk_index=chunk_index, byte_start=byte_start,
            global_line_start=global_line_start, cfg=cfg,
            expect_header=expect_header, certify_time_ms=certify_time_ms)
    from ..util import in_ts_valid_window, sha256_bytes
    from ..parse import (
        MALFORMED_EMPTY_LINE, MALFORMED_ENCODING, MALFORMED_FIELD_COUNT,
        MALFORMED_TIMESTAMP_PARSE, MALFORMED_TIMESTAMP_RANGE,
        MALFORMED_PRICE_PARSE, MALFORMED_PRICE_PRECISION,
        MALFORMED_PRICE_NONPOSITIVE, MALFORMED_BID_ASK_RELATION,
        MALFORMED_VOLUME, MALFORMED_DUPLICATE, MALFORMED_NON_MONOTONIC,
        empty_malformed_counter,
    )
    counters = empty_malformed_counter()
    malformed = []
    records = []
    rec_ord = 0
    prev_ts = None
    seen = set()
    has_header = False
    rows_parsed = 0
    global_line = global_line_start
    expected_fields = 4 if cfg["has_volume"] else 3

    for line_bytes, _term in v_split_lines(data):
        global_line += 1
        if not line_bytes:
            counters[MALFORMED_EMPTY_LINE] += 1
            continue
        rows_parsed += 1
        try:
            text = line_bytes.decode("utf-8")
        except UnicodeDecodeError:
            counters[MALFORMED_ENCODING] += 1
            malformed.append(_side(chunk_index, byte_start, global_line,
                                   MALFORMED_ENCODING, line_bytes,
                                   "vparse: undecodable"))
            continue
        if expect_header and not has_header:
            if v_is_header(text.split(",", 1)[0]):
                has_header = True
                continue
        fields = text.split(",")
        if len(fields) != expected_fields:
            counters[MALFORMED_FIELD_COUNT] += 1
            malformed.append(_side(chunk_index, byte_start, global_line,
                                   MALFORMED_FIELD_COUNT, line_bytes,
                                   "vparse: field count"))
            continue
        ts_wall, err = v_timestamp_dotted(fields[0])
        if err is not None:
            counters[MALFORMED_TIMESTAMP_PARSE] += 1
            malformed.append(_side(chunk_index, byte_start, global_line,
                                   MALFORMED_TIMESTAMP_PARSE, line_bytes, err))
            continue
        ts_ms = ts_wall - cfg["source_tz_offset_minutes"] * 60_000
        if not in_ts_valid_window(ts_ms, certify_time_ms):
            counters[MALFORMED_TIMESTAMP_RANGE] += 1
            malformed.append(_side(chunk_index, byte_start, global_line,
                                   MALFORMED_TIMESTAMP_RANGE, line_bytes,
                                   "vparse: ts out of window"))
            continue
        bid_u, be = v_price_micro(fields[1], versions.PRICE_SCALE)
        if be is not None:
            counters[be] += 1
            malformed.append(_side(chunk_index, byte_start, global_line, be,
                                   line_bytes, "vparse: bid"))
            continue
        ask_u, ae = v_price_micro(fields[2], versions.PRICE_SCALE)
        if ae is not None:
            counters[ae] += 1
            malformed.append(_side(chunk_index, byte_start, global_line, ae,
                                   line_bytes, "vparse: ask"))
            continue
        if bid_u <= 0 or ask_u <= 0:
            counters[MALFORMED_PRICE_NONPOSITIVE] += 1
            malformed.append(_side(chunk_index, byte_start, global_line,
                                   MALFORMED_PRICE_NONPOSITIVE, line_bytes,
                                   "vparse: nonpositive"))
            continue
        if ask_u < bid_u:
            counters[MALFORMED_BID_ASK_RELATION] += 1
            malformed.append(_side(chunk_index, byte_start, global_line,
                                   MALFORMED_BID_ASK_RELATION, line_bytes,
                                   "vparse: ask<bid"))
            continue
        if cfg["has_volume"]:
            vol, ve = v_volume(fields[3])
            if ve is not None:
                counters[MALFORMED_VOLUME] += 1
                malformed.append(_side(chunk_index, byte_start, global_line,
                                       MALFORMED_VOLUME, line_bytes, "vparse: vol"))
                continue
        else:
            vol = 1
        key = (ts_ms, bid_u, ask_u, vol)
        if key in seen:
            counters[MALFORMED_DUPLICATE] += 1
            malformed.append(_side(chunk_index, byte_start, global_line,
                                   MALFORMED_DUPLICATE, line_bytes,
                                   "vparse: duplicate"))
            continue
        if len(seen) < versions.DUPLICATE_TRACKER_CAP:
            seen.add(key)
        if prev_ts is not None and ts_ms < prev_ts:
            counters[MALFORMED_NON_MONOTONIC] += 1
            malformed.append(_side(chunk_index, byte_start, global_line,
                                   MALFORMED_NON_MONOTONIC, line_bytes,
                                   "vparse: regression"))
            continue
        prev_ts = ts_ms
        records.append((ts_ms, bid_u, ask_u, ask_u - bid_u, vol, global_line,
                        rec_ord))
        rec_ord += 1

    return records, malformed, counters, {
        "has_header": has_header, "rows_parsed": rows_parsed}


# ---------------------------------------------------------------------------
# Independent G_TICKSTORY_MT5 six-column parser (P3-DE-004 amendment).
# Re-implements the compact date/time grammar, the six-field layout and the
# ``last`` capture with code paths that differ from engine/parse.py. Must
# agree with the producer on every canonical row, malformed class, counter
# and preservation (last_u) value.
# ---------------------------------------------------------------------------

_CDATE_RE = re.compile(r"^(\d{4})(\d{2})(\d{2})$")
_CHMS_RE = re.compile(r"^(\d{2}):(\d{2}):(\d{2})$")


def v_compact_epoch_ms(date_text, time_text):
    """Independent compact timestamp -> (ts_ms, None) or (None, err)."""
    m = _CDATE_RE.match(date_text)
    if not m:
        return None, "vparse: date shape not recognized"
    y, mo, d = (int(g) for g in m.group(1, 2, 3))
    mt = _CHMS_RE.match(time_text)
    if not mt:
        return None, "vparse: time shape not recognized"
    hh, mi, ss = (int(g) for g in mt.group(1, 2, 3))
    if not (1 <= mo <= 12 and 1 <= d <= 31 and hh <= 23 and mi <= 59
            and ss <= 59):
        return None, "vparse: compact field out of range"
    days = v_civil_days(y, mo, d)
    return (days * 86400 + hh * 3600 + mi * 60 + ss) * 1000, None


def v_parse_chunk_tickstory(data, *, chunk_index, byte_start, global_line_start,
                            cfg, expect_header, certify_time_ms):
    """Independent six-column producer-parser counterpart.

    Same return contract as engine.parse._parse_chunk_tickstory. ``last`` is
    validated here and returned via flags["last_u"]; it is NOT part of the
    canonical record (spec 8.6).
    """
    from ..util import in_ts_valid_window, sha256_bytes
    from ..parse import (
        MALFORMED_EMPTY_LINE, MALFORMED_ENCODING, MALFORMED_FIELD_COUNT,
        MALFORMED_TIMESTAMP_PARSE, MALFORMED_TIMESTAMP_RANGE,
        MALFORMED_PRICE_PARSE, MALFORMED_PRICE_PRECISION,
        MALFORMED_PRICE_NONPOSITIVE, MALFORMED_BID_ASK_RELATION,
        MALFORMED_VOLUME, MALFORMED_DUPLICATE, MALFORMED_NON_MONOTONIC,
        empty_malformed_counter,
    )
    counters = empty_malformed_counter()
    malformed = []
    records = []
    last_values = []
    rec_ord = 0
    prev_ts = None
    seen = set()
    has_header = False
    rows_parsed = 0
    global_line = global_line_start

    for line_bytes, _term in v_split_lines(data):
        global_line += 1
        if not line_bytes:
            counters[MALFORMED_EMPTY_LINE] += 1
            continue
        rows_parsed += 1
        try:
            text = line_bytes.decode("utf-8")
        except UnicodeDecodeError:
            counters[MALFORMED_ENCODING] += 1
            malformed.append(_side(chunk_index, byte_start, global_line,
                                   MALFORMED_ENCODING, line_bytes,
                                   "vparse: undecodable"))
            continue
        if expect_header and not has_header:
            if v_is_header(text.split(",", 1)[0]):
                has_header = True
                continue
        fields = text.split(",")
        if len(fields) != 6:
            counters[MALFORMED_FIELD_COUNT] += 1
            malformed.append(_side(chunk_index, byte_start, global_line,
                                   MALFORMED_FIELD_COUNT, line_bytes,
                                   "vparse: field count != 6"))
            continue
        ts_wall, err = v_compact_epoch_ms(fields[0], fields[1])
        if err is not None:
            counters[MALFORMED_TIMESTAMP_PARSE] += 1
            malformed.append(_side(chunk_index, byte_start, global_line,
                                   MALFORMED_TIMESTAMP_PARSE, line_bytes, err))
            continue
        ts_ms = ts_wall - cfg["source_tz_offset_minutes"] * 60_000
        if not in_ts_valid_window(ts_ms, certify_time_ms):
            counters[MALFORMED_TIMESTAMP_RANGE] += 1
            malformed.append(_side(chunk_index, byte_start, global_line,
                                   MALFORMED_TIMESTAMP_RANGE, line_bytes,
                                   "vparse: ts out of window"))
            continue
        bid_u, be = v_price_micro(fields[2], versions.PRICE_SCALE)
        if be is not None:
            counters[be] += 1
            malformed.append(_side(chunk_index, byte_start, global_line, be,
                                   line_bytes, "vparse: bid"))
            continue
        ask_u, ae = v_price_micro(fields[3], versions.PRICE_SCALE)
        if ae is not None:
            counters[ae] += 1
            malformed.append(_side(chunk_index, byte_start, global_line, ae,
                                   line_bytes, "vparse: ask"))
            continue
        last_u, le = v_price_micro(fields[4], versions.PRICE_SCALE)
        if le is not None:
            counters[le] += 1
            malformed.append(_side(chunk_index, byte_start, global_line, le,
                                   line_bytes, "vparse: last"))
            continue
        if bid_u <= 0 or ask_u <= 0 or last_u <= 0:
            counters[MALFORMED_PRICE_NONPOSITIVE] += 1
            malformed.append(_side(chunk_index, byte_start, global_line,
                                   MALFORMED_PRICE_NONPOSITIVE, line_bytes,
                                   "vparse: nonpositive price"))
            continue
        if ask_u < bid_u:
            counters[MALFORMED_BID_ASK_RELATION] += 1
            malformed.append(_side(chunk_index, byte_start, global_line,
                                   MALFORMED_BID_ASK_RELATION, line_bytes,
                                   "vparse: ask<bid"))
            continue
        vol, ve = v_volume(fields[5])
        if ve is not None:
            counters[MALFORMED_VOLUME] += 1
            malformed.append(_side(chunk_index, byte_start, global_line,
                                   MALFORMED_VOLUME, line_bytes, "vparse: vol"))
            continue
        key = (ts_ms, bid_u, ask_u, vol)
        if key in seen:
            counters[MALFORMED_DUPLICATE] += 1
            malformed.append(_side(chunk_index, byte_start, global_line,
                                   MALFORMED_DUPLICATE, line_bytes,
                                   "vparse: duplicate"))
            continue
        if len(seen) < versions.DUPLICATE_TRACKER_CAP:
            seen.add(key)
        if prev_ts is not None and ts_ms < prev_ts:
            counters[MALFORMED_NON_MONOTONIC] += 1
            malformed.append(_side(chunk_index, byte_start, global_line,
                                   MALFORMED_NON_MONOTONIC, line_bytes,
                                   "vparse: regression"))
            continue
        prev_ts = ts_ms
        records.append((ts_ms, bid_u, ask_u, ask_u - bid_u, vol, global_line,
                        rec_ord))
        last_values.append(last_u)
        rec_ord += 1

    flags = {"has_header": has_header, "rows_parsed": rows_parsed,
             "last_u": last_values}
    return records, malformed, counters, flags


def v_preservation_lines(records, last_values):
    """Independent preservation serialization bytes ``src_line|last_u\\n``
    for the canonical rows in rec_ord order (spec 8.6)."""
    out = []
    for rec, last_u in zip(records, last_values):
        out.append(("%d|%d\n" % (rec[5], last_u)).encode("utf-8"))
    return b"".join(out)
        


def _side(chunk_index, byte_start, global_line, cls, raw, reason):
    from ..util import sha256_bytes
    return {
        "chunk_index": chunk_index,
        "byte_start": byte_start,
        "global_line_no": global_line,
        "class": cls,
        "raw_sha256": sha256_bytes(raw),
        "reason": reason,
    }