Spaces:
Running
Running
| """Verified, database-first journal discovery helpers for ResearchBee. | |
| The module owns deterministic candidate generation and hard filtering for both | |
| manuscript matching and subject browsing. Language models may explain or rerank | |
| only records selected from ``journals.db``; they never supply journal identity, | |
| ISSNs, metrics, indexation state, access flags, or policy facts. | |
| """ | |
| from __future__ import annotations | |
| import re | |
| from dataclasses import dataclass, field | |
| from typing import Any, Iterable, Sequence | |
| from researchbee_journals import JournalIndex, JournalRecord, Recommendation | |
| _Q_ORDER = {"Q1": 1, "Q2": 2, "Q3": 3, "Q4": 4} | |
| def normalise_title(value: str) -> str: | |
| return re.sub(r"[^a-z0-9]+", " ", str(value or "").lower()).strip() | |
| def split_title_list(value: str | Iterable[str] | None) -> list[str]: | |
| if value is None: | |
| return [] | |
| if isinstance(value, str): | |
| # Commas are part of canonical titles such as Accounting, | |
| # Organizations and Society. Separate multiple titles with newlines | |
| # or semicolons; iterable callers already provide explicit boundaries. | |
| raw = re.split(r"[;\n]+", value) | |
| else: | |
| raw = [str(v) for v in value] | |
| out: list[str] = [] | |
| seen: set[str] = set() | |
| for item in raw: | |
| item = " ".join(item.split()) | |
| key = normalise_title(item) | |
| if item and key and key not in seen: | |
| seen.add(key) | |
| out.append(item) | |
| return out | |
| def ranking_filter_to_quartiles(value: str) -> tuple[list[str], str | None]: | |
| """Return exact allowed quartiles plus any explanatory legacy note.""" | |
| v = " ".join(str(value or "No filter").split()).lower() | |
| if v in {"q1", "q1 only"}: | |
| return ["Q1"], None | |
| if v in {"q2", "q2 only"}: | |
| return ["Q2"], None | |
| if v in {"q1 or q2", "q1/q2", "q1 and q2"}: | |
| return ["Q1", "Q2"], None | |
| # Older front ends offered exact top-percent choices even though the local | |
| # source does not carry a defensible category percentile for every title. | |
| if v.startswith("top "): | |
| return ["Q1"], ( | |
| "The legacy top-percentage choice was interpreted as Q1. Exact top " | |
| "1/5/10/15/25% eligibility was not inferred from quartile data." | |
| ) | |
| return [], None | |
| def access_mode_key(value: str) -> str: | |
| v = " ".join(str(value or "Any journal").split()).lower() | |
| if "diamond" in v or "no apc" in v: | |
| return "diamond" | |
| if "open" in v and "any" not in v: | |
| return "open" | |
| return "any" | |
| def access_model(record: JournalRecord) -> dict[str, Any]: | |
| if record.oa_diamond: | |
| return { | |
| "code": "diamond", | |
| "label": "Diamond open access", | |
| "author_fee_status": "No APC indicated by the source record", | |
| "verification": "Journal-level source flag; verify the current journal website", | |
| } | |
| if record.oa_unpaywall or record.scimago_oa: | |
| return { | |
| "code": "open", | |
| "label": "Open-access journal", | |
| "author_fee_status": "Current APC not available in the local journal index", | |
| "verification": "Journal-level OA flag; verify the current fee and waiver terms", | |
| } | |
| return { | |
| "code": "other_or_unknown", | |
| "label": "Subscription, hybrid, or access model not confirmed", | |
| "author_fee_status": "Current APC not available in the local journal index", | |
| "verification": "Check the journal website and DOAJ before submission", | |
| } | |
| def matching_subjects(record: JournalRecord, asjc_codes: Sequence[str]) -> list[dict]: | |
| code_set = set(asjc_codes or []) | |
| return [dict(s) for s in record.subjects if s.get("asjc") in code_set] | |
| def quartile_for(record: JournalRecord, asjc_codes: Sequence[str]) -> tuple[str | None, list[dict]]: | |
| matches = matching_subjects(record, asjc_codes) | |
| ranked = [s for s in matches if s.get("quartile") in _Q_ORDER] | |
| if ranked: | |
| best = min(ranked, key=lambda s: _Q_ORDER[s["quartile"]]) | |
| return best["quartile"], matches | |
| # A quartile in another field is not evidence for the selected subject. | |
| return (None if asjc_codes else record.sjr_best_quartile), matches | |
| def _normalise_scores(recommendations: Sequence[Recommendation]) -> dict[str, float]: | |
| if not recommendations: | |
| return {} | |
| vals = [float(r.score) for r in recommendations] | |
| lo, hi = min(vals), max(vals) | |
| span = hi - lo | |
| if span <= 1e-9: | |
| return {r.journal.scopus_id: 0.5 for r in recommendations} | |
| return {r.journal.scopus_id: (float(r.score) - lo) / span for r in recommendations} | |
| class DiscoveryPreferences: | |
| ranking_preference: str = "No filter" | |
| access_preference: str = "Any journal" | |
| sharing_requirement: bool = False | |
| intended_repository: str = "Institutional repository" | |
| apc_budget: str = "" | |
| funder: str = "" | |
| preferred_journals: list[str] = field(default_factory=list) | |
| avoid_journals: list[str] = field(default_factory=list) | |
| def from_values( | |
| cls, | |
| *, | |
| ranking_preference: str = "No filter", | |
| access_preference: str = "Any journal", | |
| sharing_requirement: bool | str = False, | |
| intended_repository: str = "Institutional repository", | |
| apc_budget: str = "", | |
| funder: str = "", | |
| preferred_journals: str | Iterable[str] | None = None, | |
| avoid_journals: str | Iterable[str] | None = None, | |
| ) -> "DiscoveryPreferences": | |
| if isinstance(sharing_requirement, str): | |
| sharing = sharing_requirement.strip().lower() in { | |
| "1", "true", "yes", "required", "on", | |
| } | |
| else: | |
| sharing = bool(sharing_requirement) | |
| return cls( | |
| ranking_preference=" ".join(str(ranking_preference or "No filter").split()), | |
| access_preference=" ".join(str(access_preference or "Any journal").split()), | |
| sharing_requirement=sharing, | |
| intended_repository=" ".join(str(intended_repository or "Institutional repository").split()), | |
| apc_budget=" ".join(str(apc_budget or "").split()), | |
| funder=" ".join(str(funder or "").split()), | |
| preferred_journals=split_title_list(preferred_journals), | |
| avoid_journals=split_title_list(avoid_journals), | |
| ) | |
| def allowed_quartiles(self) -> list[str]: | |
| return ranking_filter_to_quartiles(self.ranking_preference)[0] | |
| def access_mode(self) -> str: | |
| return access_mode_key(self.access_preference) | |
| class CandidatePool: | |
| recommendations: list[Recommendation] | |
| score_ratios: dict[str, float] | |
| asjc_rows: list[dict] | |
| applied_criteria: list[str] | |
| unverified_criteria: list[str] | |
| legacy_notes: list[str] | |
| def resolve_asjc_rows( | |
| index: JournalIndex, | |
| *, | |
| category: str = "", | |
| area: str = "", | |
| additional_categories: Sequence[str] | None = None, | |
| ) -> list[dict]: | |
| return index.resolve_asjc( | |
| category=category, | |
| area=area, | |
| names=additional_categories or [], | |
| ) | |
| def build_candidate_pool( | |
| index: JournalIndex, | |
| *, | |
| title: str, | |
| abstract: str, | |
| asjc_rows: Sequence[dict], | |
| preferences: DiscoveryPreferences, | |
| limit: int = 50, | |
| candidate_pool: int = 500, | |
| ) -> CandidatePool: | |
| codes = [str(row.get("code", "")) for row in asjc_rows if row.get("code")] | |
| allowed_q, legacy_note = ranking_filter_to_quartiles(preferences.ranking_preference) | |
| recs = index.recommend( | |
| title=title, | |
| abstract=abstract, | |
| asjc=codes or None, | |
| limit=max(limit, 1), | |
| candidate_pool=max(candidate_pool, limit), | |
| require_english=True, | |
| oa_mode=preferences.access_mode, | |
| allowed_quartiles=allowed_q, | |
| exclude_titles=preferences.avoid_journals, | |
| preferred_titles=preferences.preferred_journals, | |
| ) | |
| applied = ["Currently active Scopus journal records only", "Journal source type only"] | |
| if codes: | |
| names = [] | |
| seen = set() | |
| for row in asjc_rows: | |
| name = row.get("name") or row.get("parent_name") or row.get("code") | |
| if name and name not in seen: | |
| seen.add(name); names.append(name) | |
| applied.append("Subject filter: " + "; ".join(names[:6])) | |
| if allowed_q: | |
| applied.append("Quartile filter: " + " or ".join(allowed_q)) | |
| if preferences.access_mode == "open": | |
| applied.append("Open-access journal flag required") | |
| elif preferences.access_mode == "diamond": | |
| applied.append("Diamond/no-APC source flag required") | |
| if preferences.avoid_journals: | |
| applied.append("Excluded titles: " + "; ".join(preferences.avoid_journals)) | |
| missing_preferred: list[str] = [] | |
| if preferences.preferred_journals: | |
| applied.append( | |
| "Preferred titles retained for comparison when eligible: " | |
| + "; ".join(preferences.preferred_journals) | |
| ) | |
| returned_keys = {normalise_title(rec.journal.title) for rec in recs} | |
| missing_preferred = [ | |
| title for title in preferences.preferred_journals | |
| if normalise_title(title) not in returned_keys | |
| ] | |
| unverified: list[str] = [] | |
| if missing_preferred: | |
| unverified.append( | |
| "Preferred title(s) not retained because no canonical active journal record met every selected subject, ranking, and access filter: " | |
| + "; ".join(missing_preferred) | |
| ) | |
| if preferences.apc_budget: | |
| unverified.append( | |
| f"APC budget ({preferences.apc_budget}) was not used as an exclusion because current title-level fees are not in the local index." | |
| ) | |
| if preferences.funder: | |
| unverified.append( | |
| f"Funder requirements for {preferences.funder} require a separate current policy check." | |
| ) | |
| if preferences.sharing_requirement: | |
| unverified.append( | |
| "Accepted-manuscript sharing will be checked against Open Policy Finder for the strongest candidates before final display." | |
| ) | |
| return CandidatePool( | |
| recommendations=recs, | |
| score_ratios=_normalise_scores(recs), | |
| asjc_rows=list(asjc_rows), | |
| applied_criteria=applied, | |
| unverified_criteria=unverified, | |
| legacy_notes=[legacy_note] if legacy_note else [], | |
| ) | |
| def ranking_component(record: JournalRecord, asjc_codes: Sequence[str], maximum: int) -> int: | |
| q, _ = quartile_for(record, asjc_codes) | |
| fractions = {"Q1": 1.0, "Q2": 0.80, "Q3": 0.55, "Q4": 0.35} | |
| if q in fractions: | |
| return round(maximum * fractions[q]) | |
| if record.currently_indexed: | |
| return round(maximum * 0.28) | |
| return 0 | |
| def practical_component( | |
| record: JournalRecord, | |
| preferences: DiscoveryPreferences, | |
| maximum: int, | |
| *, | |
| ku_agreement_candidate: bool = False, | |
| ) -> int: | |
| score = 0.0 | |
| access = access_model(record)["code"] | |
| if preferences.access_mode == "diamond": | |
| score += 0.70 if access == "diamond" else 0.0 | |
| elif preferences.access_mode == "open": | |
| score += 0.65 if access in {"open", "diamond"} else 0.0 | |
| else: | |
| score += 0.35 | |
| if access == "diamond": | |
| score += 0.20 | |
| elif access == "open": | |
| score += 0.10 | |
| if normalise_title(record.title) in { | |
| normalise_title(t) for t in preferences.preferred_journals | |
| }: | |
| score += 0.15 | |
| if ku_agreement_candidate: | |
| score += 0.10 | |
| return round(maximum * min(score, 1.0)) | |
| def policy_requirement_state( | |
| green_oa: dict[str, Any] | None, | |
| intended_repository: str = "Institutional repository", | |
| manuscript_version: str = "Accepted manuscript (AAM / postprint)", | |
| *, | |
| intended_licence: str = "", | |
| funder: str = "", | |
| ) -> str: | |
| """Evaluate one complete permission route against the requested context.""" | |
| from policy_evaluation import evaluate_policy | |
| if not green_oa: | |
| return "unknown" | |
| evaluation = evaluate_policy( | |
| green_oa, | |
| manuscript_version=manuscript_version, | |
| intended_repository=intended_repository, | |
| intended_licence=intended_licence, | |
| funder=funder, | |
| ) | |
| return evaluation.get("state", "unknown") | |
| def clamp_int(value: Any, low: int, high: int, default: int = 0) -> int: | |
| try: | |
| n = int(round(float(value))) | |
| except (TypeError, ValueError, OverflowError): | |
| n = default | |
| return max(low, min(high, n)) | |