import asyncio
import os
import sys
from typing import List, NamedTuple

import pytest

from clarabot_nano.indexer.extractor.base import (
    ALL_SPACE_SEARCH_RE,
    SPACE_SEARCH_RE,
    NEWLINE_SEARCH_RE,
    NEWLINE_STRIP_RE,
    ServerInternalExtractError,
    RetryExtractError,
    ExtractorInoperable,
)
from clarabot_nano_tests.indexer.extractor.resources import INPUT_ROOT, EXTRACT_ROOT, OUTPUT_ROOT, SAMPLES_ROOT
from clarabot_nano_tests.indexer.extractor.utility import ExtractorTesterBase, diff_extracts

HEAD_INDEX = 1


def list_path(t: str) -> List[int]:
    p = []
    for x in t.split("/"):
        if not x:
            pass
        elif x == "*":
            p.append(None)
        elif x.startswith("*[") and x.endswith("]"):
            p.append(int(x[2:-1]))
        else:
            assert False, t
    return p


def paths_match(test_path, check_path):
    if len(check_path) < len(test_path):
        return False
    for tp, cp in zip(test_path, check_path):
        if tp is None:
            if cp is not None:
                return False
        elif tp == -1:
            if cp is None:
                return False
        elif tp != cp:
            return False
    return True


def _move_head_node_around(d: List[NamedTuple]) -> int:
    if d[0].__class__.__name__ != "MoveNode":
        return 0
    if not paths_match([None, HEAD_INDEX, -1], list_path(d[0].node)):
        return 0
    if not paths_match([None, HEAD_INDEX], list_path(d[0].target)):
        return 0
    return 1


def _extra_meta_content_type_override_win(d: List[NamedTuple]) -> int:
    if d[0].__class__.__name__ != "InsertNode" or not d[0].tag.endswith("}meta"):
        return 0
    p0 = list_path(d[0].target)
    if not paths_match([None, HEAD_INDEX], p0):
        return 0
    attr_path = [*p0, d[0].position + 1]
    attrs = {}
    for i in range(1, 3):
        if d[i].__class__.__name__ != "InsertAttrib" or not paths_match(attr_path, list_path(d[i].node)):
            return 0
        attrs[d[i].name] = d[i].value
    if (
        set(attrs.keys()) != {"name", "content"}
        or attrs["name"] != "Content-Type-Override"
        or not (
            attrs["content"].startswith("application/vnd.openxmlformats-officedocument.")
            or attrs["content"].startswith("application/vnd.ms-")
        )
    ):
        return 0
    return 3


EXTRACT_DIFFERENCE_ACCEPTORS = [_move_head_node_around]

if sys.platform.startswith("win"):
    EXTRACT_DIFFERENCE_ACCEPTORS.append(_extra_meta_content_type_override_win)


def is_extract_difference_acceptable(diff: List[NamedTuple]):
    d = list(diff)
    while d:
        accepted = 0
        for acceptor in EXTRACT_DIFFERENCE_ACCEPTORS:
            accepted = acceptor(d)
            if accepted > 0:
                d = d[accepted:]
                break
        if accepted == 0:
            return False
    return True


def test_indexer_extractor_regexps():
    assert ALL_SPACE_SEARCH_RE.sub("", "  \xa0 a b \n c\nd  \r\n\t\r  \n  e ") == "abcde"
    assert SPACE_SEARCH_RE.sub("", "  a b \n c\nd  e ") == "ab\nc\nde"
    assert SPACE_SEARCH_RE.sub("", "  \xa0 a   b \n   c\nd  \xa0\t e\tx ") == "ab\nc\ndex"
    assert NEWLINE_SEARCH_RE.sub("", "a\nb\r\nc\rd\r\re\n\nf \r\n\r\r\n\r\n\n\n\r\r\ng") == "abcdef g"
    assert NEWLINE_STRIP_RE.sub("\n", "\r\n \t\n a\n \t b \r\n\t\r  \n c \n  \t") == "\na\nb\nc\n"


async def _check_extract_output(indexer_extractor_tester, ip, ep, op):
    extractor = indexer_extractor_tester.create_extractor()
    # Limit is reduced to approx because of final normalization.
    approx_length_limit = indexer_extractor_tester.ec.indexer_extract_text_limit * 0.9
    for in_path, ex_dir, out_dir in indexer_extractor_tester.walk_resources(ip, ep, op):
        # for debugging
        # if in_path.name != "tf33478217_win32.dotx":
        #     continue
        assert in_path.is_file(), f"{in_path}"
        attempt = 0
        while True:
            try:
                extraction = await extractor.extract(in_path)
            except ServerInternalExtractError as e:
                extraction = e.extraction
            except ExtractorInoperable:
                assert False, f"{in_path}"
            except RetryExtractError:  # OOM may crash the worker
                attempt += 1
                if attempt <= 2:
                    await asyncio.sleep(0.2)
                else:
                    raise
                continue
            break
        # verify/write extracts
        extracts = [extract.replace(b"\r\n", b"\n") for extract in extraction.get_extracts()]  # fix newlines
        extracts_in_dir = indexer_extractor_tester.read_dir_files(ex_dir)
        if extracts_in_dir is not None:
            assert len(extracts) == len(extracts_in_dir), f"{in_path}"
            for i, (extract, extract_in_dir) in enumerate(zip(extracts, extracts_in_dir)):
                ex_diff = diff_extracts(extract, extract_in_dir)
                if not is_extract_difference_acceptable(ex_diff):
                    raise AssertionError(
                        f"extracts mismatch {i} {in_path}:\n  {extract}\n  {extract_in_dir}\n  {ex_diff}\n  xy"
                    )
        elif extracts:
            ex_dir.mkdir(parents=True)
            for i, extract in enumerate(extracts):
                (ex_dir / f"{i}.xml").write_bytes(extract)
        # verify/write outputs
        outputs = extraction.get_outputs()
        outputs_in_dir = indexer_extractor_tester.read_dir_files(out_dir, raw=False)
        if outputs_in_dir is not None:
            assert len(outputs) == len(outputs_in_dir), f"{in_path}"
            for generated, reference in zip(outputs, outputs_in_dir):
                if len(generated) != len(reference):
                    # Tika extractor yields fragments in different increments on different platforms and maybe configs.
                    # So the end of an extract cut may be different, but match up to the existing values.
                    shorter_length = min(len(generated), len(reference))
                    assert approx_length_limit < shorter_length, f"{in_path}\n{approx_length_limit}\n{shorter_length}"
                    assert generated[:shorter_length] == reference[:shorter_length], f"{in_path}"
                else:
                    assert generated == reference, f"{in_path}"
        else:
            out_dir.mkdir(parents=True)
            for i, output in enumerate(outputs):
                (out_dir / f"{i}.txt").write_text(output, encoding="utf-8")


# NOTE: System locale affects date, currency and some number formatting. The references are made with English/US.
@pytest.mark.asyncio
async def test_indexer_extractor_included(indexer_extractor_tester: ExtractorTesterBase):
    await _check_extract_output(indexer_extractor_tester, INPUT_ROOT, EXTRACT_ROOT, OUTPUT_ROOT)


@pytest.mark.skipif(not os.path.exists(SAMPLES_ROOT), reason="can't find nano-sample-files repo")
# @pytest.mark.parametrize("run", list(range(50)))
@pytest.mark.asyncio
async def test_indexer_extractor_external(indexer_extractor_tester: ExtractorTesterBase):
    await _check_extract_output(
        indexer_extractor_tester,
        os.path.join(SAMPLES_ROOT, "input"),
        os.path.join(SAMPLES_ROOT, "extract"),
        os.path.join(SAMPLES_ROOT, "output"),
    )
