"""Reference List Manipulation (RLM) detection via Gini coefficient.

Single-query DuckDB implementation, memoized via irohds so results are
shared P2P. Any peer with the same PubMed snapshot (same FTP fingerprint)
and same parameters gets the result instantly instead of recomputing.

Usage:
    from pmq import PubMed
    from pmq.rlm import analyze

    with PubMed() as db:
        df = analyze(db, min_pubs=50, top=100)
"""
from __future__ import annotations
import os, sys, time
from pathlib import Path

import duckdb
import irohds

from ._tables import ftp_fingerprint

_RLM_SQL = """
WITH
all_authors AS (
    SELECT 'orcid:' || pi.identifier AS author_key,
           p.last_name, p.initials, p.pmid
    FROM person_identifier pi
    JOIN person p ON p.pmid = pi.pmid AND p.role = pi.role AND p.seq = pi.person_seq
    WHERE pi.source = 'ORCID' AND p.role = 'author' AND p.last_name IS NOT NULL
    UNION ALL
    SELECT 'name:' || LOWER(p.last_name) || '_' || LOWER(p.initials),
           p.last_name, p.initials, p.pmid
    FROM person p
    WHERE p.role = 'author'
      AND p.last_name IS NOT NULL AND p.initials IS NOT NULL
      AND NOT EXISTS (
          SELECT 1 FROM person_identifier pi
          WHERE pi.pmid = p.pmid AND pi.role = p.role
            AND pi.person_seq = p.seq AND pi.source = 'ORCID'
      )
),
prolific AS (
    SELECT author_key,
           MIN(last_name) AS last_name, MIN(initials) AS initials,
           COUNT(DISTINCT pmid) AS n_pubs
    FROM all_authors
    GROUP BY author_key
    HAVING COUNT(DISTINCT pmid) >= {min_pubs}
),
author_papers AS (
    SELECT DISTINCT a.author_key, a.pmid
    FROM all_authors a SEMI JOIN prolific p ON a.author_key = p.author_key
),
cite_links AS (
    SELECT DISTINCT ri.pmid AS citing_pmid,
           TRY_CAST(ri.value AS INTEGER) AS cited_pmid
    FROM reference_id ri
    WHERE ri.id_type = 'pubmed' AND TRY_CAST(ri.value AS INTEGER) IS NOT NULL
),
nsc AS (
    SELECT ap.author_key, cl.citing_pmid, COUNT(*) AS n_refs
    FROM author_papers ap
    JOIN cite_links cl ON cl.cited_pmid = ap.pmid
    ANTI JOIN author_papers self
        ON self.author_key = ap.author_key AND self.pmid = cl.citing_pmid
    GROUP BY ap.author_key, cl.citing_pmid
),
ranked AS (
    SELECT author_key, n_refs,
           ROW_NUMBER() OVER (PARTITION BY author_key ORDER BY n_refs) AS rk,
           COUNT(*) OVER (PARTITION BY author_key) AS n,
           SUM(CAST(n_refs AS BIGINT)) OVER (PARTITION BY author_key) AS total
    FROM nsc
)
SELECT
    r.author_key,
    p.last_name,
    p.initials,
    p.n_pubs,
    MAX(r.n)::INTEGER AS n_citing_papers,
    MAX(r.total)::INTEGER AS n_nsc,
    ROUND(
        (2.0 * SUM(CAST(r.rk AS BIGINT) * r.n_refs))
        / (MAX(r.n) * MAX(r.total))
        - (MAX(r.n) + 1.0) / MAX(r.n),
    4) AS gini,
    MAX(r.n_refs)::INTEGER AS max_refs_from_one_paper,
    SUM(CASE WHEN r.n_refs >= 17 THEN 1 ELSE 0 END)::INTEGER AS papers_with_17plus
FROM ranked r
JOIN prolific p ON r.author_key = p.author_key
GROUP BY r.author_key, p.last_name, p.initials, p.n_pubs
HAVING MAX(r.n) > 0 AND MAX(r.total) > 0
ORDER BY gini DESC
"""


@irohds.memo(ns="pmq", ignore=["data_dir"])
def _compute_rlm(ftp_fp: str, min_pubs: int, top: int,
                  data_dir: str = "") -> irohds.FileRef:
    """Compute RLM and write to Parquet. Memoized: shared via P2P."""
    from ._tables import create_table

    dd = Path(data_dir)

    # Only load tables the RLM query actually touches
    conn = duckdb.connect()
    for name in ("person", "person_identifier", "reference_id"):
        ref = create_table(ftp_fp, name, data_dir=data_dir)
        if os.path.exists(ref.path):
            conn.execute(
                f"CREATE OR REPLACE VIEW {name} AS "
                f"SELECT * FROM read_parquet('{ref.path}')")

    sql = _RLM_SQL.format(min_pubs=int(min_pubs))
    if top:
        sql += f"\nLIMIT {int(top)}"

    output = dd / f"rlm_min{min_pubs}_top{top}.parquet"
    tmp = output.with_suffix('.parquet.tmp')
    conn.execute(
        f"COPY ({sql}) TO '{tmp}' (FORMAT PARQUET, COMPRESSION ZSTD)")
    conn.close()
    os.replace(str(tmp), str(output))

    return irohds.FileRef(str(output))


def analyze(db, *, min_pubs: int = 50, top: int = 100,
            output: str | None = None, verbose: bool = True):
    """Run RLM analysis. Returns a DataFrame.

    First call computes and shares via P2P. Subsequent calls on any
    peer with the same PubMed snapshot return instantly.
    """
    t0 = time.time()
    log = lambda msg: print(f'[rlm] {msg}', file=sys.stderr, flush=True) if verbose else None

    log('Running RLM analysis...')

    fp = ftp_fingerprint()
    dd = str(db.data_dir)
    ref = _compute_rlm(fp, min_pubs, top, data_dir=dd)

    df = duckdb.sql(f"SELECT * FROM read_parquet('{ref.path}')").df()

    elapsed = time.time() - t0
    log(f'Done. {len(df)} authors in {elapsed:.1f}s')

    if output:
        df.to_csv(output, index=False)
        log(f'Written to {output}')
    elif verbose and len(df) > 0:
        print(f'\n{"Rank":>4}  {"Gini":>6}  {"Pubs":>6}  {"Citing":>7}  '
              f'{"NSC":>7}  {"Max":>4}  {"17+":>4}  Author')
        print('-' * 75)
        for i, r in df.iterrows():
            name = f"{r['last_name']}, {r['initials']}"
            tag = ' [ORCID]' if r['author_key'].startswith('orcid:') else ''
            print(f'{i+1:4}  {r["gini"]:6.4f}  {r["n_pubs"]:6,}  '
                  f'{r["n_citing_papers"]:7,}  {r["n_nsc"]:7,}  '
                  f'{r["max_refs_from_one_paper"]:4}  {r["papers_with_17plus"]:4}  '
                  f'{name}{tag}')

    return df
