"""Подготовка корпуса к индексации: цена, границы и слепые зоны.

Зачем отдельная операция. До неё вопрос «сколько это будет стоить и что не
покроется» задавался уже ПОСЛЕ индексации: время узнавали по факту, а слепые
зоны (динамические вызовы) — по кривым ответам. `preflight` отвечает до того,
как что-либо создано, и ничего не создаёт сам: ни `.ragraf`, ни каталога, ни
файла. Это проверяется свойством набора, а не обещанием.

Что отвечает и чем:

  * распределение по расширениям и число файлов — обход с отсечением каталогов;
  * доля сгенерированного кода — маркеры в первых килобайтах файла;
  * миграции — по путям и именам файлов;
  * ожидаемое время разбора — ЗАМЕР на выборке тем же парсером, что у индексатора,
    и экстраполяция с названной погрешностью (расхождение половин выборки);
  * дыры динамики — по языкам и шаблонам, с адресом `путь:строка`;
  * готовность векторов и Joern — по факту наличия библиотеки, модели в кэше,
    Java и скриптов Joern;
  * эффект исключений — сколько файлов не попадёт в корпус.

Чего НЕ делает: не индексирует, не пишет, не решает за человека. Числа, которые
нельзя измерить (нет парсера, нет модели), называются `unknown` с причиной, а не
нулём: ноль и «нечем измерить» — разные состояния.

Автор: Mikhail Deynekin
Сайт: https://Deynekin.com
Почта: Mikhail@Deynekin.com
"""

from __future__ import annotations

import os
import shutil
import time
from pathlib import Path
from typing import Any

from . import util

#: Пределы. Полное число всегда возвращается рядом с усечённым списком.
MAX_FILES_WALKED = 200_000
MAX_FILES_READ = 5_000
MAX_FINDINGS = 200
HEAD_BYTES = 4096

#: Маркеры сгенерированного кода. Проверяются в первых `HEAD_BYTES` файла.
GENERATED_MARKERS = (
    "auto-generated", "autogenerated", "automatically generated",
    "do not edit", "do not modify", "code generated by", "@generated",
    "сгенерирован автоматически", "не редактировать",
)

#: Признаки миграций: имена каталогов и частей пути.
MIGRATION_HINTS = ("migration", "migrations", "migrate", "миграции", "db/migrate")

#: Дыры динамики по языкам: (вид, регулярное выражение).
#:
#: Почему это важно и почему списком. Статический разбор не видит вызов, имя
#: которого собирается во время выполнения: `$obj->$method()`,
#: `call_user_func`, `getattr(obj, name)`, `eval`. Такие места — не дефект кода,
#: а ГРАНИЦА метода, и `preflight` обязан назвать их до начала работы, иначе
#: «в графе нет вызова» читается как «вызова нет в коде».
DYNAMIC_PATTERNS: dict[str, tuple[tuple[str, str], ...]] = {
    "php": (
        ("call_user_func", r"call_user_func(?:_array)?\s*\("),
        ("variable_method", r"\$[A-Za-z_]\w*\s*->\s*\$[A-Za-z_]\w*\s*\("),
        # Имя метода в фигурных скобках — тот же случай, что `$obj->$method`, и
        # встречается чаще: `$this->{$name}()`. Прежний шаблон его не ловил,
        # поэтому «динамических вызовов нет» могло быть ложью.
        ("variable_method", r"\$[A-Za-z_]\w*\s*->\s*\{[^}\n]*\}\s*\("),
        ("variable_class", r"new\s+\$[A-Za-z_]\w*"),
        ("eval", r"\beval\s*\("),
        ("reflection", r"new\s+Reflection(?:Class|Method|Function)"),
    ),
    "python": (
        ("getattr", r"\bgetattr\s*\("),
        ("eval", r"\beval\s*\("),
        ("exec", r"\bexec\s*\("),
        ("import_module", r"\bimport_module\s*\(|\b__import__\s*\("),
        ("reflection", r"\bglobals\s*\(\s*\)\s*\[|\bsetattr\s*\("),
    ),
    "javascript": (
        ("eval", r"\beval\s*\("),
        ("new_function", r"new\s+Function\s*\("),
        ("bracket_call", r"\w+\s*\[\s*['\"][A-Za-z_$][\w$]*['\"]\s*\]\s*\("),
    ),
    "typescript": (
        ("eval", r"\beval\s*\("),
        ("new_function", r"new\s+Function\s*\("),
        ("bracket_call", r"\w+\s*\[\s*['\"][A-Za-z_$][\w$]*['\"]\s*\]\s*\("),
    ),
    "csharp": (
        ("reflection", r"GetType\s*\(\s*\)|typeof\s*\(\s*\w+\s*\)\s*\.\s*GetMethod"),
        ("dynamic", r"\bdynamic\s+\w+"),
        ("invoke_member", r"InvokeMember\s*\("),
    ),
    "java": (
        ("reflection", r"\.getMethod\s*\(|\.getDeclaredMethod\s*\(|Class\.forName\s*\("),
    ),
}

#: Шаги обхода, которые не считаются: каталоги, которые не станут корпусом.
_WALK_SKIP_DIRS = (".git", ".ragraf", ".zwork", "node_modules", "vendor",
                   "__pycache__", ".venv", "venv", ".idea", ".vscode")


def _lang_of(path: Path) -> str:
    from . import config

    return str(config.LANG_BY_EXT.get(path.suffix.lower(), ""))


def _walk(root: Path, *, skip_globs: tuple[str, ...] = ()) -> tuple[list[Path], dict[str, Any]]:
    """Обход каталога без исключённых подкаталогов. Отдаёт файлы и статистику.

    Обход ручной, а не `rglob`: исключённые каталоги надо пропускать ЦЕЛИКОМ,
    иначе `node_modules` с десятками тысяч файлов съедает минуты (измерено на
    реальном проекте и записано в `init_corpus.inspect_project`).
    """
    files: list[Path] = []
    skipped_dirs: dict[str, int] = {}
    skipped_by_rule = 0
    stack = [root]
    visited = 0
    while stack:
        current = stack.pop()
        try:
            entries = sorted(current.iterdir())
        except OSError:
            continue
        for entry in entries:
            visited += 1
            if visited > MAX_FILES_WALKED:
                break
            try:
                if entry.is_dir():
                    if entry.name in _WALK_SKIP_DIRS:
                        skipped_dirs[entry.name] = skipped_dirs.get(entry.name, 0) + 1
                        continue
                    stack.append(entry)
                    continue
                if not entry.is_file():
                    continue
            except OSError:
                continue
            relative = util.to_posix(entry.relative_to(root))
            if any(util.glob_match(relative, pattern) for pattern in skip_globs):
                skipped_by_rule += 1
                continue
            files.append(entry)
    files.sort(key=lambda item: util.to_posix(item))
    return files, {"skipped_dirs": skipped_dirs, "skipped_by_rule": skipped_by_rule,
                   "truncated": visited > MAX_FILES_WALKED}


def _read_head(path: Path) -> str:
    try:
        with path.open("rb") as handle:
            raw = handle.read(HEAD_BYTES)
    except OSError:
        return ""
    return raw.decode("utf-8", errors="replace")


def _findings_in(text: str, lang: str) -> list[tuple[str, int, str]]:
    """Динамические места в тексте: (вид, строка, короткая цитата)."""
    import re

    patterns = DYNAMIC_PATTERNS.get(lang) or ()
    if not patterns:
        return []
    out: list[tuple[str, int, str]] = []
    for kind, pattern in patterns:
        try:
            compiled = re.compile(pattern)
        except re.error:  # noqa: BLE001 — шаблон в коде, а не во входе
            continue
        for match in compiled.finditer(text):
            line = text.count("\n", 0, match.start()) + 1
            start = max(0, text.rfind("\n", 0, match.start()) + 1)
            end = text.find("\n", match.end())
            quote = " ".join(text[start:end if end != -1 else len(text)].split())
            out.append((kind, line, quote[:160]))
            if len(out) >= MAX_FINDINGS * 4:
                return out
    return out


def _parse_sample(root: Path, files: list[Path], sample: int) -> dict[str, Any]:
    """Замерить разбор выборки тем же парсером, что у индексатора.

    Выборка детерминирована (равномерный шаг по отсортированному списку), иначе
    два прогона на одном каталоге давали бы разные оценки — а сравнивать оценки
    между проектами иначе нельзя.

    Погрешность — расхождение половин выборки, а не «плюс-минус на глаз»: если
    файлы разной величины, половины покажут это числом. При выборке, равной
    всему набору, погрешности нет по определению.
    """
    from . import config

    parseable = [path for path in files
                 if path.suffix.lower() in config.PARSE_LANGS_EXT]
    total = len(parseable)
    out: dict[str, Any] = {
        "parseable_files": total,
        "sample_requested": int(sample),
        "sample_files": 0,
        "sample_bytes": 0,
        "sample_ms": 0.0,
        "per_file_ms": None,
        "estimated_total_ms": None,
        "error_promille": None,
        "method": "tree_sitter на выборке, экстраполяция по числу файлов",
        "available": False,
        "reason": "",
    }
    if total == 0:
        out["reason"] = "файлов с разбором деревом нет: оценивать нечего"
        return out
    try:
        from tree_sitter_language_pack import get_parser  # type: ignore
    except Exception as exc:  # noqa: BLE001 — причина называется
        out["reason"] = f"парсер деревом недоступен ({type(exc).__name__}): оценить время нечем"
        return out

    step = max(1, total // max(1, int(sample)))
    chosen = parseable[::step][:max(1, int(sample))]
    parsers: dict[str, Any] = {}
    durations: list[float] = []
    used_bytes = 0
    used = 0
    for path in chosen:
        lang = str(config.PARSE_LANGS_EXT.get(path.suffix.lower(), ""))
        if not lang:
            continue
        try:
            parser = parsers.get(lang)
            if parser is None:
                parser = get_parser(lang)
                parsers[lang] = parser
            data = path.read_bytes()
            started = time.perf_counter()
            parser.parse(data)
            durations.append((time.perf_counter() - started) * 1000.0)
            used += 1
            used_bytes += len(data)
        except Exception:  # noqa: BLE001 — один файл не отменяет замер
            continue
    if not durations:
        out["reason"] = ("выборка не разобрана ни одним парсером: оценить время нечем "
                         "(файлы могли оказаться не того языка)")
        return out

    total_ms = sum(durations)
    per_file = total_ms / len(durations)
    out.update({
        "available": True,
        "sample_files": used,
        "sample_bytes": used_bytes,
        "sample_ms": round(total_ms, 3),
        "per_file_ms": round(per_file, 4),
        # Меньше миллисекунды — показываем 1: ноль читается как «не измерено»,
        # а состояния «нечем измерить» и «измерено, вышло мало» разные.
        "estimated_total_ms": max(1, int(round(per_file * total))),
    })
    if used >= total:
        out["error_promille"] = 0
        out["method"] = ("tree_sitter по всем файлам с разбором: это замер, "
                         "а не экстраполяция")
    else:
        half = max(1, len(durations) // 2)
        first = sum(durations[:half]) / half
        second = sum(durations[half:]) / max(1, len(durations) - half)
        mean = (first + second) / 2 or 1e-9
        out["error_promille"] = int(round(abs(first - second) / mean * 1000))
        out["half_ms"] = [round(first, 4), round(second, 4)]
    out["extra_promille"] = int(round((total - used) * 1000 / max(1, total)))
    out["reason"] = (f"замер на {used} файлах из {total} с разбором деревом; "
                     f"погрешность — расхождение половин выборки")
    return out


def _vectors_readiness() -> dict[str, Any]:
    """Готовность векторов — без создания каталогов и без загрузки модели."""
    out: dict[str, Any] = {"available": False, "library": False, "model_cached": None,
                           "cache_dir": "", "reason": ""}
    override = os.environ.get("RAGRAF_MODEL_CACHE")
    base = Path(override) if override else (Path.home() / ".ragraf" / "models")
    out["cache_dir"] = util.to_posix(base)
    try:
        import fastembed  # noqa: F401
    except Exception as exc:  # noqa: BLE001 — причина называется
        out["reason"] = (f"библиотеки векторов нет ({type(exc).__name__}): "
                         f"поиск пойдёт только лексическим путём")
        return out
    out["library"] = True
    try:
        entries = [item for item in base.iterdir()
                   if item.is_dir() and any(item.iterdir())] if base.exists() else []
    except OSError as exc:
        out["reason"] = f"каталог моделей не читается ({type(exc).__name__})"
        return out
    out["model_cached"] = bool(entries)
    if entries:
        out["available"] = True
        out["reason"] = f"моделей в кэше: {len(entries)}"
    else:
        out["reason"] = ("библиотека есть, но моделей в кэше нет: первая индексация "
                         "скачает её из сети")
    return out


def _joern_readiness() -> dict[str, Any]:
    """Готовность Joern: скрипты и Java. Ничего не запускает."""
    out: dict[str, Any] = {"available": False, "scripts": [], "java": False,
                           "home": "", "reason": ""}
    candidates = []
    env = os.environ.get("RAGRAF_JOERN_HOME")
    if env:
        candidates.append(Path(env))
    tools = Path(__file__).resolve().parents[2] / ".tools"
    candidates.extend([tools / "joern-cli", tools / "joern"])
    for base in candidates:
        if not base.exists():
            continue
        scripts = sorted(str(item.name) for item in base.iterdir()
                         if item.is_file() and item.suffix.lower() in (".bat", ".sh"))
        if scripts:
            out["home"] = util.to_posix(base)
            out["scripts"] = scripts[:10]
            break
    java = shutil.which("java")
    out["java"] = bool(java)
    if out["scripts"] and java:
        out["available"] = True
        out["reason"] = "скрипты Joern и Java найдены"
    elif not out["scripts"] and not java:
        out["reason"] = ("ни скриптов Joern, ни Java: рёбра вызовов будут выведены "
                         "по имени, режим останется partial")
    elif not out["scripts"]:
        out["reason"] = "Java есть, скриптов Joern нет: граф свойств кода не построить"
    else:
        out["reason"] = "скрипты Joern есть, Java не найдена: CPG не запустится"
    return out


def scan(root: str | Path, *, sample: int = 50,
         skip_globs: tuple[str, ...] = ()) -> dict[str, Any]:
    """Осмотреть каталог и ответить, сколько будет стоить индекс и что не покроется.

    Ничего не пишет и не создаёт. `skip_globs` — правила исключения конфига,
    если он уже есть: эффект исключений измеряется, а не обещается.
    """
    target = Path(str(root))
    report: dict[str, Any] = {
        "root": util.to_posix(target),
        "exists": target.is_dir(),
        "files": {"total": 0, "bytes": 0, "by_ext": {}, "by_language": {},
                  "read": 0},
        "generated": {"count": 0, "share_promille": 0, "read": 0, "samples": [],
                      "reason": ""},
        "migrations": {"files": 0, "dirs": [], "samples": []},
        "timing": {"available": False, "reason": "каталог не осмотрен"},
        "dynamic": {"count": 0, "by_kind": {}, "by_language": {}, "findings": [],
                    "shown": 0, "languages_covered": []},
        "readiness": {},
        "exclusions": {"skipped_dirs": {}, "skipped_by_rule": 0, "considered": 0},
        "limits": [
            "оценка времени — замер на выборке: это не гарантия, а измеренная цена",
            "динамические вызовы считаются по шаблонам языков, а не полным разбором",
        ],
    }
    if not target.is_dir():
        report["reason"] = f"каталог не найден: {util.to_posix(target)}"
        report["timing"] = {"available": False,
                            "reason": "каталог не найден: осматривать нечего"}
        return report

    from . import config

    files, walk_stats = _walk(target, skip_globs=skip_globs)
    report["exclusions"] = {
        "skipped_dirs": dict(sorted(walk_stats["skipped_dirs"].items())),
        "skipped_by_rule": int(walk_stats["skipped_by_rule"]),
        "considered": len(files),
        "walk_truncated": bool(walk_stats["truncated"]),
    }

    by_ext: dict[str, int] = {}
    by_language: dict[str, int] = {}
    total_bytes = 0
    code_files: list[Path] = []
    for path in files:
        ext = path.suffix.lower()
        by_ext[ext] = by_ext.get(ext, 0) + 1
        lang = _lang_of(path)
        if lang:
            by_language[lang] = by_language.get(lang, 0) + 1
            code_files.append(path)
        try:
            total_bytes += path.stat().st_size
        except OSError:
            continue
    report["files"].update({
        "total": len(files),
        "bytes": total_bytes,
        "by_ext": dict(sorted(by_ext.items(), key=lambda item: (-item[1], item[0]))[:40]),
        "by_language": dict(sorted(by_language.items(), key=lambda item: (-item[1], item[0]))),
        "code_files": len(code_files),
    })

    # --- маркеры сгенерированного кода и динамические места -----------------
    read_limit = min(len(code_files), MAX_FILES_READ)
    generated = 0
    samples: list[dict[str, Any]] = []
    dynamic_findings: list[dict[str, Any]] = []
    by_kind: dict[str, int] = {}
    by_lang_dyn: dict[str, int] = {}
    covered: set[str] = set()
    for path in code_files[:read_limit]:
        lang = _lang_of(path)
        text = _read_head(path)
        if not text:
            continue
        lowered = text.lower()
        if any(marker in lowered for marker in GENERATED_MARKERS):
            generated += 1
            if len(samples) < 5:
                samples.append({"path": util.to_posix(path.relative_to(target))})
        if not lang or lang not in DYNAMIC_PATTERNS:
            continue
        covered.add(lang)
        hits = _findings_in(text, lang)
        if hits:
            by_lang_dyn[lang] = by_lang_dyn.get(lang, 0) + len(hits)
        for kind, line, quote in hits:
            by_kind[kind] = by_kind.get(kind, 0) + 1
            if len(dynamic_findings) < MAX_FINDINGS:
                dynamic_findings.append({
                    "path": util.to_posix(path.relative_to(target)),
                    "line": line,
                    "kind": kind,
                    "language": lang,
                    "quote": quote,
                    "address": f"{util.to_posix(path.relative_to(target))}:{line}",
                })
    report["files"]["read"] = read_limit
    report["generated"].update({
        "count": generated,
        "read": read_limit,
        "share_promille": int(round(generated * 1000 / read_limit)) if read_limit else 0,
        "samples": samples,
        "reason": (f"маркеры искались в первых {HEAD_BYTES} байтах {read_limit} файлов "
                   f"из {len(code_files)}; это нижняя оценка"),
    })
    dynamic_findings.sort(key=lambda item: (item["path"], item["line"], item["kind"]))
    report["dynamic"].update({
        "count": sum(by_kind.values()),
        "by_kind": dict(sorted(by_kind.items(), key=lambda item: (-item[1], item[0]))),
        "by_language": dict(sorted(by_lang_dyn.items(), key=lambda item: (-item[1], item[0]))),
        "findings": dynamic_findings,
        "shown": len(dynamic_findings),
        "languages_covered": sorted(covered),
        "languages_without_patterns": sorted(set(by_language) - set(DYNAMIC_PATTERNS)),
    })
    if len(dynamic_findings) >= MAX_FINDINGS:
        report["limits"].append(
            f"мест динамики больше {MAX_FINDINGS}: список усечён, полное число — в "
            f"`dynamic.count`")

    # --- миграции ----------------------------------------------------------
    migration_dirs: dict[str, int] = {}
    migration_samples: list[str] = []
    migration_files = 0
    for path in files:
        relative = util.to_posix(path.relative_to(target))
        lowered = relative.lower()
        if any(hint in lowered for hint in MIGRATION_HINTS):
            migration_files += 1
            parts = relative.split("/")
            for part in parts[:-1]:
                if any(hint in part.lower() for hint in MIGRATION_HINTS):
                    migration_dirs[part] = migration_dirs.get(part, 0) + 1
            if len(migration_samples) < 5:
                migration_samples.append(relative)
    report["migrations"] = {
        "files": migration_files,
        "dirs": sorted(migration_dirs),
        "samples": migration_samples,
        "reason": ("миграции ищутся по путям и именам (migration/migrate/миграции); "
                   "содержимое SQL не разбирается"),
    }

    # --- время, готовность ---------------------------------------------------
    report["timing"] = _parse_sample(target, code_files, sample)
    report["readiness"] = {"vectors": _vectors_readiness(), "joern": _joern_readiness()}
    return report