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