"""Batch orchestration and aggregation helpers."""

from __future__ import annotations

import json
import os
from collections import OrderedDict, defaultdict
from dataclasses import dataclass
from datetime import date, datetime, timedelta
from typing import Iterable

from cleaner import compute_stats

DATE_FMT = "%Y-%m-%d"
MONTH_FMT = "%Y-%m"


@dataclass(frozen=True)
class BatchWindow:
    start: datetime
    end: datetime
    index: int
    chunk_days: int

    @property
    def month_key(self) -> str:
        return self.start.strftime(MONTH_FMT)

    @property
    def window_id(self) -> str:
        if self.chunk_days == 1:
            return self.start.strftime(DATE_FMT)
        return f"{self.start.strftime(DATE_FMT)}_to_{self.end.strftime(DATE_FMT)}"

    @property
    def scope_desc(self) -> str:
        return f"{self.start.strftime(DATE_FMT)} 至 {self.end.strftime(DATE_FMT)} 分片"


def parse_window_date(value: str) -> datetime:
    return datetime.strptime(value, DATE_FMT)


def build_chunk_windows(since_date: datetime, until_date: datetime, chunk_days: int) -> list[BatchWindow]:
    if chunk_days <= 0:
        raise ValueError("chunk_days must be positive")
    if since_date >= until_date:
        raise ValueError("since_date must be earlier than until_date")

    windows: list[BatchWindow] = []
    cursor = since_date
    index = 0
    while cursor < until_date:
        month_boundary = (cursor.replace(day=1) + timedelta(days=32)).replace(day=1)
        window_end = min(cursor + timedelta(days=chunk_days), until_date, month_boundary)
        windows.append(BatchWindow(start=cursor, end=window_end, index=index, chunk_days=chunk_days))
        cursor = window_end
        index += 1
    return windows


def month_key_for_window(window: BatchWindow) -> str:
    return window.month_key


def batch_root_paths(batch_root: str, month_key: str) -> dict:
    base = batch_root or os.path.join("outputs", "inquiries")
    return {
        "base": base,
        "raw_dir": os.path.join(base, "raw", month_key),
        "manifests_dir": os.path.join(base, "manifests"),
        "monthly_dir": os.path.join(base, "monthly"),
        "manifest_path": os.path.join(base, "manifests", f"{month_key}.json"),
        "monthly_path": os.path.join(base, "monthly", f"{month_key}.json"),
    }


def raw_shard_path(batch_root: str, window: BatchWindow) -> str:
    paths = batch_root_paths(batch_root, window.month_key)
    return os.path.join(paths["raw_dir"], f"{window.window_id}.json")


def ensure_parent_dir(path: str) -> None:
    os.makedirs(os.path.dirname(path), exist_ok=True)


def write_json_atomic(path: str, payload) -> None:
    ensure_parent_dir(path)
    if isinstance(payload, str):
        text = payload
    else:
        text = json.dumps(payload, ensure_ascii=False, indent=2)

    tmp_path = f"{path}.tmp"
    with open(tmp_path, "w", encoding="utf-8") as handle:
        handle.write(text)
        if not text.endswith("\n"):
            handle.write("\n")
        handle.flush()
        os.fsync(handle.fileno())
    os.replace(tmp_path, path)


def read_json(path: str):
    with open(path, encoding="utf-8") as handle:
        return json.load(handle)


def _parse_message_time(message: dict) -> datetime:
    raw = str(message.get("timestamp") or "")
    for fmt in ("%Y-%m-%d %H:%M", "%Y-%m-%d %H:%M:%S"):
        try:
            return datetime.strptime(raw, fmt)
        except (TypeError, ValueError):
            continue
    return datetime.min


def _message_identity(message: dict):
    message_id = str(
        message.get("messageId")
        or message.get("message_id")
        or message.get("id")
        or ""
    ).strip()
    if message_id:
        return ("id", message_id)

    text = " ".join(str(message.get("text") or "").split())
    return (
        "fallback",
        str(message.get("timestamp") or ""),
        str(message.get("speaker") or ""),
        text,
    )


def _message_sort_key(message: dict):
    return (
        _parse_message_time(message),
        str(message.get("speaker") or ""),
        str(message.get("text") or ""),
        str(message.get("messageId") or message.get("message_id") or message.get("id") or ""),
    )


def dedupe_messages(messages: Iterable[dict]) -> list[dict]:
    seen = set()
    kept: list[tuple[int, dict]] = []
    for index, message in enumerate(messages):
        key = _message_identity(message)
        if key in seen:
            continue
        seen.add(key)
        kept.append((index, message))

    kept.sort(key=lambda item: (_message_sort_key(item[1]), item[0]))
    return [message for _, message in kept]


def _detail_score(detail: dict) -> tuple:
    messages = detail.get("messages") or []
    return (
        len(messages),
        1 if detail.get("stats") else 0,
        int(detail.get("total_loaded") or 0),
    )


def merge_detail_records(details: Iterable[dict]) -> dict | None:
    detail_list = [detail for detail in details if detail]
    if not detail_list:
        return None

    representative = max(detail_list, key=_detail_score)
    merged_messages = dedupe_messages(
        message
        for detail in detail_list
        for message in (detail.get("messages") or [])
    )

    merged_detail = dict(representative)
    merged_detail["messages"] = merged_messages
    merged_detail["stats"] = compute_stats(merged_messages)
    merged_detail["total_loaded"] = max(int(detail.get("total_loaded") or 0) for detail in detail_list)
    return merged_detail


def _record_score(record: dict) -> tuple:
    detail = record.get("detail") or {}
    messages = detail.get("messages") or []
    completeness = sum(
        1
        for key in ("buyer_name", "product", "country", "assignee", "update_time", "create_time")
        if record.get(key)
    )
    return (
        1 if detail else 0,
        len(messages),
        completeness,
    )


def merge_inquiry_records(records: list[dict]) -> tuple[dict, int]:
    if not records:
        return {}, 0

    representative = max(records, key=_record_score)
    merged = dict(representative)
    merged["inquiry_id"] = representative.get("inquiry_id")

    for key in ("detail_url", "buyer_name", "product", "last_message", "category", "assignee", "country", "user_level_icon", "update_time", "create_time", "user_level"):
        if not merged.get(key):
            for record in records:
                value = record.get(key)
                if value:
                    merged[key] = value
                    break

    merged_detail = merge_detail_records(record.get("detail") for record in records)
    if merged_detail:
        merged["detail"] = merged_detail
        merged["detail_skipped"] = False
        merged["scrape_status"] = "success"
        merged["first_reply_time"] = merged_detail.get("stats", {}).get("first_seller_response")
    else:
        merged["detail"] = representative.get("detail")
        merged["detail_skipped"] = all(record.get("detail_skipped") for record in records)
        merged["scrape_status"] = representative.get("scrape_status", "pending")
        merged["first_reply_time"] = representative.get("first_reply_time")

    return merged, len(records)


def _inquiry_sort_key(record: dict):
    raw_time = record.get("update_time") or record.get("create_time") or ""
    for fmt in ("%Y-%m-%d %H:%M", "%Y-%m-%d"):
        try:
            return datetime.strptime(raw_time, fmt)
        except (TypeError, ValueError):
            continue
    return datetime.min


def merge_inquiries_by_id(inquiries: Iterable[dict]) -> tuple[list[dict], dict]:
    grouped: "OrderedDict[str, list[dict]]" = OrderedDict()
    for inquiry in inquiries:
        inquiry_id = str(inquiry.get("inquiry_id") or "")
        if not inquiry_id:
            continue
        grouped.setdefault(inquiry_id, []).append(inquiry)

    merged: list[dict] = []
    conflict_count = 0
    for inquiry_id, records in grouped.items():
        merged_record, record_count = merge_inquiry_records(records)
        merged.append(merged_record)
        if record_count > 1:
            conflict_count += 1

    merged.sort(key=_inquiry_sort_key, reverse=True)
    aggregation = {
        "dedupe_key": "inquiry_id",
        "raw_total": sum(len(records) for records in grouped.values()),
        "deduped_total": len(merged),
        "conflict_count": conflict_count,
    }
    return merged, aggregation


def aggregate_skipped_stats(payloads: Iterable[dict]) -> dict:
    totals: dict[str, int] = defaultdict(int)
    for payload in payloads:
        skipped = payload.get("skipped") or {}
        for key, value in skipped.items():
            if isinstance(value, int):
                totals[key] += value
    return dict(totals)


def build_batch_meta(
    *,
    role: str,
    batch_root: str,
    window: BatchWindow | None = None,
    month_key: str | None = None,
    chunk_days: int | None = None,
    resume: bool = False,
    source_shards: list[str] | None = None,
    aggregation: dict | None = None,
    manifest_path: str | None = None,
    monthly_path: str | None = None,
) -> dict:
    meta = {
        "role": role,
        "batch_root": batch_root,
        "resume": resume,
    }
    if window is not None:
        meta["window"] = {
            "start": window.start.strftime(DATE_FMT),
            "end": window.end.strftime(DATE_FMT),
            "index": window.index,
            "chunk_days": window.chunk_days,
            "window_id": window.window_id,
        }
        meta["scope_desc"] = window.scope_desc
        meta["month_key"] = window.month_key
    if month_key is not None:
        meta["month_key"] = month_key
    if chunk_days is not None:
        meta["chunk_days"] = chunk_days
    if source_shards is not None:
        meta["source_shards"] = source_shards
    if aggregation is not None:
        meta["aggregation"] = aggregation
    if manifest_path is not None:
        meta["manifest_path"] = manifest_path
    if monthly_path is not None:
        meta["monthly_path"] = monthly_path
    return meta


def build_manifest(
    *,
    report_month: str,
    batch_root: str,
    chunk_days: int,
    windows: list[dict],
    source_shards: list[str],
    monthly_path: str,
    status: str,
    failures: list[dict] | None = None,
    aggregation: dict | None = None,
) -> dict:
    completed_windows = [item["window_id"] for item in windows if item.get("status") == "ok"]
    failed_windows = [item["window_id"] for item in windows if item.get("status") != "ok"]
    manifest = {
        "status": status,
        "report_month": report_month,
        "batch_root": batch_root,
        "chunk_days": chunk_days,
        "window_count": len(windows),
        "completed_windows": completed_windows,
        "failed_windows": failed_windows,
        "missing_windows": [item["window_id"] for item in windows if item.get("status") == "missing"],
        "source_files": source_shards,
        "monthly_output": monthly_path,
    }
    if failures:
        manifest["failures"] = failures
    if aggregation:
        manifest["aggregation"] = aggregation
    return manifest
