"""Per-table Parquet creation with irohds memoization.

Each table is independently memoized: key = ftp_fingerprint + table_name.
On cache miss: ensure_build() -> compact one table -> return FileRef.
On cache hit: irohds materializes via P2P (delta-efficient).

Sorting ensures deterministic BLAKE3 hashes (same input -> same bytes ->
same iroh blob hash across machines). citation sorts by (pub_year, pmid)
so row groups form natural date ranges; everything else sorts by pmid.

OOM prevention: coren provides mem_available, DuckDB memory_limit is set
to 60% of that with temp_directory for spill-to-disk external merge sort.
"""
import os
import re
import hashlib
import ssl
import time
from pathlib import Path
from urllib.request import urlopen

import certifi
import duckdb
import irohds
from coren import MachCap

from ._build import ensure_build, flock

FTP_BASE = "https://ftp.ncbi.nlm.nih.gov/pubmed"
_FTP_PAT = re.compile(r'href="(pubmed\d+n\d+\.xml\.gz)"')
_SSL_CTX = ssl.create_default_context(cafile=certifi.where())

# Cache FTP fingerprint to avoid repeated NCBI requests.
# 30-min TTL matches BUILD_FRESH_SECS -- be gentle with NCBI servers.
_fp_cache: tuple[float, str] | None = None
_FP_TTL = 30 * 60

# Tables from the XML corpus (produced by pmq-core + DuckDB compaction)
_XML_TABLES = [
    "citation", "abstract_text", "person", "person_affiliation",
    "person_identifier", "mesh_heading", "mesh_qualifier", "keyword",
    "chemical", "suppl_mesh", "grant_info", "publication_type",
    "article_id", "elocation_id", "pub_date_history", "article_date",
    "comment_correction", "data_bank", "accession_number", "reference",
    "reference_id", "other_id", "general_note", "object_info",
    "object_param", "citation_tag", "delete_citation",
]

# Auxiliary tables (downloaded separately, parsed in Python/DuckDB)
AUX_TABLES = {"journal", "deleted_pmid_list", "computed_author"}

ALL_TABLES = _XML_TABLES + sorted(AUX_TABLES)

_CORE_TABLES = {"citation", "abstract_text", "person", "article_id"}

_SORT_ORDERS = {"citation": "pub_year, pmid"}
_DEFAULT_SORT = "1"


def ftp_fingerprint() -> str:
    """Blake2b of sorted NCBI FTP .xml.gz filenames. Cached for 30 min."""
    global _fp_cache
    now = time.time()
    if _fp_cache and (now - _fp_cache[0]) < _FP_TTL:
        return _fp_cache[1]
    names = []
    for suffix in ("baseline/", "updatefiles/"):
        html = urlopen(f"{FTP_BASE}/{suffix}", timeout=30, context=_SSL_CTX).read().decode()
        names.extend(m.group(1) for m in _FTP_PAT.finditer(html))
    names.sort()
    fp = hashlib.blake2b("\n".join(names).encode(), digest_size=32).hexdigest()
    _fp_cache = (now, fp)
    return fp


@irohds.memo(ns="pmq", ignore=["data_dir"])
def create_table(ftp_date: str, table_name: str, data_dir: str = "") -> irohds.FileRef:
    """Build one table, return FileRef. Process-safe via per-table flock."""
    dd = Path(data_dir)
    output = dd / f"{table_name}.parquet"

    if table_name in AUX_TABLES:
        with flock(dd / f".lock-{table_name}"):
            if not output.exists():
                from ._aux import build_aux_table
                build_aux_table(dd, table_name, output)
    else:
        ensure_build(dd)  # own lock internally
        with flock(dd / f".lock-{table_name}"):
            if not output.exists():
                compact_table(dd, table_name, output)

    return irohds.FileRef(str(output))


def compact_table(data_dir: Path, table_name: str, output: Path):
    """Merge per-worker parquet files into one deterministically-sorted file."""
    if output.exists():
        return  # another caller finished first
    workers = data_dir / '_workers'
    if not workers.exists():
        return

    files = [f for f in workers.glob(f'*/{table_name}.parquet')
             if f.stat().st_size > 0]
    if not files:
        return

    glob_path = str(workers / '*' / f'{table_name}.parquet')
    tmp = output.with_suffix('.parquet.tmp')
    tmp_dir = data_dir / '.duckdb_tmp'
    tmp_dir.mkdir(parents=True, exist_ok=True)

    cap = MachCap.read(str(data_dir), detect_net=False, run_bench=False)
    mem_mb = max(128, int(cap.mem_available * 0.6 / (1 << 20)))
    order = _SORT_ORDERS.get(table_name, _DEFAULT_SORT)

    conn = duckdb.connect()
    try:
        conn.execute(f"SET memory_limit='{mem_mb}MB'")
        conn.execute(f"SET temp_directory='{tmp_dir}'")
        conn.execute(f"SET threads={max(1, cap.cores_physical - 1)}")
        conn.execute("SET enable_progress_bar=true")
        conn.execute("SET enable_progress_bar_print=true")
        conn.execute(
            f"COPY (SELECT * FROM read_parquet('{glob_path}') ORDER BY {order}) "
            f"TO '{tmp}' (FORMAT PARQUET, COMPRESSION ZSTD, ROW_GROUP_SIZE 1000000)")
        os.replace(str(tmp), str(output))
    except Exception:
        tmp.unlink(missing_ok=True)
        raise
    finally:
        conn.close()


def has_all_tables(data_dir: Path) -> bool:
    """True if all core tables exist as compacted .parquet files."""
    return all((data_dir / f"{t}.parquet").exists() for t in _CORE_TABLES)


def table_has_data(data_dir: Path, table_name: str) -> bool:
    """True if a compacted file or any worker shards exist for this table."""
    if (data_dir / f"{table_name}.parquet").exists():
        return True
    workers = data_dir / '_workers'
    if not workers.exists():
        return False
    return any(f.stat().st_size > 0 for f in workers.glob(f'*/{table_name}.parquet'))
