"""Import past forecasts from a Drupal 7 / Backdrop `advisory` database.

Works for the Sierra archive and any site built on the avalanche-center
template (Gulmarg, Chile) — they share one field schema (see app/mapping.py).
Reads a LOCAL MySQL/MariaDB copy only, e.g. the Sierra archive's DDEV database:

    cd ../archive.sierraavalanchecenter.org && ddev start && ddev describe   # note the db port
    python -m cli.import_drupal --dsn mysql://db:db@127.0.0.1:<port>/db --center sac --source-key sac-archive

Idempotent: forecasts upsert on (source='drupal', source_id='<source-key>:<nid>').
Historical forecasts are tagged guidance_era='legacy' by default — they predate
the current avalanche-problem guidance and are treated as descriptive history.
"""

from __future__ import annotations

import html
import re
from collections import Counter
from datetime import UTC, datetime
from urllib.parse import unquote, urlparse

import pymysql
import pymysql.cursors
import typer
from psycopg.types.json import Jsonb

from app.mapping import (
    date_from_title,
    drupal_problem_type,
    parse_danger,
    parse_likelihood,
    parse_sizes,
    rose_cells_to_locations,
    season_for,
)
from app.phrases import rebuild as rebuild_phrases

from .common import center_by_slug, connect

app = typer.Typer()

TEXT_FIELDS = {
    "bottom_line": "field_bottom_line",
    "hazard_discussion": "field_text_discussion",
    "weather_discussion": "field_mountain_weather",
    "recent_activity": "field_recent_activity",
}
BAND_FIELDS = {"lower": "field_danger_rating_1", "middle": "field_danger_rating_2", "upper": "field_danger_rating_3"}


def html_to_text(value: str | None) -> str | None:
    if not value:
        return None
    s = re.sub(r"(?i)<\s*br\s*/?>", "\n", value)
    s = re.sub(r"(?i)</\s*(p|div|h[1-6])\s*>", "\n\n", s)
    s = re.sub(r"(?i)</\s*li\s*>", "\n", s)
    s = re.sub(r"<[^>]+>", "", s)
    s = html.unescape(s).replace("\xa0", " ")
    s = re.sub(r"[ \t]+", " ", s)
    s = re.sub(r" *\n *", "\n", s)
    s = re.sub(r"\n{3,}", "\n\n", s).strip()
    return s or None


def mysql_connect(dsn: str) -> pymysql.connections.Connection:
    u = urlparse(dsn)
    if u.scheme not in ("mysql", "mariadb"):
        raise SystemExit("DSN must look like mysql://user:pass@host:port/dbname")
    return pymysql.connect(
        host=u.hostname or "127.0.0.1",
        port=u.port or 3306,
        user=unquote(u.username or ""),
        password=unquote(u.password or ""),
        database=u.path.lstrip("/"),
        charset="utf8mb4",
        cursorclass=pymysql.cursors.DictCursor,
    )


def table_exists(cur, table: str) -> bool:
    cur.execute("SHOW TABLES LIKE %s", (table,))
    return cur.fetchone() is not None


def field_values(cur, field: str, bundle: str, column: str | None = None) -> dict[int, list]:
    """entity_id -> [values ordered by delta] for a simple field."""
    table = f"field_data_{field}"
    if not table_exists(cur, table):
        return {}
    col = column or f"{field}_value"
    cur.execute(
        f"SELECT entity_id, delta, `{col}` AS v FROM `{table}` WHERE entity_type = 'node' AND bundle = %s AND deleted = 0 ORDER BY entity_id, delta",
        (bundle,),
    )
    out: dict[int, list] = {}
    for r in cur.fetchall():
        out.setdefault(r["entity_id"], []).append(r["v"])
    return out


def rose_values(cur, field: str, bundle: str) -> dict[int, dict[int, int]]:
    table = f"field_data_{field}"
    if not table_exists(cur, table):
        return {}
    cols = ", ".join(f"`{field}_{i}` AS c{i}" for i in range(24))
    cur.execute(f"SELECT entity_id, {cols} FROM `{table}` WHERE entity_type = 'node' AND bundle = %s AND deleted = 0", (bundle,))
    return {r["entity_id"]: {i: r[f"c{i}"] for i in range(24)} for r in cur.fetchall()}


def term_names(cur) -> dict[int, str]:
    if not table_exists(cur, "taxonomy_term_data"):
        return {}
    cur.execute("SELECT tid, name FROM taxonomy_term_data")
    return {r["tid"]: r["name"] for r in cur.fetchall()}


@app.command()
def main(
    dsn: str = typer.Option(..., help="mysql://user:pass@127.0.0.1:port/db of a LOCAL copy"),
    center: str = typer.Option(..., help="Center slug in this tool"),
    source_key: str = typer.Option(..., help="Stable name for the source site, e.g. sac-archive"),
    bundle: str = typer.Option("advisory"),
    era: str = typer.Option("legacy", help="legacy | current"),
    include_unpublished: bool = typer.Option(False),
    limit: int = typer.Option(0, help="0 = all"),
    dry_run: bool = typer.Option(False),
):
    if era not in ("legacy", "current"):
        raise SystemExit("--era must be legacy or current")
    src = mysql_connect(dsn)
    counts: Counter = Counter()
    errors: list[dict] = []
    with src.cursor() as cur:
        cur.execute(
            f"SELECT nid, vid, title, status, created FROM node WHERE type = %s {'' if include_unpublished else 'AND status = 1'} ORDER BY nid"
            + (f" LIMIT {int(limit)}" if limit else ""),
            (bundle,),
        )
        nodes = cur.fetchall()
        typer.echo(f"{len(nodes)} {bundle} nodes")
        danger = {band: field_values(cur, f, bundle) for band, f in BAND_FIELDS.items()}
        overall = field_values(cur, "field_overalldanger", bundle)
        texts = {k: field_values(cur, f, bundle) for k, f in TEXT_FIELDS.items()}
        region = field_values(cur, "field_forecast_region", bundle, "field_forecast_region_tid")
        terms = term_names(cur)
        slots = {}
        for n in (1, 2, 3):
            slots[n] = {
                "type": field_values(cur, f"field_type_{n}", bundle),
                "likelihood": field_values(cur, f"field_likelihood_{n}", bundle),
                "size": field_values(cur, f"field_size_{n}", bundle),
                "rose": rose_values(cur, f"field_rose_{n}", bundle),
                "description": field_values(cur, f"field_description_{n}", bundle),
            }
    src.close()

    first = lambda d, nid: (d.get(nid) or [None])[0]  # noqa: E731

    records = []
    for node in nodes:
        nid = node["nid"]
        valid_date = date_from_title(node["title"])
        if not valid_date:
            valid_date = datetime.fromtimestamp(node["created"], UTC).date().isoformat()
            counts["date_from_created"] += 1
        problems, normal_caution, unknown = [], False, []
        for n in (1, 2, 3):
            raw_type = first(slots[n]["type"], nid)
            if raw_type in (None, "", "0"):
                continue
            afp_type, is_nc = drupal_problem_type(raw_type)
            if is_nc:
                normal_caution = True
                counts["normal_caution_slots"] += 1
                continue
            if afp_type is None:
                unknown.append(raw_type)
                continue
            size_min, size_max = parse_sizes(slots[n]["size"].get(nid))
            problems.append(
                {
                    "rank": n,
                    "problem_type": afp_type,
                    "locations": rose_cells_to_locations(slots[n]["rose"].get(nid, {})),
                    "likelihood": parse_likelihood(first(slots[n]["likelihood"], nid)),
                    "size_min": size_min,
                    "size_max": size_max,
                    "discussion": html_to_text(first(slots[n]["description"], nid)),
                    "source_problem_key": str(raw_type),
                }
            )
        if unknown:
            errors.append({"nid": nid, "error": f"unknown problem type(s) {unknown}"})
        tid = first(region, nid)
        records.append(
            {
                "source_id": f"{source_key}:{nid}",
                "valid_date": valid_date,
                "season": season_for(valid_date),
                "title": node["title"],
                "published_at": datetime.fromtimestamp(node["created"], UTC),
                "zone_name": terms.get(tid) if tid else None,
                "danger": {b: parse_danger(first(danger[b], nid)) for b in BAND_FIELDS},
                "overall": parse_danger(first(overall, nid)),
                "normal_caution": normal_caution,
                "texts": {k: html_to_text(first(v, nid)) for k, v in texts.items()},
                "raw": {"nid": nid, "vid": node["vid"], "status": node["status"], "region_tid": tid,
                        "bottom_line_html": first(texts["bottom_line"], nid)},
                "problems": problems,
            }
        )
        counts["problems"] += len(problems)

    typer.echo(f"parsed {len(records)} forecasts, {counts['problems']} problems, {counts['normal_caution_slots']} Normal Caution slots, "
               f"{counts['date_from_created']} dates from node.created, {len(errors)} errors")
    if dry_run:
        for r in records[:3]:
            typer.echo(f"  {r['valid_date']} {r['danger']} {[(p['problem_type'], p['likelihood'], p['size_min'], p['size_max'], len(p['locations'])) for p in r['problems']]}")
        return

    with connect() as conn:
        c = center_by_slug(conn, center)
        job = conn.execute(
            "INSERT INTO import_jobs (source, center_id, params) VALUES ('drupal', %s, %s) RETURNING id",
            (c["id"], Jsonb({"source_key": source_key, "bundle": bundle, "era": era, "include_unpublished": include_unpublished})),
        ).fetchone()["id"]
        zone_ids: dict[str, int] = {}
        for r in records:
            zone_id = None
            if r["zone_name"]:
                if r["zone_name"] not in zone_ids:
                    conn.execute(
                        "INSERT INTO zones (center_id, name) VALUES (%s, %s) ON CONFLICT (center_id, name) DO NOTHING",
                        (c["id"], r["zone_name"]),
                    )
                    zone_ids[r["zone_name"]] = conn.execute(
                        "SELECT id FROM zones WHERE center_id = %s AND name = %s", (c["id"], r["zone_name"])
                    ).fetchone()["id"]
                zone_id = zone_ids[r["zone_name"]]
            fid = conn.execute(
                """INSERT INTO forecasts (center_id, zone_id, source, source_id, valid_date, published_at, title,
                       danger_upper, danger_middle, danger_lower, danger_overall, normal_caution, bottom_line,
                       hazard_discussion, weather_discussion, recent_activity, season, guidance_era, raw)
                   VALUES (%(c)s, %(z)s, 'drupal', %(sid)s, %(d)s, %(pub)s, %(title)s, %(du)s, %(dm)s, %(dl)s, %(do)s, %(nc)s,
                       %(bl)s, %(hd)s, %(wd)s, %(ra)s, %(season)s, %(era)s, %(raw)s)
                   ON CONFLICT (source, source_id) DO UPDATE SET
                       zone_id = EXCLUDED.zone_id, valid_date = EXCLUDED.valid_date, published_at = EXCLUDED.published_at,
                       title = EXCLUDED.title, danger_upper = EXCLUDED.danger_upper, danger_middle = EXCLUDED.danger_middle,
                       danger_lower = EXCLUDED.danger_lower, danger_overall = EXCLUDED.danger_overall,
                       normal_caution = EXCLUDED.normal_caution, bottom_line = EXCLUDED.bottom_line,
                       hazard_discussion = EXCLUDED.hazard_discussion, weather_discussion = EXCLUDED.weather_discussion,
                       recent_activity = EXCLUDED.recent_activity, season = EXCLUDED.season,
                       guidance_era = EXCLUDED.guidance_era, raw = EXCLUDED.raw, imported_at = now()
                   RETURNING id""",
                {
                    "c": c["id"], "z": zone_id, "sid": r["source_id"], "d": r["valid_date"], "pub": r["published_at"],
                    "title": r["title"], "du": r["danger"]["upper"], "dm": r["danger"]["middle"], "dl": r["danger"]["lower"],
                    "do": r["overall"], "nc": r["normal_caution"], "bl": r["texts"]["bottom_line"],
                    "hd": r["texts"]["hazard_discussion"], "wd": r["texts"]["weather_discussion"],
                    "ra": r["texts"]["recent_activity"], "season": r["season"], "era": era, "raw": Jsonb(r["raw"]),
                },
            ).fetchone()["id"]
            conn.execute("DELETE FROM forecast_problems WHERE forecast_id = %s", (fid,))
            for p in r["problems"]:
                conn.execute(
                    """INSERT INTO forecast_problems (forecast_id, rank, problem_type, locations, likelihood, size_min, size_max,
                           discussion, source_problem_key)
                       VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)""",
                    (fid, p["rank"], p["problem_type"], p["locations"], p["likelihood"], p["size_min"], p["size_max"],
                     p["discussion"], p["source_problem_key"]),
                )
            counts["imported"] += 1
        conn.execute(
            "UPDATE import_jobs SET status = 'done', finished_at = now(), counts = %s, errors = %s WHERE id = %s",
            (Jsonb(dict(counts)), Jsonb(errors[:500]), job),
        )
        phrases = rebuild_phrases(conn, c["id"])
    typer.echo(f"imported {counts['imported']} forecasts into {center} (job {job})")
    typer.echo(f"phrase library: {phrases['phrases']} phrases ({phrases['distinct']} distinct)")


if __name__ == "__main__":
    app()