tools.py

from __future__ import annotations

import json
import math
import os
import re
import time
from collections import Counter
from pathlib import Path
from typing import Any

import numpy as np
import pandas as pd
import torch
from adapters import AutoAdapterModel
from bertopic import BERTopic
from bertopic.vectorizers import ClassTfidfTransformer
from hdbscan import HDBSCAN
from huggingface_hub import InferenceClient
from sklearn.feature_extraction.text import ENGLISH_STOP_WORDS, CountVectorizer, TfidfVectorizer
from sklearn.metrics import silhouette_score
from sklearn.cluster import AgglomerativeClustering
from transformers import AutoTokenizer
from umap import UMAP


OUTPUT_DIR = Path("outputs")
DEFAULT_INPUT_CSV = (
    Path("data")
    / "ComputersinHumanBehavior_TopicModelling_Export_7553_for_app.csv"
)

SPECTER2_BASE_MODEL = "allenai/specter2_base"
SPECTER2_ADAPTER = "allenai/specter2"
HF_MODEL = "mistralai/Mistral-7B-Instruct-v0.3"
HF_TIMEOUT_SECONDS = 300
RANDOM_STATE = 42
MAX_RETRIES = 3
API_RETRY_DELAY = 2

PAJAIS_CATEGORIES = [
    "Digital Innovation & Entrepreneurship",
    "Electronic Commerce",
    "Electronic Government",
    "Enterprise Systems",
    "Green IS & Sustainability",
    "Healthcare Information Systems",
    "Human-Computer Interaction",
    "Information Security & Privacy",
    "IS Development & Project Management",
    "IS Education",
    "IS in Developing Countries",
    "IS Strategy & Governance",
    "IT Services & Outsourcing",
    "Knowledge Management",
    "Mobile & Ubiquitous Computing",
    "Online Communities & Social Media",
    "Philosophy & Research Methods",
    "Social & Ethical Issues",
    "Big Data Analytics",
    "AI & Machine Learning",
    "Blockchain",
    "Cloud Computing",
    "Internet of Things",
    "Digital Platforms",
    "Future of Work",
]

BOILERPLATE_PATTERNS = [
    r"©\s*\d{4}.*?(?:Elsevier|Ltd|Inc|reserved)",
    r"All rights reserved\.?",
    r"PII:?\s*S?\d+[-\d]+",
    r"https?://\S+",
    r"\S+@\S+\.\S+",
    r"doi:?\s*\S+",
    r"Crown Copyright.*?reserved\.?",
    r"Published by.*?(?:Elsevier|Springer|Wiley)",
]

EXTRA_STOPWORDS = {
    "study",
    "paper",
    "research",
    "results",
    "finding",
    "findings",
    "article",
    "author",
    "authors",
    "based",
    "using",
    "used",
    "examines",
    "examined",
    "investigates",
    "investigated",
    "analysis",
    "implications",
    "effect",
    "effects",
}

TEXT_TYPE_SETTINGS = {
    "abstract": {
        "vectorizer_min_df": 1,
        "vectorizer_max_df": 1.0,
        "ngram_range": (1, 2),
        "umap_n_neighbors": 10,
        "umap_n_components": 5,
        "hdbscan_min_cluster_size": 30,
        "hdbscan_min_samples": 8,
        "target_themes": 15,
        "target_topics_soft_max": 55,
    },
    "title": {
        "vectorizer_min_df": 1,
        "vectorizer_max_df": 1.0,
        "ngram_range": (1, 2),
        "umap_n_neighbors": 8,
        "umap_n_components": 5,
        "hdbscan_min_cluster_size": 20,
        "hdbscan_min_samples": 5,
        "target_themes": 15,
        "target_topics_soft_max": 40,
    },
}


def _ensure_output_dir(output_dir: Path = OUTPUT_DIR) -> Path:
    output_dir.mkdir(parents=True, exist_ok=True)
    return output_dir


def _write_json(path: Path, payload: dict[str, Any]) -> None:
    path.write_text(json.dumps(payload, indent=2, ensure_ascii=False), encoding="utf-8")


def _read_json(path: Path) -> dict[str, Any]:
    return json.loads(path.read_text(encoding="utf-8"))

def _clean_text(text: Any) -> str:
    value = "" if pd.isna(text) else str(text)
    for pattern in BOILERPLATE_PATTERNS:
        value = re.sub(pattern, " ", value, flags=re.IGNORECASE)
    value = value.replace("\u00a9", " ")
    value = re.sub(r"\s+", " ", value).strip()
    return value


def _normalize_label(value: str) -> str:
    label = re.sub(r"\s+", " ", str(value)).strip(" -:;,.")
    label = re.sub(r"[/|]+", " ", label)
    label = re.sub(r"\s+", " ", label).strip()
    return label


def _slugify_label(value: str) -> str:
    slug = re.sub(r"[^a-z0-9]+", "_", value.lower())
    return slug.strip("_")


def _keywords_to_label(keywords: list[str], max_words: int = 4) -> str:
    cleaned = [_normalize_label(kw.replace("_", " ")) for kw in keywords if kw.strip()]
    if not cleaned:
        return "Unlabeled Topic"

    selected: list[str] = []
    covered_tokens: set[str] = set()
    for phrase in sorted(cleaned, key=lambda item: (-len(item.split()), item)):
        tokens = [token for token in phrase.lower().split() if token]
        if not tokens:
            continue
        if all(token in covered_tokens for token in tokens):
            continue
        selected.append(phrase)
        covered_tokens.update(tokens)
        if len(selected) >= 3:
            break

    if not selected:
        selected = cleaned[:max_words]

    if len(selected) == 1:
        return selected[0].title()
    if len(selected) == 2:
        return f"{selected[0].title()} and {selected[1].title()}"
    return f"{selected[0].title()}, {selected[1].title()}, and {selected[2].title()}"


def _get_llm_client() -> InferenceClient | None:
    token = os.environ.get("HF_TOKEN") or os.environ.get("HUGGINGFACEHUB_API_TOKEN")
    if not token:
        print("HF_TOKEN is not set. Using deterministic fallback where available.")
        return None
    return InferenceClient(model=HF_MODEL, token=token, timeout=HF_TIMEOUT_SECONDS)


def _extract_message_content(response: Any) -> str:
    if isinstance(response, str):
        return response.strip()
    choices = getattr(response, "choices", None)
    if choices:
        message = getattr(choices[0], "message", None)
        if message is not None:
            content = getattr(message, "content", None)
            if content:
                return str(content).strip()
    if isinstance(response, dict):
        try:
            return str(response["choices"][0]["message"]["content"]).strip()
        except (KeyError, IndexError, TypeError):
            return str(response).strip()
    return str(response).strip()


def _chat_completion(client: InferenceClient, messages: list[dict[str, str]], max_tokens: int, temperature: float) -> str:
    try:
        response = client.chat_completion(
            messages=messages,
            max_tokens=max_tokens,
            temperature=temperature,
        )
        return _extract_message_content(response)
    except Exception:
        prompt = "\n\n".join(f"{message['role'].upper()}: {message['content']}" for message in messages)
        response = client.text_generation(
            prompt,
            max_new_tokens=max_tokens,
            temperature=temperature,
            do_sample=temperature > 0,
        )
        return _extract_message_content(response)


def _call_llm_json(prompt: str, max_tokens: int = 300) -> dict[str, Any] | None:
    client = _get_llm_client()
    if client is None:
        return None
    last_error: Exception | None = None
    for _ in range(MAX_RETRIES):
        try:
            text = _chat_completion(
                client,
                [
                    {
                        "role": "system",
                        "content": "Return only valid JSON. Do not wrap it in markdown. Do not include commentary outside the JSON object.",
                    },
                    {"role": "user", "content": prompt},
                ],
                max_tokens=max_tokens,
                temperature=0.2,
            )
            match = re.search(r"\{.*\}", text, flags=re.DOTALL)
            if not match:
                return None
            return json.loads(match.group(0))
        except Exception as exc:  # nosec - explicit retry path
            last_error = exc
            time.sleep(API_RETRY_DELAY)
    if last_error:
        print(f"LLM JSON call failed: {last_error}")
    return None


def _call_llm_text(prompt: str, max_tokens: int = 700) -> str:
    client = _get_llm_client()
    if client is None:
        return ""
    last_error: Exception | None = None
    for _ in range(MAX_RETRIES):
        try:
            return _chat_completion(
                client,
                [
                    {"role": "system", "content": "Write clean academic text only."},
                    {"role": "user", "content": prompt},
                ],
                max_tokens=max_tokens,
                temperature=0.3,
            )
        except Exception as exc:  # nosec - explicit retry path
            last_error = exc
            time.sleep(API_RETRY_DELAY)
    if last_error:
        print(f"LLM text call failed: {last_error}")
    return ""


class Specter2Embedder:
    def __init__(
        self,
        model_name: str = SPECTER2_BASE_MODEL,
        adapter_name: str = SPECTER2_ADAPTER,
        batch_size: int = 8,
        max_length: int = 512,
    ) -> None:
        self.model_name = model_name
        self.adapter_name = adapter_name
        self.batch_size = batch_size
        self.max_length = max_length
        self.device = "cuda" if torch.cuda.is_available() else "cpu"
        self.tokenizer: AutoTokenizer | None = None
        self.model: AutoAdapterModel | None = None

    def _load(self) -> None:
        if self.tokenizer is not None and self.model is not None:
            return
        self.tokenizer = AutoTokenizer.from_pretrained(self.model_name)
        self.model = AutoAdapterModel.from_pretrained(self.model_name)
        self.model.load_adapter(self.adapter_name, source="hf", load_as="specter2", set_active=True)
        self.model.set_active_adapters("specter2")
        self.model.to(self.device)
        self.model.eval()

    def encode(self, texts: list[str], cache_path: Path | None = None) -> np.ndarray:
        if cache_path and cache_path.exists():
            cached = np.load(cache_path)
            if len(cached) == len(texts):
                return cached

        self._load()
        assert self.tokenizer is not None
        assert self.model is not None

        embeddings: list[np.ndarray] = []
        total_batches = math.ceil(len(texts) / self.batch_size)
        for batch_index in range(total_batches):
            start = batch_index * self.batch_size
            end = start + self.batch_size
            batch = texts[start:end]
            encoded = self.tokenizer(
                batch,
                padding=True,
                truncation=True,
                max_length=self.max_length,
                return_token_type_ids=False,
                return_tensors="pt",
            )
            encoded = {key: value.to(self.device) for key, value in encoded.items()}
            with torch.no_grad():
                outputs = self.model(**encoded)
                batch_embeddings = outputs.last_hidden_state[:, 0, :]
                batch_embeddings = torch.nn.functional.normalize(batch_embeddings, p=2, dim=1)
            embeddings.append(batch_embeddings.cpu().numpy().astype(np.float32))
            print(f"SPECTER2 batch {batch_index + 1}/{total_batches} complete")

        matrix = np.vstack(embeddings)
        if cache_path:
            np.save(cache_path, matrix)
        return matrix


def _prepare_embedding_text(df: pd.DataFrame, text_type: str, sep_token: str) -> list[str]:
    if text_type == "title":
        return df["Title_Clean"].tolist()
    return [
        f"{title}{sep_token}{abstract}".strip()
        for title, abstract in zip(df["Title_Clean"], df["Abstract_Clean"], strict=True)
    ]


def _load_clean_papers(output_dir: Path = OUTPUT_DIR) -> pd.DataFrame:
    path = output_dir / "papers_clean.csv"
    if not path.exists():
        raise FileNotFoundError("Run load_scopus_csv first.")
    return pd.read_csv(path, keep_default_na=False)


def load_scopus_csv(file_path: str, output_dir: Path = OUTPUT_DIR) -> dict[str, Any]:
    output_dir = _ensure_output_dir(output_dir)
    df = pd.read_csv(file_path, keep_default_na=False)
    df["Title_Clean"] = df["Title"].apply(_clean_text)
    df["Abstract_Clean"] = df["Abstract"].apply(_clean_text)
    df["Author Keywords"] = df["Author Keywords"].fillna("").astype(str)
    df["Cited by"] = pd.to_numeric(df["Cited by"], errors="coerce").fillna(0).astype(int)
    df["Year"] = pd.to_numeric(df["Year"], errors="coerce").astype(int)
    if "Sr No" not in df.columns:
        df.insert(0, "Sr No", range(1, len(df) + 1))
    df.to_csv(output_dir / "papers_clean.csv", index=False, encoding="utf-8-sig")

    stats = {
        "total_papers": int(len(df)),
        "year_range": f"{int(df['Year'].min())} - {int(df['Year'].max())}",
        "columns": list(df.columns),
        "sample_titles": df["Title_Clean"].head(3).tolist(),
        "sample_abstracts": df["Abstract_Clean"].head(3).tolist(),
        "timestamp": pd.Timestamp.utcnow().isoformat(),
    }
    _write_json(output_dir / "data_stats.json", stats)
    return stats


def _build_topic_model(text_type: str) -> BERTopic:
    settings = TEXT_TYPE_SETTINGS[text_type]
    stop_words = sorted(ENGLISH_STOP_WORDS.union(EXTRA_STOPWORDS))
    vectorizer_model = CountVectorizer(
        stop_words=stop_words,
        ngram_range=settings["ngram_range"],
        min_df=settings["vectorizer_min_df"],
        max_df=settings["vectorizer_max_df"],
    )
    umap_model = UMAP(
        n_neighbors=settings["umap_n_neighbors"],
        n_components=settings["umap_n_components"],
        min_dist=0.0,
        metric="cosine",
        random_state=RANDOM_STATE,
        low_memory=True,
    )
    hdbscan_model = HDBSCAN(
        min_cluster_size=settings["hdbscan_min_cluster_size"],
        min_samples=settings["hdbscan_min_samples"],
        metric="euclidean",
        cluster_selection_method="eom",
        prediction_data=True,
    )
    return BERTopic(
        umap_model=umap_model,
        hdbscan_model=hdbscan_model,
        vectorizer_model=vectorizer_model,
        ctfidf_model=ClassTfidfTransformer(reduce_frequent_words=True),
        top_n_words=10,
        verbose=True,
        calculate_probabilities=False,
        low_memory=True,
        min_topic_size=settings["hdbscan_min_cluster_size"],
    )


def _compute_topic_diversity(topic_model: BERTopic, top_n_words: int = 10) -> float:
    words: list[str] = []
    for topic_id, values in topic_model.get_topics().items():
        if topic_id == -1:
            continue
        words.extend(word for word, _ in values[:top_n_words])
    if not words:
        return 0.0
    return len(set(words)) / len(words)


def _compute_silhouette(embeddings: np.ndarray, topics: list[int]) -> float | None:
    valid_indices = [idx for idx, topic_id in enumerate(topics) if topic_id != -1]
    valid_labels = [topics[idx] for idx in valid_indices]
    if len(set(valid_labels)) < 2 or len(valid_indices) < 200:
        return None
    sample_size = min(1500, len(valid_indices))
    sampled = valid_indices[:sample_size]
    try:
        return float(
            silhouette_score(
                embeddings[sampled],
                [topics[idx] for idx in sampled],
                metric="cosine",
            )
        )
    except Exception:
        return None


def _topic_keywords(topic_model: BERTopic, topic_id: int, top_n_words: int = 8) -> list[str]:
    return [word for word, _ in topic_model.get_topic(topic_id)[:top_n_words]]


def _representative_docs_map(topic_model: BERTopic) -> dict[int, list[str]]:
    raw_docs = topic_model.get_representative_docs() or {}
    return {int(topic_id): [str(doc) for doc in docs] for topic_id, docs in raw_docs.items()}


def _relabel_topics_with_fallback(topic_model: BERTopic, text_type: str, output_dir: Path) -> dict[str, Any]:
    topic_info = topic_model.get_topic_info()
    representative_docs = _representative_docs_map(topic_model)
    labels: dict[str, Any] = {}

    for _, row in topic_info.iterrows():
        topic_id = int(row["Topic"])
        if topic_id == -1:
            continue
        keywords = _topic_keywords(topic_model, topic_id)
        base_label = _keywords_to_label(keywords)
        docs = representative_docs.get(topic_id, [])[:3]
        prompt = f"""
Create one concise human-readable research topic label for a BERTopic cluster from Computers in Human Behavior.

Rules:
1. 3 to 6 words
2. academic and specific
3. do not output generic labels like "topic", "cluster", "research topic", or repeated stopwords
4. prefer phenomenon or domain wording over raw keywords
5. return the label only, no explanation

Topic keywords: {", ".join(keywords)}
Representative texts:
{chr(10).join("- " + doc[:500] for doc in docs)}
"""
        response_text = _call_llm_text(prompt, max_tokens=40)
        label = base_label
        confidence = 0.5
        if response_text:
            candidate = _normalize_label(response_text.splitlines()[0])
            if (
                candidate
                and not re.fullmatch(r"topic\s*\d+", candidate, flags=re.IGNORECASE)
                and " and of " not in candidate.lower()
            ):
                label = candidate
                confidence = 0.8

        labels[str(topic_id)] = {
            "label": label,
            "confidence": confidence,
            "topic_size": int(row["Count"]),
            "keywords": keywords,
            "representative_texts": docs,
        }

    payload = {
        "text_type": text_type,
        "num_labeled": len(labels),
        "topics": labels,
        "timestamp": pd.Timestamp.utcnow().isoformat(),
    }
    _write_json(output_dir / f"labels_{text_type}.json", payload)
    return payload


def _fallback_theme_groups(labels_payload: dict[str, Any], target_themes: int) -> dict[str, list[str]]:
    rows = []
    for topic_id, info in labels_payload["topics"].items():
        corpus_text = " ".join([info["label"]] + info["keywords"][:6])
        rows.append((topic_id, corpus_text, info["topic_size"]))
    texts = [item[1] for item in rows]
    ids = [item[0] for item in rows]

    if len(texts) <= target_themes:
        return {labels_payload["topics"][topic_id]["label"]: [topic_id] for topic_id in ids}

    vectorizer = TfidfVectorizer(stop_words="english", ngram_range=(1, 2))
    matrix = vectorizer.fit_transform(texts)
    clustering = AgglomerativeClustering(n_clusters=target_themes, metric="cosine", linkage="average")
    cluster_ids = clustering.fit_predict(matrix.toarray())

    groups: dict[int, list[str]] = {}
    for topic_id, cluster_id in zip(ids, cluster_ids, strict=True):
        groups.setdefault(int(cluster_id), []).append(topic_id)

    theme_groups: dict[str, list[str]] = {}
    for cluster_id, topic_ids in groups.items():
        all_keywords = Counter()
        for topic_id in topic_ids:
            all_keywords.update(labels_payload["topics"][topic_id]["keywords"][:5])
        label = _keywords_to_label([word for word, _ in all_keywords.most_common(4)])
        theme_groups[f"{label} Theme {cluster_id + 1}"] = topic_ids
    return theme_groups


def consolidate_into_themes(
    text_type: str,
    target_themes: int = 15,
    output_dir: Path = OUTPUT_DIR,
) -> dict[str, Any]:
    output_dir = _ensure_output_dir(output_dir)
    labels_payload = _read_json(output_dir / f"labels_{text_type}.json")
    topics = labels_payload["topics"]
    topic_rows = []
    for topic_id, info in topics.items():
        topic_rows.append(
            {
                "topic_id": topic_id,
                "label": info["label"],
                "size": info["topic_size"],
                "keywords": ", ".join(info["keywords"][:6]),
            }
        )
    topic_rows = sorted(topic_rows, key=lambda item: item["size"], reverse=True)
    prompt = f"""
Group these CHB topic labels into approximately {target_themes} broader human-readable themes.

Topics:
{chr(10).join(f"- {row['topic_id']}: {row['label']} | size={row['size']} | keywords={row['keywords']}" for row in topic_rows)}

Return JSON only:
{{
  "themes": [
    {{
      "theme_name": "Human-readable theme",
      "topic_ids": ["0", "4", "18"]
    }}
  ]
}}
"""
    response = _call_llm_json(prompt, max_tokens=1200)
    if response and isinstance(response.get("themes"), list):
        theme_groups = {
            _normalize_label(item["theme_name"]): [str(topic_id) for topic_id in item["topic_ids"]]
            for item in response["themes"]
            if item.get("theme_name") and item.get("topic_ids")
        }
    else:
        theme_groups = _fallback_theme_groups(labels_payload, target_themes)

    themes: dict[str, Any] = {}
    for theme_name, topic_ids in theme_groups.items():
        member_labels = [topics[str(topic_id)]["label"] for topic_id in topic_ids if str(topic_id) in topics]
        member_keywords = []
        total_size = 0
        for topic_id in topic_ids:
            info = topics.get(str(topic_id))
            if not info:
                continue
            member_keywords.extend(info["keywords"][:4])
            total_size += int(info["topic_size"])
        themes[theme_name] = {
            "topic_ids": [str(topic_id) for topic_id in topic_ids],
            "member_labels": member_labels,
            "keywords": [word for word, _ in Counter(member_keywords).most_common(8)],
            "total_size": total_size,
        }

    payload = {
        "text_type": text_type,
        "num_themes": len(themes),
        "themes": dict(sorted(themes.items(), key=lambda item: item[1]["total_size"], reverse=True)),
        "timestamp": pd.Timestamp.utcnow().isoformat(),
    }
    _write_json(output_dir / f"themes_{text_type}.json", payload)
    return {
        "num_themes": payload["num_themes"],
        "themes": {name: info["total_size"] for name, info in payload["themes"].items()},
    }


def _fallback_taxonomy(theme_name: str, keywords: list[str]) -> dict[str, Any]:
    tokens = set(re.findall(r"[a-z]+", f"{theme_name} {' '.join(keywords)}".lower()))
    category_tokens = {
        category: set(re.findall(r"[a-z]+", category.lower()))
        for category in PAJAIS_CATEGORIES
    }
    scored = []
    for category, candidate_tokens in category_tokens.items():
        overlap = len(tokens & candidate_tokens)
        scored.append((category, overlap))
    scored.sort(key=lambda item: item[1], reverse=True)
    best_category, score = scored[0]
    if score == 0:
        return {
            "mapping_type": "NOVEL",
            "pajais_category": None,
            "confidence": 0.45,
            "reasoning": "No direct lexical overlap with PAJAIS categories.",
        }
    return {
        "mapping_type": "MAPPED",
        "pajais_category": best_category,
        "confidence": round(min(0.85, 0.45 + 0.10 * score), 2),
        "reasoning": "Fallback lexical match against PAJAIS categories.",
    }


def compare_with_taxonomy(text_type: str, output_dir: Path = OUTPUT_DIR) -> dict[str, Any]:
    output_dir = _ensure_output_dir(output_dir)
    payload = _read_json(output_dir / f"themes_{text_type}.json")
    themes = payload["themes"]
    taxonomy_map: dict[str, Any] = {}
    pajais_str = "\n".join(f"{index + 1}. {category}" for index, category in enumerate(PAJAIS_CATEGORIES))

    for theme_name, info in themes.items():
        prompt = f"""
Map this CHB research theme to the PAJAIS 25-category taxonomy.

Theme: {theme_name}
Member labels: {", ".join(info["member_labels"][:8])}
Keywords: {", ".join(info["keywords"][:8])}

PAJAIS categories:
{pajais_str}

Return JSON only:
{{
  "mapping_type": "MAPPED or NOVEL",
  "pajais_category": "Category name or null",
  "confidence": 0.0,
  "reasoning": "One short sentence"
}}
"""
        response = _call_llm_json(prompt, max_tokens=220)
        mapping = response if response else _fallback_taxonomy(theme_name, info["keywords"])
        taxonomy_map[theme_name] = {
            "mapping_type": mapping["mapping_type"],
            "pajais_category": mapping.get("pajais_category"),
            "confidence": float(mapping.get("confidence", 0.5)),
            "reasoning": mapping.get("reasoning", ""),
            "size": int(info["total_size"]),
            "member_labels": info["member_labels"],
            "keywords": info["keywords"],
            "topic_ids": info["topic_ids"],
        }

    mapped_count = sum(1 for item in taxonomy_map.values() if item["mapping_type"] == "MAPPED")
    novel_count = len(taxonomy_map) - mapped_count
    result = {
        "text_type": text_type,
        "total_themes": len(taxonomy_map),
        "mapped_count": mapped_count,
        "novel_count": novel_count,
        "taxonomy_map": taxonomy_map,
        "timestamp": pd.Timestamp.utcnow().isoformat(),
    }
    _write_json(output_dir / f"taxonomy_map_{text_type}.json", result)
    return {
        "mapped_count": mapped_count,
        "novel_count": novel_count,
        "mapped_themes": [name for name, info in taxonomy_map.items() if info["mapping_type"] == "MAPPED"],
        "novel_themes": [name for name, info in taxonomy_map.items() if info["mapping_type"] == "NOVEL"],
    }


def generate_comparison_csv(output_dir: Path = OUTPUT_DIR) -> dict[str, Any]:
    output_dir = _ensure_output_dir(output_dir)
    abstract_tax = _read_json(output_dir / "taxonomy_map_abstract.json")
    title_tax = _read_json(output_dir / "taxonomy_map_title.json")
    abstract_themes = abstract_tax["taxonomy_map"]
    title_themes = title_tax["taxonomy_map"]

    normalized_abstract = {_slugify_label(name): name for name in abstract_themes}
    normalized_title = {_slugify_label(name): name for name in title_themes}
    overlap_keys = set(normalized_abstract) & set(normalized_title)

    rows = []
    for norm_key in sorted(set(normalized_abstract) | set(normalized_title)):
        abstract_name = normalized_abstract.get(norm_key)
        title_name = normalized_title.get(norm_key)
        abstract_info = abstract_themes.get(abstract_name, {})
        title_info = title_themes.get(title_name, {})
        rows.append(
            {
                "Normalized_Theme_Key": norm_key,
                "Abstract_Theme": abstract_name or "",
                "Title_Theme": title_name or "",
                "In_Abstracts": "Yes" if abstract_name else "No",
                "In_Titles": "Yes" if title_name else "No",
                "Abstract_Size": abstract_info.get("size", 0),
                "Title_Size": title_info.get("size", 0),
                "Abstract_Mapping": abstract_info.get("mapping_type", ""),
                "Title_Mapping": title_info.get("mapping_type", ""),
                "Abstract_PAJAIS": abstract_info.get("pajais_category", ""),
                "Title_PAJAIS": title_info.get("pajais_category", ""),
            }
        )

    comparison_df = pd.DataFrame(rows).sort_values(
        ["In_Abstracts", "In_Titles", "Abstract_Size", "Title_Size"],
        ascending=[False, False, False, False],
    )
    comparison_df.to_csv(output_dir / "comparison.csv", index=False, encoding="utf-8-sig")

    result = {
        "total_unique_themes": int(len(rows)),
        "themes_in_both": int(len(overlap_keys)),
        "abstract_only": int(len(set(normalized_abstract) - overlap_keys)),
        "title_only": int(len(set(normalized_title) - overlap_keys)),
        "abstract_only_themes": [normalized_abstract[key] for key in sorted(set(normalized_abstract) - overlap_keys)],
        "title_only_themes": [normalized_title[key] for key in sorted(set(normalized_title) - overlap_keys)],
        "themes_in_both_labels": [
            {
                "normalized_key": key,
                "abstract_theme": normalized_abstract[key],
                "title_theme": normalized_title[key],
            }
            for key in sorted(overlap_keys)
        ],
        "timestamp": pd.Timestamp.utcnow().isoformat(),
    }
    _write_json(output_dir / "comparison_summary.json", result)
    return result


def export_narrative(output_dir: Path = OUTPUT_DIR) -> dict[str, Any]:
    output_dir = _ensure_output_dir(output_dir)
    stats = _read_json(output_dir / "data_stats.json")
    abstract_tax = _read_json(output_dir / "taxonomy_map_abstract.json")
    title_tax = _read_json(output_dir / "taxonomy_map_title.json")
    comparison = _read_json(output_dir / "comparison_summary.json")

    prompt = f"""
Write a concise 500-word academic results narrative for the CHB BERTopic V3 pipeline.

Facts to use:
- Corpus size: {stats['total_papers']} papers
- Year range: {stats['year_range']}
- Abstract themes: {abstract_tax['total_themes']}
- Abstract mapped themes: {abstract_tax['mapped_count']}
- Abstract novel themes: {abstract_tax['novel_count']}
- Title themes: {title_tax['total_themes']}
- Title mapped themes: {title_tax['mapped_count']}
- Title novel themes: {title_tax['novel_count']}
- Theme overlap between abstract and title sets: {comparison['themes_in_both']}

Requirements:
1. Mention SPECTER2, UMAP, HDBSCAN, BERTopic, and PAJAIS.
2. Stay factual.
3. Do not invent percentages not implied by the data.
4. Use simple academic English.
"""
    narrative = _call_llm_text(prompt, max_tokens=900)
    if not narrative:
        narrative = (
            f"This CHB BERTopic V3 pipeline analyzed {stats['total_papers']} CHB papers across {stats['year_range']}. "
            "The workflow used SPECTER2 embeddings, UMAP reduction, HDBSCAN clustering, BERTopic topic representation, "
            "and PAJAIS taxonomy mapping. The resulting abstract-side and title-side themes are saved in the output folder "
            "for direct inspection and later paper writing."
        )
    (output_dir / "narrative.txt").write_text(narrative, encoding="utf-8")
    meta = {
        "word_count": len(narrative.split()),
        "timestamp": pd.Timestamp.utcnow().isoformat(),
    }
    _write_json(output_dir / "narrative_meta.json", meta)
    return {"word_count": meta["word_count"], "preview": narrative[:500]}


def run_bertopic_discovery(text_type: str, output_dir: Path = OUTPUT_DIR) -> dict[str, Any]:
    text_type = text_type.lower()
    if text_type not in TEXT_TYPE_SETTINGS:
        raise ValueError("text_type must be 'abstract' or 'title'.")

    output_dir = _ensure_output_dir(output_dir)
    papers = _load_clean_papers(output_dir)
    settings = TEXT_TYPE_SETTINGS[text_type]

    analysis_docs = papers["Abstract_Clean"].tolist() if text_type == "abstract" else papers["Title_Clean"].tolist()
    embedder = Specter2Embedder(batch_size=8)
    sep_token = AutoTokenizer.from_pretrained(SPECTER2_BASE_MODEL).sep_token
    embedding_texts = _prepare_embedding_text(papers, text_type, sep_token)
    embeddings = embedder.encode(embedding_texts, cache_path=output_dir / f"embeddings_{text_type}.npy")

    topic_model = _build_topic_model(text_type)
    topics, _ = topic_model.fit_transform(analysis_docs, embeddings)

    noise_rate = float(sum(topic_id == -1 for topic_id in topics) / len(topics))
    discovered_topics = len(set(topic_id for topic_id in topics if topic_id != -1))
    if noise_rate > 0.10 and discovered_topics >= 10:
        reduced_topics = topic_model.reduce_outliers(
            analysis_docs,
            topics,
            embeddings=embeddings,
            strategy="embeddings",
        )
        if reduced_topics != topics:
            topics = reduced_topics
            topic_model.update_topics(
                analysis_docs,
                topics=topics,
                vectorizer_model=topic_model.vectorizer_model,
                ctfidf_model=topic_model.ctfidf_model,
            )

    topic_info = topic_model.get_topic_info()
    representative_docs = _representative_docs_map(topic_model)

    assignment_df = pd.DataFrame(
        {
            "paper_id": papers["Sr No"],
            "year": papers["Year"],
            "cited_by": papers["Cited by"],
            "title": papers["Title_Clean"],
            "text": analysis_docs,
            "topic": topics,
        }
    )
    assignment_df["topic_size"] = assignment_df["topic"].map(topic_info.set_index("Topic")["Count"]).fillna(0).astype(int)
    assignment_df.to_csv(output_dir / f"topic_assignments_{text_type}.csv", index=False, encoding="utf-8-sig")

    topic_rows = []
    for _, row in topic_info.iterrows():
        topic_id = int(row["Topic"])
        topic_rows.append(
            {
                "topic_id": topic_id,
                "count": int(row["Count"]),
                "name": row.get("Name", ""),
                "keywords": ", ".join(_topic_keywords(topic_model, topic_id)) if topic_id != -1 else "",
                "representative_doc_1": representative_docs.get(topic_id, [""])[0] if representative_docs.get(topic_id) else "",
            }
        )
    pd.DataFrame(topic_rows).to_csv(output_dir / f"topic_info_{text_type}.csv", index=False, encoding="utf-8-sig")

    topic_model.save(output_dir / f"bertopic_model_{text_type}", serialization="safetensors", save_ctfidf=True)

    silhouette = _compute_silhouette(embeddings, topics)
    quality = {
        "text_type": text_type,
        "total_documents": int(len(analysis_docs)),
        "num_topics_excluding_noise": int(sum(topic_id != -1 for topic_id in topic_info["Topic"])),
        "noise_documents": int(sum(topic_id == -1 for topic_id in topics)),
        "noise_rate": round(float(sum(topic_id == -1 for topic_id in topics) / len(topics)), 4),
        "largest_topic_size": int(topic_info[topic_info["Topic"] != -1]["Count"].max()) if (topic_info["Topic"] != -1).any() else 0,
        "topic_diversity": round(_compute_topic_diversity(topic_model), 4),
        "silhouette_sample": round(silhouette, 4) if silhouette is not None else None,
        "umap_n_neighbors": settings["umap_n_neighbors"],
        "hdbscan_min_cluster_size": settings["hdbscan_min_cluster_size"],
        "timestamp": pd.Timestamp.utcnow().isoformat(),
    }
    _write_json(output_dir / f"quality_{text_type}.json", quality)

    summary_topics = topic_info[topic_info["Topic"] != -1].head(10)
    summaries = {}
    for _, row in summary_topics.iterrows():
        topic_id = int(row["Topic"])
        summaries[str(topic_id)] = {
            "size": int(row["Count"]),
            "keywords": _topic_keywords(topic_model, topic_id),
            "sample": (representative_docs.get(topic_id, [""])[0] or "")[:200],
        }

    payload = {
        "column": "Abstract" if text_type == "abstract" else "Title",
        "text_type": text_type,
        "total_texts": int(len(analysis_docs)),
        "num_topics": int(sum(topic_info["Topic"] != -1)),
        "top_10_topics": summaries,
        "timestamp": pd.Timestamp.utcnow().isoformat(),
    }
    _write_json(output_dir / f"summaries_{text_type}.json", payload)
    return payload


def label_topics_with_llm(text_type: str, output_dir: Path = OUTPUT_DIR) -> dict[str, Any]:
    output_dir = _ensure_output_dir(output_dir)
    topic_model = BERTopic.load(output_dir / f"bertopic_model_{text_type}")
    payload = _relabel_topics_with_fallback(topic_model, text_type, output_dir)
    return {
        "num_labeled": payload["num_labeled"],
        "sample_labels": {
            topic_id: info["label"]
            for topic_id, info in list(payload["topics"].items())[:10]
        },
    }


def run_full_pipeline(file_path: str | None = None, output_dir: Path = OUTPUT_DIR) -> dict[str, Any]:
    output_dir = _ensure_output_dir(output_dir)
    input_path = file_path or str(DEFAULT_INPUT_CSV)
    stats = load_scopus_csv(input_path, output_dir=output_dir)
    abstract = run_bertopic_discovery("abstract", output_dir=output_dir)
    title = run_bertopic_discovery("title", output_dir=output_dir)
    abstract_labels = label_topics_with_llm("abstract", output_dir=output_dir)
    title_labels = label_topics_with_llm("title", output_dir=output_dir)
    abstract_themes = consolidate_into_themes("abstract", TEXT_TYPE_SETTINGS["abstract"]["target_themes"], output_dir)
    title_themes = consolidate_into_themes("title", TEXT_TYPE_SETTINGS["title"]["target_themes"], output_dir)
    abstract_taxonomy = compare_with_taxonomy("abstract", output_dir=output_dir)
    title_taxonomy = compare_with_taxonomy("title", output_dir=output_dir)
    comparison = generate_comparison_csv(output_dir=output_dir)
    narrative = export_narrative(output_dir=output_dir)

    manifest = {
        "input_file": str(input_path),
        "output_dir": str(output_dir),
        "method": {
            "topic_model": "BERTopic",
            "embedding_model": "SPECTER2",
            "embedding_input_title_abstract_for_abstract_run": True,
            "embedding_input_title_only_for_title_run": True,
            "dimension_reduction": "UMAP",
            "clustering": "HDBSCAN",
            "llm_backend": f"Hugging Face Inference: {HF_MODEL}",
        },
        "stats": stats,
        "abstract": abstract,
        "title": title,
        "abstract_labels": abstract_labels,
        "title_labels": title_labels,
        "abstract_themes": abstract_themes,
        "title_themes": title_themes,
        "abstract_taxonomy": abstract_taxonomy,
        "title_taxonomy": title_taxonomy,
        "comparison": comparison,
        "narrative": narrative,
        "timestamp": pd.Timestamp.utcnow().isoformat(),
    }
    _write_json(output_dir / "run_manifest.json", manifest)
    return manifest