/// PubMed XML parser -- DTD 250101 complete.
///
/// Streams .xml.gz files, writing records directly to Arrow builders
/// via WorkerWriter. No intermediate Vec<Val> or ArticleData -- values
/// flow from XML attributes/text straight into columnar builders.
use crate::tree::{build_subtree, Element};
use crate::writer::{RowWriter, WorkerWriter};

use quick_xml::events::Event;
use quick_xml::Reader;
use std::io::{BufRead, BufReader};
use std::path::Path;
use flate2::read::GzDecoder;

// -- Helpers ------------------------------------------------------------------

fn date_str(el: Option<&Element>) -> Option<String> {
    let el = el?;
    let y = el.find("Year").and_then(|e| e.text())?;
    let mut s = y.to_owned();
    if let Some(m) = el.find("Month").and_then(|e| e.text()) {
        s.push('-'); if m.len() < 2 { s.push('0'); } s.push_str(m);
        if let Some(d) = el.find("Day").and_then(|e| e.text()) {
            s.push('-'); if d.len() < 2 { s.push('0'); } s.push_str(d);
        }
    }
    Some(s)
}

fn push_ymd(r: &mut RowWriter, el: Option<&Element>) {
    let (y, m, d) = match el {
        Some(e) => (
            e.find("Year").and_then(|e| e.text()),
            e.find("Month").and_then(|e| e.text()),
            e.find("Day").and_then(|e| e.text()),
        ),
        None => (None, None, None),
    };
    r.str(y); r.str(m); r.str(d);
}

// -- Core: parse one article directly into writer -----------------------------

fn parse_article(article: &Element, w: &mut WorkerWriter) -> anyhow::Result<()> {
    let mc = match article.find("MedlineCitation") { Some(e) => e, None => return Ok(()) };
    let art = match mc.find("Article") { Some(e) => e, None => return Ok(()) };
    let pd = article.find("PubmedData");

    let pmid: i32 = mc.find("PMID")
        .and_then(|e| e.text()).and_then(|t| t.parse().ok()).unwrap_or(0);
    let pmid_ver: i32 = mc.find("PMID")
        .and_then(|e| e.attr("Version")).and_then(|v| v.parse().ok()).unwrap_or(1);

    // -- Journal (inlined into citation, natural keys) --
    let jour = art.find("Journal");
    let ji = jour.and_then(|j| j.find("JournalIssue"));
    let mji = mc.find("MedlineJournalInfo");

    let issn_el = jour.and_then(|j| j.find("ISSN"));
    let issn_val = issn_el.and_then(|e| e.text());
    let issn_type = issn_el.and_then(|e| e.attr("IssnType"));

    let journal_nlm_id = mji.and_then(|m| m.find("NlmUniqueID")).and_then(|e| e.text());
    let journal_medline_ta = mji.and_then(|m| m.find("MedlineTA")).and_then(|e| e.text())
        .or_else(|| jour.and_then(|j| j.find("Title")).and_then(|e| e.text()))
        .unwrap_or("Unknown");
    let journal_title = jour.and_then(|j| j.find("Title")).and_then(|e| e.text());
    let journal_iso_abbr = jour.and_then(|j| j.find("ISOAbbreviation")).and_then(|e| e.text());
    let issn_print = if issn_type == Some("Print") { issn_val } else { None };
    let issn_electronic = if issn_type == Some("Electronic") { issn_val } else { None };
    let issn_linking = mji.and_then(|m| m.find("ISSNLinking")).and_then(|e| e.text());
    let journal_country = mji.and_then(|m| m.find("Country")).and_then(|e| e.text());

    // PubDate
    let pubdate = ji.and_then(|j| j.find("PubDate"));
    let pub_y_raw = pubdate.and_then(|p| p.find("Year")).and_then(|e| e.text());
    let pub_m = pubdate.and_then(|p| p.find("Month")).and_then(|e| e.text());
    let pub_d = pubdate.and_then(|p| p.find("Day")).and_then(|e| e.text());
    let medline_date = pubdate.and_then(|p| p.find("MedlineDate")).and_then(|e| e.text());
    let pub_y = pub_y_raw.or_else(||
        medline_date.filter(|md| md.len() >= 4).map(|md| &md[..4])
    );

    // Abstract (pre-compute inner_text values that allocate)
    let abstract_el = art.find("Abstract");
    let copyright_info = abstract_el
        .and_then(|a| a.find("CopyrightInformation")).and_then(|e| e.inner_text());
    if let Some(abs) = abstract_el {
        for (i, at) in abs.find_all("AbstractText").enumerate() {
            let text = at.inner_text();
            w.row("abstract_text", |r| {
                r.int(pmid); r.str(Some("primary")); r.str(None); r.int(i as i32);
                r.str(at.attr("Label")); r.str(at.attr("NlmCategory"));
                r.str(text.as_deref());
            })?;
        }
    }

    // AuthorList
    let al = art.find("AuthorList");
    let al_complete = al.map(|a| a.attr("CompleteYN").unwrap_or("Y")).unwrap_or("Y");
    if let Some(al) = al {
        parse_persons(w, pmid, "author", al, "Author")?;
    }

    // -- Citation row (33 columns) --
    let title = art.find("ArticleTitle").and_then(|e| e.inner_text());
    let vern_title = art.find("VernacularTitle").and_then(|e| e.inner_text());
    let coi = mc.find("CoiStatement").and_then(|e| e.inner_text());
    let date_comp = date_str(mc.find("DateCompleted"));
    let date_rev = date_str(mc.find("DateRevised"));
    let pagination = art.find("Pagination");

    w.row("citation", |r| {
        r.int(pmid); r.int(pmid_ver);
        r.str(mc.attr("Status")); r.str(Some(mc.attr("Owner").unwrap_or("NLM")));
        r.str(mc.attr("IndexingMethod"));
        r.str(journal_nlm_id); r.str(Some(journal_medline_ta));
        r.str(journal_title); r.str(journal_iso_abbr);
        r.str(issn_print); r.str(issn_electronic); r.str(issn_linking);
        r.str(journal_country);
        r.str(art.attr("PubModel"));
        r.str(ji.and_then(|j| j.find("Volume")).and_then(|e| e.text()));
        r.str(ji.and_then(|j| j.find("Issue")).and_then(|e| e.text()));
        r.str(ji.and_then(|j| j.attr("CitedMedium")));
        r.str(pub_y); r.str(pub_m); r.str(pub_d); r.str(medline_date);
        r.str(title.as_deref()); r.str(vern_title.as_deref());
        r.str(pagination.and_then(|p| p.find("MedlinePgn")).and_then(|e| e.text()));
        r.str(pagination.and_then(|p| p.find("StartPage")).and_then(|e| e.text()));
        r.str(pagination.and_then(|p| p.find("EndPage")).and_then(|e| e.text()));
        r.str(copyright_info.as_deref()); r.str(coi.as_deref());
        r.int_opt(mc.find("NumberOfReferences").and_then(|e| e.text()).and_then(|s| s.parse().ok()));
        r.str(date_comp.as_deref()); r.str(date_rev.as_deref());
        r.str(pd.and_then(|p| p.find("PublicationStatus")).and_then(|e| e.text()));
        r.str(Some(al_complete));
    })?;

    // -- Languages --
    for (i, lang) in art.find_all("Language").enumerate() {
        w.row("citation_tag", |r| {
            r.int(pmid); r.str(Some("language")); r.int(i as i32);
            r.str(Some(lang.text().unwrap_or("eng")));
        })?;
    }

    // -- ELocationID --
    for (i, el) in art.find_all("ELocationID").enumerate() {
        w.row("elocation_id", |r| {
            r.int(pmid); r.int(i as i32);
            r.str(Some(el.attr("EIdType").unwrap_or("")));
            r.str(Some(el.attr("ValidYN").unwrap_or("Y")));
            r.str(Some(el.text().unwrap_or("")));
        })?;
    }

    // -- ArticleDate --
    for (i, ad) in art.find_all("ArticleDate").enumerate() {
        w.row("article_date", |r| {
            r.int(pmid); r.int(i as i32); r.str(ad.attr("DateType"));
            push_ymd(r, Some(ad));
        })?;
    }

    // -- DataBankList --
    if let Some(dbl) = art.find("DataBankList") {
        let complete = dbl.attr("CompleteYN").unwrap_or("Y");
        for (i, db) in dbl.find_all("DataBank").enumerate() {
            w.row("data_bank", |r| {
                r.int(pmid); r.int(i as i32);
                r.str(db.find("DataBankName").and_then(|e| e.text()));
                r.str(Some(complete));
            })?;
            if let Some(anl) = db.find("AccessionNumberList") {
                for (j, an) in anl.find_all("AccessionNumber").enumerate() {
                    w.row("accession_number", |r| {
                        r.int(pmid); r.int(i as i32); r.int(j as i32);
                        r.str(Some(an.text().unwrap_or("")));
                    })?;
                }
            }
        }
    }

    // -- GrantList --
    if let Some(gl) = art.find("GrantList") {
        for (i, g) in gl.find_all("Grant").enumerate() {
            w.row("grant_info", |r| {
                r.int(pmid); r.int(i as i32);
                r.str(g.find("GrantID").and_then(|e| e.text()));
                r.str(g.find("Acronym").and_then(|e| e.text()));
                r.str(g.find("Agency").and_then(|e| e.text()));
                r.str(g.find("Country").and_then(|e| e.text()));
            })?;
        }
    }

    // -- PublicationTypeList --
    if let Some(ptl) = art.find("PublicationTypeList") {
        for (i, pt) in ptl.find_all("PublicationType").enumerate() {
            let text = pt.inner_text();
            w.row("publication_type", |r| {
                r.int(pmid); r.int(i as i32);
                r.str(Some(pt.attr("UI").unwrap_or("")));
                r.str(Some(text.as_deref().unwrap_or("")));
            })?;
        }
    }

    // -- ChemicalList --
    if let Some(cl) = mc.find("ChemicalList") {
        for (i, c) in cl.find_all("Chemical").enumerate() {
            let nos = c.find("NameOfSubstance");
            let name = nos.and_then(|n| n.inner_text());
            w.row("chemical", |r| {
                r.int(pmid); r.int(i as i32);
                r.str(c.find("RegistryNumber").and_then(|e| e.text()));
                r.str(Some(nos.and_then(|n| n.attr("UI")).unwrap_or("")));
                r.str(Some(name.as_deref().unwrap_or("")));
            })?;
        }
    }

    // -- SupplMeshList --
    if let Some(sml) = mc.find("SupplMeshList") {
        for (i, sm) in sml.find_all("SupplMeshName").enumerate() {
            let name = sm.inner_text();
            w.row("suppl_mesh", |r| {
                r.int(pmid); r.int(i as i32);
                r.str(Some(sm.attr("Type").unwrap_or("")));
                r.str(Some(sm.attr("UI").unwrap_or("")));
                r.str(Some(name.as_deref().unwrap_or("")));
            })?;
        }
    }

    // -- MeshHeadingList --
    if let Some(mhl) = mc.find("MeshHeadingList") {
        for (i, mh) in mhl.find_all("MeshHeading").enumerate() {
            let dn = mh.find("DescriptorName");
            let dn_text = dn.and_then(|d| d.inner_text());
            w.row("mesh_heading", |r| {
                r.int(pmid); r.int(i as i32);
                r.str(Some(dn.and_then(|d| d.attr("UI")).unwrap_or("")));
                r.str(Some(dn_text.as_deref().unwrap_or("")));
                r.str(Some(dn.and_then(|d| d.attr("MajorTopicYN")).unwrap_or("N")));
                r.str(dn.and_then(|d| d.attr("Type")));
                r.str(dn.and_then(|d| d.attr("AutoHM")));
            })?;
            for (j, qn) in mh.find_all("QualifierName").enumerate() {
                let qn_text = qn.inner_text();
                w.row("mesh_qualifier", |r| {
                    r.int(pmid); r.int(i as i32); r.int(j as i32);
                    r.str(Some(qn.attr("UI").unwrap_or("")));
                    r.str(Some(qn_text.as_deref().unwrap_or("")));
                    r.str(Some(qn.attr("MajorTopicYN").unwrap_or("N")));
                    r.str(qn.attr("AutoHM"));
                })?;
            }
        }
    }

    // -- KeywordList --
    for kwl in mc.find_all("KeywordList") {
        let owner = kwl.attr("Owner").unwrap_or("NLM");
        for (i, kw) in kwl.find_all("Keyword").enumerate() {
            let text = kw.inner_text();
            w.row("keyword", |r| {
                r.int(pmid); r.str(Some(owner)); r.int(i as i32);
                r.str(text.as_deref()); r.str(Some(kw.attr("MajorTopicYN").unwrap_or("N")));
            })?;
        }
    }

    // -- CommentsCorrectionsList --
    if let Some(ccl) = mc.find("CommentsCorrectionsList") {
        for (i, cc) in ccl.find_all("CommentsCorrections").enumerate() {
            let ref_source = cc.find("RefSource").and_then(|e| e.inner_text());
            let note = cc.find("Note").and_then(|e| e.inner_text());
            w.row("comment_correction", |r| {
                r.int(pmid); r.int(i as i32);
                r.str(Some(cc.attr("RefType").unwrap_or("")));
                r.str(ref_source.as_deref());
                r.int_opt(cc.find("PMID").and_then(|e| e.text()).and_then(|s| s.parse().ok()));
                r.str(note.as_deref());
            })?;
        }
    }

    // -- InvestigatorList --
    if let Some(il) = mc.find("InvestigatorList") {
        parse_persons(w, pmid, "investigator", il, "Investigator")?;
    }

    // -- PersonalNameSubjectList --
    if let Some(pnsl) = mc.find("PersonalNameSubjectList") {
        parse_persons(w, pmid, "subject", pnsl, "PersonalNameSubject")?;
    }

    // -- OtherAbstract --
    for oa in mc.find_all("OtherAbstract") {
        let oa_type_s = format!("other_{}", oa.attr("Type").unwrap_or("other"));
        let oa_lang = oa.attr("Language");
        for (i, at) in oa.find_all("AbstractText").enumerate() {
            let text = at.inner_text();
            w.row("abstract_text", |r| {
                r.int(pmid); r.str(Some(&oa_type_s)); r.str(oa_lang); r.int(i as i32);
                r.str(at.attr("Label")); r.str(at.attr("NlmCategory"));
                r.str(text.as_deref());
            })?;
        }
    }

    // -- OtherID --
    for (i, oid) in mc.find_all("OtherID").enumerate() {
        w.row("other_id", |r| {
            r.int(pmid); r.int(i as i32);
            r.str(Some(oid.attr("Source").unwrap_or("")));
            r.str(Some(oid.text().unwrap_or("")));
        })?;
    }

    // -- GeneralNote --
    for (i, gn) in mc.find_all("GeneralNote").enumerate() {
        let text = gn.inner_text();
        w.row("general_note", |r| {
            r.int(pmid); r.int(i as i32); r.str(gn.attr("Owner"));
            r.str(Some(text.as_deref().unwrap_or("")));
        })?;
    }

    // -- CitationSubset --
    for (i, cs) in mc.find_all("CitationSubset").enumerate() {
        w.row("citation_tag", |r| {
            r.int(pmid); r.str(Some("subset")); r.int(i as i32);
            r.str(Some(cs.text().unwrap_or("")));
        })?;
    }

    // -- SpaceFlightMission --
    for (i, sfm) in mc.find_all("SpaceFlightMission").enumerate() {
        w.row("citation_tag", |r| {
            r.int(pmid); r.str(Some("mission")); r.int(i as i32);
            r.str(Some(sfm.text().unwrap_or("")));
        })?;
    }

    // -- GeneSymbolList --
    if let Some(gsl) = mc.find("GeneSymbolList") {
        for (i, gs) in gsl.find_all("GeneSymbol").enumerate() {
            w.row("citation_tag", |r| {
                r.int(pmid); r.str(Some("gene")); r.int(i as i32);
                r.str(Some(gs.text().unwrap_or("")));
            })?;
        }
    }

    // -- PubmedData --
    if let Some(pd) = pd {
        // History
        if let Some(hist) = pd.find("History") {
            for ppd in hist.find_all("PubMedPubDate") {
                w.row("pub_date_history", |r| {
                    r.int(pmid); r.str(Some(ppd.attr("PubStatus").unwrap_or("")));
                    push_ymd(r, Some(ppd));
                    r.str(ppd.find("Hour").and_then(|e| e.text()));
                    r.str(ppd.find("Minute").and_then(|e| e.text()));
                })?;
            }
        }

        // ArticleIdList
        if let Some(ail) = pd.find("ArticleIdList") {
            for aid in ail.find_all("ArticleId") {
                w.row("article_id", |r| {
                    r.int(pmid);
                    r.str(Some(aid.attr("IdType").unwrap_or("")));
                    r.str(Some(aid.text().unwrap_or("")));
                })?;
            }
        }

        // ObjectList
        if let Some(ol) = pd.find("ObjectList") {
            for (i, obj) in ol.find_all("Object").enumerate() {
                w.row("object_info", |r| {
                    r.int(pmid); r.int(i as i32);
                    r.str(Some(obj.attr("Type").unwrap_or("")));
                })?;
                for (j, p) in obj.find_all("Param").enumerate() {
                    w.row("object_param", |r| {
                        r.int(pmid); r.int(i as i32); r.int(j as i32);
                        r.str(Some(p.attr("Name").unwrap_or("")));
                        r.str(p.text());
                    })?;
                }
            }
        }

        // ReferenceList
        for (li, rl) in pd.find_all("ReferenceList").enumerate() {
            let rl_title = rl.find("Title").and_then(|e| e.text());
            for (ri, refr) in rl.find_all("Reference").enumerate() {
                let cit_text = refr.find("Citation").and_then(|c| c.inner_text());
                w.row("reference", |r| {
                    r.int(pmid); r.int(li as i32); r.int(ri as i32);
                    r.str(cit_text.as_deref()); r.str(rl_title);
                })?;
                if let Some(rail) = refr.find("ArticleIdList") {
                    for raid in rail.find_all("ArticleId") {
                        w.row("reference_id", |r| {
                            r.int(pmid); r.int(li as i32); r.int(ri as i32);
                            r.str(Some(raid.attr("IdType").unwrap_or("")));
                            r.str(Some(raid.text().unwrap_or("")));
                        })?;
                    }
                }
            }
        }
    }

    Ok(())
}

// -- Person parsing (author/investigator/subject) -----------------------------

fn parse_persons(
    w: &mut WorkerWriter, pmid: i32, role: &str,
    parent: &Element, child_tag: &str,
) -> anyhow::Result<()> {
    for (i, el) in parent.find_all(child_tag).enumerate() {
        let cn = el.find("CollectiveName").and_then(|e| e.inner_text());
        w.row("person", |r| {
            r.int(pmid); r.str(Some(role)); r.int(i as i32);
            r.str(Some(el.attr("ValidYN").unwrap_or("Y")));
            r.str(el.attr("EqualContrib"));
            r.str(el.find("LastName").and_then(|e| e.text()));
            r.str(el.find("ForeName").and_then(|e| e.text()));
            r.str(el.find("Initials").and_then(|e| e.text()));
            r.str(el.find("Suffix").and_then(|e| e.text()));
            r.str(cn.as_deref());
        })?;

        for (j, ai) in el.find_all("AffiliationInfo").enumerate() {
            let aff = ai.find("Affiliation").and_then(|a| a.inner_text());
            if aff.is_some() {
                w.row("person_affiliation", |r| {
                    r.int(pmid); r.str(Some(role)); r.int(i as i32);
                    r.int(j as i32); r.str(aff.as_deref());
                })?;
            }
            for ident in ai.find_all("Identifier") {
                let val = ident.text().unwrap_or("");
                if !val.is_empty() {
                    w.row("person_identifier", |r| {
                        r.int(pmid); r.str(Some(role)); r.int(i as i32);
                        r.str(Some(ident.attr("Source").unwrap_or("")));
                        r.str(Some(val));
                    })?;
                }
            }
        }
        // Author-level identifiers (outside AffiliationInfo)
        for ident in el.find_all("Identifier") {
            let val = ident.text().unwrap_or("");
            if !val.is_empty() {
                w.row("person_identifier", |r| {
                    r.int(pmid); r.str(Some(role)); r.int(i as i32);
                    r.str(Some(ident.attr("Source").unwrap_or("")));
                    r.str(Some(val));
                })?;
            }
        }
    }
    Ok(())
}

// -- Streaming file parser ----------------------------------------------------

/// Parse a .xml.gz file, writing records directly into the worker writer.
/// Returns the number of articles parsed. Streams one article at a time
/// for constant memory (peak ~5KB per article, not ~90MB per file).
pub fn parse_file(path: &Path, w: &mut WorkerWriter) -> anyhow::Result<u64> {
    let file = std::fs::File::open(path)?;
    let reader: Box<dyn BufRead> = if path.extension().is_some_and(|e| e == "gz") {
        Box::new(BufReader::with_capacity(256 * 1024, GzDecoder::new(BufReader::new(file))))
    } else {
        Box::new(BufReader::with_capacity(256 * 1024, file))
    };

    let mut xml = Reader::from_reader(reader);
    xml.config_mut().trim_text(true);
    let mut buf = Vec::with_capacity(4096);
    let mut count: u64 = 0;

    loop {
        buf.clear();
        match xml.read_event_into(&mut buf) {
            Ok(Event::Start(ref e)) => {
                let name = e.name();
                let tag = name.as_ref();
                if tag == b"PubmedArticle" || tag == b"PubmedBookArticle" {
                    match build_subtree(&mut xml, e) {
                        Ok(tree) => {
                            parse_article(&tree, w)?;
                            count += 1;
                        }
                        Err(err) => eprintln!("warn: skipping malformed article: {err}"),
                    }
                } else if tag == b"DeleteCitation" {
                    if let Ok(tree) = build_subtree(&mut xml, e) {
                        for pmid_el in tree.find_all("PMID") {
                            if let Some(id) = pmid_el.text().and_then(|t| t.parse().ok()) {
                                w.row("delete_citation", |r| r.int(id))?;
                            }
                        }
                    }
                }
            }
            Ok(Event::Eof) => break,
            Err(e) => {
                eprintln!("warn: XML error at position {}: {e}", xml.error_position());
                break;
            }
            _ => {}
        }
    }
    Ok(count)
}
