from __future__ import annotations

import argparse
import json
import mimetypes
import socket
import sqlite3
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import parse_qs, unquote, urlparse

import toyota_epc_query_validation_db


ROOT = Path(__file__).resolve().parent
DB_PATH = ROOT / "toyota_epc_output" / "toyota_epc_validation.sqlite"
STATIC_DIR = ROOT / "toyota_epc_frontend"
IMAGE_DIR = ROOT / "toyota_epc_output" / "part_images"


def row_dict(row: sqlite3.Row) -> dict:
    return {key: row[key] for key in row.keys()}


def normalize_oe(value: str) -> str:
    compact = "".join(ch for ch in value.upper() if ch.isalnum())
    if len(compact) != 10:
        return value.strip().upper()
    return f"{compact[:5]}-{compact[5:]}"


def parse_int(value: str | None, default: int, maximum: int) -> int:
    try:
        parsed = int(value or default)
    except ValueError:
        return default
    return max(1, min(parsed, maximum))


def image_url(market: str, catalog: str, figure_group: str, sequence: str, image_code: str) -> str:
    filename = f"{market}_{catalog}_{figure_group}_{sequence}_{image_code}.png"
    if (IMAGE_DIR / filename).exists():
        return f"/images/part_images/{filename}"
    return ""


class ToyotaEpcRepository:
    def __init__(self, db_path: Path):
        self.db_path = db_path
        self.conn = sqlite3.connect(db_path, check_same_thread=False)
        self.conn.row_factory = sqlite3.Row

    def stats(self) -> dict:
        tables = [
            "epc_part_index",
            "epc_part_name",
            "epc_vehicle_detail",
            "epc_vehicle_name",
            "epc_frame_month_point",
            "epc_figure_image",
        ]
        counts = {}
        for table in tables:
            counts[table] = self.conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0]
        return {
            "database": str(self.db_path),
            "database_size_bytes": self.db_path.stat().st_size,
            "counts": counts,
            "image_samples": len(list(IMAGE_DIR.glob("*.png"))) if IMAGE_DIR.exists() else 0,
        }

    def part_names(self, market: str, reference_code: str) -> list[dict]:
        rows = self.conn.execute(
            """
            SELECT language_file, language_code, reference_code, part_name,
                   source_file, row_no, byte_offset
            FROM epc_part_name
            WHERE market = ? AND reference_code = ?
            ORDER BY
              CASE language_file
                WHEN 'HINMEI01.DAT' THEN 0
                WHEN 'HINMEI03.DAT' THEN 1
                WHEN 'HINMEI04.DAT' THEN 2
                WHEN 'HINMEI05.DAT' THEN 3
                ELSE 9
              END,
              row_no
            LIMIT 6
            """,
            (market, reference_code),
        ).fetchall()
        return [row_dict(row) for row in rows]

    def vehicle_names(self, market: str, catalog: str) -> list[dict]:
        rows = self.conn.execute(
            """
            SELECT DISTINCT vehicle_name_epc, model_family, vehicle_range_start, vehicle_range_end, source_file, row_no, byte_offset
            FROM epc_vehicle_name
            WHERE market = ? AND catalog = ?
            ORDER BY row_no
            LIMIT 5
            """,
            (market, catalog),
        ).fetchall()
        return [row_dict(row) for row in rows]

    def catalog_scope(self, market: str, catalog: str, sample_limit: int) -> dict:
        counts = self.conn.execute(
            """
            SELECT
              COUNT(DISTINCT model) AS model_count,
              COUNT(DISTINCT frame_code) AS frame_count,
              COUNT(DISTINCT engine_epc) AS engine_count
            FROM epc_vehicle_detail
            WHERE market = ? AND catalog = ?
            """,
            (market, catalog),
        ).fetchone()
        frames = [
            row["frame_code"]
            for row in self.conn.execute(
                """
                SELECT DISTINCT frame_code
                FROM epc_vehicle_detail
                WHERE market = ? AND catalog = ? AND frame_code <> ''
                ORDER BY frame_code
                LIMIT 80
                """,
                (market, catalog),
            )
        ]
        engines = [
            row["engine_epc"]
            for row in self.conn.execute(
                """
                SELECT DISTINCT engine_epc
                FROM epc_vehicle_detail
                WHERE market = ? AND catalog = ? AND engine_epc <> ''
                ORDER BY engine_epc
                LIMIT 60
                """,
                (market, catalog),
            )
        ]
        samples = [
            row_dict(row)
            for row in self.conn.execute(
                """
                SELECT DISTINCT model, frame_code, engine_epc, production_start,
                       production_end, steering, destination
                FROM epc_vehicle_detail
                WHERE market = ? AND catalog = ?
                ORDER BY frame_code, model, production_start
                LIMIT ?
                """,
                (market, catalog, sample_limit),
            )
        ]
        return {
            "model_count": counts["model_count"] or 0,
            "frame_count": counts["frame_count"] or 0,
            "engine_count": counts["engine_count"] or 0,
            "frames": frames,
            "engines": engines,
            "samples": samples,
        }

    def candidates(self, market: str, catalog: str, reference_code: str, figure_group: str) -> list[str]:
        rows = self.conn.execute(
            """
            SELECT DISTINCT oe_no
            FROM epc_part_index
            WHERE market = ? AND catalog = ? AND reference_code = ? AND figure_group = ?
            ORDER BY oe_no
            LIMIT 120
            """,
            (market, catalog, reference_code, figure_group),
        ).fetchall()
        return [row["oe_no"] for row in rows]

    def images(self, market: str, catalog: str, figure_group: str) -> list[dict]:
        rows = self.conn.execute(
            """
            SELECT b.sequence, b.image_code, b.start_ym, b.end_ym, b.option_code,
                   b.source_file AS bzi_file, b.row_no AS bzi_row_no,
                   MIN(i.image_index) AS image_index
            FROM epc_figure_image b
            LEFT JOIN epc_image_index i
              ON i.market = b.market
             AND i.catalog = b.catalog
             AND i.figure_group = b.figure_group
             AND i.image_code = b.image_code
            WHERE b.market = ? AND b.catalog = ? AND b.figure_group = ?
            GROUP BY b.id
            ORDER BY b.sequence, b.image_code
            LIMIT 8
            """,
            (market, catalog, figure_group),
        ).fetchall()
        out = []
        for row in rows:
            item = row_dict(row)
            item["url"] = image_url(market, catalog, figure_group, item["sequence"], item["image_code"])
            out.append(item)
        return out

    def oe_substitutions(self, oe: str) -> dict:
        forward_rows = self.conn.execute(
            """
            SELECT market, old_oe, new_oe, effective_to_raw, relation_code,
                   effective_from_raw, sequence_code, source_file, row_no, byte_offset
            FROM epc_part_substitution
            WHERE old_oe = ?
            ORDER BY market, new_oe, effective_from_raw
            LIMIT 200
            """,
            (oe,),
        ).fetchall()
        reverse_rows = self.conn.execute(
            """
            SELECT market, old_oe, new_oe, effective_to_raw, relation_code,
                   effective_from_raw, sequence_code, source_file, row_no, byte_offset
            FROM epc_part_substitution
            WHERE new_oe = ?
            ORDER BY market, old_oe, effective_from_raw
            LIMIT 200
            """,
            (oe,),
        ).fetchall()
        forward = [row_dict(row) for row in forward_rows]
        reverse = [row_dict(row) for row in reverse_rows]
        related = sorted({row["new_oe"] for row in forward} | {row["old_oe"] for row in reverse} | {oe})
        return {
            "source_table": "epc_part_substitution",
            "source_file": "DAIHIN.DAT",
            "input_oe": oe,
            "replaced_by": forward,
            "replaces": reverse,
            "related_oe": related,
            "note": "This is the EPC part-number substitution/translation relation, separate from same-figure candidates.",
        }
    def query_oe(self, value: str, sample_limit: int = 4, market: str = "") -> dict:
        oe = normalize_oe(value)
        substitutions = self.oe_substitutions(oe)
        market_filter = market.strip().upper()
        rows = self.conn.execute(
            """
            SELECT market, catalog, reference_code, figure_group,
                   COUNT(*) AS hit_rows,
                   MIN(source_file) AS source_file,
                   MIN(row_no) AS row_no,
                   MIN(byte_offset) AS byte_offset
            FROM epc_part_index
            WHERE oe_no = ? AND (? = '' OR market = ?)
            GROUP BY market, catalog, reference_code, figure_group
            ORDER BY market, catalog, reference_code, figure_group
            """,
            (oe, market_filter, market_filter),
        ).fetchall()
        groups = []
        application_rows = []
        for row in rows:
            group = row_dict(row)
            group["part_names"] = self.part_names(group["market"], group["reference_code"])
            group["vehicle_names"] = self.vehicle_names(group["market"], group["catalog"])
            group["catalog_scope"] = self.catalog_scope(group["market"], group["catalog"], sample_limit)
            group["candidate_same_position_oe"] = self.candidates(
                group["market"],
                group["catalog"],
                group["reference_code"],
                group["figure_group"],
            )
            group["images"] = self.images(group["market"], group["catalog"], group["figure_group"])
            vehicle = group["vehicle_names"][0] if group["vehicle_names"] else {}
            part_name = group["part_names"][0]["part_name"] if group["part_names"] else ""
            scope = group["catalog_scope"]
            application_rows.append({
                "market": group["market"],
                "oe_no": oe,
                "reference_code": group["reference_code"],
                "quantity": "01",
                "catalog": group["catalog"],
                "vehicle_name_epc": vehicle.get("vehicle_name_epc", ""),
                "vehicle_range_start": vehicle.get("vehicle_range_start", ""),
                "vehicle_range_end": vehicle.get("vehicle_range_end", ""),
                "model_family": vehicle.get("model_family", ""),
                "figure_group": group["figure_group"],
                "part_name": part_name,
                "frames": scope.get("frames", []),
                "engines": scope.get("engines", []),
            })
            groups.append(group)

        markets = sorted({group["market"] for group in groups})
        vehicle_names = sorted(
            {
                vehicle["vehicle_name_epc"]
                for group in groups
                for vehicle in group["vehicle_names"]
                if vehicle.get("vehicle_name_epc")
            }
        )
        frames = sorted(
            {frame for group in groups for frame in group["catalog_scope"].get("frames", [])}
        )
        engines = sorted(
            {engine for group in groups for engine in group["catalog_scope"].get("engines", [])}
        )
        candidate_oe = sorted(
            {
                candidate
                for group in groups
                for candidate in group.get("candidate_same_position_oe", [])
            }
        )
        part_names = sorted(
            {
                name["part_name"]
                for group in groups
                for name in group.get("part_names", [])
                if name.get("part_name")
            }
        )
        accurate_forward_oe = sorted({row["new_oe"] for row in substitutions["replaced_by"]})
        accurate_reverse_oe = sorted({row["old_oe"] for row in substitutions["replaces"]})
        accurate_related_oe = sorted((set(accurate_forward_oe) | set(accurate_reverse_oe)) - {oe})
        return {
            "type": "oe",
            "input": value,
            "normalized": oe,
            "source": "local Toyota EPC SQLite",
            "market_filter": market_filter,
            "hit_groups": len(groups),
            "application_count": len(application_rows),
            "summary": {
                "markets": markets,
                "vehicle_names": vehicle_names,
                "frames": frames,
                "engines": engines,
                "part_names": part_names[:20],
                "candidate_oe": candidate_oe,
                "accurate_replaced_by_oe": accurate_forward_oe,
                "accurate_replaces_oe": accurate_reverse_oe,
                "accurate_related_oe": accurate_related_oe,
            },
            "substitutions": substitutions,
            "application_rows": application_rows,
            "groups": groups,
            "accuracy_note": "准确 OE 对照只来自 DAIHIN 原始替换表；同图号/同PNC号码只能作为应用位置参考，不能当作准确通用。",
        }

    def query_vin(self, value: str, model: str = "") -> dict:
        compact = "".join(ch for ch in value.upper() if ch.isalnum())
        vin9 = compact[:9]
        raw_results = toyota_epc_query_validation_db.query(self.conn, value, model)
        results = [
            row
            for row in raw_results
            if row.get("vin_key") and vin9.startswith(str(row.get("vin_key")))
        ]
        if not results and model:
            results = raw_results
        return {
            "type": "vin",
            "input": value,
            "source": "local Toyota EPC SQLite",
            "vin9": vin9,
            "raw_count_before_prefix_filter": len(raw_results),
            "count": len(results),
            "results": results,
        }

    def query_model(self, value: str, limit: int = 80) -> dict:
        q = value.strip().upper()
        like = q.replace("#", "%").replace("*", "%")
        if "%" not in like:
            like = f"%{like}%"
        rows = self.conn.execute(
            """
            SELECT DISTINCT
              d.market, d.catalog, n.vehicle_name_epc, n.model_family,
              d.model, d.vin_key, d.production_start, d.production_end,
              d.frame_code, d.engine_epc, d.body, d.transmission, d.gear,
              d.steering, d.door, d.grade, d.turbo, d.destination,
              d.source_file, d.row_no, d.byte_offset
            FROM epc_vehicle_detail d
            LEFT JOIN epc_vehicle_name n
              ON n.market = d.market AND n.catalog = d.catalog
            WHERE d.model = ? OR d.model LIKE ? OR d.frame_code = ?
            ORDER BY d.market, d.catalog, d.model, d.production_start
            LIMIT ?
            """,
            (q, like, q, limit),
        ).fetchall()
        results = [row_dict(row) for row in rows]
        return {
            "type": "model",
            "input": value,
            "source": "local Toyota EPC SQLite",
            "count": len(results),
            "summary": {
                "markets": sorted({row["market"] for row in results}),
                "vehicle_names": sorted({row["vehicle_name_epc"] for row in results if row.get("vehicle_name_epc")}),
                "frames": sorted({row["frame_code"] for row in results if row.get("frame_code")}),
                "engines": sorted({row["engine_epc"] for row in results if row.get("engine_epc")}),
            },
            "results": results,
        }


def send_bytes(handler: BaseHTTPRequestHandler, body: bytes, content_type: str, status: int = 200) -> None:
    handler.send_response(status)
    handler.send_header("Content-Type", content_type)
    handler.send_header("Content-Length", str(len(body)))
    handler.end_headers()
    handler.wfile.write(body)


def make_handler(repo: ToyotaEpcRepository):
    class Handler(BaseHTTPRequestHandler):
        def send_json(self, payload: dict, status: int = 200) -> None:
            send_bytes(
                self,
                json.dumps(payload, ensure_ascii=False, indent=2).encode("utf-8"),
                "application/json; charset=utf-8",
                status,
            )

        def send_file(self, path: Path) -> None:
            if not path.is_file():
                self.send_json({"error": "not_found", "path": str(path)}, status=404)
                return
            content_type = mimetypes.guess_type(path.name)[0] or "application/octet-stream"
            send_bytes(self, path.read_bytes(), content_type)

        def do_GET(self) -> None:
            parsed = urlparse(self.path)
            params = {key: values[0] for key, values in parse_qs(parsed.query).items() if values}
            try:
                if parsed.path == "/":
                    self.send_file(STATIC_DIR / "index.html")
                elif parsed.path.startswith("/static/"):
                    name = unquote(parsed.path.removeprefix("/static/"))
                    self.send_file(STATIC_DIR / name)
                elif parsed.path.startswith("/images/part_images/"):
                    name = Path(unquote(parsed.path)).name
                    self.send_file(IMAGE_DIR / name)
                elif parsed.path == "/api/health":
                    self.send_json({"ok": True, "service": "toyota_epc_compare", "stats": repo.stats()})
                elif parsed.path == "/api/toyota/oe":
                    self.send_json(repo.query_oe(params.get("q", ""), parse_int(params.get("sample"), 4, 20), params.get("market", "")))
                elif parsed.path == "/api/toyota/vin":
                    self.send_json(repo.query_vin(params.get("q", ""), params.get("model", "")))
                elif parsed.path == "/api/toyota/model":
                    self.send_json(repo.query_model(params.get("q", ""), parse_int(params.get("limit"), 80, 500)))
                else:
                    self.send_json({"error": "not_found", "path": parsed.path}, status=404)
            except Exception as exc:
                self.send_json({"error": type(exc).__name__, "message": str(exc)}, status=500)

        def log_message(self, format: str, *args) -> None:
            return

    return Handler


def port_available(port: int, host: str = "127.0.0.1") -> bool:
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
        sock.settimeout(0.2)
        return sock.connect_ex((host, port)) != 0


def choose_port(preferred: int) -> int:
    for port in range(preferred, preferred + 30):
        if port_available(port):
            return port
    raise RuntimeError("No available port found")


def main() -> None:
    parser = argparse.ArgumentParser(description="Toyota EPC local compare frontend")
    parser.add_argument("--host", default="127.0.0.1")
    parser.add_argument("--port", type=int, default=8787)
    parser.add_argument("--db", default=str(DB_PATH))
    args = parser.parse_args()

    db_path = Path(args.db)
    if not db_path.exists():
        raise SystemExit(f"Database not found: {db_path}")
    if not STATIC_DIR.exists():
        raise SystemExit(f"Frontend files not found: {STATIC_DIR}")

    port = choose_port(args.port)
    repo = ToyotaEpcRepository(db_path)
    server = ThreadingHTTPServer((args.host, port), make_handler(repo))
    print(f"Toyota EPC compare app: http://{args.host}:{port}/")
    print(f"Database: {db_path}")
    server.serve_forever()


if __name__ == "__main__":
    main()