"""Main three-pane desktop window for the adsfolio PDF library."""
from __future__ import annotations
import json
import logging
import sqlite3
import time
from collections.abc import Callable, Mapping, Sequence
from copy import deepcopy
from dataclasses import dataclass, replace
from pathlib import Path
from typing import Any, Generic, TypeVar
from PySide6.QtCore import (
QItemSelectionModel,
QModelIndex,
QSignalBlocker,
Qt,
QTimer,
)
from PySide6.QtGui import QAction, QActionGroup, QColor, QFont, QKeySequence, QShortcut
from PySide6.QtWidgets import (
QApplication,
QComboBox,
QDialog,
QDialogButtonBox,
QDockWidget,
QFileDialog,
QHBoxLayout,
QHeaderView,
QInputDialog,
QLabel,
QLineEdit,
QMainWindow,
QMenu,
QMessageBox,
QPlainTextEdit,
QSplitter,
QToolBar,
QToolButton,
QTreeView,
QTreeWidget,
QTreeWidgetItem,
QVBoxLayout,
QWidget,
)
from .. import __version__
from ..citations import CitationStyle, citation_preference_key
from ..client import ADSClient
from ..config import AuthenticationError, Config
from ..pdf.annotation_merge import (
AnnotationMergeError,
MergeInput,
MergeReport,
MissingDependencyError,
)
from ..pdf.catalog import PdfCatalog
from ..pdf.doi_providers import DOI_PROVIDER_LABELS, DoiProvider
from ..pdf.files import sha256_file
from ..pdf.identify_settings import IdentifySettings
from ..pdf.identity import (
IDENTITY_KEY,
IdentityValidationError,
assert_mutation_allowed,
extract_pdf_own_identity,
)
from ..pdf.readiness import (
METADATA_SHA256_KEY,
RECHECK_UNVERIFIED_META_KEY,
metadata_ready,
stored_bibcode_matches_file,
)
from ..pdf.recheck_worker import has_external_providers
from ..pdf.registry import (
LOCKED_METADATA_FIELDS_KEY,
CentralPdfRegistryState,
LibraryRefresh,
PdfRegistry,
RecordFact,
)
from ..pdf.rename_settings import RenameSettings
from ..pdf.search import FIELD_SEPARATOR, SEARCH_FIELDS
from ..pdf.watch_session import FileSnapshot, WatchEvent, WatchOutcome, WatchState
from ..pdf.watcher import WatchIdentifyPolicy
from ..references import LocalReferences
from .about import AboutDialog
from .ads_library_dialog import AdsLibraryDialog
from .ads_library_export_dialog import ExportAdsLibraryDialog
from .ads_recheck_job import AdsRecheckJob, AdsTokenCheck
from .appearance import AppearanceDialog
from .bibtex_export import (
CitationKeySettingsDialog,
document_ids_of,
export_bibtex_asking_keys,
fetch_ads_bibtex_now,
)
from .citation_styles import CitationStylesDialog
from .combined_view import CombinedViewController
from .config import RootEntry, load_config, save_config
from .cv_dialog import CVDialog
from .detail_pane import DetailPane
from .dialogs import (
IdentifyDialog,
ManualMetadataDialog,
RenameDialog,
UserProfileDialog,
)
from .duplicate_review import DuplicateCopy, DuplicateReviewDialog
from .edition_merge import (
EditionMergeDialog,
EditionMergeRequest,
edition_candidates,
edition_role,
)
from .export_dialog import ExportPapersDialog, ExportScope
from .icons import (
ICON_SIZE,
ActionIconBinder,
BadgeToolButton,
RoundIconButton,
vector_icon,
)
from .jobs import CommandJob, cli_argv
from .library import (
Column,
FileQuery,
LibraryView,
ads_url,
arxiv_url,
build_columns,
count_matching,
verification_stale,
)
from .models import (
FileTableModel,
FirstAuthorDelegate,
LibraryTableView,
LinkDelegate,
ResolutionDelegate,
RowHeightDelegate,
SearchHighlight,
)
from .notifications import Notice, NoticeAction, NotificationButton, NotificationPanel
from .osint import open_path, open_url, reveal_in_file_manager
from .references_dialog import ReferencesDialog
from .refresh_job import LibraryRefreshJob
from .search_completion import SearchCompleter
from .token_dialog import TokenDialog
from .update_dialog import UpdateButton, UpdateDialog, quiet_progress_bar
from .updates import (
REPOSITORY_URL,
Release,
UpdateChecker,
UpdateDownloader,
installation_kind,
)
from .watch import LibraryWatchController, WatchFailure
from .watch_sounds import (
SILENT_SOUND,
WatchSoundChoice,
WatchSoundPlayer,
normalize_sound_id,
)
from .welcome import WelcomePanel
logger = logging.getLogger(__name__)
_SettingsT = TypeVar("_SettingsT")
DEFAULT_WINDOW_SIZE = (1280, 780)
DEFAULT_SPLITTER_SIZES = [220, 700, 360]
ADS_RECHECK_INTERVAL_MS = 24 * 60 * 60 * 1000
# How often a pending close checks whether every watcher has stopped. This
# sets only how soon the window closes after the last watcher stops.
WATCH_STOP_CHECK_MS = 200
# Cost note for checks that read every skipped or unproven PDF again.
_RETRY_COST_TEXT = (
"The check reads each of these PDFs again, so it can take a long time in a "
"large library. New PDFs in a watched library wait until its check finishes. "
"Cancel in the status bar stops it and keeps its progress."
)
# Message labels for operations that wait for a watcher to pause.
_FOREGROUND_OPERATION_LABELS = {
"ads_recheck": "automatic identifier recheck of",
"ads_refresh": "ADS update in",
"local_refresh": "local refresh of",
"identify": "identification in",
"identify_rename": "identify and rename in",
"guided_identify": "identification in",
"manual_metadata": "paper-details entry in",
"edition_merge": "annotation merge review in",
"rename": "rename in",
"rename_file": "rename in",
}
_MERGE_AUDIT_KEY = "_adsfolio_annotation_merge"
class _SettingsPersistenceError(OSError):
"""Report a settings transaction or compensating rollback failure."""
class _SettingsPersistenceCoordinator(Generic[_SettingsT]):
"""Coordinate domain settings and GUI scope as a compensated transaction."""
def __init__(
self,
*,
scopes: dict[str, str],
root_key: str,
load_global: Callable[[], _SettingsT],
save_global: Callable[[_SettingsT], object],
load_library: Callable[[], _SettingsT | None],
save_library: Callable[[_SettingsT | None], object],
save_gui: Callable[[], object],
) -> None:
self._scopes = scopes
self._root_key = root_key
self._load_global = load_global
self._save_global = save_global
self._load_library = load_library
self._save_library = save_library
self._save_gui = save_gui
def persist(self, scope: str, settings: _SettingsT) -> None:
"""Persist settings and scope, compensating every attempted store.
Parameters
----------
scope
`global` or `library` destination selected by the dialog.
settings
Normalized settings value to persist.
Raises
------
OSError
If a store or any compensating rollback fails.
ValueError
If `scope` is unsupported.
"""
if scope not in {"global", "library"}:
raise ValueError(f"Unsupported settings scope: {scope!r}")
had_scope = self._root_key in self._scopes
prior_scope = self._scopes.get(self._root_key)
try:
prior_global = self._load_global()
prior_library = self._load_library()
except (OSError, ValueError, sqlite3.Error) as exc:
raise _SettingsPersistenceError(
f"Could not snapshot prior settings: {exc}"
) from exc
domain_attempted = False
gui_attempted = False
try:
domain_attempted = True
if scope == "global":
self._save_global(settings)
else:
self._save_library(settings)
self._scopes[self._root_key] = scope
gui_attempted = True
self._save_gui()
except (OSError, ValueError, sqlite3.Error) as exc:
if had_scope and prior_scope is not None:
self._scopes[self._root_key] = prior_scope
else:
self._scopes.pop(self._root_key, None)
rollback_failures: list[str] = []
if domain_attempted:
try:
if scope == "global":
self._save_global(prior_global)
else:
self._save_library(prior_library)
except (OSError, ValueError, sqlite3.Error) as rollback_exc:
rollback_failures.append(f"domain settings: {rollback_exc}")
if gui_attempted:
try:
self._save_gui()
except (OSError, ValueError, sqlite3.Error) as rollback_exc:
rollback_failures.append(f"GUI settings: {rollback_exc}")
if rollback_failures:
raise _SettingsPersistenceError(
f"{exc}; rollback failed: {'; '.join(rollback_failures)}"
) from exc
raise _SettingsPersistenceError(str(exc)) from exc
# Column that only the Duplicates view shows.
DUPLICATE_COLUMN = "duplicate_kind"
[docs]
@dataclass(frozen=True)
class RenameConflictRequest:
"""Store source and collision destination paths for UI comparison."""
root: Path
source_path: Path
conflict_path: Path
event: WatchEvent | None = None
@dataclass(frozen=True)
class _LibraryViewContext:
"""Retain the originating view while a library job runs."""
root: Path
selected: tuple[dict[str, Any], ...]
current: dict[str, Any] | None
query: FileQuery
loaded: int
scroll: int
@dataclass(frozen=True)
class _PendingPriorityEdit:
"""Keep the selected PDF and requested action while a recheck stops."""
root: Path
path: Path
location_id: int | None
attachment_id: int | None
catalog_sha256: str | None
content_sha256: str
identifier: str | None
reviewed_candidate: Notice | None
notice_key: str | None = None
@dataclass(frozen=True)
class _ConflictReviewContext:
"""Keep the selected catalog location while its lookup runs."""
root: Path
relative_path: str
current_path: str
location_id: int
COMBINED_VIEW_REASON = (
"Unavailable while Combine libraries is checked. "
"Clear Combine libraries in the sidebar to use it."
)
[docs]
class MainWindow(QMainWindow):
"""Browse registered PDF roots with a paged table and metadata pane."""
def __init__(
self,
parent: QWidget | None = None,
*,
watch_controller: LibraryWatchController | None = None,
) -> None:
super().__init__(parent)
self.setWindowTitle("ADSfolio")
self.config = load_config()
self._sort_save_pending = False
self._saved_sort_selection = (
self.config.sort_column,
self.config.sort_descending,
self.config.secondary_sort_column,
self.config.tertiary_sort_column,
)
self._saved_sort_roots = deepcopy(self.config.roots)
self._restoring_config = True
self._unrestored_roots: list[RootEntry] = []
self.columns = build_columns(self.config.citation_styles)
self.registries: dict[str, PdfRegistry] = {}
self.root_entries: dict[str, RootEntry] = {}
self.current_root: Path | None = None
self.current_view = (
LibraryView(self.config.last_view)
if self.config.last_view in {view.value for view in LibraryView}
else LibraryView.ALL
)
sortable_columns = {
column.key for column in self.columns if column.sortable is True
}
sort_column_is_valid = self.config.sort_column in sortable_columns
sort_column = (
self.config.sort_column if sort_column_is_valid else "first_seen_at"
)
self.current_query = FileQuery(
view=self.current_view,
search_exact=self.config.search_exact,
search_wildcards=self.config.search_wildcards,
search_fields=tuple(self.config.search_fields),
order_by=sort_column,
descending=(self.config.sort_descending if sort_column_is_valid else True),
secondary_order_by=(
self.config.secondary_sort_column
if self.config.secondary_sort_column in sortable_columns
else None
),
tertiary_order_by=(
self.config.tertiary_sort_column
if self.config.tertiary_sort_column in sortable_columns
else None
),
)
self.update_checker = UpdateChecker(self)
self.update_checker.finished.connect(self._update_check_finished)
self.update_downloader = UpdateDownloader(self)
self._about_dialog: AboutDialog | None = None
self._update_dialog: UpdateDialog | None = None
self.job: Any | None = None
# Short noun phrase that names the running job in messages.
self._job_description = ""
self._ads_library_dialog: AdsLibraryDialog | None = None
self._ads_library_export_dialog: ExportAdsLibraryDialog | None = None
self._close_requested = False
self._refresh_selection_id: int | None = None
self._candidate_update_notice: Notice | None = None
self._ads_match_candidates: list[dict[str, str]] = []
self._ambiguous_ads_match_context: tuple[Path, str, int | None] | None = None
self._conflict_review_context: _ConflictReviewContext | None = None
self._conflict_review_payload: dict[str, str] | None = None
self._conflict_review_line_seen = False
self._pending_watch_refresh_roots: set[Path] = set()
self._watch_refresh_selection_id: int | None = None
self._pending_watch_dialog: tuple[str, Path] | None = None
# PDF to rename after a duplicate copy moved to the Trash.
self._pending_file_rename: Path | None = None
self._manual_metadata_target: tuple[Path, Path, str | None] | None = None
self._detail_document_snapshot: (
tuple[str, int, int | None, dict[str, Any] | None] | None
) = None
self._edition_merge_target: EditionMergeRequest | None = None
self._pending_edition_merges: dict[str, EditionMergeRequest] = {}
self._merge_output_paths: set[Path] = set()
self._merge_inputs_by_sha256: dict[str, tuple[MergeInput, ...]] = {}
self._pending_ads_refresh: (
tuple[Path, tuple[str, ...], int | None, str | None] | None
) = None
self._pending_priority_edit: _PendingPriorityEdit | None = None
# Extra refresh-ads argument lists run after the pending refresh.
self._pending_ads_follow_ups: tuple[tuple[str, ...], ...] = ()
self._ads_refresh_root: Path | None = None
# PDFs that the next Identify dialog targets; empty means the library.
self._identify_files: tuple[Path, ...] = ()
self._identify_rename_files: tuple[Path, ...] = ()
self._identify_rename_context: _LibraryViewContext | None = None
self._local_refresh_context: _LibraryViewContext | None = None
self._local_refresh_result: LibraryRefresh | None = None
self._guided_identify_context: _LibraryViewContext | None = None
self._guided_identify_commands: list[list[str]] = []
self._foreground_paused_root: Path | None = None
self._ads_recheck_job: AdsRecheckJob | None = None
self._ads_token_check: AdsTokenCheck | None = None
self._ads_recheck_token_available: bool | None = None
self._ads_recheck_active_root: Path | None = None
self._ads_recheck_roots: list[Path] | None = None
self._ads_recheck_again = False
self._ads_recheck_startup_pending = True
self._ads_recheck_failures: list[str] = []
# The current cycle checks remembered failures again.
self._ads_recheck_retry_remembered = False
# The next cycle checks remembered failures again.
self._ads_recheck_retry_requested = False
# Remembered failures from another ADSfolio version in this cycle.
self._ads_recheck_outdated = 0
self._ads_recheck_started_at = 0.0
self._ads_recheck_timer = QTimer(self)
self._ads_recheck_timer.setInterval(ADS_RECHECK_INTERVAL_MS)
self._ads_recheck_timer.timeout.connect(self._offer_ads_recheck)
self._ads_recheck_offer: QMessageBox | None = None
self._ads_recheck_retry_timer = QTimer(self)
self._ads_recheck_retry_timer.setSingleShot(True)
self._ads_recheck_retry_timer.setInterval(1000)
self._ads_recheck_retry_timer.timeout.connect(self._continue_ads_rechecks)
self._watch_refresh_timer = QTimer(self)
self._watch_refresh_timer.setSingleShot(True)
self._watch_refresh_timer.setInterval(0)
self._watch_refresh_timer.timeout.connect(self._run_watch_refresh)
self.watch_controller = watch_controller or LibraryWatchController(self)
self.watch_sound_player = WatchSoundPlayer(self)
self._build_ui()
self._connect_watch_controller()
self._restore_config()
self.combined.finish_restoring()
self._restoring_config = False
self._sync_welcome()
self._ads_recheck_timer.start()
QTimer.singleShot(0, self._start_configured_watchers)
def _build_ui(self) -> None:
self.sidebar = QTreeWidget(self)
self.sidebar.setHeaderHidden(True)
self.sidebar.setIndentation(self.sidebar.fontMetrics().horizontalAdvance(" "))
self.sidebar.setMinimumWidth(0)
self.sidebar.setTextElideMode(Qt.ElideRight)
self.sidebar.header().setSectionResizeMode(0, QHeaderView.Stretch)
self.sidebar.itemSelectionChanged.connect(self._sidebar_changed)
self.sidebar.setContextMenuPolicy(Qt.CustomContextMenu)
self.sidebar.customContextMenuRequested.connect(self._sidebar_menu)
self.libraries_item = QTreeWidgetItem(["LIBRARIES"])
self.libraries_item.setData(0, Qt.UserRole + 1, "libraries")
self.views_item = QTreeWidgetItem(["VIEWS"])
self.views_item.setData(0, Qt.UserRole + 1, "views")
self.views_item.setToolTip(
0,
"ADS is the primary identification service. Verified non-ADS identities have a separate view. Changed PDFs need identification again.",
)
self.sidebar.addTopLevelItem(self.libraries_item)
self.sidebar.addTopLevelItem(self.views_item)
self.combined = CombinedViewController(self)
self.search_highlight = SearchHighlight()
self.search_highlight.enabled = self.config.search_highlight
self.model = FileTableModel(
None,
self.current_query,
columns=self.columns,
parent=self,
)
self.model.rowsInserted.connect(lambda *_args: self._update_status())
self.model.query_failed.connect(self.statusBar().showMessage)
self.table = LibraryTableView(self)
self.table.setModel(self.model)
self.table.setRootIsDecorated(False)
self.table.setUniformRowHeights(True)
self._sync_table_sort_indicator()
self.table.setSortingEnabled(True)
self.table.setSelectionBehavior(QTreeView.SelectionBehavior.SelectRows)
self.table.setSelectionMode(QTreeView.SelectionMode.ExtendedSelection)
self.table.setAlternatingRowColors(True)
header = self.table.header()
header.setSectionsMovable(True)
header.sortIndicatorChanged.connect(self._remember_table_sort)
header.setFirstSectionMovable(True)
header.setToolTip(
"Drag a header to reorder columns. Right-click a header to change alignment."
)
self._applying_table_layout = False
header.sectionMoved.connect(self._remember_table_column_order)
self.model.modelReset.connect(self._sync_sort_after_column_reset)
header.setContextMenuPolicy(Qt.CustomContextMenu)
header.customContextMenuRequested.connect(self._show_header_context_menu)
self._rebuild_table_delegates()
self.model.modelReset.connect(self._sync_detail_and_selection_actions)
self.model.modelReset.connect(self._reset_link_activation)
self.table.doubleClicked.connect(self._open_index)
self.table.clicked.connect(self._table_clicked)
self.table.selectionModel().currentChanged.connect(self._current_changed)
self.table.selectionModel().selectionChanged.connect(
self._table_selection_changed
)
self.table.setContextMenuPolicy(Qt.CustomContextMenu)
self.table.customContextMenuRequested.connect(self._table_menu)
self.detail = DetailPane(self)
self.detail.set_compact_rows(self.config.detail_compact_rows)
self.detail.set_label_width(self.config.detail_label_width)
self.detail.label_width_changed.connect(self._set_detail_label_width)
self.detail.set_citation_styles(self.config.citation_styles)
self.detail.set_row_order(self.config.detail_row_order)
self.detail.set_hidden_rows(self.config.detail_hidden_rows)
self.detail.set_expand_abstracts(self.config.expand_abstracts)
self.detail.set_row_spacing(self.config.detail_row_spacing)
self.detail.set_label_bold(self.config.detail_label_bold)
self.detail.set_link_activation(self.config.link_activation)
self._set_table_row_height(self.config.table_row_height)
self.detail.copy_feedback.connect(
lambda field: self.status.showMessage(f"Copied: {field}", 1500)
)
self.detail.metadata_changed.connect(self._save_detail_metadata)
self.detail.ads_update_requested.connect(self._update_from_identifier)
self.detail.manual_metadata_requested.connect(self._enter_manual_metadata)
self.detail.row_order_changed.connect(self._set_detail_row_order)
splitter = QSplitter(Qt.Horizontal, self)
splitter.addWidget(self.sidebar)
splitter.addWidget(self.table)
splitter.addWidget(self.detail)
splitter.setStretchFactor(0, 0)
splitter.setStretchFactor(1, 1)
splitter.setStretchFactor(2, 0)
self.notification_panel = NotificationPanel(self)
self.notice_bar = self.notification_panel.notice_bar
self.lookup_guidance = self.notification_panel.guidance
self.view_explanation = self.notification_panel.view_explanation
self.check_papers_button = self.notification_panel.identify_button
self.check_papers_button.clicked.connect(lambda: self._guide_identification())
central = QWidget(self)
central_layout = QVBoxLayout(central)
central_layout.setContentsMargins(0, 0, 0, 0)
central_layout.setSpacing(0)
self.welcome = WelcomePanel(central)
self.welcome.add_folder_requested.connect(self._add_root)
self.welcome.token_requested.connect(self._edit_token)
self.welcome.profile_requested.connect(self._edit_user_profile)
self.start_library_button = self.welcome.add_folder_button
central_layout.addWidget(self.welcome, 1)
central_layout.addWidget(splitter, 1)
self.setCentralWidget(central)
self.splitter = splitter
splitter.splitterMoved.connect(self._remember_splitter_sizes)
self.search = QLineEdit(self)
self.search.setClearButtonEnabled(True)
self.search.setPlaceholderText(
"Search, or type a field: author:doe year:2019 title:comet"
)
self._search_value_cache: tuple[tuple[Any, str], list[str]] | None = None
self.search_completer = SearchCompleter(self.search, self._search_field_values)
self.search_timer = QTimer(self)
self.search_timer.setSingleShot(True)
self.search_timer.setInterval(150)
self.search.textChanged.connect(self.search_timer.start)
self.search_timer.timeout.connect(self._search_changed)
toolbar = QToolBar("Find", self)
self.search_toolbar = toolbar
toolbar.visibilityChanged.connect(self._find_bar_visibility_changed)
toolbar.setObjectName("search_toolbar")
toolbar.addWidget(QLabel("Find: "))
toolbar.addWidget(self.search)
self.search_fields_button = QToolButton(self)
self.search_fields_button.setText("Search Fields")
self.search_fields_button.setAccessibleName("Search fields")
self.search_fields_button.setPopupMode(QToolButton.InstantPopup)
self.search_fields_button.setAutoRaise(True)
self.search_fields_button.setIconSize(ICON_SIZE)
self.search_fields_button.setIcon(
vector_icon("search_fields", self.search_fields_button)
)
fields_menu = QMenu(self.search_fields_button)
self.search_all_fields_action = fields_menu.addAction("All Fields")
self.search_all_fields_action.setCheckable(True)
fields_menu.addSeparator()
self.search_field_actions: dict[str, QAction] = {}
for key, label in SEARCH_FIELDS.items():
action = fields_menu.addAction(f"{label} ({key}:)")
action.setCheckable(True)
action.setData(key)
self.search_field_actions[key] = action
self.search_fields_button.setMenu(fields_menu)
self._set_search_fields(self.config.search_fields)
self.search_all_fields_action.triggered.connect(
lambda: self._search_fields_chosen(None)
)
for key, action in self.search_field_actions.items():
action.triggered.connect(
lambda _checked=False, field=key: self._search_fields_chosen(field)
)
toolbar.addWidget(self.search_fields_button)
self.search_highlight_action = QAction("Highlight Matches", self)
self.search_exact_action = QAction("Exact Match", self)
self.search_wildcards_action = QAction("Wildcards", self)
for action, checked, tooltip in (
(
self.search_highlight_action,
self.config.search_highlight,
"Highlight matches: mark the found text in the table",
),
(
self.search_exact_action,
self.config.search_exact,
"Exact match: match accents and letter case exactly\n"
"When off, “doremont” finds “Dorémont”.",
),
(
self.search_wildcards_action,
self.config.search_wildcards,
"Wildcards: read % as any text and _ as any one character\n"
"When off, % and _ match only themselves.",
),
):
action.setCheckable(True)
action.setChecked(checked)
action.setToolTip(tooltip)
action.toggled.connect(self._search_options_changed)
toolbar.addAction(action)
self.addToolBar(toolbar)
self.log_dock = QDockWidget("Job log", self)
self.log_dock.setObjectName("job_log_dock")
self.log = QPlainTextEdit(self.log_dock)
self.log.setReadOnly(True)
self.log_dock.setWidget(self.log)
self.addDockWidget(Qt.BottomDockWidgetArea, self.log_dock)
self.log_dock.hide()
self._last_link_activation: tuple[str, float] | None = None
self.status = self.statusBar()
self.job_label = QLabel(self)
self.job_label.hide()
self.status.addPermanentWidget(self.job_label)
self.job_progress = quiet_progress_bar(self)
self.job_progress.setRange(0, 0)
self.job_progress.setTextVisible(False)
self.job_progress.setFixedWidth(96)
self.job_progress.hide()
self.status.addPermanentWidget(self.job_progress)
self._build_actions()
# The menu keeps the full action name. The status bar shows "Cancel".
self.cancel_job_action.setIconText("Cancel")
self.job_cancel_button = QToolButton(self)
self.job_cancel_button.setDefaultAction(self.cancel_job_action)
self.job_cancel_button.setToolButtonStyle(Qt.ToolButtonTextOnly)
self.job_cancel_button.hide()
self.status.addPermanentWidget(self.job_cancel_button)
toolbar.addSeparator()
toolbar.setToolButtonStyle(Qt.ToolButtonIconOnly)
toolbar.setIconSize(ICON_SIZE)
self._toolbar_icons = ActionIconBinder(toolbar)
self._toolbar_icons.bind(self.refresh_action, "refresh")
self._toolbar_icons.bind(self.identify_action, "identify")
self._toolbar_icons.bind(self.rename_action, "rename")
self._toolbar_icons.bind(self.search_highlight_action, "highlight")
self._toolbar_icons.bind(self.search_exact_action, "match_case")
self._toolbar_icons.bind(self.search_wildcards_action, "wildcard")
for action in (self.refresh_action, self.identify_action, self.rename_action):
# Menus keep their text-only labels.
action.setIconVisibleInMenu(False)
self._set_action_tooltip(
self.refresh_action,
"Refresh: scan local PDFs in the current library and reconcile "
"added, changed, moved, and missing files",
)
self._set_action_tooltip(
self.rename_action,
"Rename: preview and apply saved rename rules to PDFs in the "
"current library",
)
toolbar.addAction(self.refresh_action)
self.identify_button = BadgeToolButton(self)
self.identify_button.setDefaultAction(self.identify_action)
self.identify_button.setToolButtonStyle(Qt.ToolButtonIconOnly)
self.identify_button.setIconSize(ICON_SIZE)
self.identify_button.setAutoRaise(True)
toolbar.addWidget(self.identify_button)
self._set_identify_badge(None)
toolbar.addAction(self.rename_action)
self.ads_refresh_button = QToolButton(self)
self.ads_refresh_button.setText("ADS Refresh")
self.ads_refresh_button.setIcon(
vector_icon("ads_refresh", self.ads_refresh_button)
)
self.ads_refresh_button.setToolButtonStyle(Qt.ToolButtonIconOnly)
self.ads_refresh_button.setIconSize(ICON_SIZE)
self.ads_refresh_button.setAutoRaise(True)
self.ads_refresh_button.setAccessibleName("ADS Refresh")
self.ads_refresh_button.setPopupMode(QToolButton.InstantPopup)
self.ads_refresh_button.setToolTip(
"ADS Refresh menu: look up stored bibcodes, update ADS metadata, "
"or apply saved rename rules"
)
ads_menu = QMenu(self.ads_refresh_button)
ads_menu.addAction(self.refresh_selected_ads_action)
ads_menu.addAction(self.update_selected_ads_action)
ads_menu.addSeparator()
ads_menu.addAction(self.link_bibcodes_action)
ads_menu.addAction(self.refresh_all_ads_action)
ads_menu.addAction(self.refresh_all_citation_metrics_action)
self.ads_refresh_button.setMenu(ads_menu)
toolbar.addWidget(self.ads_refresh_button)
self.notifications_button = NotificationButton(self)
self.notifications_button.clicked.connect(self._toggle_notifications)
toolbar.addWidget(self.notifications_button)
self.notice_bar.changed.connect(self._sync_notifications)
self._sync_notifications()
# Round icon buttons sit together at the toolbar's end with a small gap.
round_buttons = QWidget(self)
round_layout = QHBoxLayout(round_buttons)
round_layout.setContentsMargins(6, 0, 4, 0)
round_layout.setSpacing(6)
# "App Update" avoids confusion with the ADS metadata update actions.
self.update_button = UpdateButton(round_buttons)
self.update_button.clicked.connect(lambda: self._show_update_dialog())
round_layout.addWidget(self.update_button)
self.github_button = RoundIconButton(
"github", "Open ADSfolio on GitHub", round_buttons
)
self.github_button.clicked.connect(self._open_github)
round_layout.addWidget(self.github_button)
toolbar.addWidget(round_buttons)
self.update_checker.changed.connect(self._sync_update_button)
self._sync_update_button()
self._sync_refresh_actions()
self._apply_table_appearance()
def _build_actions(self) -> None:
file_menu = self.menuBar().addMenu("File")
self.add_action = QAction("Add Library Root…", self)
self.local_references_action = QAction("Local References…", self)
self.compare_selected_action = QAction("Compare Selected PDFs…", self)
self.compare_selected_action.setToolTip(
"Compare two or more selected PDFs. Move a copy to the Trash or merge "
"annotations."
)
self.compare_selected_action.triggered.connect(
lambda: self._compare_selected(self._selected_records())
)
self.merge_annotations_action = QAction("Merge Annotations…", self)
self.merge_annotations_action.triggered.connect(
lambda: self._merge_annotations()
)
self.local_references_action.triggered.connect(self._open_local_references)
self.remove_action = QAction("Remove Root", self)
self.refresh_action = QAction("Refresh", self)
refresh_shortcuts = [QKeySequence("F5")]
for shortcut in QKeySequence.keyBindings(QKeySequence.StandardKey.Refresh):
if shortcut not in refresh_shortcuts:
refresh_shortcuts.append(shortcut)
self.refresh_action.setShortcuts(refresh_shortcuts)
self.refresh_action.setToolTip(
"Scan local PDFs and reconcile added, changed, moved, and missing files"
)
self.reconcile_file_facts_action = QAction("Reconcile File Facts…", self)
self.load_more_action = QAction("Load Next Page", self)
self.remove_missing_action = QAction("Remove Missing Registry Records…", self)
self.preferences_action = QAction("Settings…", self)
self.preferences_action.setMenuRole(QAction.PreferencesRole)
preferences_shortcuts = [QKeySequence("Ctrl+,")]
for shortcut in QKeySequence.keyBindings(QKeySequence.StandardKey.Preferences):
if shortcut not in preferences_shortcuts:
preferences_shortcuts.append(shortcut)
self.preferences_action.setShortcuts(preferences_shortcuts)
self.preferences_action.setShortcutContext(Qt.WindowShortcut)
self.preferences_action.triggered.connect(self._open_preferences)
self.quit_action = QAction("Quit", self)
self.add_action.triggered.connect(self._add_root)
self.remove_action.triggered.connect(self._remove_root)
self.refresh_action.triggered.connect(self._refresh_visible_library)
self.reconcile_file_facts_action.triggered.connect(
lambda: self._refresh_local_files()
)
self.load_more_action.triggered.connect(self._load_next_page)
self.remove_missing_action.triggered.connect(self._remove_missing_records)
self.quit_action.triggered.connect(self.close)
for action in (
self.add_action,
self.local_references_action,
self.remove_action,
self.refresh_action,
self.reconcile_file_facts_action,
self.load_more_action,
self.remove_missing_action,
self.preferences_action,
self.quit_action,
):
file_menu.addAction(action)
tools = self.menuBar().addMenu("Tools")
# Disabled items explain their reason in a tooltip.
tools.setToolTipsVisible(True)
self.token_action = QAction("NASA ADS Token…", self)
self.token_action.triggered.connect(self._edit_token)
tools.addAction(self.token_action)
self.profile_action = QAction("My Publication Profile…", self)
self.cv_action = QAction("CV Generator…", self)
self.cv_action.setToolTip(
"Make a publication list for your CV in LaTeX, HTML, Markdown, or text"
)
self.cv_action.triggered.connect(self._open_cv_generator)
self.export_selected_action = QAction("Export Selection As…", self)
self.export_selected_action.setToolTip(
"Export Papers: write the selected papers as BibTeX, citation text, "
"or a publication list, from their stored details without ADS"
)
self.export_selected_action.triggered.connect(lambda: self._export_papers())
self.export_all_action = QAction("Export All As…", self)
self.export_all_action.setToolTip(
"Export Papers: write every current PDF in this library as BibTeX, "
"citation text, or a publication list"
)
self.export_all_action.triggered.connect(
lambda: self._export_papers(all_papers=True)
)
self.citation_keys_action = QAction("Citation Keys…", self)
self.citation_keys_action.setToolTip(
"Choose how BibTeX export makes citation keys for ADS papers and "
"for papers outside ADS"
)
self.citation_keys_action.triggered.connect(self._edit_citation_keys)
self.citation_styles_action = QAction("Custom Citation Styles…", self)
self.identify_action = QAction("Identify…", self)
self.identify_action.setShortcut("Ctrl+Shift+I")
self.rename_action = QAction("Rename…", self)
self.rename_action.setShortcut("Ctrl+Shift+R")
self.cancel_job_action = QAction("Cancel Current Job", self)
self.check_ads_updates_action = QAction("Check for ADS Updates", self)
self.check_ads_updates_action.setToolTip(
"Look up identified papers in ADS and apply changed bibcodes"
)
self.recheck_remembered_action = QAction("Recheck PDFs That Need Review…", self)
self.recheck_remembered_action.setToolTip(
"Check again the PDFs that the automatic identifier recheck skips "
"because an earlier check failed"
)
self.refresh_selected_ads_action = QAction(
"Refresh and Rename Selected from ADS", self
)
self.update_selected_ads_action = QAction(
"Update and Rename Selected by Identifier…", self
)
self.update_selected_ads_action.triggered.connect(
lambda: self._update_from_identifier("")
)
self.link_bibcodes_action = QAction("Look Up Stored Bibcodes…", self)
self.link_bibcodes_action.triggered.connect(self._link_stored_bibcodes)
self.refresh_all_ads_action = QAction(
"Refresh and Rename All ADS Records…", self
)
self.refresh_all_citation_metrics_action = QAction(
"Refresh and Rename All Citation Splits…", self
)
self.add_to_ads_library_action = QAction("Add to ADS Library…", self)
self.export_ads_library_action = QAction("Export ADS Library…", self)
self.watch_current_action = QAction("Watch Current Library", self)
self.watch_on_start_action = QAction("Autowatch at Start", self)
self.auto_process_action = QAction("Automatically Identify && Rename", self)
self.watch_sound_action = QAction("New PDF Sound…", self)
self.watch_sound_action.triggered.connect(self._choose_watch_sound)
for action in (
self.watch_current_action,
self.watch_on_start_action,
self.auto_process_action,
):
action.setCheckable(True)
self.cancel_job_action.setEnabled(False)
self.profile_action.triggered.connect(self._edit_user_profile)
self.citation_styles_action.triggered.connect(self._edit_citation_styles)
self.identify_action.triggered.connect(lambda: self._guide_identification())
self.rename_action.triggered.connect(self._rename)
self.cancel_job_action.triggered.connect(self._cancel_job)
self.check_ads_updates_action.triggered.connect(
lambda: self._queue_ads_rechecks()
)
self.recheck_remembered_action.triggered.connect(
self._confirm_remembered_recheck
)
self.refresh_selected_ads_action.triggered.connect(self._refresh_selected_ads)
self.refresh_all_ads_action.triggered.connect(self._refresh_all_ads)
self.refresh_all_citation_metrics_action.triggered.connect(
self._refresh_all_citation_metrics
)
self.add_to_ads_library_action.triggered.connect(self._add_to_ads_library)
self.export_ads_library_action.triggered.connect(self._export_ads_library)
self.watch_current_action.triggered.connect(self._toggle_current_watch)
self.watch_on_start_action.triggered.connect(
lambda checked: self._set_current_watch_policy("watch_on_start", checked)
)
self.auto_process_action.triggered.connect(
lambda checked: self._set_current_watch_policy(
"auto_process_new_pdfs", checked
)
)
tools.addAction(self.profile_action)
tools.addAction(self.cv_action)
tools.addAction(self.export_selected_action)
tools.addAction(self.export_all_action)
tools.addAction(self.citation_keys_action)
tools.addAction(self.citation_styles_action)
tools.addSeparator()
tools.addAction(self.compare_selected_action)
tools.addAction(self.merge_annotations_action)
tools.addAction(self.identify_action)
tools.addAction(self.rename_action)
tools.addAction(self.cancel_job_action)
tools.addSeparator()
tools.addAction(self.check_ads_updates_action)
tools.addAction(self.recheck_remembered_action)
tools.addAction(self.refresh_selected_ads_action)
tools.addAction(self.link_bibcodes_action)
tools.addAction(self.refresh_all_ads_action)
tools.addAction(self.refresh_all_citation_metrics_action)
tools.addAction(self.add_to_ads_library_action)
tools.addAction(self.export_ads_library_action)
tools.addSeparator()
tools.addAction(self.watch_current_action)
tools.addAction(self.watch_on_start_action)
tools.addAction(self.auto_process_action)
tools.addAction(self.watch_sound_action)
view = self.menuBar().addMenu("View")
self.log_dock.toggleViewAction().setShortcut("Ctrl+Shift+J")
view.addAction(self.log_dock.toggleViewAction())
self.sidebar_action = QAction("Show Libraries", self)
self.sidebar_action.setCheckable(True)
self.sidebar_action.setChecked(self.config.sidebar_visible)
self.sidebar_action.setShortcut("Ctrl+Shift+L")
self.sidebar_action.triggered.connect(self._toggle_sidebar)
view.addAction(self.sidebar_action)
self.search_toolbar.toggleViewAction().setText("Show Find Bar")
self.search_toolbar.toggleViewAction().setShortcut("Ctrl+Shift+F")
view.addAction(self.search_toolbar.toggleViewAction())
self.detail_action = QAction("Show Detail Pane", self)
self.detail_action.setCheckable(True)
self.detail_action.setChecked(self.config.detail_visible)
self.detail_action.setShortcut("Ctrl+Shift+D")
self.detail_action.triggered.connect(self._toggle_detail)
view.addAction(self.detail_action)
self.appearance_action = QAction("Appearance…", self)
self.appearance_action.triggered.connect(self._edit_appearance)
self.sort_priorities_action = QAction("Sort Priorities…", self)
self.sort_priorities_action.triggered.connect(self._edit_sort_priorities)
view.addAction(self.appearance_action)
view.addAction(self.sort_priorities_action)
self.open_selected_action = QAction(self.table)
self.open_selected_action.setShortcutContext(Qt.WidgetWithChildrenShortcut)
self.table.addAction(self.open_selected_action)
self.open_selected_action.setShortcuts(["Return", "Enter"])
self.open_selected_action.triggered.connect(self._open_current)
self.search_focus_action = QAction("Find…", self)
self.search_focus_action.setShortcut(QKeySequence.StandardKey.Find)
self.search_focus_action.triggered.connect(self._focus_search)
self.clear_search_action = QAction("Clear Find", self)
self._clear_search_shortcuts = []
for widget in (self.search, self.table):
shortcut = QShortcut(QKeySequence("Escape"), widget)
shortcut.setContext(Qt.WidgetWithChildrenShortcut)
shortcut.activated.connect(self._clear_search)
self._clear_search_shortcuts.append(shortcut)
self.clear_search_action.triggered.connect(self._clear_search)
view.addSeparator()
view.addAction(self.search_focus_action)
view.addAction(self.clear_search_action)
self.addActions(
[
self.search_focus_action,
self.clear_search_action,
]
)
help_menu = self.menuBar().addMenu("Help")
self.about_action = QAction("About ADSfolio", self)
self.about_action.setMenuRole(QAction.AboutRole)
self.about_action.triggered.connect(self._show_about)
self.check_updates_action = QAction("Check for Updates…", self)
self.check_updates_action.triggered.connect(self._check_for_updates)
self.automatic_updates_action = QAction("Automatically Check for Updates", self)
self.automatic_updates_action.setCheckable(True)
self.automatic_updates_action.setChecked(
self.config.check_updates_automatically
)
self.automatic_updates_action.toggled.connect(self._set_automatic_updates)
help_menu.addAction(self.about_action)
help_menu.addAction(self.check_updates_action)
help_menu.addAction(self.automatic_updates_action)
self._sync_ads_library_action()
[docs]
def check_updates_on_start(self) -> None:
"""Check for updates when startup preferences allow it."""
if self.config.check_updates_automatically and not self._close_requested:
self.update_checker.check()
def _show_about(self) -> None:
if self._about_dialog is None:
self._about_dialog = AboutDialog(
self.update_checker,
self.automatic_updates_action,
self,
browser_command=self.config.web_browser_command,
)
self._about_dialog.browser_command = self.config.web_browser_command
self._about_dialog.show()
self._about_dialog.raise_()
self._about_dialog.activateWindow()
def _check_for_updates(self) -> None:
self._show_update_dialog(check=True)
def _open_github(self) -> None:
"""Open the project repository in the configured browser."""
if not open_url(REPOSITORY_URL, application=self.config.web_browser_command):
self.status.showMessage(
f"Could not open the browser. Open this address: {REPOSITORY_URL}"
)
def _update_is_new(self) -> bool:
"""Return whether an available update has not been skipped."""
release = self.update_checker.latest
return (
release is not None
and self.update_checker.update_available
and str(release.version) != self.config.skipped_update_version
)
def _sync_update_button(self) -> None:
release = self.update_checker.latest
dot = self._update_is_new()
self.update_button.set_dot_visible(dot)
if dot and release is not None:
tip = f"Update available: ADSfolio {release.version}"
elif self.update_checker.busy:
tip = "Checking for updates…"
else:
tip = "Check for ADSfolio updates"
self.update_button.setToolTip(tip)
self.update_button.setAccessibleName(tip)
def _show_update_dialog(self, *, check: bool = False) -> None:
"""Open the update window, checking first when no result exists."""
if self._update_dialog is None:
self._update_dialog = UpdateDialog(
self.update_checker,
self.update_downloader,
installation_kind(),
self,
browser_command=self.config.web_browser_command,
)
self._update_dialog.skip_requested.connect(self._skip_update)
dialog = self._update_dialog
dialog.browser_command = self.config.web_browser_command
dialog.skipped_version = self.config.skipped_update_version
if check or (
self.update_checker.latest is None and not self.update_checker.busy
):
self.update_checker.check()
dialog.sync_state()
dialog.show()
dialog.raise_()
dialog.activateWindow()
def _skip_update(self, version: str) -> None:
self.config.skipped_update_version = version
self.notice_bar.dismiss("application-update")
self._sync_update_button()
try:
save_config(self.config)
except OSError as exc:
self.notice_bar.enqueue(
Notice(
"update-preference",
f"Could not save the skipped version: {exc}",
severity="error",
)
)
def _set_automatic_updates(self, enabled: bool) -> None:
self.config.check_updates_automatically = enabled
try:
save_config(self.config)
except OSError as exc:
self.notice_bar.enqueue(
Notice(
"update-preference",
f"Could not save the update preference: {exc}",
severity="error",
)
)
def _update_check_finished(self) -> None:
release = self.update_checker.latest
if release is not None:
self.notice_bar.dismiss("application-update")
if release is not None and self._update_is_new():
self.notice_bar.enqueue(
Notice(
"application-update",
f"ADSfolio {release.version} is available.",
action=NoticeAction.VIEW_RELEASE,
payload=release,
)
)
def _restore_config(self) -> None:
if self.config.window_geometry:
from PySide6.QtCore import QByteArray
self.restoreGeometry(
QByteArray.fromBase64(self.config.window_geometry.encode())
)
else:
self.resize(*DEFAULT_WINDOW_SIZE)
if self.config.splitter_sizes:
self.splitter.setSizes(self.config.splitter_sizes)
else:
self.splitter.setSizes(list(DEFAULT_SPLITTER_SIZES))
for entry in self.config.roots:
try:
self._register_root(
Path(entry.path),
entry.label,
entry.recursive,
watch_on_start=entry.watch_on_start,
auto_process_new_pdfs=entry.auto_process_new_pdfs,
)
except (OSError, ValueError) as exc:
self._unrestored_roots.append(entry)
self.notice_bar.enqueue(
Notice(
f"root-restore:{entry.path}",
f"Could not restore library root {entry.path}: {exc}",
severity="error",
)
)
if self.config.window_state:
from PySide6.QtCore import QByteArray
self.restoreState(QByteArray.fromBase64(self.config.window_state.encode()))
for index, column in enumerate(self.columns):
width = self.config.column_widths.get(column.key)
if width is not None:
self.table.setColumnWidth(index, width)
self._apply_table_layout()
last_view = self.current_view
if self.config.last_root:
try:
self._select_root(Path(self.config.last_root))
except (OSError, ValueError) as exc:
self.notice_bar.enqueue(
Notice(
"last-root-restore",
f"Could not restore the last selected root: {exc}",
severity="error",
)
)
if self.current_root is None and self.root_entries:
self._select_root(Path(next(iter(self.root_entries))))
if self.current_root is not None and last_view is not LibraryView.ALL:
self._select_view(last_view)
self.sidebar.setVisible(self.config.sidebar_visible)
self.detail.setVisible(self.config.detail_visible)
self._sync_watch_actions()
def _set_detail_label_width(self, width: int | None) -> None:
"""Retain the user's shared label-column width."""
self.config.detail_label_width = width
def _remember_splitter_sizes(self) -> None:
"""Retain panel widths when Qt reports zero for a hidden panel."""
previous = self.config.splitter_sizes or list(DEFAULT_SPLITTER_SIZES)
self.config.splitter_sizes = [
width
or (
previous[index]
if index < len(previous)
else DEFAULT_SPLITTER_SIZES[index]
)
for index, width in enumerate(self.splitter.sizes())
]
def _toggle_sidebar(self, visible: bool) -> None:
"""Show or hide libraries and retain the preference."""
self._remember_splitter_sizes()
self.config.sidebar_visible = visible
self.sidebar.setVisible(visible)
if visible and self.splitter.sizes()[0] == 0:
self.splitter.setSizes(self.config.splitter_sizes)
def _toggle_detail(self, visible: bool) -> None:
"""Show or hide metadata and retain the preference."""
self._remember_splitter_sizes()
self.config.detail_visible = visible
self.detail.setVisible(visible)
if visible and self.splitter.sizes()[2] == 0:
sizes = list(self.config.splitter_sizes)
if sizes[2] == 0:
sizes[2] = DEFAULT_SPLITTER_SIZES[2]
self.splitter.setSizes(sizes)
def _find_bar_visibility_changed(self, visible: bool) -> None:
"""Keep keyboard focus usable after hiding Find."""
if not visible and self.search.hasFocus():
self.table.setFocus()
def _focus_search(self) -> None:
"""Reveal Find and select its current query for replacement."""
self.search_toolbar.show()
self.search.setFocus()
self.search.selectAll()
def _register_root(
self,
path: Path,
label: str = "",
recursive: bool = False,
*,
watch_on_start: bool = False,
auto_process_new_pdfs: bool = False,
) -> None:
if "\0" in str(path):
raise ValueError("Library root contains a null character")
path = path.expanduser().resolve()
key = str(path)
if key not in self.registries:
registry = PdfRegistry(path)
if path.is_dir():
registry.catalog.discover_root(path, recursive=recursive)
self.registries[key] = registry
self.root_entries[key] = RootEntry(
path=key,
label=label,
recursive=recursive,
watch_on_start=watch_on_start,
auto_process_new_pdfs=auto_process_new_pdfs,
)
else:
return
self._unrestored_roots = [
entry
for entry in self._unrestored_roots
if self._config_root_key(entry.path) != key
]
count = self._safe_count(self.registries[key])
count_text = str(count) if count is not None else "?"
root_item = QTreeWidgetItem(
[f"{label or path.name or str(path)} ({count_text})"]
)
root_item.setToolTip(0, key)
root_item.setData(0, Qt.UserRole, key)
root_item.setData(0, Qt.UserRole + 1, "root")
self.libraries_item.addChild(root_item)
self.combined.register_root(root_item, key)
if self.combined.enabled:
self.combined.update_columns()
self.libraries_item.setExpanded(True)
if not self._restoring_config and not self._ads_recheck_startup_pending:
self._queue_added_ads_recheck(path)
def _connect_watch_controller(self) -> None:
"""Connect controller events after the notice surface exists."""
self.watch_controller.started.connect(self._handle_watch_started)
self.watch_controller.stopped.connect(self._handle_watch_stopped)
self.watch_controller.detected.connect(self._handle_watch_detection)
self.watch_controller.outcome.connect(self._handle_watch_outcome)
self.watch_controller.failed.connect(self._handle_watch_failure)
self.watch_controller.paused.connect(self._handle_watch_paused)
reconciled = getattr(self.watch_controller, "reconciled", None)
if reconciled is not None:
reconciled.connect(self._schedule_watch_refresh)
self.notice_bar.action_requested.connect(self._handle_notice_action)
self.notice_bar.manual_entry_requested.connect(self._enter_manual_metadata)
self.notice_bar.update_by_id_requested.connect(self._update_notice_by_id)
self.notice_bar.open_requested.connect(self._open_notice_pdf)
self.notice_bar.ads_entry_requested.connect(self._open_notice_ads_entry)
self.notice_bar.use_id_requested.connect(self._use_notice_id)
self.notice_bar.reveal_requested.connect(self._reveal_notice_path)
self.notice_bar.dismiss_requested.connect(self._handle_notice_dismiss)
self.notice_bar.action_all_requested.connect(self._handle_notice_actions)
self.notice_bar.dismiss_all_requested.connect(self._handle_notice_dismissals)
def _sync_config_roots(self) -> None:
self.config.roots = [
self.root_entries[key]
for key in self.registries
if key in self.root_entries
]
registered_paths = set(self.registries)
for entry in self._unrestored_roots:
key = self._config_root_key(entry.path)
if key not in registered_paths:
self.config.roots.append(entry)
registered_paths.add(key)
@staticmethod
def _config_root_key(path: str) -> str:
"""Normalize a saved root for comparison, retaining invalid paths."""
try:
return str(Path(path).expanduser().resolve())
except (OSError, RuntimeError, ValueError):
return path
def _sync_watch_actions(self) -> None:
root = self.current_root.resolve() if self.current_root is not None else None
entry = self.root_entries.get(str(root)) if root is not None else None
actions = (
self.watch_current_action,
self.watch_on_start_action,
self.auto_process_action,
)
blockers = [QSignalBlocker(action) for action in actions]
try:
for action in actions:
action.setEnabled(entry is not None and not self.combined.enabled)
action.setToolTip(COMBINED_VIEW_REASON if self.combined.enabled else "")
self.watch_current_action.setChecked(
root is not None and self.watch_controller.is_watching(root)
)
self.watch_on_start_action.setChecked(
entry.watch_on_start if entry is not None else False
)
self.auto_process_action.setChecked(
entry.auto_process_new_pdfs if entry is not None else False
)
if root is not None and (
self._ads_recheck_active_root == root
or (
self._local_refresh_context is not None
and self._local_refresh_context.root == root
)
or self._detail_edit_block_reason()
):
self.watch_current_action.setEnabled(False)
if not self.combined.enabled:
self.watch_current_action.setToolTip(self._operation_wait_message())
finally:
del blockers
self._update_root_watch_tooltips()
def _update_root_watch_tooltips(self) -> None:
"""Show each library's watch state in its sidebar tooltip."""
for index in range(self.libraries_item.childCount()):
item = self.libraries_item.child(index)
key = str(item.data(0, Qt.UserRole))
entry = self.root_entries.get(key)
if entry is None:
continue
watching = "Yes" if self.watch_controller.is_watching(Path(key)) else "No"
autowatch = "On" if entry.watch_on_start else "Off"
automatic = "On" if entry.auto_process_new_pdfs else "Off"
lines = (
key,
f"Watching now: {watching}",
f"Autowatch at Start: {autowatch}",
f"Automatically Identify & Rename: {automatic}",
)
item.setToolTip(0, "\n".join(lines))
def _sync_ads_library_action(self) -> None:
"""Synchronize ADS library actions with current window activity."""
if hasattr(self, "add_to_ads_library_action"):
self.add_to_ads_library_action.setEnabled(self._can_add_to_ads_library())
reason = self._add_to_ads_library_block_reason()
self._set_action_tooltip(
self.add_to_ads_library_action,
"Add papers from one library to an ADS library",
reason,
)
if hasattr(self, "export_ads_library_action"):
self.export_ads_library_action.setEnabled(self._can_export_ads_library())
self._sync_refresh_actions()
@staticmethod
def _set_action_tooltip(
action: QAction, description: str, detail: str = ""
) -> None:
"""Set an action tooltip that ends its first line with the shortcut.
Parameters
----------
action : QAction
Action whose tooltip is replaced.
description : str
First tooltip line.
detail : str, optional
Extra lines shown below the first line.
"""
shortcuts = action.shortcuts()
if shortcuts:
shortcut = shortcuts[0].toString(QKeySequence.NativeText)
description = f"{description} ({shortcut})"
action.setToolTip(f"{description}\n{detail}" if detail else description)
def _set_identify_badge(
self, count: int | None, *, unavailable: bool = False
) -> None:
"""Show how many catalogued PDFs need identification or review.
Parameters
----------
count : int or None
Records without current identification in the whole current library,
or `None` when no library is open or the count is unavailable.
unavailable : bool, optional
Say in the tooltip that the count could not be read.
"""
if not hasattr(self, "identify_button"):
return
detail = ""
if unavailable:
detail = (
"The number of catalogued PDFs that need identification is unavailable."
)
elif count is not None:
subject = "catalogued PDF needs" if count == 1 else "catalogued PDFs need"
detail = (
f"{count} {subject} identification or review. ADS is the primary service. "
"Verified non-ADS identities are excluded from this count. Identify can "
"skip PDFs that already have usable metadata or are in the "
"configured unresolved folder, and also checks PDFs that are "
"not catalogued yet."
)
if self.combined.enabled:
detail += (
"\nThis action applies to one whole library. "
"Right-click its library folder to identify it."
)
self._set_action_tooltip(
self.identify_action,
"Identify: find publication details for PDFs in the current library",
detail,
)
self.identify_button.set_badge(count)
self.identify_button.setAccessibleDescription(self.identify_action.toolTip())
def _library_name(self, root: Path) -> str:
"""Return the sidebar name of a library root."""
entry = self.root_entries.get(str(root))
return (entry.label if entry is not None else "") or root.name or str(root)
def _operation_text(self) -> str:
"""Name the operation that holds the library lock.
Returns
-------
str
A noun phrase such as ``"automatic identifier recheck of Papers"``,
or ``""`` when no operation runs or waits.
"""
if self.job is not None:
return self._job_description or "library job"
if self._pending_priority_edit is not None:
return "stop of the automatic identifier recheck"
if self._ads_recheck_active_root is not None:
root = self._ads_recheck_active_root
return f"automatic identifier recheck of {self._library_name(root)}"
if self._pending_watch_dialog is not None:
kind, root = self._pending_watch_dialog
label = _FOREGROUND_OPERATION_LABELS.get(kind, "library operation in")
return f"{label} {self._library_name(root)}"
if self._manual_metadata_target is not None:
return f"paper-details entry for {self._manual_metadata_target[1].name}"
if self._edition_merge_target is not None:
return "annotation merge review"
if self._local_refresh_context is not None:
root = self._local_refresh_context.root
return f"local refresh of {self._library_name(root)}"
if self._guided_identify_context is not None:
root = self._guided_identify_context.root
return f"identification in {self._library_name(root)}"
if self._pending_ads_refresh is not None:
return f"ADS update in {self._library_name(self._pending_ads_refresh[0])}"
if self._foreground_paused_root is not None:
root = self._foreground_paused_root
return f"library operation in {self._library_name(root)}"
if self._close_requested:
return "application shutdown"
return ""
def _operation_wait_message(self, before: str = "") -> str:
"""Tell the user which operation to wait for, and how to cancel it.
Parameters
----------
before
Optional action phrase, such as ``"starting a watcher"``.
"""
operation = self._operation_text() or "current operation"
suffix = f" before {before}" if before else ""
message = f"Wait for the {operation} to finish{suffix}."
if self.job is not None and self.cancel_job_action.isEnabled():
message += " To stop it, click Cancel in the status bar."
return message
def _sync_job_indicator(self) -> None:
"""Show the running job's name beside the status-bar progress bar."""
running = self.job is not None
if running:
text = self._operation_text()
self.job_label.setText(text[:1].upper() + text[1:] + "…")
self.job_progress.setToolTip(self._operation_wait_message())
else:
self._job_description = ""
self.job_label.clear()
self.job_progress.setToolTip("")
self.job_label.setVisible(running)
self.job_progress.setVisible(running)
self.job_cancel_button.setVisible(running)
def _detail_edit_block_reason(self) -> str:
"""Explain why metadata editing is temporarily unavailable."""
if self._pending_priority_edit is not None:
return "Wait for the automatic identifier recheck to stop."
if self._ads_recheck_active_root is not None:
return self._operation_wait_message()
if (
self.job is not None
or self._pending_watch_dialog is not None
or self._pending_ads_refresh is not None
or self._local_refresh_context is not None
or self._manual_metadata_target is not None
or self._edition_merge_target is not None
or self._close_requested
):
return self._operation_wait_message()
return ""
def _can_preempt_ads_recheck(self) -> bool:
"""Allow a selected-PDF edit when the recheck is the sole blocker."""
root = self._ads_recheck_active_root
recheck_job = self._ads_recheck_job
pending_watch = self._pending_watch_dialog
recheck_running = recheck_job is not None and self.job is recheck_job
recheck_waiting = self.job is None and pending_watch == ("ads_recheck", root)
return bool(
root is not None
and (recheck_running or recheck_waiting)
and (pending_watch is None or pending_watch == ("ads_recheck", root))
and self._pending_priority_edit is None
and self._pending_ads_refresh is None
and self._local_refresh_context is None
and self._manual_metadata_target is None
and self._edition_merge_target is None
and not self._close_requested
)
def _sync_refresh_actions(self) -> None:
"""Enable refresh controls only when their root and selection are usable."""
if not hasattr(self, "refresh_selected_ads_action"):
return
detail_block_reason = self._detail_edit_block_reason()
identifier_priority = self._can_preempt_ads_recheck()
self.detail.set_read_only(
bool(detail_block_reason),
reason=detail_block_reason,
allow_identifier_update=identifier_priority,
allow_manual_metadata=identifier_priority,
)
available = (
self.current_root is not None
and not self.combined.enabled
and self.job is None
and self._pending_watch_dialog is None
and self._pending_ads_refresh is None
and self._local_refresh_context is None
and self._manual_metadata_target is None
and self._edition_merge_target is None
and not self._close_requested
and not detail_block_reason
)
record = self._selected_record()
bibcode = (
(record.get("metadata") or {}).get("bibcode") or record.get("bibcode")
if record
else None
)
self.check_papers_button.setEnabled(available)
self._sync_export_action()
self.identify_action.setEnabled(available)
self.rename_action.setEnabled(available)
if self.combined.enabled:
library_reason = (
"This action applies to one whole library. "
"Right-click a library folder to choose the action."
)
self._set_action_tooltip(
self.identify_action, "Identify papers in a library", library_reason
)
self._set_action_tooltip(
self.rename_action, "Rename PDFs in a library", library_reason
)
else:
self._set_action_tooltip(
self.rename_action,
"Rename: preview and apply saved rename rules to PDFs in the "
"current library",
)
self.remove_action.setEnabled(
self.current_root is not None and not self.combined.enabled
)
self.merge_annotations_action.setEnabled(
not self.combined.enabled and self._ads_recheck_active_root is None
)
self.merge_annotations_action.setToolTip(
COMBINED_VIEW_REASON
if self.combined.enabled
else self._operation_wait_message()
if self._ads_recheck_active_root is not None
else ""
)
self.refresh_action.setEnabled(
available or (self.combined.enabled and self.job is None)
)
self._set_action_tooltip(
self.refresh_action,
"Refresh: reload the combined view from the catalog"
if self.combined.enabled
else "Refresh: scan local PDFs in the current library and reconcile "
"added, changed, moved, and missing files",
)
self.reconcile_file_facts_action.setEnabled(available)
self.remove_missing_action.setEnabled(available)
selected_available = available or (
self.combined.enabled
and not detail_block_reason
and self._selected_edit_target() is not None
)
self.refresh_selected_ads_action.setEnabled(
selected_available
and bool(bibcode)
and not bool((record.get("metadata") or {}).get("_adsfolio_metadata_stale"))
)
identifier_available = selected_available or (
identifier_priority and self._selected_edit_target() is not None
)
self.update_selected_ads_action.setEnabled(
identifier_available and record is not None
)
selected_reason = detail_block_reason
if not selected_reason and self.combined.enabled and not selected_available:
selected_reason = "Select exactly one PDF."
elif not selected_reason and record is None:
selected_reason = "Select a PDF."
identifier_reason = selected_reason
if identifier_priority and identifier_available:
identifier_reason = (
"The automatic identifier recheck will stop before this update."
)
self._set_action_tooltip(
self.update_selected_ads_action,
"Update and rename the selected PDF by identifier",
identifier_reason
or "The update uses the selected PDF's library and its Rename settings.",
)
refresh_reason = selected_reason
if not refresh_reason and not bibcode:
refresh_reason = "The selected PDF has no saved Bibcode. Use Update by ID."
elif not refresh_reason and bool(
(record.get("metadata") or {}).get("_adsfolio_metadata_stale")
):
refresh_reason = (
"The PDF changed. Identify it before refreshing its ADS record."
)
self._set_action_tooltip(
self.refresh_selected_ads_action,
"Refresh and rename the selected PDF from ADS",
refresh_reason,
)
self.link_bibcodes_action.setEnabled(available)
self.refresh_all_ads_action.setEnabled(available)
self.refresh_all_citation_metrics_action.setEnabled(available)
for action in (
self.link_bibcodes_action,
self.refresh_all_ads_action,
self.refresh_all_citation_metrics_action,
):
action.setToolTip(COMBINED_VIEW_REASON if self.combined.enabled else "")
if hasattr(self, "ads_refresh_button"):
self.ads_refresh_button.setEnabled(available or identifier_available)
self.ads_refresh_button.setToolTip(
"ADS Refresh menu: selected actions use the PDF's library. "
"All-library actions require a single library view."
if self.combined.enabled
else "ADS Refresh menu: look up stored bibcodes, update ADS metadata, "
"or apply saved rename rules"
)
self._sync_watch_actions()
def _can_add_to_ads_library(self) -> bool:
"""Return whether this window can open an ADS library dialog."""
return not self._add_to_ads_library_block_reason()
def _add_to_ads_library_block_reason(self) -> str:
"""Explain why the ADS library dialog cannot open, or return ``""``."""
if self.job is not None:
return "Wait for the current job to finish."
if (
self._ads_library_dialog is not None
or self._ads_library_export_dialog is not None
):
return "An ADS library dialog is already open."
if self.combined.enabled:
if self._combined_selection_root() is None:
return "Select PDFs from one available library."
return ""
if self.current_root is None:
return "Open a library first."
return ""
def _combined_selection_root(self) -> Path | None:
"""Return the one library that owns every selected combined-view PDF."""
records = self._selected_record_refs()
roots = {self.combined.root_for(record) for record in records}
if len(roots) != 1 or None in roots:
return None
return next(iter(roots))
def _can_export_ads_library(self) -> bool:
"""Return whether this window can open the remote export dialog."""
return (
not self._close_requested
and self.job is None
and self._ads_library_dialog is None
and self._ads_library_export_dialog is None
)
def _selected_view_rows(self) -> list[QModelIndex]:
"""Return one column-0 view index for each selected table row.
The table selects whole rows. This reads the selection ranges, so the
cost depends on the number of selected rows, not on rows times columns.
``QItemSelectionModel.selectedRows`` checks every cell through the model.
"""
selection = self.table.selectionModel()
model = self.table.model()
if selection is None or model is None:
return []
seen: set[tuple[int, int]] = set()
rows: list[QModelIndex] = []
for selected_range in selection.selection():
parent = selected_range.parent()
parent_key = id(parent.internalPointer()) if parent.isValid() else 0
for row in range(selected_range.top(), selected_range.bottom() + 1):
if (parent_key, row) in seen:
continue
seen.add((parent_key, row))
index = model.index(row, 0, parent)
if index.isValid():
rows.append(index)
return rows
def _selected_source_rows(self) -> list[int]:
"""Return selected source-model rows in ascending order."""
return sorted(
{
source.row()
for index in self._selected_view_rows()
if (source := self._source_table_index(index)).isValid()
}
)
def _selected_record_refs(self) -> tuple[dict[str, Any], ...]:
"""Return the model's selected records without copies, in row order.
Do not change these records. Use `_selected_records` to pass records to
a dialog or a job.
"""
return tuple(
record
for row in self._selected_source_rows()
if (record := self.model.record_at(row)) is not None
)
def _selected_records(self) -> tuple[dict[str, Any], ...]:
"""Return deep-copied selected records in visual table-row order."""
return tuple(deepcopy(record) for record in self._selected_record_refs())
def _ads_library_dialog_closed(self, dialog: AdsLibraryDialog) -> None:
"""Release the window's ADS library dialog after it closes."""
if self._ads_library_dialog is dialog:
self._ads_library_dialog = None
self._sync_ads_library_action()
def _ads_library_export_dialog_closed(self, dialog: ExportAdsLibraryDialog) -> None:
"""Release the window's ADS export dialog after it closes."""
if self._ads_library_export_dialog is dialog:
self.config.export_exclude_fields = dialog.excluded_fields
self._ads_library_export_dialog = None
self._sync_ads_library_action()
if self._close_requested and self.job is None:
QTimer.singleShot(0, self.close)
def _add_to_ads_library(self) -> None:
"""Open an ADS library dialog with immutable local record snapshots."""
if not self._can_add_to_ads_library():
self._sync_ads_library_action()
return
owner = (
self._combined_selection_root()
if self.combined.enabled
else self.current_root
)
assert owner is not None
root = owner.expanduser().resolve()
registry = self.registries[str(root)]
selected_records = self._selected_records()
try:
all_records = tuple(
deepcopy(record) for record in registry.query_files(resolved=True)
)
except sqlite3.OperationalError as exc:
if "locked" not in str(exc).lower() and "busy" not in str(exc).lower():
raise
self.status.showMessage("Library is busy; try again in a moment.")
return
dialog = AdsLibraryDialog(
root,
selected_records,
all_records,
parent=self,
)
self._ads_library_dialog = dialog
dialog.finished.connect(lambda _result: self._ads_library_dialog_closed(dialog))
self._sync_ads_library_action()
dialog.show()
def _open_local_references(self) -> None:
"""Open local bibliography intake and export without requiring a PDF root."""
try:
dialog = ReferencesDialog(parent=self)
except (OSError, ValueError, sqlite3.Error) as exc:
QMessageBox.warning(self, "Local references", str(exc))
return
dialog.references_changed.connect(self.refresh)
dialog.exec()
dialog.deleteLater()
def _export_ads_library(self) -> None:
"""Open the remote ADS library export dialog."""
if not self._can_export_ads_library():
self._sync_ads_library_action()
return
dialog = ExportAdsLibraryDialog(
parent=self, exclude_fields=self.config.export_exclude_fields
)
self._ads_library_export_dialog = dialog
dialog.finished.connect(
lambda _result: self._ads_library_export_dialog_closed(dialog)
)
self._sync_ads_library_action()
dialog.show()
def _start_root_watch(self, root: Path) -> bool:
normalized = root.expanduser().resolve()
if self._detail_edit_block_reason():
self.status.showMessage(self._operation_wait_message("starting a watcher"))
return False
entry = self.root_entries.get(str(normalized))
if entry is None:
return False
if not normalized.is_dir():
self.notice_bar.enqueue(
Notice(
f"watch-missing:{normalized}",
f"Watch root does not exist: {normalized}",
severity="error",
root=normalized,
)
)
return False
try:
self.watch_controller.start_root(normalized, recursive=entry.recursive)
except (OSError, RuntimeError, ValueError) as exc:
self.notice_bar.enqueue(
Notice(
f"watch-start:{normalized}",
f"Could not start watching {normalized}: {exc}",
severity="error",
root=normalized,
)
)
return False
return self.watch_controller.is_watching(normalized)
def _start_configured_watchers(self) -> None:
if self._close_requested:
return
for entry in tuple(self.root_entries.values()):
if self._close_requested:
return
if entry.watch_on_start:
self._start_root_watch(Path(entry.path))
self._sync_watch_actions()
QTimer.singleShot(0, self._queue_startup_ads_rechecks)
def _toggle_current_watch(self, checked: bool) -> None:
if self.current_root is None:
self._sync_watch_actions()
return
root = self.current_root.resolve()
if (
self._local_refresh_context is not None
and self._local_refresh_context.root == root
) or root == self._ads_recheck_active_root:
self._sync_watch_actions()
return
if checked:
self._start_root_watch(root)
elif not self.watch_controller.stop_root(root):
self.notice_bar.enqueue(
Notice(
f"watch-stop:{root}",
f"Could not stop watching {root}",
severity="error",
root=root,
)
)
else:
self.notice_bar.clear_root(root)
self._sync_watch_actions()
def _set_current_watch_policy(self, field: str, checked: bool) -> None:
if self.current_root is None:
self._sync_watch_actions()
return
root = self.current_root.resolve()
key = str(root)
previous = self.root_entries[key]
self.root_entries[key] = replace(previous, **{field: checked})
self._sync_config_roots()
try:
save_config(self.config)
except OSError as exc:
self.root_entries[key] = previous
self._sync_config_roots()
self.notice_bar.enqueue(
Notice(
f"watch-save:{root}:{field}",
f"Could not save watch setting: {exc}",
severity="error",
root=root,
)
)
self._sync_watch_actions()
def _set_root_recursive(self, root: Path, recursive: bool) -> bool:
if root.expanduser().resolve() == self._ads_recheck_active_root:
self.status.showMessage(self._operation_wait_message())
return False
normalized = root.expanduser().resolve()
key = str(normalized)
entry = self.root_entries.get(key)
if entry is None:
return False
if self.watch_controller.is_watching(normalized):
self.notice_bar.enqueue(
Notice(
f"watch-recursive:{normalized}",
"Stop watching this library before changing recursive scope.",
severity="warning",
root=normalized,
)
)
return False
previous = entry
self.root_entries[key] = replace(entry, recursive=recursive)
self._sync_config_roots()
try:
save_config(self.config)
except OSError as exc:
self.root_entries[key] = previous
self._sync_config_roots()
self.notice_bar.enqueue(
Notice(
f"recursive-save:{normalized}",
f"Could not save recursive setting: {exc}",
severity="error",
root=normalized,
)
)
return False
return True
def _handle_watch_started(self, _session_id: str) -> None:
self._update_root_watch_labels()
self._sync_watch_actions()
def _handle_watch_stopped(self, _session_id: str) -> None:
self._update_root_watch_labels()
self._sync_watch_actions()
def _handle_watch_paused(self, session_id: str) -> None:
pending = self._pending_watch_dialog
if pending is None:
return
kind, root = pending
if self.watch_controller.session_id(root) != session_id:
return
self._pending_watch_dialog = None
self._foreground_paused_root = root
if kind == "ads_recheck":
if self._pending_priority_edit is not None:
self._end_ads_recheck_root()
else:
self._launch_ads_recheck(root)
elif kind == "ads_refresh":
self._start_pending_ads_refresh(root)
elif kind == "local_refresh":
self._launch_local_refresh(root)
elif kind == "identify":
self._open_identify_dialog(root)
elif kind == "identify_rename":
self._launch_identify_and_rename(root)
elif kind == "guided_identify":
self._launch_guided_identification(root)
elif kind == "manual_metadata":
self._open_manual_metadata(root)
elif kind == "edition_merge":
self._open_edition_merge(root)
elif kind == "rename_file":
self._launch_file_rename(root)
else:
self._open_rename_dialog(root)
def _handle_watch_failure(self, failure: WatchFailure) -> None:
event = failure.event
current = event is not None and self._is_current_watch_event(event)
self.notice_bar.enqueue(
Notice(
self._watch_notice_key(event)
if event is not None
else f"watch-failure:{failure.root}:{failure.session_id}",
failure.message,
severity="error",
action=NoticeAction.RETRY if current else None,
payload=event if current else failure,
root=failure.root,
manual_entry=current,
update_by_id=current,
reveal_path=event.path if event is not None else None,
)
)
self._sync_watch_actions()
def _is_current_watch_event(self, event: WatchEvent) -> bool:
root = event.root.expanduser().resolve()
return self.watch_controller.session_id(root) == event.session_id
@staticmethod
def _watch_notice_key(event: WatchEvent) -> str:
identity = event.identity
return (
f"watch:{event.root.resolve()}:{event.session_id}:"
f"{identity.device}:{identity.inode}:{identity.birth_time_ns}:"
f"{identity.generation}"
)
def _effective_identify_settings(self, root: Path) -> IdentifySettings:
normalized = root.expanduser().resolve()
global_settings = Config.load_identify_settings()
if self.config.identify_scopes.get(str(normalized), "global") != "library":
return global_settings
registry = self.registries.get(str(normalized))
if registry is None:
return global_settings
return CentralPdfRegistryState(registry).identify_settings or global_settings
def _effective_rename_settings(self, root: Path) -> RenameSettings:
normalized = root.expanduser().resolve()
global_settings = Config.load_rename_settings()
if self.config.rename_scopes.get(str(normalized), "global") != "library":
return global_settings
registry = self.registries.get(str(normalized))
if registry is None:
return global_settings
return CentralPdfRegistryState(registry).rename_settings or global_settings
def _queue_ads_rechecks(self, retry_remembered: bool = False) -> None:
"""Queue the registered roots for one serial recheck cycle.
Parameters
----------
retry_remembered
Check again the PDFs whose earlier check failed. A running cycle
passes the request to the next cycle.
"""
if self._close_requested:
return
if retry_remembered:
self._ads_recheck_retry_requested = True
if self._ads_recheck_roots is not None:
self._ads_recheck_again = True
return
self._ads_recheck_roots = [Path(key) for key in self.root_entries]
self._ads_recheck_retry_remembered = self._ads_recheck_retry_requested
self._ads_recheck_retry_requested = False
self._ads_recheck_outdated = 0
self._ads_recheck_failures = []
self._ads_recheck_token_available = None
self._continue_ads_rechecks()
def _queue_startup_ads_rechecks(self) -> None:
self._ads_recheck_startup_pending = False
self._offer_ads_recheck()
def _offer_ads_recheck(self) -> None:
"""Ask whether to start a recheck cycle. A cycle never starts by itself."""
if (
self._close_requested
or not self.root_entries
or self._ads_recheck_roots is not None
or self._ads_recheck_offer is not None
):
return
box = QMessageBox(self)
box.setIcon(QMessageBox.Question)
box.setWindowTitle("Check for ADS Updates")
box.setText("Check your libraries for ADS updates now?")
box.setInformativeText(
"ADSfolio looks up your identified papers in ADS in batches. It "
"updates changed bibcodes, for example an arXiv paper that is now "
"published, and renames those PDFs. It does not open a PDF when ADS "
"returns its stored bibcode. New PDFs in a watched library wait until its "
"check finishes.\n\nYou can also start the check from Tools > "
"Check for ADS Updates."
)
start = box.addButton("Check Now", QMessageBox.AcceptRole)
later = box.addButton("Not Now", QMessageBox.RejectRole)
box.setDefaultButton(later)
box.setEscapeButton(later)
box.setWindowModality(Qt.NonModal)
box.finished.connect(lambda _result: self._ads_recheck_offer_closed(box, start))
self._ads_recheck_offer = box
box.show()
def _ads_recheck_offer_closed(self, box: QMessageBox, start: Any) -> None:
"""Start the cycle only when the user chose Check Now."""
self._ads_recheck_offer = None
box.deleteLater()
if box.clickedButton() is start:
self._queue_ads_rechecks()
def _queue_added_ads_recheck(self, root: Path) -> None:
"""Add a new root to a running cycle. Without a cycle, start nothing."""
if self._close_requested:
return
roots = self._ads_recheck_roots
if roots is None:
return
if (
not self._ads_recheck_again
and root != self._ads_recheck_active_root
and root not in roots
):
roots.append(root)
def _ads_token_check_finished(
self, check: AdsTokenCheck, available: bool, error: str | None
) -> None:
if self._ads_token_check is not check:
return
self._ads_token_check = None
check.deleteLater()
if self._close_requested:
self.close()
return
if error:
self._ads_recheck_failures.append(f"Could not read the ADS token: {error}")
self._ads_recheck_token_available = available
self._continue_ads_rechecks()
def _continue_ads_rechecks(self) -> None:
"""Start the next eligible root after other GUI operations finish."""
roots = self._ads_recheck_roots
if roots is None or self._close_requested:
return
if self._ads_token_check is not None:
return
if self._ads_recheck_active_root is not None:
return
if (
self.job is not None
or self._pending_watch_dialog is not None
or self._pending_ads_refresh is not None
or self._pending_priority_edit is not None
or self._foreground_paused_root is not None
or self._local_refresh_context is not None
or self._manual_metadata_target is not None
or self._edition_merge_target is not None
or QApplication.activeModalWidget() is not None
):
if not self._ads_recheck_retry_timer.isActive():
self._ads_recheck_retry_timer.start()
return
while roots:
root = roots[0]
if str(root) not in self.registries:
roots.pop(0)
continue
try:
settings = self._effective_identify_settings(root)
except (OSError, ValueError, sqlite3.Error) as exc:
self._ads_recheck_failures.append(f"{root}: {exc}")
roots.pop(0)
continue
if (
DoiProvider.ADS in settings.providers
and self._ads_recheck_token_available is None
):
check = AdsTokenCheck(self)
self._ads_token_check = check
check.finished.connect(
lambda available, error, current=check: (
self._ads_token_check_finished(current, available, error)
)
)
check.start()
return
roots.pop(0)
if (
self._ads_recheck_token_available is False
and not has_external_providers(settings)
):
continue
self._ads_recheck_active_root = root
self._request_foreground_dialog("ads_recheck", root)
return
self._finish_ads_recheck_cycle()
def _ads_recheck_summary(self, root: Path, result: Any, error: str | None) -> str:
"""Return one log line that summarizes a library's recheck."""
name = self._library_name(root)
elapsed = time.monotonic() - self._ads_recheck_started_at
prefix = f"Automatic identifier recheck of {name}"
if result is None:
return f"{prefix} stopped after {elapsed:.0f} s: {error or 'no result'}"
failures = result.failures
failure_count = failures if isinstance(failures, int) else len(failures)
counts = (
f"{result.checked} checked, "
f"{getattr(result, 'remembered_skipped', 0)} skipped because an "
"earlier check failed, "
f"{result.updated} updated, {result.renamed} renamed, "
f"{failure_count} failed"
)
state = "was cancelled" if result.cancelled else "finished"
return f"{prefix} {state} after {elapsed:.0f} s: {counts}."
def _remembered_failure_counts(self) -> dict[Path, int]:
"""Count remembered recheck failures in each registered library."""
counts: dict[Path, int] = {}
for key, registry in self.registries.items():
try:
raw = registry.get_meta(RECHECK_UNVERIFIED_META_KEY)
entries = json.loads(raw) if raw else {}
except (sqlite3.Error, json.JSONDecodeError):
continue
if isinstance(entries, dict) and entries:
counts[Path(key)] = len(entries)
return counts
def _ask_outdated_recheck(self, outdated: int) -> None:
"""Ask once per version to recheck failures remembered by an older version."""
if self._close_requested:
return
self.config.remembered_recheck_prompt_version = __version__
answer = QMessageBox.question(
self,
"Recheck PDFs That Need Review",
f"{outdated} PDFs failed the identity check in an earlier ADSfolio "
"version. The automatic recheck skips them. This version can check "
"them again.\n\n"
f"{_RETRY_COST_TEXT}\n\nCheck them now? You can also start this "
"later from Tools > Recheck PDFs That Need Review.",
QMessageBox.Yes | QMessageBox.No,
QMessageBox.No,
)
if answer == QMessageBox.Yes:
self._queue_ads_rechecks(retry_remembered=True)
def _confirm_remembered_recheck(self) -> None:
"""Ask before the next recheck cycle checks remembered failures again."""
counts = self._remembered_failure_counts()
lines = "\n".join(
f"{self._library_name(root)}: {count}" for root, count in counts.items()
)
skipped = (
f"The automatic recheck skips {sum(counts.values())} PDFs because an "
f"earlier check failed:\n{lines}"
if counts
else "The automatic recheck skips no PDFs because of an earlier failure."
)
when = (
"after the current recheck"
if self._ads_recheck_roots is not None
else "now"
)
answer = QMessageBox.question(
self,
"Recheck PDFs That Need Review",
f"{skipped}\n\nThis check also adds a missing identity proof to "
"identified PDFs whose bibcode did not change.\n\n"
f"{_RETRY_COST_TEXT}\n\nStart the check {when}?",
QMessageBox.Yes | QMessageBox.No,
QMessageBox.No,
)
if answer == QMessageBox.Yes:
self._queue_ads_rechecks(retry_remembered=True)
def _launch_ads_recheck(self, root: Path) -> None:
"""Snapshot current settings after the watcher acknowledges its pause."""
try:
if self._close_requested or str(root) not in self.registries:
self._end_ads_recheck_root()
return
identify_settings = self._effective_identify_settings(root)
ads_enabled = (
DoiProvider.ADS in identify_settings.providers
and self._ads_recheck_token_available is not False
)
if not ads_enabled and not has_external_providers(identify_settings):
self._end_ads_recheck_root()
return
rename_settings = self._effective_rename_settings(root)
recursive = self.root_entries[str(root)].recursive
job = AdsRecheckJob(
root,
identify_settings,
rename_settings,
recursive,
self,
ads_enabled=ads_enabled,
retry_remembered=self._ads_recheck_retry_remembered,
)
self._ads_recheck_job = job
self.job = job
self._job_description = (
f"automatic identifier recheck of {self._library_name(root)}"
)
job.finished.connect(
lambda result, error: self._ads_recheck_finished(
root, job, result, error
)
)
self._sync_refresh_actions()
self._sync_ads_library_action()
self.cancel_job_action.setEnabled(True)
self._sync_job_indicator()
self.status.showMessage(f"Checking identifiers for {root.name}…")
self._ads_recheck_started_at = time.monotonic()
retry_note = (
" PDFs that failed before are checked again."
if self._ads_recheck_retry_remembered
else ""
)
self.log.appendPlainText(
f"Automatic identifier recheck of {self._library_name(root)} "
f"started.{retry_note}"
)
job.start()
except (OSError, ValueError, sqlite3.Error, RuntimeError) as exc:
self._ads_recheck_failures.append(f"{root}: {exc}")
self._ads_recheck_job = None
self.job = None
self._sync_job_indicator()
self._end_ads_recheck_root()
def _ads_recheck_finished(
self, root: Path, job: AdsRecheckJob, result: Any, error: str | None
) -> None:
"""Collect one result after the worker releases its catalog."""
if self._ads_recheck_job is not job:
return
self._ads_recheck_job = None
self.job = None
self._sync_job_indicator()
job.deleteLater()
if not self._close_requested:
self.log.appendPlainText(self._ads_recheck_summary(root, result, error))
if error:
self._ads_recheck_failures.append(f"{root}: {error}")
if result is not None:
self._ads_recheck_outdated += int(
getattr(result, "remembered_outdated", 0) or 0
)
activities = getattr(result, "activities", ())
if activities:
details = "\n".join(activities)
self.log.appendPlainText(
f"Automatic identifier recheck for {root}:\n{details}"
)
failures = result.failures
if isinstance(failures, int):
if failures:
self._ads_recheck_failures.append(
f"{root}: {failures} lookup failures"
)
elif failures:
self._ads_recheck_failures.extend(
f"{root}: {failure}" for failure in failures
)
if (result.updated or result.renamed) and str(root) in self.registries:
self._schedule_watch_refresh(root)
if (
self.combined.enabled
and str(root) not in self.combined.selected
):
for key in self.combined.selected:
if key in self.registries:
self._schedule_watch_refresh(Path(key))
break
elif self.current_root is not None and self.current_root != root:
self._schedule_watch_refresh(self.current_root)
if result.cancelled:
self._ads_recheck_roots = []
self._ads_recheck_again = False
self._ads_recheck_retry_requested = False
self._end_ads_recheck_root()
def _end_ads_recheck_root(self) -> None:
root = self._ads_recheck_active_root
self._ads_recheck_active_root = None
self.cancel_job_action.setEnabled(False)
if self._close_requested:
self._foreground_paused_root = None
self.close()
return
self._resume_foreground_watch()
self._sync_refresh_actions()
self._sync_ads_library_action()
if root is not None and self.status.currentMessage() == (
f"Checking identifiers for {root.name}…"
):
self._update_status()
if self._pending_priority_edit is not None:
QTimer.singleShot(0, self._run_pending_priority_edit)
else:
QTimer.singleShot(0, self._continue_ads_rechecks)
def _finish_ads_recheck_cycle(self) -> None:
failures = self._ads_recheck_failures
retried = self._ads_recheck_retry_remembered
outdated = self._ads_recheck_outdated
self._ads_recheck_roots = None
self._ads_recheck_failures = []
self._ads_recheck_token_available = None
self._ads_recheck_retry_remembered = False
self._ads_recheck_outdated = 0
if failures:
details = "\n".join(failures)
self.log.appendPlainText(
f"Automatic identifier recheck failures:\n{details}"
)
self.notice_bar.enqueue(
Notice(
"ads-recheck-failures",
"Some automatic identifier rechecks failed. See details for the affected libraries.",
severity="warning",
details=details,
)
)
if self._ads_recheck_again:
self._ads_recheck_again = False
QTimer.singleShot(0, self._queue_ads_rechecks)
elif (
outdated
and not retried
and self.config.remembered_recheck_prompt_version != __version__
):
QTimer.singleShot(0, lambda: self._ask_outdated_recheck(outdated))
def _choose_watch_sound(self) -> None:
"""Choose and preview the sound for new PDFs in watched libraries."""
dialog = QDialog(self)
dialog.setWindowTitle("New PDF Sound")
layout = QVBoxLayout(dialog)
layout.addWidget(
QLabel("Play a sound when a new PDF arrives in a watched library.", dialog)
)
choice = WatchSoundChoice(
dialog, sound_id=self.config.watch_sound, player=self.watch_sound_player
)
layout.addWidget(choice)
feedback = QLabel("", dialog)
feedback.setWordWrap(True)
layout.addWidget(feedback)
choice.preview_failed.connect(
lambda: feedback.setText(
"Could not play the sound. Check the audio output."
)
)
choice.sound_changed.connect(lambda _sound: feedback.clear())
buttons = QDialogButtonBox(
QDialogButtonBox.Save | QDialogButtonBox.Cancel, dialog
)
buttons.accepted.connect(dialog.accept)
buttons.rejected.connect(dialog.reject)
layout.addWidget(buttons)
if dialog.exec() == QDialog.Accepted:
previous = self.config.watch_sound
self.config.watch_sound = choice.sound_id()
self._sync_config_roots()
try:
save_config(self.config)
except OSError as exc:
self.config.watch_sound = previous
QMessageBox.warning(self, "Cannot save sound", str(exc))
def _handle_watch_detection(self, event: WatchEvent) -> None:
if not self._is_current_watch_event(event):
return
entry = self.root_entries.get(str(event.root.resolve()))
if entry is None:
return
sound = normalize_sound_id(self.config.watch_sound)
if (
event.newly_added
and sound != SILENT_SOUND
and not self.watch_sound_player.play(sound)
):
self.notice_bar.enqueue(
Notice(
"watch-sound-unavailable",
"Could not play the new PDF sound. Check the audio output.",
severity="warning",
)
)
identify_settings = self._effective_identify_settings(event.root)
try:
excluded = WatchIdentifyPolicy.from_settings(
event.root, identify_settings
).excludes(event.path)
except ValueError:
excluded = False
if excluded:
self.watch_controller.dismiss(event)
return
if entry.auto_process_new_pdfs:
self._process_watch_event(event)
return
self.notice_bar.enqueue(
Notice(
self._watch_notice_key(event),
f"New PDF found: {event.path.name}",
action=NoticeAction.IDENTIFY_RENAME,
payload=event,
root=event.root,
reveal_path=event.path,
)
)
def _process_watch_event(self, event: WatchEvent) -> None:
if not self._is_current_watch_event(event):
return
self.watch_controller.process(
event,
self._effective_identify_settings(event.root),
self._effective_rename_settings(event.root),
)
def _handle_watch_outcome(self, outcome: WatchOutcome) -> None:
event = outcome.event
if not self._is_current_watch_event(event):
return
if outcome.annotation_update:
self.notice_bar.dismiss(self._watch_notice_key(event))
self._schedule_watch_refresh(event.root)
return
state = event.state
destination = outcome.destination or event.path
message = outcome.message or f"{state.value.title()}: {destination.name}"
if outcome.message and not message.startswith(
(f"{destination.name}:", f"{destination.name} ")
):
message = f"{destination.name}: {message}"
conflict_request: RenameConflictRequest | None = None
if outcome.conflict_path is not None:
conflict_request = RenameConflictRequest(
root=event.root,
source_path=destination,
conflict_path=outcome.conflict_path,
event=event,
)
message = self._conflict_notice_text(
event.root, destination, outcome.conflict_path
)
self.notice_bar.enqueue(
Notice(
self._watch_notice_key(event),
message,
severity=(
"error"
if state is WatchState.FAILED
else (
"success"
if state in {WatchState.RESOLVED, WatchState.RENAMED}
else "warning"
)
),
action=(
NoticeAction.SHOW_CONFLICT
if conflict_request is not None
else (NoticeAction.RETRY if state is WatchState.FAILED else None)
),
payload=(
conflict_request
if conflict_request is not None
else (
replace(event, path=destination)
if state in {WatchState.UNRESOLVED, WatchState.FAILED}
else event
)
),
root=event.root,
details=(
f"{destination} | {outcome.conflict_path}"
if outcome.conflict_path
else str(destination)
),
manual_entry=state in {WatchState.UNRESOLVED, WatchState.FAILED},
update_by_id=state in {WatchState.UNRESOLVED, WatchState.FAILED},
reveal_path=destination,
candidate_bibcode=outcome.candidate_bibcode,
candidate_sha256=outcome.candidate_sha256,
candidate_fingerprint=outcome.candidate_fingerprint,
candidate_confirmable=outcome.candidate_confirmable,
)
)
if state in {
WatchState.RESOLVED,
WatchState.RENAMED,
WatchState.UNRESOLVED,
WatchState.FAILED,
}:
self._schedule_watch_refresh(event.root)
if state in {WatchState.RESOLVED, WatchState.RENAMED}:
self._offer_edition_merge(event, destination)
def _offer_edition_merge(self, event: WatchEvent, destination: Path) -> None:
"""Prompt for identifier-linked editions or an explicit preprint selection."""
for path in (event.path, destination):
if path in self._merge_output_paths:
self._merge_output_paths.discard(path)
return
root = event.root.expanduser().resolve()
registry = self.registries.get(str(root))
if registry is None:
return
try:
relative_path = destination.resolve().relative_to(root).as_posix()
arrival = registry.find_file_by_current_path(root, relative_path)
if arrival is None:
return
arrival = deepcopy(arrival)
bibcode = self._record_value(arrival, "bibcode")
if bibcode:
arrival["ads"] = registry.find_ads_document_by_bibcode(bibcode) or {}
records = registry.query_files()
candidates, linked = edition_candidates(arrival, records)
candidates = tuple(
record
for record in candidates
if (
root / str(record.get("current_path") or record["relative_path"])
).is_file()
)
except (OSError, ValueError, sqlite3.Error) as exc:
self.log.appendPlainText(
f"Could not check PDF editions for {destination}: {exc}"
)
return
if not candidates:
return
choices = (arrival, *(deepcopy(record) for record in candidates))
target_record = next(
(record for record in choices if edition_role(record) == "published"),
arrival,
)
request = EditionMergeRequest(
root=root,
records=choices,
target=root
/ str(target_record.get("current_path") or target_record["relative_path"]),
choose_source=not linked or len(choices) != 2,
linked=linked,
event=event,
)
key = self._watch_notice_key(event)
# Post a notice only. A modal window that opens by itself blocks the
# main window and pauses the watcher. **Review Merge…** opens it.
message = (
f"{destination.name}: another edition shares its identifiers. Review annotation merging?"
if linked
else f"{destination.name}: ADS has not linked the editions. Choose an existing preprint or published PDF to review annotation merging."
)
self.notice_bar.enqueue(
Notice(
key,
message,
action=NoticeAction.MERGE_EDITIONS,
payload=request,
root=root,
reveal_path=destination,
details="\n".join(
str(
root
/ str(record.get("current_path") or record["relative_path"])
)
for record in choices
),
)
)
def _review_next_edition_merge(self) -> None:
"""Open the next merge review that the user asked for, when nothing else runs."""
if (
not self.isVisible()
or self._close_requested
or self.job is not None
or self._pending_watch_dialog is not None
or self._pending_ads_refresh is not None
or self._pending_priority_edit is not None
or self._edition_merge_target is not None
or self._foreground_paused_root is not None
or QApplication.activeModalWidget() is not None
):
return
while self._pending_edition_merges:
key = next(iter(self._pending_edition_merges))
request = self._pending_edition_merges.pop(key)
if (
request.event is not None
and self._is_current_watch_event(request.event)
and str(request.root) in self.registries
):
self._edition_merge_target = request
self._request_foreground_dialog("edition_merge", request.root)
return
def _open_edition_merge(self, root: Path) -> None:
"""Confirm the source and target, then review annotations with writes paused."""
request = self._edition_merge_target
try:
if request is None or request.root != root:
return
dialog = EditionMergeDialog(request, self)
if dialog.exec() != QDialog.Accepted:
return
target, source = dialog.target, dialog.source
if target is None or source is None:
return
report = self._review_annotation_pair(target, source)
if report is not None and report.complete and request.event is not None:
self.notice_bar.dismiss(self._watch_notice_key(request.event))
finally:
self._edition_merge_target = None
self._resume_foreground_watch()
def _reveal_notice_path(self, path: Path) -> None:
"""Reveal the notice's file and report a failed launch in the status bar."""
if not reveal_in_file_manager(path):
self.status.showMessage(f"Could not reveal file: {path}")
def _open_notice_pdf(self, path: Path) -> None:
"""Open the notice's file with the configured PDF viewer."""
if not path.is_file() or not open_path(path, self.config.pdf_viewer_command):
self.status.showMessage(f"Could not open PDF: {path}")
def _open_notice_ads_entry(self, bibcode: str) -> None:
"""Open the proposed ADS record for comparison with the notice's PDF."""
self._open_external_url(ADSClient.abstract_url(bibcode))
def _use_notice_id(self, notice: Notice) -> None:
"""Apply the displayed candidate without asking the user to enter its ID."""
event = notice.payload
if (
not isinstance(event, WatchEvent)
or not self._is_current_watch_event(event)
or not notice.candidate_bibcode
or not notice.candidate_sha256
or not notice.candidate_fingerprint
):
return
self._update_from_identifier(
notice.candidate_bibcode,
root=event.root,
path=event.path,
reviewed_candidate=notice,
)
def _update_notice_by_id(self, payload: object) -> None:
"""Look up the notice's PDF independently of the selected library row."""
if not isinstance(payload, WatchEvent) or not self._is_current_watch_event(
payload
):
return
if self._update_from_identifier("", root=payload.root, path=payload.path):
self.notice_bar.dismiss(self._watch_notice_key(payload))
def _conflict_copies(
self, root: Path, paths: Sequence[Path]
) -> list[DuplicateCopy]:
"""Return catalog facts for PDFs in a name conflict."""
root = root.expanduser().resolve()
registry = self.registries.get(str(root))
copies = []
for path in paths:
try:
relative = path.resolve().relative_to(root).as_posix()
except ValueError:
continue
record = (
registry.find_file_by_current_path(root, relative)
if registry is not None
else None
)
copy = (
self._duplicate_copy(root, record)
if record is not None
else DuplicateCopy(root=root, relative_path=relative)
)
if copy.sha256 is None:
# Hash only a PDF that the catalog has no digest for.
try:
copy = replace(copy, sha256=sha256_file(copy.path))
except OSError:
pass
copies.append(copy)
return copies
def _conflict_notice_text(
self, root: Path, destination: Path, conflict: Path
) -> str:
"""Say why a new PDF kept its name, from the stored catalog facts."""
copies = self._conflict_copies(root, (conflict, destination))
if (
len(copies) == 2
and copies[0].sha256
and copies[0].sha256 == copies[1].sha256
):
reason = "It has the same bytes as"
elif (
len(copies) == 2
and copies[0].document_id is not None
and copies[0].document_id == copies[1].document_id
):
reason = "It is another copy of the paper in"
else:
reason = "Another PDF already has its filename:"
return (
f"{destination.name}: not renamed. {reason} {conflict.name}. "
"Review both PDFs to keep one."
)
def _duplicate_copy(self, root: Path, record: Mapping[str, Any]) -> DuplicateCopy:
"""Return the duplicate-window facts of one catalog record."""
return DuplicateCopy(
root=root,
relative_path=str(record.get("current_path") or record["relative_path"]),
sha256=str(record.get("sha256") or "") or None,
first_seen_at=record.get("first_seen_at"),
document_id=record.get("document_id"),
)
def _review_conflict(self, request: RenameConflictRequest) -> None:
"""Open the duplicate window for a new PDF whose name is taken."""
copies = self._conflict_copies(
request.root, (request.conflict_path, request.source_path)
)
if len(copies) != 2:
self._show_collision_pair(request)
return
old, new = copies
if old.sha256 and old.sha256 == new.sha256:
message = (
f"{new.path.name} has the same bytes as {old.path.name}. "
"ADSfolio did not rename it, because that name is taken."
)
elif old.document_id is not None and old.document_id == new.document_id:
message = (
f"{new.path.name} is another copy of the paper in {old.path.name}. "
"The bytes differ, for example by annotations. ADSfolio did not "
"rename it, because that name is taken."
)
else:
message = (
f"{new.path.name} and {old.path.name} are different papers that get "
f"the same filename from the Rename settings. ADSfolio did not "
f"rename {new.path.name}."
)
self._open_duplicate_review(
copies, message, lambda: self._show_collision_pair(request)
)
def _review_duplicates(self, record: Mapping[str, Any]) -> None:
"""Open the duplicate window for the group of one Duplicates-view row."""
registry = (
self.combined.registry()
if self.combined.enabled
else self.registries.get(str(self.current_root))
)
if registry is None:
return
group_key = record.get("duplicate_group")
try:
group = [
item
for item in registry.query_records(duplicates_only=True)
if group_key and item.get("duplicate_group") == group_key
]
except (OSError, sqlite3.Error) as exc:
self.status.showMessage(f"Could not read the duplicate PDFs: {exc}")
return
copies = [
self._duplicate_copy(Path(str(item.get("root") or self.current_root)), item)
for item in group
]
if len(copies) < 2:
self.status.showMessage("This PDF has no other copy.")
return
title = self._record_value(dict(record), "title")
bibcode = self._record_value(dict(record), "bibcode")
same_bytes = len({copy.sha256 for copy in copies}) == 1 and copies[0].sha256
message = (
"These PDFs have the same bytes."
if same_bytes
else "These PDFs are copies of one paper"
+ (f": {title}" if title else "")
+ (f" ({bibcode})" if bibcode else "")
+ ". The bytes differ, for example by annotations."
)
self._open_duplicate_review(copies, message, lambda: None)
def _open_duplicate_review(
self,
copies: Sequence[DuplicateCopy],
message: str,
show_in_table: Callable[[], None],
) -> None:
"""Show the duplicate window, then rename or merge as the user chose."""
records = self._copy_records(copies)
can_merge = (
len(copies) == 2
and len(records) == 2
and copies[0].root == copies[1].root
and copies[0].sha256 != copies[1].sha256
)
dialog = DuplicateReviewDialog(self, copies, message, can_merge=can_merge)
moved: list[DuplicateCopy] = []
dialog.trashed.connect(
lambda copy: (moved.append(copy), self._duplicate_trashed(copy))
)
dialog.show_in_table.connect(show_in_table)
dialog.exec()
remaining = list(dialog.copies)
merge_chosen = dialog.merge_chosen
dialog.deleteLater()
if merge_chosen and can_merge:
self._merge_annotations(records, root=copies[0].root)
elif moved and len(remaining) == 1:
self._rename_kept_copy(remaining[0])
def _copy_records(self, copies: Sequence[DuplicateCopy]) -> list[dict[str, Any]]:
"""Return the catalog records of the copies that the catalog tracks."""
records = []
for copy in copies:
registry = self.registries.get(str(copy.root))
record = (
registry.find_file_by_current_path(copy.root, copy.relative_path)
if registry is not None
else None
)
if record is not None:
records.append(record)
return records
def _compare_selected(self, records: Sequence[Mapping[str, Any]]) -> None:
"""Open the duplicate window for two or more selected PDFs."""
if len(records) < 2:
self.status.showMessage("Select two or more PDFs to compare.")
return
copies = [
self._duplicate_copy(
Path(str(record.get("root") or self.current_root)), record
)
for record in records
]
digests = {copy.sha256 for copy in copies}
documents = {copy.document_id for copy in copies}
count = len(copies)
if len(digests) == 1 and None not in digests:
message = (
f"These {count} PDFs have the same bytes. Keep one and move the "
"others to the Trash."
)
elif len(documents) == 1 and None not in documents:
title = self._record_value(dict(records[0]), "title")
message = (
"These PDFs are copies of one paper"
+ (f": {title}" if title else "")
+ ". The bytes differ, for example by annotations."
)
else:
message = (
"These PDFs are not copies of one paper record. Check each PDF "
"before you move one to the Trash."
)
self._open_duplicate_review(copies, message, lambda: None)
def _duplicate_trashed(self, copy: DuplicateCopy) -> None:
"""Record a copy that moved to the Trash and refresh its library."""
registry = self.registries.get(str(copy.root))
if registry is not None:
try:
registry.catalog.mark_location_missing(copy.root, copy.relative_path)
except (ValueError, sqlite3.Error) as exc:
self.log.appendPlainText(
f"Could not mark {copy.path} as removed: {exc}"
)
self.log.appendPlainText(f"Moved duplicate PDF to the Trash: {copy.path}")
self._schedule_watch_refresh(copy.root)
def _rename_kept_copy(self, copy: DuplicateCopy) -> None:
"""Give the remaining identified copy its name from the Rename settings."""
registry = self.registries.get(str(copy.root))
record = (
registry.find_file_by_current_path(copy.root, copy.relative_path)
if registry is not None
else None
)
if record is None or record.get("status") != "resolved":
return
if (
self.job is not None
or self._pending_watch_dialog is not None
or self._pending_ads_refresh is not None
):
self.status.showMessage(
"Another operation is running. Use Tools → Rename later to rename "
f"{copy.path.name}."
)
return
self._pending_file_rename = copy.path
self._request_foreground_dialog("rename_file", copy.root)
def _launch_file_rename(self, root: Path) -> None:
"""Rename one PDF with the saved Rename settings. No ADS request."""
path = self._pending_file_rename
self._pending_file_rename = None
if path is None or not path.is_file() or str(root) not in self.registries:
self._resume_foreground_watch()
return
job = CommandJob(
cli_argv(
"pdf",
"rename",
str(root),
"--settings-scope",
self.config.rename_scopes.get(str(root), "global"),
"--file",
str(path),
"--apply",
),
parent=self,
working_directory=root,
)
self._ads_refresh_root = root
self._watch_job(job, f"rename in {self._library_name(root)}")
self.status.showMessage(f"Renaming {path.name}")
job.start()
def _show_collision_pair(self, request: RenameConflictRequest) -> None:
"""Select the two exact conflict paths in the main table."""
root = request.root.expanduser().resolve()
if str(root) not in self.registries:
return
if self.combined.enabled:
self.combined.set_enabled(False)
try:
paths = {
request.source_path.resolve().relative_to(root).as_posix(),
request.conflict_path.resolve().relative_to(root).as_posix(),
}
except ValueError:
self.status.showMessage(
"Both PDFs must be inside this library for comparison."
)
return
self._select_root(root)
self._select_view(LibraryView.ALL)
if self.search.text():
self.search.clear()
self.search_timer.stop()
self._search_changed()
# Register a destination added after the last catalog refresh.
registry = self.registries.get(str(root))
if registry is not None:
try:
registry.reconcile_file_facts(root)
self.model.set_registry(registry, self.current_query)
except (OSError, sqlite3.Error) as exc:
self.status.showMessage(f"Could not reconcile files: {exc}")
return
selection = self.table.selectionModel()
selection.clearSelection()
# A requested path can be outside the model's first page.
while self.model.canFetchMore():
self.model.fetchMore()
first_index = None
for row in range(self.model.rowCount()):
rec = self.model.record_at(row)
if not rec:
continue
rec_path = str(rec.get("current_path") or rec.get("relative_path") or "")
if rec_path in paths:
idx = self.model.index(row, 0)
selection.select(
idx, QItemSelectionModel.Select | QItemSelectionModel.Rows
)
if first_index is None:
first_index = idx
if first_index is not None:
selection.setCurrentIndex(first_index, QItemSelectionModel.NoUpdate)
self.table.scrollTo(first_index)
self._sync_detail_and_selection_actions()
def _handle_notice_action(self, payload: object) -> None:
if isinstance(payload, RenameConflictRequest):
self._review_conflict(payload)
return
if isinstance(payload, EditionMergeRequest):
if payload.event is not None and self._is_current_watch_event(
payload.event
):
self._pending_edition_merges[self._watch_notice_key(payload.event)] = (
payload
)
self._review_next_edition_merge()
return
if isinstance(payload, Release):
self._show_update_dialog()
return
if not isinstance(payload, WatchEvent) or not self._is_current_watch_event(
payload
):
return
if payload.state is WatchState.FAILED:
self.watch_controller.retry(payload)
elif payload.state is WatchState.DETECTED:
self._process_watch_event(payload)
elif payload.state is WatchState.UNRESOLVED:
self._update_notice_by_id(payload)
def _handle_notice_dismiss(self, payload: object) -> None:
if isinstance(payload, EditionMergeRequest) and payload.event is not None:
self._pending_edition_merges.pop(
self._watch_notice_key(payload.event), None
)
if (
isinstance(payload, WatchEvent)
and payload.state is WatchState.DETECTED
and self._is_current_watch_event(payload)
):
self.watch_controller.dismiss(payload)
def _handle_notice_actions(self, payloads: tuple[object, ...]) -> None:
for payload in payloads:
self._handle_notice_action(payload)
def _handle_notice_dismissals(self, payloads: tuple[object, ...]) -> None:
for payload in payloads:
self._handle_notice_dismiss(payload)
def _schedule_watch_refresh(self, root: Path) -> None:
normalized = root.expanduser().resolve()
if not self._pending_watch_refresh_roots:
record = self._selected_record()
self._watch_refresh_selection_id = (
int(record["location_id"])
if record is not None and record.get("location_id") is not None
else None
)
self._pending_watch_refresh_roots.add(normalized)
if not self._watch_refresh_timer.isActive():
self._watch_refresh_timer.start()
def _run_watch_refresh(self) -> None:
roots = tuple(self._pending_watch_refresh_roots)
self._pending_watch_refresh_roots.clear()
selection_id = self._watch_refresh_selection_id
self._watch_refresh_selection_id = None
for root in roots:
if str(root) in self.registries:
self.refresh(root)
if (
selection_id is not None
and not self.combined.enabled
and self.current_root is not None
and self.current_root.resolve() in roots
):
registry = self.registries[str(self.current_root)]
if registry.find_location_by_id(selection_id) is None:
previous = registry.catalog.query_files(
root=self.current_root,
location_id=selection_id,
include_missing=True,
)
if previous:
current = registry.catalog.query_files(
root=self.current_root,
attachment_id=previous[0]["attachment_id"],
)
if len(current) == 1:
selection_id = int(current[0]["location_id"])
self._select_file_id(selection_id)
def _update_root_watch_labels(self) -> None:
for index in range(self.libraries_item.childCount()):
item = self.libraries_item.child(index)
key = str(item.data(0, Qt.UserRole))
entry = self.root_entries.get(key)
registry = self.registries.get(key)
if entry is None or registry is None:
continue
count = self._safe_count(registry)
count_text = str(count) if count is not None else "?"
marker = "● " if self.watch_controller.is_watching(Path(key)) else ""
item.setText(
0,
f"{marker}{entry.label or Path(key).name or key} ({count_text})",
)
def _populate_views(self, registry: PdfRegistry) -> None:
"""Show virtual-view counts for the selected library root."""
self.views_item.takeChildren()
facts = self._safe_facts(registry)
for view in LibraryView:
count = count_matching(facts, view) if facts is not None else None
item = QTreeWidgetItem(
[f"{view.label} ({count if count is not None else '?'})"]
)
item.setData(0, Qt.UserRole, str(self.current_root))
item.setData(0, Qt.UserRole + 1, view.value)
item.setToolTip(0, view.description)
self.views_item.addChild(item)
self.views_item.setExpanded(True)
def _safe_facts(self, registry: Any) -> tuple[RecordFact, ...] | None:
"""Return cached count facts, reporting an active SQLite writer safely."""
try:
return registry.record_facts()
except sqlite3.OperationalError as exc:
if "locked" not in str(exc).lower() and "busy" not in str(exc).lower():
raise
self.status.showMessage("Library is busy; try again in a moment.")
return None
def _safe_count(self, registry: PdfRegistry, **filters: Any) -> int | None:
"""Return a registry count, reporting an active SQLite writer safely."""
try:
return registry.count_files_matching(**filters)
except sqlite3.OperationalError as exc:
if "locked" not in str(exc).lower() and "busy" not in str(exc).lower():
raise
self.status.showMessage("Library is busy; try again in a moment.")
return None
def _select_root(self, path: Path) -> None:
key = str(path.expanduser().resolve())
for index in range(self.libraries_item.childCount()):
item = self.libraries_item.child(index)
if item.data(0, Qt.UserRole) == key:
self.sidebar.setCurrentItem(item)
return
def _select_view(self, view: LibraryView) -> None:
"""Select `view` for the current root after populating the sidebar."""
for index in range(self.views_item.childCount()):
item = self.views_item.child(index)
if item.data(0, Qt.UserRole + 1) == view.value:
self.sidebar.setCurrentItem(item)
return
def _sidebar_changed(self) -> None:
item = self.sidebar.currentItem()
if item is None:
return
kind = item.data(0, Qt.UserRole + 1)
if kind == "root":
key = item.data(0, Qt.UserRole)
view = LibraryView.ALL
elif kind in {view.value for view in LibraryView}:
key = item.data(0, Qt.UserRole)
view = LibraryView(str(kind))
else:
return
self.current_root = Path(str(key))
self.current_view = view
self.current_query = replace(
self.current_query, view=self.current_view, search=self.search.text()
)
registry = (
self.combined.registry()
if self.combined.enabled
else self.registries[str(self.current_root)]
)
self.model.set_registry(registry, self.current_query)
self._sync_duplicate_column()
self.combined.sync_tree_mode()
if kind == "root":
self._populate_views(registry)
self._update_status()
self._sync_watch_actions()
self._sync_ads_library_action()
def _search_options_changed(self) -> None:
"""Save the search options and run the search again."""
self.config.search_highlight = self.search_highlight_action.isChecked()
self.config.search_exact = self.search_exact_action.isChecked()
self.config.search_wildcards = self.search_wildcards_action.isChecked()
self._search_changed()
def _set_search_fields(self, fields: Sequence[str]) -> None:
"""Check the menu items for the fields that plain terms search."""
chosen = {field for field in fields if field in self.search_field_actions}
self.search_all_fields_action.setChecked(not chosen)
for key, action in self.search_field_actions.items():
action.setChecked(key in chosen)
labels = [SEARCH_FIELDS[key] for key in SEARCH_FIELDS if key in chosen]
self.search_fields_button.setToolTip(
"Search fields: choose the fields that words without a field name "
"search\n"
f"Now: {', '.join(labels) if labels else 'all fields'}.\n"
"Type field:text, for example author:doe or year:2019, to search "
"one field."
)
def _search_fields_chosen(self, field: str | None) -> None:
"""Apply a Search Fields menu choice and run the search again."""
if field is None:
chosen: list[str] = []
else:
chosen = [
key
for key, action in self.search_field_actions.items()
if action.isChecked()
]
self.config.search_fields = chosen
self._set_search_fields(chosen)
self._search_changed()
def _search_source(self) -> Any:
"""Return the registry or combined adapter that the table searches."""
if self.combined.enabled:
return self.combined.registry()
if self.current_root is None:
return None
return self.registries.get(str(self.current_root))
def _search_field_values(self, field: str) -> list[str]:
"""Return sorted distinct stored values of `field` for autocomplete."""
source = self._search_source()
if source is None:
return []
try:
key = (source.change_token(), field)
cached = self._search_value_cache
if cached is not None and cached[0] == key:
return cached[1]
values: set[str] = set()
for _location, document, _folded in source.search_documents():
values.update(
value
for value in document.get(field, "").split(FIELD_SEPARATOR)
if value
)
except sqlite3.OperationalError:
return []
result = sorted(values, key=str.casefold)
self._search_value_cache = (key, result)
return result
def _search_changed(self) -> None:
self.current_query = replace(
self.current_query,
view=self.current_view,
search=self.search.text(),
search_exact=self.search_exact_action.isChecked(),
search_wildcards=self.search_wildcards_action.isChecked(),
search_fields=tuple(self.config.search_fields),
)
highlight = self.search_highlight
highlight.enabled = self.search_highlight_action.isChecked()
highlight.query = self.current_query.search
highlight.exact = self.current_query.search_exact
highlight.wildcards = self.current_query.search_wildcards
highlight.default_fields = self.current_query.search_fields
self.model.set_query(self.current_query)
self.table.viewport().update()
self._update_status()
def _current_changed(self, current: QModelIndex, previous: QModelIndex) -> None:
del previous
selected = self._selected_view_rows()
source = self._source_table_index(current)
selected_source = (
self._source_table_index(selected[0])
if len(selected) == 1
else QModelIndex()
)
record = (
self.model.record_at(source.row())
if source.isValid()
and (
(selected_source.isValid() and selected_source.row() == source.row())
if self.combined.enabled
else len(selected) <= 1
)
else None
)
root = (
self.combined.root_for(record)
if record is not None and self.combined.enabled
else self.current_root
)
if (
self.combined.enabled
and record is not None
and (root is None or self.combined.path_for(record) is None)
):
record = None
if record is None or root is None or record.get("location_id") is None:
self._detail_document_snapshot = None
else:
document_id = record.get("document_id")
snapshot_key = (str(root), int(record["location_id"]), document_id)
if (
self._detail_document_snapshot is None
or self._detail_document_snapshot[:3] != snapshot_key
):
try:
registry = self.registries.get(str(root))
if registry is None:
raise ValueError("The PDF's library is unavailable.")
document = (
registry.catalog.get_document(int(document_id))
if document_id is not None
else None
)
if document_id is not None and document is None:
raise ValueError("The paper record is no longer available.")
self._detail_document_snapshot = (*snapshot_key, deepcopy(document))
except (ValueError, sqlite3.Error) as exc:
self._detail_document_snapshot = None
self.status.showMessage(f"Cannot check paper details: {exc}")
record = None
cached_document = None
if record is not None and root is not None:
metadata = record.get("metadata")
bibcode = (
metadata.get("bibcode")
if isinstance(metadata, dict)
else record.get("bibcode")
)
if bibcode:
try:
cached_document = self.registries[
str(root)
].find_ads_document_by_bibcode(str(bibcode))
except sqlite3.OperationalError as exc:
if (
"locked" not in str(exc).lower()
and "busy" not in str(exc).lower()
):
raise
self.status.showMessage("Library is busy; try again in a moment.")
if not self.detail.shows(record, root or Path(), cached_document):
self.detail.set_record(
record,
root or Path(),
cached_document,
)
self._sync_ads_library_action()
def _sync_detail_and_selection_actions(self) -> None:
"""Synchronize the detail pane and actions after selection or model changes."""
self._current_changed(self.table.currentIndex(), QModelIndex())
self._sync_export_action()
def _table_selection_changed(self, *_args: Any) -> None:
"""Update detail actions after a table selection changes."""
if self.combined.restoring_tree:
# The combined view restores a selection in one step and then
# synchronizes once.
return
self.combined.selection_changed()
self._sync_detail_and_selection_actions()
def _set_table_model(self, model: Any) -> None:
"""Replace the visible table model and reconnect selection signals."""
old_selection = self.table.selectionModel()
if old_selection is not None:
for signal, callback in (
(old_selection.currentChanged, self._current_changed),
(old_selection.selectionChanged, self._table_selection_changed),
):
try:
signal.disconnect(callback)
except (RuntimeError, TypeError):
pass
self.table.setModel(model)
selection = self.table.selectionModel()
selection.currentChanged.connect(self._current_changed)
selection.selectionChanged.connect(self._table_selection_changed)
self._sync_table_sort_indicator()
self._apply_table_layout()
def _sync_export_action(self) -> None:
"""Enable export for the current library and its selected rows."""
if hasattr(self, "export_selected_action"):
has_library = self.current_root is not None or self.combined.enabled
self.export_selected_action.setEnabled(
has_library and self.table.selectionModel().hasSelection()
)
self.export_all_action.setEnabled(has_library)
self.export_all_action.setToolTip(
"Export Papers: write every current PDF in the checked libraries "
"as BibTeX, citation text, or a publication list"
if self.combined.enabled
else "Export Papers: write every current PDF in this library as "
"BibTeX, citation text, or a publication list"
)
def _open_index(self, index: QModelIndex) -> None:
if QApplication.keyboardModifiers() != Qt.NoModifier:
return
record = self._record_at_table_index(index)
if record is None or (self.current_root is None and not self.combined.enabled):
return
columns = self._columns_for_table_model()
if columns is not None and index.column() < len(columns):
column_key = columns[index.column()].key
if column_key in {"bibcode", "arxiv", "filename"}:
if self.config.link_activation == "double":
self._activate_link_cell(index, record, column_key)
return
self._open_pdf_record(record)
def _merge_annotations(
self,
records: Sequence[dict[str, Any]] | None = None,
*,
root: Path | None = None,
) -> None:
"""Review the two selected editions, or choose PDFs from file dialogs."""
if self._ads_recheck_active_root is not None:
self.status.showMessage(self._operation_wait_message())
return
if self.combined.enabled and root is None:
self.status.showMessage("Select one library to merge PDFs.")
return
root = root or self.current_root
selected = tuple(
deepcopy(record)
for record in (records if records is not None else self._selected_records())
)
if len(selected) == 2 and root is not None:
if (
self.job is not None
or self._pending_watch_dialog is not None
or self._pending_ads_refresh is not None
or self._foreground_paused_root is not None
or self._edition_merge_target is not None
or self._close_requested
):
self.status.showMessage(self._operation_wait_message("merging PDFs"))
return
target_record = next(
(record for record in selected if edition_role(record) == "published"),
selected[0],
)
_candidates, linked = edition_candidates(selected[0], selected)
self._edition_merge_target = EditionMergeRequest(
root=root,
records=selected,
target=root
/ str(
target_record.get("current_path") or target_record["relative_path"]
),
linked=linked,
)
self._request_foreground_dialog("edition_merge", root)
return
if len(selected) > 2:
self.status.showMessage(
"Select exactly two PDFs to review annotation merging."
)
return
start = str(root) if root else ""
target, _ = QFileDialog.getOpenFileName(
self, "Choose the canonical (target) PDF", start, "PDF (*.pdf)"
)
if not target:
return
source, _ = QFileDialog.getOpenFileName(
self,
"Choose the PDF whose annotations to copy (source)",
str(Path(target).parent),
"PDF (*.pdf)",
)
if not source:
return
self._review_annotation_pair(Path(target), Path(source))
def _review_annotation_pair(self, target: Path, source: Path) -> MergeReport | None:
"""Open annotation review and remember outputs created by this application."""
try:
if not target.is_file() or not source.is_file():
raise AnnotationMergeError(
"A selected PDF is no longer available. Refresh the library and select the files again."
)
if target.samefile(source):
raise AnnotationMergeError("Choose two distinct PDF files.")
from .annotation_merge import AnnotationMergeDialog
dialog = AnnotationMergeDialog(target, source, self)
dialog.copy_saved.connect(
lambda output: self._merge_output_paths.add(Path(output).resolve())
)
dialog.copy_saved.connect(
lambda _output: self._remember_annotation_merge(dialog.report)
)
dialog.originals_trashed.connect(self._record_trashed_merge_originals)
dialog.exec()
if dialog.report is not None:
self._merge_output_paths.add(Path(dialog.report.output_path).resolve())
return dialog.report
except (
MissingDependencyError,
AnnotationMergeError,
OSError,
RuntimeError,
) as exc:
QMessageBox.warning(self, "Cannot merge annotations", str(exc))
return None
def _remember_annotation_merge(self, report: MergeReport | None) -> None:
"""Carry earlier inputs forward and register the saved merged PDF."""
if report is None or not report.verified or not report.output_sha256:
return
direct_inputs = report.input_files
records = {}
for original in direct_inputs:
path = Path(original.path)
for key, registry in self.registries.items():
root = Path(key)
if path.is_relative_to(root):
record = registry.find_file_by_current_path(
root, path.relative_to(root).as_posix()
)
if record is not None:
records[original.path] = record
break
inputs: dict[str, MergeInput] = {}
for original in direct_inputs:
ancestors = self._merge_inputs_by_sha256.get(original.sha256, ())
record = records.get(original.path) or {}
audit = (record.get("metadata") or {}).get(_MERGE_AUDIT_KEY)
if (
not ancestors
and isinstance(audit, dict)
and audit.get("complete") is True
and audit.get("output_sha256") == original.sha256
and record.get("sha256") == original.sha256
):
stored = audit.get("inputs")
if isinstance(stored, list):
ancestors = tuple(
MergeInput(value["path"], value["sha256"])
for value in stored
if isinstance(value, dict)
and isinstance(value.get("path"), str)
and Path(value["path"]).is_absolute()
and isinstance(value.get("sha256"), str)
)
for ancestor in ancestors:
inputs[ancestor.path] = ancestor
inputs[original.path] = original
report.input_files = tuple(inputs.values())
if report.complete:
self._merge_inputs_by_sha256[report.output_sha256] = report.input_files
output = Path(report.output_path).absolute()
canonical = records.get(direct_inputs[0].path) if direct_inputs else None
metadata = deepcopy((canonical or {}).get("metadata") or {})
if (
canonical
and direct_inputs
and canonical.get("sha256") == direct_inputs[0].sha256
):
# Verified annotation copying preserves the canonical page content.
for key in (METADATA_SHA256_KEY, "sha256"):
if metadata.get(key) == direct_inputs[0].sha256:
metadata[key] = report.output_sha256
metadata[_MERGE_AUDIT_KEY] = {
"complete": report.complete,
"output_sha256": report.output_sha256,
"inputs": [
{"path": original.path, "sha256": original.sha256}
for original in report.input_files
],
}
try:
for key, registry in list(self.registries.items()):
root = Path(key)
if not output.is_relative_to(root):
continue
relative = output.relative_to(root).as_posix()
stat = output.stat()
registry.upsert_file(
root=root,
relative_path=relative,
current_path=relative,
sha256=report.output_sha256,
size_bytes=stat.st_size,
mtime_ns=stat.st_mtime_ns,
status=str((canonical or {}).get("status") or "unresolved"),
metadata=metadata,
)
self.refresh(root)
except (OSError, RuntimeError, sqlite3.Error) as exc:
self.status.showMessage(
f"Merged PDF saved, but the library could not update: {exc}. Refresh local files after closing the review."
)
def _record_trashed_merge_originals(self, paths: tuple[Path, ...]) -> None:
"""Hide trashed locations while retaining their catalog history."""
removed = {path.absolute() for path in paths if not path.exists()}
for key, registry in list(self.registries.items()):
root = Path(key)
relative = {
path.relative_to(root).as_posix()
for path in removed
if path.is_relative_to(root)
}
if not relative:
continue
seen = {
record["current_path"] for record in registry.iter_files()
} - relative
try:
registry.catalog.mark_missing_locations(root, seen)
self.refresh(root)
except (OSError, RuntimeError, sqlite3.Error) as exc:
self.status.showMessage(
f"Original PDFs moved to Trash, but the library could not update: {exc}. Refresh local files."
)
def _reset_link_activation(self) -> None:
"""Forget the last activated target when the table contents are replaced."""
self._last_link_activation = None
def _columns_for_table_model(self) -> tuple[Column, ...] | None:
"""Return column definitions matching the model installed on the table."""
table_model = self.table.model()
if table_model is None:
return None
model_columns = getattr(table_model, "columns", None)
if model_columns is not None:
return tuple(model_columns)
base_columns = build_columns()
if table_model.columnCount() == len(base_columns):
return base_columns
if table_model.columnCount() == len(self.columns):
return self.columns
return None
def _table_clicked(self, index: QModelIndex) -> None:
"""Show status details, copy citations, or open links for a table click."""
if not index.isValid():
return
record = self._record_at_table_index(index)
if record is None or (self.current_root is None and not self.combined.enabled):
return
columns = self._columns_for_table_model()
if columns is None or index.column() >= len(columns):
return
column = columns[index.column()]
if column.key.startswith("citation:"):
display_text = index.data(Qt.DisplayRole)
if display_text is not None and str(display_text) != "":
self._copy_text(str(display_text))
self.status.showMessage("Copied", 1500)
return
if column.key == "status" and QApplication.keyboardModifiers() == Qt.NoModifier:
self.table.selectionModel().setCurrentIndex(
index,
QItemSelectionModel.ClearAndSelect | QItemSelectionModel.Rows,
)
self.detail_action.setChecked(True)
self._toggle_detail(True)
return
if self.config.link_activation != "single":
return
if QApplication.keyboardModifiers() != Qt.NoModifier:
return
self._activate_link_cell(index, record, column.key)
def _activate_link_cell(
self, index: QModelIndex, record: dict[str, Any], column_key: str
) -> None:
"""Open the ADS, arXiv, or filename target of a table cell."""
if column_key == "filename":
path = self._path_for_record(record)
if path is None:
return
target = str(path)
elif column_key in {"bibcode", "arxiv"}:
columns = self._columns_for_table_model()
column = next((c for c in columns or () if c.key == column_key), None)
value = column.getter(record) if column is not None else None
if not value:
return
target = ads_url(value) if column_key == "bibcode" else arxiv_url(value)
else:
return
# Qt delivers the second click of a double-click after "doubleClicked";
# ignore a repeat of the same target within the system interval.
now = time.monotonic()
previous = self._last_link_activation
interval = QApplication.doubleClickInterval() / 1000
if (
previous is not None
and previous[0] == target
and now - previous[1] < interval
):
return
self._last_link_activation = (target, now)
if column_key == "filename":
self._open_filename(record)
else:
self._open_external_url(target)
def _open_external_url(self, url: str) -> None:
if open_url(url, self.config.web_browser_command):
return
QMessageBox.warning(self, "Cannot open link", f"Could not open:\n{url}")
def _open_filename(self, record: dict[str, Any]) -> None:
path = self._path_for_record(record)
if path is None:
return
if self.config.filename_click_action == "reveal":
if reveal_in_file_manager(path):
return
QMessageBox.warning(
self,
"Cannot reveal file",
f"File is missing or cannot be revealed:\n{path}",
)
return
self._open_pdf_record(record)
def _open_pdf_record(self, record: dict[str, Any]) -> None:
path = self._path_for_record(record)
if path is None:
return
if not open_path(path, self.config.pdf_viewer_command):
QMessageBox.warning(
self, "Cannot open PDF", f"File is missing or cannot be opened:\n{path}"
)
def _path_for_record(self, record: dict[str, Any]) -> Path | None:
"""Resolve a table record through its owning library."""
if self.combined.enabled:
return self.combined.path_for(record)
if self.current_root is None:
return None
return self.current_root / str(record.get("current_path") or "")
def _open_current(self) -> None:
"""Open the current table selection using the standard PDF action."""
focused = QApplication.focusWidget()
if focused is not None and self.detail.isAncestorOf(focused):
return
index = self.table.currentIndex()
record = self._record_at_table_index(index)
if record is not None:
self._open_pdf_record(record)
def _clear_search(self) -> None:
"""Clear search text and return keyboard navigation to the table."""
self.search.clear()
self.table.setFocus()
def _sidebar_menu(self, position: Any) -> None:
item = self.sidebar.itemAt(position)
if item is None or item.data(0, Qt.UserRole + 1) != "root":
return
root = Path(str(item.data(0, Qt.UserRole)))
menu = QMenu(self)
menu.setToolTipsVisible(True)
watch_action = menu.addAction("Watch Current Library")
watch_action.setCheckable(True)
watch_action.setChecked(self.watch_controller.is_watching(root))
watch_reason = self._detail_edit_block_reason()
watch_action.setEnabled(not watch_reason)
if watch_reason:
watch_action.setToolTip(watch_reason)
watch_action.triggered.connect(
lambda checked: self._set_root_watching(root, checked)
)
watch_on_start = menu.addAction("Autowatch at Start")
watch_on_start.setCheckable(True)
watch_on_start.setChecked(self.root_entries[str(root)].watch_on_start)
watch_on_start.triggered.connect(
lambda checked: self._set_root_watch_policy(root, "watch_on_start", checked)
)
auto_process = menu.addAction("Automatically Identify && Rename")
auto_process.setCheckable(True)
auto_process.setChecked(self.root_entries[str(root)].auto_process_new_pdfs)
auto_process.triggered.connect(
lambda checked: self._set_root_watch_policy(
root, "auto_process_new_pdfs", checked
)
)
recursive_action = menu.addAction("Include Subdirectories")
recursive_action.setCheckable(True)
recursive_action.setChecked(self.root_entries[str(root)].recursive)
recursive_action.setEnabled(not self.watch_controller.is_watching(root))
recursive_action.triggered.connect(
lambda checked: self._set_root_recursive(root, checked)
)
menu.addSeparator()
refresh_action = menu.addAction(
"Refresh", lambda: self._refresh_local_files(root)
)
refresh_action.setEnabled(
self.job is None and self._pending_watch_dialog is None
)
menu.addAction("Remove from list", lambda: self._remove_root(root))
menu.addAction(
"Reveal in File Manager",
lambda: reveal_in_file_manager(root),
)
menu.exec(self.sidebar.viewport().mapToGlobal(position))
def _set_root_watching(self, root: Path, checked: bool) -> None:
previous = self.current_root
self.current_root = root.resolve()
try:
self._toggle_current_watch(checked)
finally:
self.current_root = previous
self._sync_watch_actions()
def _set_root_watch_policy(self, root: Path, field: str, checked: bool) -> None:
previous = self.current_root
self.current_root = root.resolve()
try:
self._set_current_watch_policy(field, checked)
finally:
self.current_root = previous
self._sync_watch_actions()
def _table_menu(self, position: Any) -> None:
index = self.table.indexAt(position)
if (
self.combined.enabled
and self.combined.tree_model is not None
and self.combined.tree_model.is_folder_index(index)
):
self.combined.folder_menu(index, position)
return
record = self._record_at_table_index(index)
if record is None or (self.current_root is None and not self.combined.enabled):
return
selection = self.table.selectionModel()
if not selection.isRowSelected(index.row(), index.parent()):
selection.clearSelection()
selection.select(
index,
QItemSelectionModel.SelectionFlag.Select
| QItemSelectionModel.SelectionFlag.Rows,
)
if self.combined.enabled:
selection.setCurrentIndex(index, QItemSelectionModel.NoUpdate)
records = self._selected_records() or (deepcopy(record),)
if self.combined.enabled:
self.combined.table_menu(records, position)
return
count = len(records)
root = self.current_root
paths = [root / str(item.get("current_path") or "") for item in records]
menu = QMenu(self)
menu.addAction(
"Open" if count == 1 else f"Open {count} PDFs",
lambda: [self._open_pdf_record(item) for item in records],
)
menu.addAction(
"Reveal in File Manager" if count == 1 else f"Reveal {count} Files",
lambda: self._reveal_paths(paths),
)
if (
count == 1
and self.current_view is LibraryView.DUPLICATES
and records[0].get("duplicate_kind")
):
menu.addAction(
"Review Duplicates…", lambda: self._review_duplicates(records[0])
)
if count > 1:
compare = menu.addAction(
"Compare Selected PDFs…", lambda: self._compare_selected(records)
)
compare.setToolTip(
"Compare the PDFs. Move a copy to the Trash or merge annotations."
)
merge = menu.addAction(
"Merge Selected PDFs…",
lambda: self._merge_annotations(records, root=root),
)
merge.setEnabled(count == 2)
merge.setToolTip(
"Review source annotations on the published PDF and save a new copy."
if count == 2
else "Select exactly two PDF files."
)
for label, values in (
("Bibcode", [self._record_value(item, "bibcode") for item in records]),
("DOI", [self._record_value(item, "doi") for item in records]),
("Path", [str(path) for path in paths]),
):
present = [value for value in values if value]
plural = "DOIs" if label == "DOI" else f"{label}s"
text = (
f"Copy {label}"
if count == 1
else f"Copy {len(present)} {plural}"
if present
else f"Copy {plural}"
)
action = menu.addAction(
text, lambda present=present: self._copy_text("\n".join(present))
)
action.setEnabled(bool(present))
if count > 1 and not present:
action.setToolTip(f"None of the selected papers has a {label}.")
elif count > 1 and len(present) < count:
action.setToolTip(
f"{count - len(present)} selected papers have no {label}."
)
menu.addSeparator()
identify = menu.addAction(
"Identify Selected PDF…"
if count == 1
else f"Identify {count} Selected PDFs…",
lambda: self._identify(paths),
)
identify.setToolTip(
"Identify only these PDFs, with the strategies and providers you choose"
)
identify.setEnabled(self.identify_action.isEnabled())
identify_rename = menu.addAction(
"Identify && Rename Selected PDF"
if count == 1
else f"Identify && Rename {count} Selected PDFs",
lambda: self._identify_and_rename(paths),
)
identify_rename.setToolTip(
"Identify these PDFs, then rename verified matches with saved settings. "
"Leave unverified files unchanged."
)
identify_rename.setEnabled(
self.identify_action.isEnabled() and self.rename_action.isEnabled()
)
if count == 1:
menu.addAction(self.refresh_selected_ads_action)
menu.addAction(self.update_selected_ads_action)
else:
refreshable = [
item for item in records if self._record_value(item, "bibcode")
]
refresh = menu.addAction(
f"Refresh {len(refreshable)} Selected ADS Records",
lambda: self._refresh_records_ads(refreshable),
)
refresh.setEnabled(
bool(refreshable) and self.refresh_all_ads_action.isEnabled()
)
menu.addSeparator()
add_to_ads_library_action = menu.addAction(
"Add to ADS Library…", self._add_to_ads_library
)
add_to_ads_library_action.setEnabled(self._can_add_to_ads_library())
menu.addAction("Export Selection As…", lambda: self._export_papers(records))
menu.addAction(self.export_all_action)
with_details = document_ids_of(records)
for label, handler in (("Copy BibTeX", self._copy_records_bibtex),):
action = menu.addAction(
label
if count == 1
else f"{label.removesuffix('…')} of {count} Papers"
+ ("…" if label.endswith("…") else ""),
lambda handler=handler: handler(records),
)
action.setEnabled(bool(with_details))
if not with_details:
action.setToolTip(
"Identify the PDF first, or use Update and Rename Selected "
"by Identifier… with its DOI or arXiv ID."
)
elif len(with_details) < count:
action.setToolTip(
f"{count - len(with_details)} selected PDFs have no paper "
"details yet and are left out."
)
menu.addSeparator()
menu.addAction(self.appearance_action)
menu.exec(self.table.viewport().mapToGlobal(position))
def _show_header_context_menu(self, position: Any) -> None:
"""Show column visibility and layout actions for the table header."""
menu = self._create_header_context_menu(
self.table.header().logicalIndexAt(position)
)
menu.exec(self.table.header().mapToGlobal(position))
def _create_header_context_menu(self, logical_index: int | None = None) -> QMenu:
"""Build the context menu for the table's column header."""
menu = QMenu(self)
menu.addAction(self.appearance_action)
menu.addSeparator()
header = self.table.header()
if logical_index is not None and 0 <= logical_index < len(self.columns):
column = self.columns[logical_index]
label = column.header or "Resolution marker"
alignment_menu = QMenu(f"Align {label}", menu)
menu.addMenu(alignment_menu)
alignment_group = QActionGroup(alignment_menu)
alignment_group.setExclusive(True)
selected = self.config.column_alignments.get(column.key, "default")
for text, value in (
("Default", "default"),
("Left", "left"),
("Center", "center"),
("Right", "right"),
):
action = alignment_menu.addAction(text)
action.setCheckable(True)
alignment_group.addAction(action)
action.setChecked(value == selected)
action.triggered.connect(
lambda _checked, key=column.key, choice=value: (
self._set_column_alignment(key, choice)
)
)
menu.addSeparator()
visible_count = sum(
not self.table.isColumnHidden(header.logicalIndex(visual))
for visual in range(header.count())
)
for visual in range(header.count()):
logical = header.logicalIndex(visual)
column = self.columns[logical]
if column.key == DUPLICATE_COLUMN:
# The Duplicates view shows this column by itself.
continue
label = column.header or "Resolution marker"
action = menu.addAction(label)
action.setCheckable(True)
action.setChecked(not self.table.isColumnHidden(logical))
action.setEnabled(action.isChecked() is False or visible_count > 1)
action.triggered.connect(
lambda checked, column_key=column.key: self._set_column_visibility(
column_key, checked
)
)
menu.addSeparator()
show_all = menu.addAction("Show all columns")
show_all.triggered.connect(self._show_all_columns)
resize = menu.addAction("Resize columns to contents")
resize.triggered.connect(
lambda: header.resizeSections(QHeaderView.ResizeToContents)
)
restore = menu.addAction("Restore default column order")
restore.triggered.connect(self._restore_default_column_order)
return menu
def _set_column_alignment(self, key: str, alignment: str) -> None:
"""Apply a column's alignment or restore its default."""
previous = self.config.column_alignments.get(key)
if alignment == "default":
self.config.column_alignments.pop(key, None)
elif alignment in {"left", "center", "right"}:
self.config.column_alignments[key] = alignment
else:
return
self.model.set_column_alignments(self.config.column_alignments)
if self.config.column_alignments.get(key) != previous:
self._save_column_preferences()
def _remember_table_column_order(self, *_args: Any) -> None:
"""Keep column preferences synchronized with header drags."""
if self._restoring_config or self._applying_table_layout:
return
header = self.table.header()
if header.count() != len(self.columns):
return
logical_indices = [
header.logicalIndex(index) for index in range(header.count())
]
if any(index < 0 or index >= len(self.columns) for index in logical_indices):
return
order = [self.columns[index].key for index in logical_indices]
if order != self.config.table_column_order:
self.config.table_column_order = order
self._save_column_preferences()
def _save_column_preferences(self) -> None:
"""Save column changes and report errors while keeping the live view."""
self._sync_config_roots()
try:
save_config(self.config)
except OSError as exc:
self.notice_bar.enqueue(
Notice(
"column-preference",
f"Could not save the column settings: {exc}. "
"The changes apply for this session.",
severity="error",
)
)
else:
self.notice_bar.dismiss("column-preference")
def _remember_table_sort(self, logical_index: int, order: Qt.SortOrder) -> None:
"""Keep window search and reloads consistent with the table sort."""
if (
self._restoring_config
or self._applying_table_layout
or not 0 <= logical_index < len(self.columns)
or not self.columns[logical_index].sortable
):
self._sync_table_sort_indicator()
return
column = self.columns[logical_index]
self.apply_sort(
column.key,
order == Qt.DescendingOrder,
self.current_query.secondary_order_by,
self.current_query.tertiary_order_by,
)
[docs]
def apply_sort(
self,
primary: str,
descending: bool,
secondary: str | None = None,
tertiary: str | None = None,
) -> bool:
"""Apply and save sortable column keys and their direction.
Parameters
----------
primary : str
Key of the primary sortable column.
descending : bool
Use descending order for every key when `True`.
secondary, tertiary : str or None, optional
Keys that order ties, in priority order.
Returns
-------
bool
`True` when ADSfolio saves the selection. `False` if a key is invalid,
initialization is incomplete, or saving fails. A save failure
leaves the selection active and shows a notice.
"""
sortable = {column.key for column in self.columns if column.sortable}
if (
self._restoring_config
or primary not in sortable
or any(
key is not None and key not in sortable for key in (secondary, tertiary)
)
):
return False
self.current_query = replace(
self.current_query,
view=self.current_view,
search=self.search.text(),
order_by=primary,
descending=descending,
secondary_order_by=secondary,
tertiary_order_by=tertiary,
)
self._sync_table_sort_indicator()
if self.model.query != self.current_query:
self.model.set_query(self.current_query)
saved = self._save_sort_preferences()
self._update_status()
return saved
def _sync_table_sort_indicator(self) -> None:
"""Show the current primary sort without starting another sort."""
logical_index = next(
(
index
for index, column in enumerate(self.columns)
if column.key == self.current_query.order_by and column.sortable
),
None,
)
if logical_index is None:
return
order = (
Qt.DescendingOrder if self.current_query.descending else Qt.AscendingOrder
)
header = self.table.header()
blocker = QSignalBlocker(header)
header.setSortIndicator(logical_index, order)
del blocker
def _sync_sort_after_column_reset(self) -> None:
"""Keep the sort indicator on its column when definitions change."""
if self._applying_table_layout:
self._sync_table_sort_indicator()
def _save_sort_preferences(self) -> bool:
"""Save the current sort and report failures without changing the view."""
if self._restoring_config:
return False
saved = (
self.config.sort_column,
self.config.sort_descending,
self.config.secondary_sort_column,
self.config.tertiary_sort_column,
)
selected = (
self.current_query.order_by,
self.current_query.descending,
self.current_query.secondary_order_by,
self.current_query.tertiary_order_by,
)
self._sync_config_roots()
if (
selected == saved
and selected == self._saved_sort_selection
and self.config.roots == self._saved_sort_roots
and not self._sort_save_pending
):
return True
self.config.sort_column = self.current_query.order_by
self.config.sort_descending = self.current_query.descending
self.config.secondary_sort_column = self.current_query.secondary_order_by
self.config.tertiary_sort_column = self.current_query.tertiary_order_by
try:
save_config(self.config)
except OSError as exc:
self._sort_save_pending = True
self.notice_bar.enqueue(
Notice(
"sort-preference",
f"Could not save the sort selection: {exc}. "
"The selection applies for this session.",
severity="error",
)
)
return False
else:
self._sort_save_pending = False
self._saved_sort_selection = selected
self._saved_sort_roots = deepcopy(self.config.roots)
self.notice_bar.dismiss("sort-preference")
return True
def _set_column_visibility(self, key: str, visible: bool) -> None:
"""Set one column's visibility while keeping one column visible."""
logical = next(
(index for index, column in enumerate(self.columns) if column.key == key),
None,
)
if logical is None:
return
if not visible:
visible_count = sum(
not self.table.isColumnHidden(index)
for index in range(len(self.columns))
)
if visible_count <= 1:
return
self.table.setColumnHidden(logical, not visible)
self.config.hidden_table_columns = [
column.key
for index, column in enumerate(self.columns)
if self.table.isColumnHidden(index)
]
self.combined.sync_tree_column()
def _show_all_columns(self) -> None:
"""Make every table column visible and update persisted preferences."""
for index in range(len(self.columns)):
self.table.setColumnHidden(index, False)
self.config.hidden_table_columns = []
self.combined.sync_tree_column()
def _restore_default_column_order(self) -> None:
"""Restore the declaration order of table columns."""
self.config.table_column_order = [column.key for column in self.columns]
self._apply_table_layout()
self._save_column_preferences()
def _edit_citation_keys(self) -> None:
"""Choose and save the citation key styles."""
samples = []
for record in self._selected_records()[:5]:
metadata = record.get("metadata")
if isinstance(metadata, dict) and metadata.get("title"):
samples.append({**metadata, "bibcode": record.get("bibcode")})
dialog = CitationKeySettingsDialog(self, Config.load_key_policy(), samples)
if dialog.exec() != QDialog.DialogCode.Accepted:
return
Config.save_key_policy(dialog.policy())
self.status.showMessage("Saved the citation key styles.", 5000)
def _records_bibtex(self, records: Sequence[dict[str, Any]]) -> str | None:
"""Return BibTeX for table records, asking for citation keys if needed.
Every library uses the same central catalog, so one catalog connection
can export papers from several libraries of the combined view.
"""
ids = document_ids_of(records)
registry = self._export_registry(records)
if not ids or registry is None:
self.status.showMessage(
"The selected PDFs have no paper details yet. Identify them first.",
8000,
)
return None
# Use an own connection to the central catalog. The key window below
# lets a watcher refresh close and replace the library's connection.
catalog = PdfCatalog()
try:
service = LocalReferences(catalog, Config.load_key_policy())
QApplication.setOverrideCursor(Qt.WaitCursor)
try:
message = fetch_ads_bibtex_now(service, ids)
except (ValueError, sqlite3.Error) as exc:
message = f"Could not save ADS BibTeX: {exc}"
finally:
QApplication.restoreOverrideCursor()
if message:
self.status.showMessage(message, 8000)
try:
return export_bibtex_asking_keys(self, service, ids)
except (ValueError, sqlite3.Error) as exc:
QMessageBox.warning(self, "BibTeX export", str(exc))
return None
finally:
catalog.close()
def _export_registry(self, records: Sequence[dict[str, Any]]) -> PdfRegistry | None:
"""Return an open registry whose central catalog holds `records`."""
if not self.combined.enabled:
if self.current_root is None:
return None
return self.registries.get(str(self.current_root))
for record in records:
root = self.combined.root_for(record)
if root is not None:
return self.registries.get(str(root))
return None
def _copy_records_bibtex(self, records: Sequence[dict[str, Any]]) -> None:
"""Copy the BibTeX of the selected papers to the clipboard."""
text = self._records_bibtex(records)
if text is not None:
self._copy_text(text)
self.status.showMessage(
f"Copied BibTeX for {len(document_ids_of(records))} papers.", 5000
)
def _copy_text(self, text: str) -> None:
QApplication.clipboard().setText(text)
def _record_value(self, record: dict[str, Any], key: str) -> str:
"""Return one scalar field from a record or its metadata."""
value = record.get(key)
if value in (None, "", []):
metadata = record.get("metadata")
value = metadata.get(key) if isinstance(metadata, dict) else None
return self._scalar_metadata_value(value) if value is not None else ""
def _reveal_paths(self, paths: Sequence[Path]) -> None:
"""Reveal each distinct folder once, selecting its first file."""
seen: set[Path] = set()
for path in paths:
if path.parent not in seen:
seen.add(path.parent)
reveal_in_file_manager(path)
def _export_papers(
self,
records: Sequence[dict[str, Any]] | None = None,
*,
all_papers: bool = False,
) -> None:
"""Open Export Papers for the selection, the table view, or all PDFs.
In the combined view, "all" means every current PDF in the checked
libraries. Every library uses the same central catalog.
"""
if self.current_root is None and not self.combined.enabled:
return
source = self._search_source()
if source is None:
return
query = self.model.query
try:
everything = source.query_files(
order_by=query.order_by,
descending=query.descending,
secondary_order_by=query.secondary_order_by,
tertiary_order_by=query.tertiary_order_by,
)
except sqlite3.Error as exc:
QMessageBox.warning(self, "Export papers", str(exc))
return
selected = list(records if records is not None else self._selected_records())
scopes = [ExportScope("selected", "Selected rows", selected)]
filtered = self.current_view is not LibraryView.ALL or bool(query.search)
if filtered:
scopes.append(
ExportScope(
"view", "Rows in the current table view", self.model.records
)
)
scopes.append(
ExportScope(
"all",
"All PDFs in the checked libraries"
if self.combined.enabled
else "All PDFs in this library",
everything,
)
)
registry = self._export_registry(everything)
if registry is None:
self.status.showMessage("This library has no current PDFs to export.", 5000)
return
# The window uses its own connection to the central catalog. A watcher
# refresh closes and replaces the library's connection while the window
# is open.
catalog = PdfCatalog()
try:
dialog = ExportPapersDialog(
self,
scopes,
catalog,
self.config.citation_styles,
initial_scope="all" if all_papers or not selected else "selected",
initial_format=self.config.export_format,
browser_command=self.config.web_browser_command,
show_unidentified=None
if self.combined.enabled
else lambda: self._select_view(LibraryView.UNRESOLVED),
exclude_fields=self.config.export_exclude_fields,
)
dialog.exec()
self.config.export_format = dialog.format_key
self.config.export_exclude_fields = dialog.excluded_fields
dialog.deleteLater()
finally:
catalog.close()
def _refresh_records_ads(self, records: Sequence[dict[str, Any]]) -> None:
"""Refresh several records from ADS, one CLI run per record, in order."""
if self.combined.enabled:
return
commands = [
(
"--bibcode",
self._record_value(record, "bibcode"),
"--relative-path",
str(record["relative_path"]),
"--citation-metrics",
)
for record in records
if self._record_value(record, "bibcode")
]
if not commands:
return
self._start_ads_refresh(
*commands[0],
selection_id=int(records[0]["location_id"]),
follow_ups=tuple(commands[1:]),
)
def _load_next_page(self) -> None:
"""Load another page without materializing the whole registry."""
self.model.fetchMore()
self._update_status()
def _remove_missing_records(self) -> None:
"""Remove stale registry entries after explicit user confirmation."""
if self.current_root is None or self.combined.enabled:
return
registry = self.registries[str(self.current_root)]
self.status.showMessage("Checking registry paths for missing files…")
QApplication.setOverrideCursor(Qt.WaitCursor)
try:
missing_ids = [
int(record["location_id"])
for record in registry.iter_files(include_missing=True)
if not (self.current_root / str(record["current_path"])).is_file()
]
finally:
QApplication.restoreOverrideCursor()
if not missing_ids:
QMessageBox.information(self, "Registry cleanup", "No missing files found.")
self._update_status()
return
answer = QMessageBox.question(
self,
"Remove missing registry records",
(
f"Remove {len(missing_ids)} record(s) whose files are absent from "
"this library root? This does not delete any files."
),
)
if answer != QMessageBox.Yes:
self._update_status()
return
removed = registry.delete_files_by_ids(missing_ids)
self.log.appendPlainText(f"Removed {removed} missing registry record(s).")
self.refresh()
def _refresh_visible_library(self) -> None:
"""Refresh the combined table or scan the selected library."""
if self.combined.enabled:
self.combined.refresh()
else:
self._refresh_local_files()
def _refresh_local_files(self, root: Path | None = None) -> None:
"""Queue a filesystem scan after any watcher write has finished."""
if self.combined.enabled and root is None:
return
root = root or self.current_root
if (
root is None
or self.job is not None
or self._pending_watch_dialog is not None
or self._pending_ads_refresh is not None
or self._local_refresh_context is not None
):
return
root = root.expanduser().resolve()
if str(root) not in self.registries:
return
self._local_refresh_context = _LibraryViewContext(
root=root,
selected=self._selected_records() if root == self.current_root else (),
current=deepcopy(self._selected_record())
if root == self.current_root
else None,
query=self.current_query,
loaded=self.model.rowCount(),
scroll=self.table.verticalScrollBar().value(),
)
self._request_foreground_dialog("local_refresh", root)
self._sync_refresh_actions()
def _launch_local_refresh(self, root: Path) -> None:
"""Start a thread-owned scan of the originally requested root."""
if self._close_requested or str(root) not in self.root_entries:
self._local_refresh_context = None
self._resume_foreground_watch()
return
self._local_refresh_result = None
job = LibraryRefreshJob(root, self.root_entries[str(root)].recursive, self)
job.refreshed.connect(self._capture_local_refresh)
self._watch_job(job, f"local refresh of {self._library_name(root)}")
self.status.showMessage(f"Refreshing local files: {root}")
job.start()
def _capture_local_refresh(self, result: LibraryRefresh) -> None:
"""Retain the scan summary until the normal job completion reload."""
self._local_refresh_result = result
def _restore_library_selection(self, context: _LibraryViewContext) -> None:
"""Restore selected locations, or their unique moved attachments."""
if (
self.combined.enabled
or self.current_root != context.root
or self.current_query != context.query
):
return
records = self.registries[str(context.root)].query_files(
**self.current_query.to_filters()
)
by_id = {record["location_id"]: record for record in records}
by_attachment: dict[int, list[int]] = {}
for record in records:
by_attachment.setdefault(record["attachment_id"], []).append(
record["location_id"]
)
targets: dict[int, int] = {}
for record in (
*context.selected,
*((context.current,) if context.current else ()),
):
old_id = record["location_id"]
if old_id in by_id:
targets[old_id] = old_id
else:
matches = by_attachment.get(record["attachment_id"], [])
if len(matches) == 1:
targets[old_id] = matches[0]
needed = set(targets.values())
while self.model.canFetchMore():
loaded = {
self.model.record_at(row)["location_id"]
for row in range(self.model.rowCount())
}
if self.model.rowCount() >= context.loaded and needed <= loaded:
break
previous = self.model.rowCount()
self.model.fetchMore()
if self.model.rowCount() == previous:
break
selection = self.table.selectionModel()
selection.clearSelection()
selected_ids = {
targets[record["location_id"]]
for record in context.selected
if record["location_id"] in targets
}
current_id = (
targets.get(context.current["location_id"]) if context.current else None
)
current_index = QModelIndex()
for row in range(self.model.rowCount()):
location_id = self.model.record_at(row)["location_id"]
index = self.model.index(row, 0)
if location_id in selected_ids:
selection.select(
index, QItemSelectionModel.Select | QItemSelectionModel.Rows
)
if location_id == current_id:
current_index = index
selection.setCurrentIndex(current_index, QItemSelectionModel.NoUpdate)
self._sync_detail_and_selection_actions()
self.table.verticalScrollBar().setValue(context.scroll)
def _edit_appearance(self) -> None:
"""Configure table and detail-pane appearance preferences."""
original_config = deepcopy(self.config)
dialog = AppearanceDialog(
self.config,
self.columns,
self.detail.available_rows(),
self,
)
dialog.settings_changed.connect(lambda: self._apply_appearance_dialog(dialog))
result = dialog.exec()
if result == QDialog.DialogCode.Accepted:
self._apply_appearance_dialog(dialog)
return
self.config = original_config
self._apply_appearance_config()
def _open_preferences(self) -> None:
"""Open the unified settings window."""
from .preferences import PreferencesDialog
dialog = PreferencesDialog(self)
dialog.exec()
dialog.deleteLater()
def _edit_citation_styles(self) -> None:
"""Edit global citation styles and apply accepted definitions live."""
selected_rows = self._selected_view_rows()
preview_record = (
self._record_at_table_index(selected_rows[0])
if len(selected_rows) == 1
else None
)
dialog = CitationStylesDialog(
self.config.citation_styles,
preview_record=preview_record,
parent=self,
)
dialog.saved.connect(self._apply_citation_styles)
dialog.exec()
def _save_citation_preferences(self, styles: Sequence[CitationStyle]) -> None:
"""Save citation styles and preserve the in-memory settings on failure."""
original = deepcopy(self.config)
try:
self._sync_config_roots()
self._apply_citation_styles(styles)
except (OSError, ValueError):
self.config = original
raise
@staticmethod
def _merge_citation_order(
order: Sequence[str],
deleted_keys: set[str],
citation_keys: Sequence[str],
added_keys: Sequence[str],
) -> list[str]:
"""Reorder active citation keys and place new ones after Bibcode."""
citations = tuple(dict.fromkeys(citation_keys))
citation_set = set(citations)
added_set = set(added_keys)
retained = [
key for key in order if key not in deleted_keys and key not in added_set
]
present = {key for key in retained if key in citation_set}
citation_slots = [
index for index, key in enumerate(retained) if key in citation_set
]
existing = [key for key in citations if key in present]
for index, key in zip(citation_slots, existing):
retained[index] = key
missing = [key for key in citations if key not in present]
if not missing or "bibcode" not in retained:
return retained
insertion = retained.index("bibcode") + 1
return [*retained[:insertion], *missing, *retained[insertion:]]
def _apply_citation_styles(self, styles: Sequence[CitationStyle]) -> None:
"""Apply accepted citation styles and update all dependent UI state."""
prior_widths = {
column.key: self.table.columnWidth(index)
for index, column in enumerate(self.columns)
}
current_index = self.table.currentIndex()
source_index = self._source_table_index(current_index)
selected_row = source_index.row() if source_index.isValid() else None
selected_record = (
self.model.record_at(selected_row) if selected_row is not None else None
)
selected_location_id = (
selected_record.get("location_id") if selected_record is not None else None
)
old_keys = {
citation_preference_key(style.id) for style in self.config.citation_styles
}
normalized_styles = tuple(styles)
new_keys = [citation_preference_key(style.id) for style in normalized_styles]
new_key_set = set(new_keys)
deleted_keys = old_keys - new_key_set
added_keys = [key for key in new_keys if key not in old_keys]
added_key_set = set(added_keys)
for preferences in (
self.config.column_widths,
self.config.column_fonts,
self.config.column_alignments,
self.config.detail_fonts,
):
for key in deleted_keys:
preferences.pop(key, None)
self.config.table_column_order = self._merge_citation_order(
self.config.table_column_order,
deleted_keys,
new_keys,
added_keys,
)
self.config.detail_row_order = self._merge_citation_order(
self.config.detail_row_order,
deleted_keys,
new_keys,
added_keys,
)
self.config.hidden_table_columns = [
key
for key in self.config.hidden_table_columns
if key not in deleted_keys and key not in added_key_set
]
self.config.detail_hidden_rows = [
key
for key in self.config.detail_hidden_rows
if key not in deleted_keys and key not in added_key_set
]
self.config.citation_styles = list(normalized_styles)
save_config(self.config)
self.combined.update_columns()
self.detail.set_row_order(self.config.detail_row_order)
self.detail.set_hidden_rows(self.config.detail_hidden_rows)
self.detail.set_citation_styles(self.config.citation_styles)
self._rebuild_table_delegates(prior_widths)
self._apply_table_appearance()
if selected_record is not None:
restored_row = selected_row
if selected_location_id is not None:
restored_row = next(
(
row
for row in range(self.model.rowCount())
if (
self.model.record_at(row) is not None
and self.model.record_at(row).get("location_id")
== selected_location_id
)
),
None,
)
if restored_row is not None and restored_row < self.model.rowCount():
restored_index = self._view_table_index(
self.model.index(restored_row, 0)
)
if restored_index.isValid():
self.table.setCurrentIndex(restored_index)
self._current_changed(restored_index, QModelIndex())
def _apply_appearance_dialog(self, dialog: AppearanceDialog) -> None:
"""Apply the current Appearance dialog values to the running window."""
self.config.column_fonts = {
key: dict(value) for key, value in dialog.column_fonts.items()
}
self.config.column_alignments = dict(dialog.column_alignments)
self.config.detail_fonts = {
key: dict(value) for key, value in dialog.detail_fonts.items()
}
self.config.timestamp_format = dialog.timestamp_format_value
self.config.table_column_order = list(dialog.table_column_order)
self.config.hidden_table_columns = list(dialog.hidden_table_columns)
self.config.expand_abstracts = dialog.expand_abstracts
self.config.table_row_height = dialog.table_row_height
self.config.detail_row_spacing = dialog.detail_row_spacing
self.config.detail_label_bold = dialog.detail_label_bold
self.config.detail_compact_rows = dialog.detail_compact_rows
self.config.show_edit_buttons = dialog.show_edit_buttons
self.config.show_lock_buttons = dialog.show_lock_buttons
self.config.filename_click_action = dialog.filename_click_action
self.config.link_activation = dialog.link_activation
self.config.pdf_viewer_command = dialog.pdf_viewer_command
self.config.web_browser_command = dialog.web_browser_command
self.config.detail_row_order = list(dialog.detail_row_order)
self.config.detail_hidden_rows = list(dialog.detail_hidden_rows)
self._apply_appearance_config()
def _apply_appearance_config(self) -> None:
"""Apply all table and detail appearance settings to the live UI."""
self.detail.set_row_order(self.config.detail_row_order)
self.detail.set_hidden_rows(self.config.detail_hidden_rows)
self.detail.set_compact_rows(self.config.detail_compact_rows)
self.detail.set_label_width(self.config.detail_label_width)
self.detail.set_link_activation(self.config.link_activation)
self._apply_table_appearance()
self._current_changed(self.table.currentIndex(), QModelIndex())
def _edit_sort_priorities(self) -> None:
"""Configure persisted primary, secondary, and tertiary sort keys."""
choices = [
"(none)",
*(column.key for column in self.columns if column.sortable is True),
]
dialog = QDialog(self)
dialog.setWindowTitle("Sort Priorities")
layout = QVBoxLayout(dialog)
selectors: list[QComboBox] = []
for label, current in (
("Primary sort", self.current_query.order_by),
("Secondary sort", self.current_query.secondary_order_by),
("Tertiary sort", self.current_query.tertiary_order_by),
):
selector = QComboBox(dialog)
selector.addItems(choices[1:] if label == "Primary sort" else choices)
selector.setCurrentText(current if current in choices else "(none)")
layout.addWidget(QLabel(label, dialog))
layout.addWidget(selector)
selectors.append(selector)
buttons = QDialogButtonBox(
QDialogButtonBox.Ok | QDialogButtonBox.Cancel, dialog
)
buttons.accepted.connect(dialog.accept)
buttons.rejected.connect(dialog.reject)
layout.addWidget(buttons)
if dialog.exec() != QDialog.DialogCode.Accepted:
return
selected = [
None if selector.currentText() == "(none)" else selector.currentText()
for selector in selectors
]
primary = selected[0] or self.current_query.order_by
secondary = selected[1]
tertiary = selected[2]
self.apply_sort(
primary,
self.current_query.descending,
secondary,
tertiary,
)
def _apply_table_appearance(self) -> None:
"""Apply persisted typography and timestamp display to the table."""
fonts: dict[str, QFont] = {}
colors: dict[str, QColor] = {}
for key, settings in self.config.column_fonts.items():
family = settings.get("family")
point_size = settings.get("point_size")
if not isinstance(family, str) or not isinstance(point_size, int):
continue
font = QFont(family)
font.setPointSize(point_size)
fonts[key] = font
color = QColor(str(settings.get("color", "")))
if color.isValid():
colors[key] = color
self.model.set_column_fonts(fonts, colors)
self.model.set_column_alignments(self.config.column_alignments)
detail_fonts: dict[str, QFont] = {}
detail_colors: dict[str, QColor] = {}
for key, settings in self.config.detail_fonts.items():
family = settings.get("family")
point_size = settings.get("point_size")
if not isinstance(family, str) or not isinstance(point_size, int):
continue
font = QFont(family)
font.setPointSize(point_size)
detail_fonts[key] = font
color = QColor(str(settings.get("color", "")))
if color.isValid():
detail_colors[key] = color
self.model.set_timestamp_format(self.config.timestamp_format)
self.detail.set_timestamp_format(self.config.timestamp_format)
self.detail.set_content_font(fonts.get("detail"), colors.get("detail"))
self.detail.set_row_styles(detail_fonts, detail_colors)
self.detail.set_expand_abstracts(self.config.expand_abstracts)
self.detail.set_row_spacing(self.config.detail_row_spacing)
self.detail.set_label_bold(self.config.detail_label_bold)
self.detail.set_action_buttons_visible(
show_edit=self.config.show_edit_buttons,
show_lock=self.config.show_lock_buttons,
)
self.detail.set_application_commands(
pdf_viewer_command=self.config.pdf_viewer_command,
web_browser_command=self.config.web_browser_command,
)
self._set_table_row_height(self.config.table_row_height)
self._apply_table_layout()
def _set_table_row_height(self, height: int | None) -> None:
"""Apply table-row height to every active item delegate."""
for delegate in self._table_row_delegates:
delegate.set_row_height(height)
self.table.doItemsLayout()
self.table.viewport().update()
def _rebuild_table_delegates(
self,
preserved_widths: dict[str, int] | None = None,
) -> None:
"""Recreate delegates and widths for the active runtime columns."""
self._table_row_delegates = []
self.search_highlight.column_keys = tuple(column.key for column in self.columns)
default_delegate = RowHeightDelegate(self.table)
default_delegate.highlight = self.search_highlight
self.table.setItemDelegate(default_delegate)
self._table_row_delegates.append(default_delegate)
widths = preserved_widths or {}
for index, column in enumerate(self.columns):
width = widths.get(
column.key,
self.config.column_widths.get(column.key, column.width),
)
self.table.setColumnWidth(index, width)
self.table.setItemDelegateForColumn(index, default_delegate)
delegate: RowHeightDelegate | None = None
if column.key == "author":
delegate = FirstAuthorDelegate(self.table)
elif column.key == "status":
delegate = ResolutionDelegate(self.table)
elif column.key in {"bibcode", "arxiv"}:
delegate = LinkDelegate(
"ADS" if column.key == "bibcode" else "arXiv",
self.table,
)
if delegate is not None:
delegate.highlight = self.search_highlight
self.table.setItemDelegateForColumn(index, delegate)
self._table_row_delegates.append(delegate)
def _apply_table_layout(self) -> None:
"""Apply persisted column order and visibility to the library table."""
positions = {column.key: index for index, column in enumerate(self.columns)}
order = list(
dict.fromkeys(
[
*(
key
for key in self.config.table_column_order
if key in positions
),
*(column.key for column in self.columns),
]
)
)
header = self.table.header()
self._applying_table_layout = True
try:
for visual_index, key in enumerate(order):
logical_index = positions[key]
header.moveSection(header.visualIndex(logical_index), visual_index)
self.table.setColumnHidden(
logical_index,
key in self.config.hidden_table_columns
or (
key == DUPLICATE_COLUMN
and self.current_view is not LibraryView.DUPLICATES
),
)
finally:
self._applying_table_layout = False
self.combined.sync_tree_column()
def _sync_duplicate_column(self) -> None:
"""Show the Duplicate column only in the Duplicates view."""
for logical_index, column in enumerate(self.columns):
if column.key == DUPLICATE_COLUMN:
self._applying_table_layout = True
try:
self.table.setColumnHidden(
logical_index, self.current_view is not LibraryView.DUPLICATES
)
finally:
self._applying_table_layout = False
def _add_root(self) -> None:
selected = QFileDialog.getExistingDirectory(self, "Add Library Root")
if not selected:
return
path = Path(selected)
recursive = False
watch, auto_process = False, False
is_new = str(path.expanduser().resolve()) not in self.registries
if is_new:
choice = self._ask_include_subfolders(path)
if choice is None:
return
recursive = choice
watch, auto_process = self._ask_watch_new_library(path)
self._register_root(
path,
recursive=recursive,
watch_on_start=watch,
auto_process_new_pdfs=auto_process,
)
if is_new:
self._sync_config_roots()
try:
save_config(self.config)
except OSError as exc:
self.notice_bar.enqueue(
Notice(
f"watch-save:{path}",
f"Could not save the new library settings: {exc}",
severity="error",
)
)
block_reason = self._detail_edit_block_reason()
if watch and not self._start_root_watch(path) and block_reason:
self.notice_bar.enqueue(
Notice(
f"watch-start:{path.expanduser().resolve()}",
f"Watching {path.name or path} did not start. {block_reason} "
"Then choose Watch Current Library. Autowatch at Start "
"is on, so watching also starts when ADSfolio opens.",
severity="warning",
root=path.expanduser().resolve(),
)
)
self._select_root(path)
self._sync_watch_actions()
if self.model.total:
self._guide_identification()
def _ask_watch_new_library(self, path: Path) -> tuple[bool, bool]:
"""Ask whether ADSfolio watches a new library for added PDFs.
Returns
-------
tuple of bool
``(watch, auto_process)``. ``watch`` turns on Autowatch at Start
and starts the watcher now. ``auto_process`` turns on
Automatically Identify & Rename.
"""
box = QMessageBox(self)
box.setIcon(QMessageBox.Question)
box.setWindowTitle("Watch This Library?")
box.setText(f"Watch {path.name or path} for new PDFs?")
box.setInformativeText(
"A watched library finds PDFs that you add to its folder. "
"With automatic processing, ADSfolio also identifies and renames them. "
"Each library has its own watch settings. Change them later in the "
"Tools menu or on the library's context menu."
)
automatic = box.addButton("Watch, Identify, and Rename", QMessageBox.YesRole)
notify = box.addButton("Watch and Notify Only", QMessageBox.AcceptRole)
skip = box.addButton("Do Not Watch", QMessageBox.NoRole)
box.setDefaultButton(skip)
box.setEscapeButton(skip)
box.exec()
clicked = box.clickedButton()
if clicked is automatic:
return True, True
if clicked is notify:
return True, False
return False, False
def _ask_include_subfolders(self, path: Path) -> bool | None:
"""Ask whether a new library includes subfolders.
Returns
-------
bool or `None`
`True` to include subfolders, `False` for this folder only, or
`None` when the user cancels. A folder without visible subfolders
returns `False` without asking.
"""
try:
has_subfolders = any(
entry.is_dir() and not entry.name.startswith(".")
for entry in path.iterdir()
)
except OSError:
has_subfolders = False
if not has_subfolders:
return False
box = QMessageBox(self)
box.setIcon(QMessageBox.Question)
box.setWindowTitle("Include Subfolders?")
box.setText(f"Include PDFs in the subfolders of {path.name or path}?")
box.setInformativeText(
"You can change this later with Include Subdirectories on the library."
)
include = box.addButton("Include Subfolders", QMessageBox.YesRole)
only = box.addButton("This Folder Only", QMessageBox.NoRole)
box.addButton(QMessageBox.Cancel)
box.setDefaultButton(only)
box.exec()
clicked = box.clickedButton()
if clicked is include:
return True
if clicked is only:
return False
return None
def _remove_root(self, root: Path | None = None) -> None:
root = root or self.current_root
if root is None:
return
root = root.expanduser().resolve()
key = str(root)
if (
self._pending_priority_edit is not None
and self._pending_priority_edit.root == root
):
self.status.showMessage("Wait for the PDF update to start.")
return
if root == self._ads_recheck_active_root:
self.status.showMessage(self._operation_wait_message())
return
if (
self._manual_metadata_target is not None
and self._manual_metadata_target[0] == root
):
self.status.showMessage(
"Close the paper details dialog before removing this library."
)
return
if any(
context is not None and context.root == root
for context in (self._local_refresh_context, self._guided_identify_context)
):
self.status.showMessage(
self._operation_wait_message("removing this library")
)
return
if (
self._pending_ads_refresh is not None
and self._pending_ads_refresh[0] == root
):
self._pending_ads_refresh = None
self._pending_watch_dialog = None
self.watch_controller.resume_writes(root)
if self.watch_controller.is_watching(
root
) and not self.watch_controller.stop_root(root):
self.notice_bar.enqueue(
Notice(
f"remove-watch:{root}",
f"Could not stop watching {root}; the library was not removed.",
severity="error",
root=root,
)
)
return
registry = self.registries.pop(key, None)
if registry is not None:
registry.close()
self.root_entries.pop(key, None)
self.combined.remove_root(key)
self.notice_bar.clear_root(root)
for index in range(self.libraries_item.childCount()):
item = self.libraries_item.child(index)
if item.data(0, Qt.UserRole) == key:
self.libraries_item.takeChild(index)
break
if self.current_root == root:
self.current_root = None
self.model.set_registry(None)
self.views_item.takeChildren()
if self.registries:
self._select_root(Path(next(iter(self.registries))))
elif self.combined.enabled:
self.combined.set_enabled(False)
self._update_status()
elif self.combined.enabled:
self.combined.update_columns()
self.combined.refresh()
self._sync_config_roots()
self._sync_watch_actions()
self._sync_ads_library_action()
[docs]
def refresh(self, root: Path | None = None) -> None:
"""Reload the current registry and sidebar counts."""
root = root or self.current_root
if root is None:
return
key = str(root)
self.registries[key].close()
self.registries[key] = PdfRegistry(root)
entry = self.root_entries[key]
for index in range(self.libraries_item.childCount()):
item = self.libraries_item.child(index)
if item.data(0, Qt.UserRole) == key:
count = self._safe_count(self.registries[key])
marker = "● " if self.watch_controller.is_watching(root) else ""
item.setText(
0,
f"{marker}{entry.label or root.name or key} ({count if count is not None else '?'})",
)
break
if self.combined.enabled and key in self.combined.selected:
self.combined.refresh()
elif not self.combined.enabled and self.current_root == root:
self.model.set_registry(self.registries[key], self.current_query)
self._populate_views(self.registries[key])
self._update_status()
def _toggle_notifications(self) -> None:
"""Open or close notifications without resizing the main panes."""
if self.notification_panel.isVisible():
self.notification_panel.hide()
else:
self.notification_panel.show_for(self.notifications_button)
def _sync_notifications(self) -> None:
"""Update the permanent notification indicator without opening its panel."""
notices = self.notice_bar.queued_notices()
count = len(notices)
severities = {notice.severity for notice in notices}
icon = (
"error"
if "error" in severities
else "warning"
if "warning" in severities
else "info"
if not count and self.view_explanation.text()
else "notifications"
)
self.notifications_button.setIcon(vector_icon(icon, self.notifications_button))
self.notifications_button.set_badge(count)
summary = (
"Errors need review."
if "error" in severities
else "Warnings need review."
if "warning" in severities
else "Open messages and current-view help."
)
self.notifications_button.setAccessibleName(
f"Notifications: {count} pending. {summary}"
)
tooltip = f"Notifications: {count} pending.\n{summary}"
if self.view_explanation.text():
tooltip += f"\n{self.view_explanation.text()}"
self.notifications_button.setToolTip(tooltip)
def _update_status(self) -> None:
self._sync_welcome()
explains_ads_group = self.current_root is not None and self.current_view in {
LibraryView.RESOLVED,
LibraryView.IDENTIFIED_WITHOUT_ADS,
LibraryView.UNRESOLVED,
}
explanation = self.current_view.description if explains_ads_group else ""
if self.combined.enabled:
explanation = (
"Combined view shows the checked libraries. "
"Select a PDF to edit its details in its own library. "
"Existing watchers keep running."
)
self.notification_panel.set_guidance(
explanation,
identify=self.current_view is LibraryView.UNRESOLVED
and not self.combined.enabled,
)
self._sync_notifications()
if self.current_root is None:
self.status.setToolTip("")
self.status.showMessage("No library selected")
self._set_identify_badge(None)
return
total = self.model.total
loaded = self.model.rowCount()
registry = (
self.combined.registry()
if self.combined.enabled
else self.registries[str(self.current_root)]
)
facts = self._safe_facts(registry)
library_unresolved = (
count_matching(facts, group="needs_identification")
if facts is not None
else None
)
self._set_identify_badge(
library_unresolved, unavailable=library_unresolved is None
)
if facts is None:
return
query = self.current_query
try:
search_keys = (
registry.search_location_keys(
query.search,
exact=query.search_exact,
wildcards=query.search_wildcards,
fields=query.search_fields,
)
if query.search
else None
)
except sqlite3.OperationalError as exc:
if "locked" not in str(exc).lower() and "busy" not in str(exc).lower():
raise
self.status.showMessage("Library is busy; try again in a moment.")
return
# A group filter replaces the group of an identification view.
view = (
self.current_view
if self.current_view in {LibraryView.ARXIV_ONLY, LibraryView.DUPLICATES}
else LibraryView.ALL
)
resolved, external, unresolved = (
count_matching(facts, view, group=group, search_keys=search_keys)
for group in (
"ads_linked",
"identified_without_ads",
"needs_identification",
)
)
prefix = "Combined libraries · " if self.combined.enabled else ""
partial = (
" · some libraries unavailable"
if self.combined.enabled and registry.unavailable_roots
else ""
)
self.status.showMessage(
f"{prefix}{loaded} loaded of {total} · {resolved} ADS linked · "
f"{external} identified without ADS · {unresolved} need identification{partial}"
)
self.load_more_action.setEnabled(self.model.canFetchMore())
self.status.setToolTip(self.current_view.description)
def _guide_identification(self) -> None:
"""Explain identification and start it with one confirmation."""
root = self.current_root
if (
root is None
or self.combined.enabled
or self.job is not None
or self._pending_watch_dialog is not None
or self._pending_ads_refresh is not None
or self._close_requested
):
return
settings = self._effective_identify_settings(root)
providers = " → ".join(DOI_PROVIDER_LABELS[p] for p in settings.providers)
skipped_folder = (
" The configured unresolved folder is skipped."
if settings.skip_unresolved_directory
else ""
)
prompt = QMessageBox(self)
prompt.setWindowTitle("Identify papers")
prompt.setTextFormat(Qt.PlainText)
prompt.setIcon(QMessageBox.Information)
prompt.setText(f"Find publication details for PDFs in {root.name}?")
cache_step = (
"1. Link stored bibcodes from the cache or ADS, including tracked subfolders.\n"
if DoiProvider.ADS in settings.providers
else "1. Use the selected metadata providers. ADS lookup is disabled.\n"
)
scope = (
"this folder and its subfolders"
if self.root_entries[str(root)].recursive
else "this folder only"
)
prompt.setInformativeText(
cache_step
+ f"2. Identify papers that still need metadata in {scope}.{skipped_folder}\n\n"
f"DOI provider order: {providers}. Search and category filters do not limit the check.\n\n"
"Locked fields and history stay intact. Unlocked publication fields can be updated. PDFs keep their names and locations.\n\n"
"Select OK to start. If ADS needs a token, setup opens first."
)
prompt.setToolTip(str(root))
prompt.setStandardButtons(QMessageBox.Ok | QMessageBox.Cancel)
prompt.setDefaultButton(QMessageBox.Ok)
prompt.setEscapeButton(QMessageBox.Cancel)
advanced = prompt.addButton("Advanced options…", QMessageBox.ActionRole)
prompt.exec()
if prompt.clickedButton() is advanced:
self._identify()
return
if prompt.standardButton(prompt.clickedButton()) != QMessageBox.Ok:
return
if DoiProvider.ADS in settings.providers:
try:
Config.get_token(prompt=False)
except AuthenticationError:
registry = self.registries[str(root)]
records = list(registry.iter_files())
codes = {
self._scalar_metadata_value(
record.get("bibcode") or record["metadata"].get("bibcode")
)
for record in records
if record.get("status") != "resolved"
and stored_bibcode_matches_file(record)
}
cached = registry.find_ads_documents_by_bibcodes(list(codes))
needs_token = (
settings.providers == (DoiProvider.ADS,)
or any(code.lower() not in cached for code in codes)
or any(
not stored_bibcode_matches_file(record)
and (verification_stale(record) or not metadata_ready(record))
for record in records
)
)
if needs_token and TokenDialog(self).exec() != QDialog.Accepted:
return
except OSError as exc:
QMessageBox.warning(self, "Could not read ADS token", str(exc))
return
self._guided_identify_context = _LibraryViewContext(
root=root,
selected=self._selected_records(),
current=deepcopy(self._selected_record()),
query=self.current_query,
loaded=self.model.rowCount(),
scroll=self.table.verticalScrollBar().value(),
)
flags = [
"--settings-scope",
self.config.identify_scopes.get(str(root), "global"),
"--recursive"
if self.root_entries[str(root)].recursive
else "--no-recursive",
]
for strategy in settings.strategies:
flags.extend(("--strategy", strategy.value))
for provider in settings.providers:
flags.extend(("--provider", provider.value))
if settings.skip_unresolved_directory and settings.output_unresolved:
flags.extend(("--output-unresolved", settings.output_unresolved))
else:
flags.append("--clear-output-unresolved")
flags.extend(
(
"--skip-unresolved-directory"
if settings.skip_unresolved_directory
else "--include-unresolved-directory",
"--allow-ads",
)
)
commands = []
if DoiProvider.ADS in settings.providers:
commands.append(
cli_argv(
"pdf",
"refresh-ads",
str(root),
"--all",
"--no-rename",
"--pending-only",
)
)
commands.append(cli_argv("pdf", "identify", str(root), *flags))
self._guided_identify_commands = commands
self._request_foreground_dialog("guided_identify", root)
self._sync_refresh_actions()
def _launch_guided_identification(self, root: Path) -> None:
"""Run the approved lookup steps after watcher writes pause."""
if self._close_requested or str(root) not in self.registries:
self._guided_identify_context = None
self._guided_identify_commands.clear()
self._resume_foreground_watch()
return
commands = self._guided_identify_commands
self._guided_identify_commands = []
job = CommandJob(
commands[0],
follow_up_argv=commands[1:],
parent=self,
working_directory=root,
)
self._refresh_selection_id = None
self._watch_job(job, f"identification in {self._library_name(root)}")
self._sync_refresh_actions()
self.status.showMessage(f"Identifying papers: {root.name}")
job.start()
def _identify(
self, files: Sequence[Path] = (), *, root: Path | None = None
) -> None:
"""Open Identify for the library, or for `files` only when given."""
if self.combined.enabled and root is None:
return
if self.combined.enabled and self._detail_edit_block_reason():
return
root = root or self.current_root
if root is None or str(root.resolve()) not in self.registries:
return
self._refresh_selection_id = None
root = root.resolve()
if self.combined.enabled and str(root) not in self.combined.selected:
return
if self.combined.enabled and any(
not path.resolve().is_relative_to(root) for path in files
):
return
self._identify_files = tuple(files)
self._request_foreground_dialog("identify", root)
def _identify_and_rename(
self, files: Sequence[Path], *, root: Path | None = None
) -> None:
"""Identify and rename selected PDFs with saved settings."""
if (
(self.combined.enabled and root is None)
or not files
or self._detail_edit_block_reason()
):
return
root = root or self.current_root
if root is None or str(root.resolve()) not in self.registries:
return
root = root.resolve()
if self.combined.enabled and str(root) not in self.combined.selected:
return
if self.combined.enabled and any(
not path.resolve().is_relative_to(root) for path in files
):
return
self._refresh_selection_id = None
self._identify_rename_files = tuple(files)
self._identify_rename_context = _LibraryViewContext(
root=root,
selected=self._selected_records(),
current=deepcopy(self._selected_record()),
query=self.current_query,
loaded=self.model.rowCount(),
scroll=self.table.verticalScrollBar().value(),
)
self._request_foreground_dialog("identify_rename", root)
def _identify_combined_selection(
self, records: Sequence[dict[str, Any]], *, rename: bool = False
) -> None:
"""Identify the captured selection only when it still has one owner."""
if not self.combined.enabled or self._detail_edit_block_reason():
return
selected = self._selected_records()
def identity(item: dict[str, Any]) -> tuple[str, Any]:
return str(item.get("root") or ""), item.get("location_id")
if (
len(selected) != len(records)
or len(self._selected_view_rows()) != len(records)
or {identity(item) for item in selected}
!= {identity(item) for item in records}
):
self.status.showMessage(
"The selection changed. Select the PDFs and try again."
)
return
roots = {self.combined.root_for(item) for item in selected}
paths = [self.combined.path_for(item) for item in selected]
if len(roots) != 1 or None in roots or any(path is None for path in paths):
self.status.showMessage("Select PDFs from one available library.")
return
root = next(iter(roots))
assert root is not None
selected_paths = tuple(path for path in paths if path is not None)
if rename:
self._identify_and_rename(selected_paths, root=root)
else:
self._identify(selected_paths, root=root)
def _launch_identify_and_rename(self, root: Path) -> None:
"""Run selected identification and renaming after watcher writes pause."""
files = self._identify_rename_files
self._identify_rename_files = ()
if self._close_requested or str(root) not in self.registries or not files:
self._identify_rename_context = None
self._resume_foreground_watch()
return
file_flags = [argument for path in files for argument in ("--file", str(path))]
identify = cli_argv(
"pdf",
"identify",
str(root),
"--settings-scope",
self.config.identify_scopes.get(str(root), "global"),
"--clear-output-unresolved",
*file_flags,
)
settings = self._effective_rename_settings(root)
rename = cli_argv(
"pdf",
"rename",
str(root),
"--scheme",
settings.template,
"--author-format",
settings.author_format,
"--max-authors",
str(settings.max_authors),
"--authors-when-truncated",
str(settings.authors_when_truncated),
"--char-policy",
settings.char_policy,
"--author-separator",
settings.author_separator,
"--author-truncation-suffix",
settings.author_truncation_suffix,
"--output-unresolved",
str(self._effective_unresolved_directory(root)),
*file_flags,
"--apply",
)
job = CommandJob(
identify, follow_up_argv=[rename], parent=self, working_directory=root
)
self._watch_job(job, f"identify and rename in {self._library_name(root)}")
self.status.showMessage(f"Identifying and renaming {len(files)} selected PDFs…")
job.start()
def _request_foreground_dialog(
self,
kind: str,
root: Path,
*,
ads_refresh: (
tuple[Path, tuple[str, ...], int | None, str | None] | None
) = None,
) -> None:
if (
self.job is not None
or self._pending_watch_dialog is not None
or self._pending_ads_refresh is not None
):
return
if kind == "ads_refresh":
if ads_refresh is None:
return
self._pending_ads_refresh = ads_refresh
if self.watch_controller.is_watching(root):
self._pending_watch_dialog = (kind, root)
self._sync_refresh_actions()
self.watch_controller.pause_writes(root)
return
if kind == "ads_recheck":
self._launch_ads_recheck(root)
elif kind == "ads_refresh":
self._start_pending_ads_refresh(root)
elif kind == "local_refresh":
self._launch_local_refresh(root)
elif kind == "identify":
self._open_identify_dialog(root)
elif kind == "identify_rename":
self._launch_identify_and_rename(root)
elif kind == "guided_identify":
self._launch_guided_identification(root)
elif kind == "manual_metadata":
self._open_manual_metadata(root)
elif kind == "edition_merge":
self._open_edition_merge(root)
elif kind == "rename_file":
self._launch_file_rename(root)
else:
self._open_rename_dialog(root)
def _open_identify_dialog(self, root: Path) -> None:
registry = self.registries[str(root)]
state = CentralPdfRegistryState(registry)
def save_identify_settings(scope: str, settings: IdentifySettings) -> None:
self._persist_identify_defaults(root, scope, settings)
library_settings = state.identify_settings
dialog = IdentifyDialog(
root,
Config.load_identify_settings(),
library_settings,
initial_scope=self.config.identify_scopes.get(str(root), "global"),
save_settings=save_identify_settings,
files=self._identify_files,
parent=self,
)
self._identify_files = ()
def start_identify_job(job: Any) -> None:
self._ads_refresh_root = root
self._watch_job(job, f"identification in {self._library_name(root)}")
dialog.job_started.connect(start_identify_job)
dialog.exec()
if self.job is None:
self._resume_foreground_watch()
def _persist_identify_defaults(
self, root: Path | None, scope: str, settings: IdentifySettings
) -> None:
"""Save Identify defaults and retain the existing scope transaction."""
self._guard_ads_recheck_settings()
if root is None:
if scope != "global":
raise ValueError("Choose a library first.")
if settings.skip_unresolved_directory and settings.output_unresolved:
output = Path(settings.output_unresolved).expanduser()
if not output.is_absolute() and all(
part == ".." for part in output.parts
):
raise ValueError(
"Cannot skip the unresolved directory because it contains "
"the watched root."
)
Config.save_identify_settings(settings)
return
root = root.expanduser().resolve()
registry = self.registries.get(str(root))
if registry is None:
raise ValueError("The selected library is unavailable.")
state = CentralPdfRegistryState(registry)
def save_gui() -> None:
self._sync_config_roots()
save_config(self.config)
persistence = _SettingsPersistenceCoordinator[IdentifySettings](
scopes=self.config.identify_scopes,
root_key=str(root),
load_global=Config.load_identify_settings,
save_global=Config.save_identify_settings,
load_library=lambda: state.identify_settings,
save_library=state.set_identify_settings,
save_gui=save_gui,
)
saved = settings
if scope == "library" and settings.output_unresolved:
output = Path(settings.output_unresolved).expanduser()
if output.is_absolute():
try:
output = output.resolve().relative_to(root)
except ValueError:
pass
saved = settings.with_overrides(output_unresolved=str(output))
WatchIdentifyPolicy.from_settings(root, saved)
persistence.persist(scope, saved)
def _edit_user_profile(self) -> None:
"""Edit the global identity used for publication prefetching."""
dialog = UserProfileDialog(
Config.load_user_profile(),
save_profile=Config.save_user_profile,
parent=self,
)
if dialog.exec() == QDialog.Accepted:
self.status.showMessage("Publication profile saved", 5000)
self._sync_welcome()
def _open_cv_generator(self) -> None:
"""Open the CV Generator, which saves its settings in cv.yaml."""
CVDialog(self).exec()
def _edit_token(self) -> None:
"""Open the ADS token dialog and refresh the welcome status."""
TokenDialog(self).exec()
self._sync_welcome()
def _sync_welcome(self) -> None:
"""Show the welcome guide instead of empty panes when no library exists."""
empty = self.current_root is None
self.welcome.setVisible(empty)
self.splitter.setVisible(not empty)
if not empty:
return
try:
profile = Config.load_user_profile()
profile_saved = bool(profile.orcid or profile.author_names)
except (OSError, ValueError):
profile_saved = False
try:
source = Config.token_source()
except OSError:
source = None
self.welcome.set_status(token_source=source, profile_saved=profile_saved)
def _refresh_selected_ads(self) -> None:
"""Refresh and rename one selected record from ADS metadata."""
if self.combined.enabled:
target = self._selected_edit_target()
if target is None:
return
root, record, _path = target
else:
root = self.current_root
if root is None:
return
record = self._selected_record()
metadata = record.get("metadata") if isinstance(record, dict) else {}
bibcode = metadata.get("bibcode") if isinstance(metadata, dict) else None
if bibcode is None and isinstance(record, dict):
bibcode = record.get("bibcode")
if not bibcode:
QMessageBox.information(
self,
"Refresh ADS record",
"Select a record with a bibcode first.",
)
return
self._start_ads_refresh(
"--bibcode",
self._scalar_metadata_value(bibcode),
"--relative-path",
str(record["relative_path"]),
"--citation-metrics",
selection_id=int(record["location_id"]),
root=root,
)
def _update_from_identifier(
self,
initial_identifier: str,
*,
root: Path | None = None,
path: Path | None = None,
reviewed_candidate: Notice | None = None,
_identifier_entered: bool = False,
) -> bool:
"""Ask for an identifier to update and rename one PDF.
Parameters
----------
initial_identifier
Identifier to show in the input field.
root
Library that owns the PDF. Default is the current library.
path
PDF path inside `root`. Default is the selected file.
reviewed_candidate
Explicit Use This ID action bound to the reviewed PDF and candidate.
Skip text entry and the duplicate identity confirmation.
_identifier_entered
Use an identifier already entered before the recheck stopped.
Returns
-------
bool
`True` when the update starts or waits for another operation to stop.
"""
if self.combined.enabled and root is None and path is None:
target = self._selected_edit_target()
if target is None:
return False
root, _record, path = target
target_root = (
root.expanduser().resolve() if root is not None else self.current_root
)
if self.combined.enabled and (root is None or path is None):
return False
if (
target_root is None
or str(target_root) not in self.registries
or self._close_requested
):
return False
priority_update = self._can_preempt_ads_recheck()
if not priority_update and (
self.job is not None
or self._ads_recheck_active_root is not None
or self._pending_watch_dialog is not None
or self._pending_ads_refresh is not None
or self._pending_priority_edit is not None
or self._local_refresh_context is not None
or self._manual_metadata_target is not None
or self._edition_merge_target is not None
):
self.status.showMessage(self._operation_wait_message("identifying a PDF"))
return False
untracked = False
if path is None:
record = self._selected_record()
else:
try:
relative_path = path.resolve().relative_to(target_root).as_posix()
except ValueError:
return False
record = self.registries[str(target_root)].find_file_by_current_path(
target_root, relative_path
)
if not path.is_file() or (record is not None and record.get("missing_at")):
QMessageBox.information(
self,
"PDF unavailable",
f"This PDF is no longer available at {path}. Refresh the library.",
)
return False
if record is None:
record = {"relative_path": relative_path, "current_path": relative_path}
untracked = True
if record is None:
return False
if reviewed_candidate is None and not _identifier_entered:
bibcode, accepted = QInputDialog.getText(
self,
"Update by identifier",
f"PDF: {record['relative_path']}\n\n"
"Enter a Bibcode, DOI, arXiv ID, or one of these reference formats:\n\n"
"1. {authorLastName}-{year}-{bibstem}\n"
" {authorLastName} = an author's last name\n"
" {year} = the four-digit publication year\n"
" {bibstem} = the ADS journal abbreviation\n"
" Example: Last-2020-Natur (Natur means Nature)\n\n"
"2. {bibstem}-{volume}-{page}\n"
" {volume} = the journal volume number\n"
" {page} = the starting page or article number\n"
" Example: Sci-123-45 (Sci means Science)\n\n"
"You can replace {bibstem} with the full journal name in either format.\n"
"Examples: Last-2020-Nature or Science-123-45\n"
"Separate fields with hyphens, underscores, or spaces.\n"
"Reference hints search ADS. Review any conflict after the lookup.\n"
"Select the correct paper from the ADS matches.\n"
"If no reference matches, enter a DOI or Bibcode.\n\n"
"The update applies saved Rename settings to this PDF.\n"
"DOI and arXiv lookups can use Crossref or DataCite if ADS has no match.",
text=initial_identifier,
)
else:
bibcode, accepted = initial_identifier, True
normalized = bibcode.strip()
if not accepted or not normalized:
return False
pdf_path = target_root / str(
record.get("current_path") or record["relative_path"]
)
if priority_update:
return self._queue_edit_after_recheck(
target_root, pdf_path, record, normalized, reviewed_candidate
)
try:
own_identity = extract_pdf_own_identity(pdf_path)
except IdentityValidationError as exc:
QMessageBox.warning(self, "PDF identity unavailable", str(exc))
return False
if reviewed_candidate is not None and (
normalized != reviewed_candidate.candidate_bibcode
or own_identity.sha256 != reviewed_candidate.candidate_sha256
):
QMessageBox.warning(
self,
"PDF changed",
"The PDF changed after this candidate was found. Retry identification before using its ID.",
)
return False
arguments = [
"--bibcode" if reviewed_candidate is not None else "--identifier",
normalized,
"--relative-path",
str(record["relative_path"]),
]
if reviewed_candidate is None:
arguments.append("--offer-conflict-review")
if reviewed_candidate is not None:
if reviewed_candidate.candidate_confirmable:
arguments.append("--confirm-identity")
arguments.extend(
[
"--expected-sha256",
str(reviewed_candidate.candidate_sha256),
"--expected-identity",
str(reviewed_candidate.candidate_fingerprint),
]
)
elif not own_identity.has_identifier:
if own_identity.blocking_errors:
QMessageBox.warning(
self,
"PDF identity unavailable",
"The PDF could not be read for identification. "
"Open the PDF and check that it is readable before retrying.",
)
return False
# The user chose this PDF and supplied the identifier. Known
# conflicts are still rejected by the identity validator.
arguments.extend(("--confirm-identity", "--review-reference"))
if (
untracked
or not record.get("sha256")
or (
reviewed_candidate is not None
and record.get("sha256") != own_identity.sha256
)
):
try:
if sha256_file(pdf_path) != own_identity.sha256:
raise ValueError(
"The PDF changed during review. Open Update by ID again."
)
registry = self.registries[str(target_root)]
registry.upsert_file(
root=target_root,
relative_path=str(record["relative_path"]),
current_path=str(record["current_path"]),
sha256=own_identity.sha256,
size_bytes=own_identity.size,
mtime_ns=own_identity.mtime_ns,
status="unresolved",
)
record = registry.find_file_by_current_path(
target_root, str(record["current_path"])
)
if record is None:
raise ValueError(
"The PDF could not be registered for the identifier lookup."
)
except (OSError, ValueError, sqlite3.Error) as exc:
QMessageBox.warning(self, "Could not update PDF", str(exc))
return False
self._candidate_update_notice = reviewed_candidate
self._start_ads_refresh(
*arguments,
selection_id=int(record["location_id"]),
ambiguous_relative_path=str(record["relative_path"]),
root=target_root,
)
started = self.job is not None or self._pending_ads_refresh is not None
if not started:
self._candidate_update_notice = None
return started
def _queue_edit_after_recheck(
self,
root: Path,
path: Path,
record: dict[str, Any],
identifier: str | None,
reviewed_candidate: Notice | None,
notice_key: str | None = None,
) -> bool:
"""Stop a recheck and retain the exact PDF selected for an edit."""
if self._ads_recheck_active_root is not None:
if not self._can_preempt_ads_recheck():
self.status.showMessage(self._operation_wait_message())
return False
elif self._detail_edit_block_reason():
self.status.showMessage(self._operation_wait_message())
return False
try:
content_sha256 = sha256_file(path)
except (OSError, ValueError) as exc:
QMessageBox.warning(self, "PDF unavailable", str(exc))
return False
self._pending_priority_edit = _PendingPriorityEdit(
root=root,
path=path,
location_id=record.get("location_id"),
attachment_id=record.get("attachment_id"),
catalog_sha256=record.get("sha256"),
content_sha256=content_sha256,
identifier=identifier,
reviewed_candidate=reviewed_candidate,
notice_key=notice_key,
)
self._ads_recheck_roots = []
self._ads_recheck_again = False
self._ads_recheck_retry_requested = False
self._sync_refresh_actions()
self.status.showMessage(
"Stopping the automatic ADS check before editing this PDF…"
)
if self._ads_recheck_job is not None:
self._ads_recheck_job.cancel()
elif self._ads_recheck_active_root is None:
QTimer.singleShot(0, self._run_pending_priority_edit)
return True
def _run_pending_priority_edit(self) -> None:
"""Revalidate the selected PDF after the recheck releases its catalog."""
pending = self._pending_priority_edit
if pending is None or self._close_requested:
return
if self._ads_recheck_active_root is not None:
return
if (
self.job is not None
or self._pending_watch_dialog is not None
or self._pending_ads_refresh is not None
or self._manual_metadata_target is not None
or self._edition_merge_target is not None
or self._local_refresh_context is not None
):
self._pending_priority_edit = None
self._sync_refresh_actions()
QMessageBox.warning(
self,
"Edit cancelled",
"Another operation started. Wait for it to finish, then try again.",
)
QTimer.singleShot(0, self._continue_ads_rechecks)
return
self._pending_priority_edit = None
self._sync_refresh_actions()
try:
registry = self.registries.get(str(pending.root))
if registry is None:
raise ValueError("The PDF's library is no longer available.")
relative = pending.path.resolve().relative_to(pending.root).as_posix()
record = registry.find_file_by_current_path(pending.root, relative)
expected = (
pending.location_id,
pending.attachment_id,
pending.catalog_sha256,
)
actual = (
(
record.get("location_id"),
record.get("attachment_id"),
record.get("sha256"),
)
if record is not None
else (None, None, None)
)
if (
actual != expected
or (record is not None and record.get("missing_at"))
or sha256_file(pending.path) != pending.content_sha256
):
raise ValueError(
"The PDF or its catalog entry changed during the recheck."
)
except (OSError, ValueError, sqlite3.Error) as exc:
QMessageBox.warning(
self,
"Edit cancelled",
f"{exc}\n\nRefresh the library and try again.",
)
else:
if pending.identifier is None:
self._manual_metadata_target = (
pending.root,
pending.path,
pending.notice_key,
)
self._request_foreground_dialog("manual_metadata", pending.root)
else:
self._update_from_identifier(
pending.identifier,
root=pending.root,
path=pending.path,
reviewed_candidate=pending.reviewed_candidate,
_identifier_entered=True,
)
QTimer.singleShot(0, self._continue_ads_rechecks)
def _enter_manual_metadata(self, payload: object = None) -> None:
"""Open local metadata entry for a warned or selected PDF."""
priority_edit = self._can_preempt_ads_recheck()
if not priority_edit and (
self._close_requested
or self.job is not None
or self._ads_recheck_active_root is not None
or self._pending_watch_dialog is not None
or self._pending_ads_refresh is not None
or self._pending_priority_edit is not None
or self._manual_metadata_target is not None
or self._local_refresh_context is not None
or self._edition_merge_target is not None
):
self.status.showMessage(
self._operation_wait_message("entering paper details")
)
return
if isinstance(payload, WatchEvent):
root = payload.root.expanduser().resolve()
path = payload.path
notice_key = self._watch_notice_key(payload)
else:
if self.combined.enabled:
target = self._selected_edit_target()
if target is None:
return
root, _record, path = target
else:
record = self._selected_record()
if record is None or self.current_root is None:
return
root = self.current_root.resolve()
path = root / str(record.get("current_path") or record["relative_path"])
notice_key = None
if str(root) not in self.registries:
return
if priority_edit:
try:
relative = path.resolve().relative_to(root).as_posix()
record = self.registries[str(root)].find_file_by_current_path(
root, relative
)
except (OSError, ValueError, sqlite3.Error) as exc:
QMessageBox.warning(self, "PDF unavailable", str(exc))
return
self._queue_edit_after_recheck(
root, path, record or {}, None, None, notice_key
)
return
self._manual_metadata_target = (root, path, notice_key)
self._request_foreground_dialog("manual_metadata", root)
def _open_manual_metadata(self, root: Path) -> None:
"""Edit one PDF's metadata while its watcher cannot write to the catalog."""
target = self._manual_metadata_target
try:
if target is None or target[0] != root or str(root) not in self.registries:
return
_, path, notice_key = target
path = path.resolve()
relative_path = path.relative_to(root).as_posix()
if not path.is_file():
raise ValueError(f"This PDF is no longer available at {path}.")
snapshot = FileSnapshot.from_path(path)
# A same-size rewrite can keep every stat field on filesystems with
# coarse timestamps. The content digest detects that change.
opened_digest = sha256_file(path)
record = self.registries[str(root)].find_file_by_current_path(
root, relative_path
)
initial_metadata = (
deepcopy(record["metadata"]) if record is not None else {}
)
document_id = record.get("document_id") if record is not None else None
original_document = (
self.registries[str(root)].catalog.get_document(int(document_id))
if document_id is not None
else None
)
if document_id is not None and original_document is None:
raise ValueError("The PDF's paper record is no longer available.")
def save_manual_metadata(changes: dict[str, Any]) -> None:
"""Merge changed fields with current metadata and preserve the PDF."""
registry = self.registries[str(root)]
latest = registry.find_file_by_current_path(root, relative_path)
if (record is None) != (latest is None) or (
record is not None
and latest is not None
and any(
latest.get(key) != record.get(key)
for key in ("location_id", "attachment_id", "document_id")
)
):
raise ValueError(
"The PDF's catalog record changed. Reopen the details dialog."
)
if (
FileSnapshot.from_path(path) != snapshot
or sha256_file(path) != opened_digest
):
raise ValueError(
"The PDF changed while this dialog was open. Reopen the details dialog."
)
if latest is not None:
missing = object()
current_metadata = latest.get("metadata") or {}
if any(
initial_metadata.get(key, missing)
!= current_metadata.get(key, missing)
for key in changes
):
raise ValueError(
"A field changed while this dialog was open. "
"Reopen the details dialog and retry."
)
metadata = deepcopy(latest["metadata"] if latest is not None else {})
metadata.update(changes)
metadata[LOCKED_METADATA_FIELDS_KEY] = sorted(
set(metadata.get(LOCKED_METADATA_FIELDS_KEY, [])) | set(changes)
)
bibcode = self._scalar_metadata_value(metadata.get("bibcode"))
provider_document = (
registry.find_ads_document_by_bibcode(bibcode) if bibcode else None
)
metadata[IDENTITY_KEY] = assert_mutation_allowed(
path,
metadata,
expected_sha256=opened_digest,
manual_confirmation=True,
provider_document=provider_document,
)
if latest is None:
digest = sha256_file(path)
if (
digest != opened_digest
or FileSnapshot.from_path(path) != snapshot
):
raise ValueError(
"The PDF changed while saving. Reopen the details dialog."
)
registry.upsert_file(
root=root,
relative_path=relative_path,
current_path=relative_path,
sha256=digest,
size_bytes=snapshot.size_bytes,
mtime_ns=snapshot.mtime_ns,
status="unresolved",
metadata=metadata,
)
else:
metadata["status"] = latest["status"]
try:
saved_metadata = registry.replace_file_metadata_if_unchanged(
root,
relative_path,
metadata,
latest,
expected_document=original_document,
changed_fields=set(changes) | {LOCKED_METADATA_FIELDS_KEY},
)
except RuntimeError as exc:
raise ValueError("The library is busy. Try again.") from exc
if not saved_metadata:
raise ValueError(
"The PDF or its metadata changed. Reopen the details dialog."
)
saved = registry.find_file_by_current_path(root, relative_path)
self.refresh(root)
if (
self.combined.enabled or self.current_root == root
) and saved is not None:
self._select_file_id(int(saved["location_id"]))
if notice_key is not None:
self.notice_bar.dismiss(notice_key)
self.status.showMessage(f"Saved paper details: {path.name}")
dialog = ManualMetadataDialog(
path,
initial_metadata,
save_metadata=save_manual_metadata,
parent=self,
)
dialog.exec()
except (OSError, ValueError, RuntimeError, sqlite3.Error) as exc:
QMessageBox.warning(self, "Could not enter paper details", str(exc))
finally:
self._manual_metadata_target = None
self._resume_foreground_watch()
self._sync_refresh_actions()
def _save_detail_metadata(self, metadata: dict[str, Any]) -> None:
"""Persist one selected file's edited metadata and retain its selection."""
block_reason = self._detail_edit_block_reason()
if block_reason:
self.status.showMessage(block_reason)
self._sync_detail_and_selection_actions()
return
if self.combined.enabled:
target = self._selected_edit_target()
if target is None:
return
root, record, _path = target
else:
record = self._selected_record()
root = self.current_root
if record is None or root is None:
return
location_id = int(record["location_id"])
registry = self.registries[str(root)]
snapshot = self._detail_document_snapshot
if snapshot is None or snapshot[:3] != (
str(root),
location_id,
record.get("document_id"),
):
QMessageBox.warning(
self,
"Metadata update failed",
"The paper record could not be checked. Select the PDF and retry.",
)
return
latest = registry.find_location_by_id(location_id)
if latest is None:
QMessageBox.warning(
self,
"Metadata update failed",
"The selected file is no longer present in the registry.",
)
return
if any(
latest.get(key) != record.get(key)
for key in ("attachment_id", "current_path", "sha256", "metadata")
):
QMessageBox.warning(
self,
"Metadata update failed",
"The PDF or its metadata changed. Select it again and retry the edit.",
)
self._detail_document_snapshot = None
self.refresh(root)
self._select_file_id(location_id)
return
relative_path = str(latest["relative_path"])
try:
changed_fields = self.detail._pending_metadata_changed_fields
previous = latest.get("metadata") or {}
if changed_fields is not None:
edited = metadata
metadata = deepcopy(previous)
for key in changed_fields:
if key in edited:
metadata[key] = deepcopy(edited[key])
else:
metadata.pop(key, None)
identity_fields = (
"bibcode",
"doi",
"arxiv",
"arxiv_id",
"identifier",
"alternate_bibcode",
)
identity_changed = any(
metadata.get(key) != previous.get(key) for key in identity_fields
)
if identity_changed:
bibcode = self._scalar_metadata_value(metadata.get("bibcode"))
provider_document = (
registry.find_ads_document_by_bibcode(bibcode) if bibcode else None
)
metadata = deepcopy(metadata)
metadata[IDENTITY_KEY] = assert_mutation_allowed(
root / str(latest["current_path"]),
metadata,
expected_sha256=str(latest.get("sha256") or "") or None,
provider_document=provider_document,
)
else:
metadata = deepcopy(metadata)
metadata["status"] = latest["status"]
saved = registry.replace_file_metadata_if_unchanged(
root,
relative_path,
metadata,
record,
expected_document=snapshot[3],
changed_fields=changed_fields,
)
except (ValueError, OSError, RuntimeError, sqlite3.Error) as exc:
QMessageBox.warning(self, "Metadata update failed", str(exc))
self._detail_document_snapshot = None
self.refresh(root)
self._select_file_id(location_id)
return
if not saved:
QMessageBox.warning(
self,
"Metadata update failed",
"The PDF or its metadata changed. Select it again and retry the edit.",
)
self._detail_document_snapshot = None
self.refresh(root)
self._select_file_id(location_id)
return
self._detail_document_snapshot = None
self.refresh(root)
self._select_file_id(location_id)
def _set_detail_row_order(self, order: list[str]) -> None:
"""Retain the user-selected detail row order for the next launch."""
self.config.detail_row_order = list(order)
def _selected_record(self) -> dict[str, Any] | None:
"""Return the single current table record, if one is selected."""
return self._record_at_table_index(self.table.currentIndex())
def _source_table_index(self, index: QModelIndex) -> QModelIndex:
"""Map a visible table index to the underlying PDF model."""
if not index.isValid():
return QModelIndex()
model = index.model()
if model is self.model:
return index
mapper = getattr(model, "mapToSource", None)
if callable(mapper):
return mapper(index)
return self.model.index(index.row(), index.column())
def _view_table_index(self, source: QModelIndex) -> QModelIndex:
"""Map an underlying PDF index to its visible table index."""
if not source.isValid():
return QModelIndex()
model = self.table.model()
if model is self.model:
return source
mapper = getattr(model, "mapFromSource", None)
if callable(mapper):
return mapper(source)
return model.index(source.row(), source.column())
def _record_at_table_index(self, index: QModelIndex) -> dict[str, Any] | None:
"""Return the PDF record for a visible table index."""
if (
index.isValid()
and index.model() is not self.model
and not callable(getattr(index.model(), "mapToSource", None))
):
return self.model.record_at(index.row())
source = self._source_table_index(index)
return self.model.record_at(source.row()) if source.isValid() else None
def _selected_edit_target(
self,
) -> tuple[Path, dict[str, Any], Path] | None:
"""Return one selected PDF and its owning library, or ``None``."""
rows = self._selected_view_rows()
index = self.table.currentIndex()
if len(rows) != 1:
return None
selected = self._source_table_index(rows[0])
current = self._source_table_index(index)
if (
not selected.isValid()
or not current.isValid()
or selected.row() != current.row()
):
return None
record = self.model.record_at(current.row())
if record is None:
return None
if self.combined.enabled:
root = self.combined.root_for(record)
path = self.combined.path_for(record)
else:
root = self.current_root
relative = str(record.get("current_path") or "")
path = root / relative if root is not None and relative else None
if root is None or path is None or str(root) not in self.registries:
return None
return root, record, path
@staticmethod
def _scalar_metadata_value(value: Any) -> str:
"""Return a single metadata value from a scalar or ADS-style list."""
if isinstance(value, (list, tuple)):
return str(value[0]).strip() if value else ""
return str(value).strip()
def _select_file_id(self, location_id: int) -> None:
"""Restore selection after a registry reload by location id."""
for row in range(self.model.rowCount()):
record = self.model.record_at(row)
if record is not None and record.get("location_id") == location_id:
index = self._view_table_index(self.model.index(row, 0))
if index.isValid():
self.table.setCurrentIndex(index)
return
def _link_stored_bibcodes(self) -> None:
"""Look up all known bibcodes without renaming or moving PDFs."""
if self.current_root is None or self.combined.enabled:
return
records = [
record
for record in self.registries[str(self.current_root)].iter_files()
if record.get("status") != "resolved" and record.get("bibcode")
]
count = sum(stored_bibcode_matches_file(record) for record in records)
skipped = len(records) - count
explanation = (
f" {skipped} other records need Identify because their saved metadata has no matching PDF digest."
if skipped
else ""
)
if (
QMessageBox.question(
self,
"Look Up Stored Bibcodes",
f"Look up stored bibcodes for {count} pending records? Cached ADS records are used first. "
"Existing metadata, locked fields, and history are retained. "
"PDFs are not renamed or moved." + explanation,
)
!= QMessageBox.Yes
):
return
self._start_ads_refresh("--all", "--no-rename", "--pending-only")
def _refresh_all_ads(self) -> None:
"""Confirm a batched ADS document refresh without per-paper metrics."""
if self.current_root is None or self.combined.enabled:
return
count = sum(
bool(record.get("bibcode"))
for record in self.registries[str(self.current_root)].iter_files()
)
if (
QMessageBox.question(
self,
"Refresh all ADS records",
(
f"Refresh ADS metadata for up to {count} records with bibcodes (including pending ADS lookups) and "
"rename their PDFs using saved settings?"
),
)
!= QMessageBox.Yes
):
return
self._start_ads_refresh("--all")
def _refresh_all_citation_metrics(self) -> None:
"""Confirm explicit per-paper requests for citation splits."""
if self.current_root is None or self.combined.enabled:
return
count = sum(
bool(record.get("bibcode"))
for record in self.registries[str(self.current_root)].iter_files()
)
if (
QMessageBox.question(
self,
"Refresh all citation splits",
(
f"Refresh ADS metadata, rename up to {count} records, and fetch "
"their refereed/non-refereed citation splits? ADS Metrics supplies "
f"these only per paper, so this sends up to {count} requests."
),
)
!= QMessageBox.Yes
):
return
self._start_ads_refresh("--all", "--citation-metrics")
def _start_ads_refresh(
self,
*args: str,
selection_id: int | None = None,
ambiguous_relative_path: str | None = None,
root: Path | None = None,
follow_ups: tuple[tuple[str, ...], ...] = (),
) -> None:
"""Start the CLI-backed ADS refresh and rename for the selected root.
Each argument list in `follow_ups` runs as another ``refresh-ads``
command after the first one succeeds.
"""
refresh_root = (
root.expanduser().resolve()
if root is not None
else (
self.current_root.expanduser().resolve() if self.current_root else None
)
)
if (
self._close_requested
or refresh_root is None
or str(refresh_root) not in self.registries
):
return
if (
self.job is not None
or self._pending_watch_dialog is not None
or self._pending_ads_refresh is not None
):
return
rename_scope = self.config.rename_scopes.get(str(refresh_root), "global")
output_unresolved = self._effective_unresolved_directory(refresh_root)
common = (
"--rename-scope",
rename_scope,
"--identify-scope",
self.config.identify_scopes.get(str(refresh_root), "global"),
"--output-unresolved",
str(output_unresolved),
)
refresh_args = (*common, *args)
self._pending_ads_follow_ups = tuple((*common, *extra) for extra in follow_ups)
self._request_foreground_dialog(
"ads_refresh",
refresh_root,
ads_refresh=(
refresh_root,
refresh_args,
selection_id,
ambiguous_relative_path,
),
)
def _effective_unresolved_directory(self, root: Path) -> Path:
"""Resolve the Identify destination used by an ADS refresh."""
settings = self._effective_identify_settings(root)
directory = (
Path(settings.output_unresolved).expanduser()
if settings.output_unresolved
else root / "unresolved"
)
if not directory.is_absolute():
directory = root / directory
return directory.resolve()
def _start_pending_ads_refresh(self, root: Path) -> None:
"""Launch a queued ADS refresh after its watcher pause is acknowledged."""
pending = self._pending_ads_refresh
self._pending_ads_refresh = None
if pending is None:
self._resume_foreground_watch()
return
pending_root, args, selection_id, ambiguous_relative_path = pending
if (
pending_root != root
or str(root) not in self.registries
or self._close_requested
):
self._resume_foreground_watch()
return
follow_ups = self._pending_ads_follow_ups
self._pending_ads_follow_ups = ()
self._launch_ads_refresh(
pending_root,
args,
selection_id=selection_id,
ambiguous_relative_path=ambiguous_relative_path,
follow_ups=follow_ups,
)
def _launch_ads_refresh(
self,
root: Path,
args: tuple[str, ...],
*,
selection_id: int | None,
ambiguous_relative_path: str | None,
follow_ups: tuple[tuple[str, ...], ...] = (),
) -> None:
"""Launch one ADS refresh with its snapshotted root and arguments."""
self._refresh_selection_id = selection_id
self._ads_refresh_root = root
self._ads_match_candidates = []
self._ambiguous_ads_match_context = (
(root, ambiguous_relative_path, selection_id)
if ambiguous_relative_path is not None
else None
)
self._conflict_review_context = None
self._conflict_review_payload = None
self._conflict_review_line_seen = False
if "--offer-conflict-review" in args:
try:
relative_path = args[args.index("--relative-path") + 1]
record = self.registries[str(root)].find_file_by_relative_path(
root, relative_path
)
if (
record is not None
and not record.get("missing_at")
and str(record.get("relative_path")) == relative_path
and (
selection_id is None
or record.get("location_id") == selection_id
)
):
self._conflict_review_context = _ConflictReviewContext(
root=root,
relative_path=relative_path,
current_path=str(record.get("current_path") or relative_path),
location_id=int(record["location_id"]),
)
except (IndexError, KeyError, TypeError, ValueError, sqlite3.Error):
logger.warning("Could not bind the PDF conflict review to its record")
job = CommandJob(
cli_argv("pdf", "refresh-ads", str(root), *args),
parent=self,
working_directory=root,
follow_up_argv=[
cli_argv("pdf", "refresh-ads", str(root), *extra)
for extra in follow_ups
],
)
job.output.connect(self._capture_ads_match_candidates)
job.output.connect(
lambda line: (
self._capture_identity_conflict(line) if self.job is job else None
)
)
self._watch_job(job, f"ADS update in {self._library_name(root)}")
job.start()
def _capture_ads_match_candidates(self, line: str) -> None:
"""Capture structured ADS ambiguity candidates emitted by the CLI."""
prefix = "ADS_MATCH_CANDIDATES_JSON:"
if not line.startswith(prefix):
return
try:
payload = json.loads(line.removeprefix(prefix))
except json.JSONDecodeError:
logger.warning("Could not parse ADS ambiguity candidates: %s", line)
return
if not isinstance(payload, list):
return
candidates: list[dict[str, str]] = []
for value in payload:
if not isinstance(value, dict):
continue
bibcode = value.get("bibcode")
if not isinstance(bibcode, str) or not bibcode:
continue
candidates.append(
{
"bibcode": bibcode,
"first_author": str(value.get("first_author") or ""),
"title": str(value.get("title") or ""),
"pdf_sha256": str(value.get("pdf_sha256") or ""),
"identity_fingerprint": str(
value.get("identity_fingerprint") or ""
),
"review_digest": str(value.get("review_digest") or ""),
}
)
self._ads_match_candidates = candidates
def _capture_identity_conflict(self, line: str) -> None:
"""Keep one complete conflict result from the active lookup."""
prefix = "PDF_IDENTITY_CONFLICT_JSON:"
if self._conflict_review_context is None or not line.startswith(prefix):
return
if self._conflict_review_line_seen:
self._conflict_review_payload = None
return
self._conflict_review_line_seen = True
def unique_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
result: dict[str, Any] = {}
for key, value in pairs:
if key in result:
raise ValueError("Duplicate key in conflict result")
result[key] = value
return result
try:
payload = json.loads(
line.removeprefix(prefix), object_pairs_hook=unique_object
)
except (ValueError, RecursionError):
logger.warning("Could not parse the PDF conflict result")
return
required = (
"identifier",
"identifier_option",
"identifiers",
"conflict",
"pdf_sha256",
"identity_fingerprint",
"review_digest",
"catalog_digest",
)
if not isinstance(payload, dict) or any(
not isinstance(payload.get(key), str) or not payload[key].strip()
for key in required
):
return
if payload["identifier_option"] not in {"--identifier", "--bibcode"}:
return
digest = payload["pdf_sha256"]
if len(digest) != 64 or any(c not in "0123456789abcdefABCDEF" for c in digest):
return
result = {key: payload[key] for key in required}
for field, fallback in (
("title", "Unknown title"),
("author", "Unknown author"),
):
value = payload.get(field)
if value is not None and not isinstance(value, str):
return
result[field] = value if value and value.strip() else fallback
if "year" in payload:
year = payload["year"]
if isinstance(year, bool) or not isinstance(year, (str, int)):
return
result["year"] = str(year)
self._conflict_review_payload = result
def _review_identity_conflict(
self,
context: _ConflictReviewContext,
payload: dict[str, str],
candidate_notice: Notice | None,
) -> None:
"""Ask about a conflict, then retry only while the selected PDF is unchanged."""
year = f"\nYear: {payload['year']}" if payload.get("year") else ""
message = (
f"PDF conflict:\n{payload['conflict']}\n\n"
f"Retrieved paper:\n{payload['title']}\n{payload['author']}"
f"{year}\n{payload['identifiers']}\n\nUse this result?"
)
question = QMessageBox(
QMessageBox.Question,
"Conflicting PDF identifiers",
message,
QMessageBox.Yes | QMessageBox.No,
self,
)
question.setTextFormat(Qt.PlainText)
question.setDefaultButton(QMessageBox.No)
if question.exec() != QMessageBox.Yes:
return
registry = self.registries.get(str(context.root))
if self._close_requested or registry is None or not context.root.is_dir():
QMessageBox.warning(
self,
"PDF unavailable",
"The PDF's library is unavailable. Add the library again. "
"Try Update by ID again.",
)
return
try:
record = registry.find_location_by_id(context.location_id)
if (
record is None
or record.get("missing_at")
or record.get("relative_path") != context.relative_path
or (record.get("current_path") or record.get("relative_path"))
!= context.current_path
):
raise FileNotFoundError
pdf_path = context.root / context.current_path
if not pdf_path.is_file():
raise FileNotFoundError
if sha256_file(pdf_path).lower() != payload["pdf_sha256"].lower():
QMessageBox.warning(
self,
"PDF changed",
"The PDF changed during review. Open Update by ID again.",
)
return
except (OSError, sqlite3.Error):
QMessageBox.warning(
self,
"PDF unavailable",
"The PDF moved or is unavailable. Refresh the library. "
"Try Update by ID again.",
)
return
self._candidate_update_notice = candidate_notice
self._start_ads_refresh(
payload["identifier_option"],
payload["identifier"],
"--relative-path",
context.relative_path,
"--confirm-conflicting-identity",
"--expected-sha256",
payload["pdf_sha256"],
"--expected-identity",
payload["identity_fingerprint"],
"--expected-review-digest",
payload["review_digest"],
"--expected-catalog-digest",
payload["catalog_digest"],
selection_id=context.location_id,
root=context.root,
)
if self.job is None and self._pending_ads_refresh is None:
self._candidate_update_notice = None
def _choose_ambiguous_ads_match(
self,
candidates: list[dict[str, str]],
relative_path: str,
selection_id: int | None,
*,
root: Path | None = None,
) -> None:
"""Let the user choose one coordinate-match candidate by Bibcode."""
labels = [
"{bibcode} — {first_author} — {title}".format(
bibcode=candidate["bibcode"],
first_author=candidate["first_author"] or "Unknown author",
title=candidate["title"] or "Untitled",
)
for candidate in candidates
]
choice, accepted = QInputDialog.getItem(
self,
"Review ADS match" if len(candidates) == 1 else "Multiple ADS matches",
f"Choose the correct record for {relative_path}:",
labels,
editable=False,
)
if not accepted:
return
candidate = candidates[labels.index(choice)]
bibcode = candidate["bibcode"]
arguments = [
"--bibcode",
bibcode,
"--relative-path",
relative_path,
"--offer-conflict-review",
]
target_root = (
root.expanduser().resolve()
if root is not None
else (
self.current_root.expanduser().resolve() if self.current_root else None
)
)
if candidate.get("pdf_sha256") and (
target_root is None or str(target_root) not in self.registries
):
QMessageBox.warning(
self,
"PDF unavailable",
"The PDF's library is no longer available. Refresh it and retry.",
)
return
if target_root is not None and str(target_root) in self.registries:
record = self.registries[str(target_root)].find_file_by_relative_path(
target_root, relative_path
)
pdf_path = target_root / str(
(record.get("current_path") if record else None) or relative_path
)
if candidate.get("pdf_sha256") and (
record is None or not pdf_path.is_file()
):
QMessageBox.warning(
self,
"PDF unavailable",
"The PDF moved or is no longer available. Refresh its library and retry.",
)
return
if pdf_path.is_file():
try:
own_identity = extract_pdf_own_identity(pdf_path)
except IdentityValidationError as exc:
QMessageBox.warning(self, "PDF identity unavailable", str(exc))
return
expected_sha256 = candidate.get("pdf_sha256") or ""
expected_identity = candidate.get("identity_fingerprint") or ""
if expected_sha256 and expected_identity:
if own_identity.sha256 != expected_sha256:
QMessageBox.warning(
self,
"PDF changed",
"The PDF changed during the lookup. Open Update by ID again.",
)
return
arguments.extend(
[
"--expected-sha256",
expected_sha256,
"--expected-identity",
expected_identity,
]
)
if candidate.get("review_digest"):
arguments.extend(
["--expected-review-digest", candidate["review_digest"]]
)
if not own_identity.has_identifier:
arguments.append("--confirm-identity")
self._start_ads_refresh(
*arguments,
selection_id=selection_id,
root=root,
)
def _rename(self) -> None:
if self.current_root is None or self.combined.enabled:
return
self._rename_library(self.current_root)
def _rename_library(self, root: Path) -> None:
"""Open Rename for one explicitly identified library."""
root = root.expanduser().resolve()
if (
str(root) not in self.registries
or (self.combined.enabled and str(root) not in self.combined.selected)
or self._detail_edit_block_reason()
):
return
self._refresh_selection_id = None
self._request_foreground_dialog("rename", root)
def _open_rename_dialog(self, root: Path) -> None:
registry = self.registries[str(root)]
state = CentralPdfRegistryState(registry)
def save_rename_settings(scope: str, settings: RenameSettings) -> None:
self._persist_rename_defaults(root, scope, settings)
dialog = RenameDialog(
root,
Config.load_rename_settings(),
state.rename_settings,
initial_scope=self.config.rename_scopes.get(str(root), "global"),
identify_settings=self._effective_identify_settings(root),
save_settings=save_rename_settings,
parent=self,
)
def start_rename_job(job: Any) -> None:
self._ads_refresh_root = root
self._watch_job(job, f"rename in {self._library_name(root)}")
dialog.job_started.connect(start_rename_job)
dialog.exec()
if self.job is None:
self._resume_foreground_watch()
def _persist_rename_defaults(
self, root: Path | None, scope: str, settings: RenameSettings
) -> None:
"""Save Rename defaults and retain the existing scope transaction."""
self._guard_ads_recheck_settings()
if root is None:
if scope != "global":
raise ValueError("Choose a library first.")
Config.save_rename_settings(settings)
return
root = root.expanduser().resolve()
registry = self.registries.get(str(root))
if registry is None:
raise ValueError("The selected library is unavailable.")
state = CentralPdfRegistryState(registry)
def save_gui() -> None:
self._sync_config_roots()
save_config(self.config)
persistence = _SettingsPersistenceCoordinator[RenameSettings](
scopes=self.config.rename_scopes,
root_key=str(root),
load_global=Config.load_rename_settings,
save_global=Config.save_rename_settings,
load_library=lambda: state.rename_settings,
save_library=state.set_rename_settings,
save_gui=save_gui,
)
persistence.persist(scope, settings)
def _guard_ads_recheck_settings(self) -> None:
"""Protect the settings snapshot until the recheck worker stops."""
if self._ads_recheck_job is not None:
raise ValueError(
f"{self._operation_wait_message()} Then save these settings."
)
def _resume_foreground_watch(self) -> None:
QTimer.singleShot(0, self._review_next_edition_merge)
root = self._foreground_paused_root
if root is None:
return
self._foreground_paused_root = None
self.watch_controller.resume_writes(root)
self._sync_refresh_actions()
def _watch_job(self, job: Any, description: str = "") -> None:
job.setParent(self)
self.job = job
self._job_description = description
self._sync_ads_library_action()
self.log_dock.show()
self.identify_action.setEnabled(False)
self.rename_action.setEnabled(False)
self.refresh_selected_ads_action.setEnabled(False)
self.link_bibcodes_action.setEnabled(False)
self.refresh_all_ads_action.setEnabled(False)
self.refresh_all_citation_metrics_action.setEnabled(False)
self.cancel_job_action.setEnabled(True)
self._sync_job_indicator()
job.output.connect(self.log.appendPlainText)
job.failed.connect(
lambda message: self.log.appendPlainText(f"Failed to start: {message}")
)
job.failed.connect(self._job_failed)
job.finished.connect(lambda code: self._job_finished(code))
def _cancel_job(self) -> None:
"""Request cancellation of the active job."""
if self.job is not None:
self._conflict_review_context = None
self._conflict_review_payload = None
if self.job is self._ads_recheck_job:
if not self._close_requested:
self._ads_recheck_roots = []
self._ads_recheck_again = False
self._ads_recheck_retry_requested = False
self.job.cancel()
self.cancel_job_action.setEnabled(False)
def _job_failed(self, message: str) -> None:
"""Restore controls when a subprocess cannot be started."""
self.identify_action.setEnabled(True)
self.rename_action.setEnabled(True)
self.refresh_selected_ads_action.setEnabled(True)
self.link_bibcodes_action.setEnabled(True)
self.refresh_all_ads_action.setEnabled(True)
self.refresh_all_citation_metrics_action.setEnabled(True)
self.cancel_job_action.setEnabled(False)
self.job = None
self._sync_job_indicator()
self._local_refresh_context = None
self._local_refresh_result = None
self._identify_rename_context = None
self._identify_rename_files = ()
if self._guided_identify_context is not None:
root = self._guided_identify_context.root
self.notice_bar.enqueue(
Notice(
f"identify:{root}",
f"Identification could not start: {message}",
severity="error",
root=root,
)
)
self._guided_identify_context = None
self._guided_identify_commands.clear()
self._sync_ads_library_action()
self._refresh_selection_id = None
self._ads_refresh_root = None
self._ads_match_candidates = []
self._ambiguous_ads_match_context = None
self._conflict_review_context = None
self._conflict_review_payload = None
self._conflict_review_line_seen = False
self._resume_foreground_watch()
if self._close_requested:
self.close()
def _job_finished(self, code: int) -> None:
identify_rename_context = self._identify_rename_context
self._identify_rename_context = None
self._identify_rename_files = ()
candidate_notice = self._candidate_update_notice
self._candidate_update_notice = None
identify_context = self._guided_identify_context
self._guided_identify_context = None
self._guided_identify_commands.clear()
local_context = self._local_refresh_context
local_result = self._local_refresh_result
self._local_refresh_context = None
self._local_refresh_result = None
candidates = self._ads_match_candidates
context = self._ambiguous_ads_match_context
conflict_context = self._conflict_review_context
conflict_payload = self._conflict_review_payload
refresh_root = (
identify_rename_context.root
if identify_rename_context
else identify_context.root
if identify_context
else local_context.root
if local_context
else self._ads_refresh_root
)
self._ads_refresh_root = None
self._ads_match_candidates = []
self._ambiguous_ads_match_context = None
self._conflict_review_context = None
self._conflict_review_payload = None
self._conflict_review_line_seen = False
if code != 0:
self.log.appendPlainText(f"Command exited with status {code}")
self.identify_action.setEnabled(True)
self.rename_action.setEnabled(True)
self.refresh_selected_ads_action.setEnabled(True)
self.link_bibcodes_action.setEnabled(True)
self.refresh_all_ads_action.setEnabled(True)
self.refresh_all_citation_metrics_action.setEnabled(True)
self.cancel_job_action.setEnabled(False)
self.job = None
self._sync_job_indicator()
self._sync_ads_library_action()
self._resume_foreground_watch()
if refresh_root is None or str(refresh_root) in self.registries:
self.refresh(refresh_root)
if (
identify_rename_context is not None
and str(identify_rename_context.root) in self.registries
):
self._restore_library_selection(identify_rename_context)
self.status.showMessage(
"Identify and rename finished. See the job log for changes and skipped files."
if code == 0
else "Identify and rename stopped. See the job log for details."
)
if local_context is not None and str(local_context.root) in self.registries:
self._restore_library_selection(local_context)
if local_result is not None:
summary = (
f"Refreshed {local_context.root.name}: {local_result.added} added, "
f"{local_result.updated} modified, {local_result.moved} moved, "
f"{local_result.restored} restored, {local_result.missing} missing "
"(history retained)."
)
self.status.showMessage(summary)
self.log.appendPlainText(summary)
elif code != 0:
self.status.showMessage(
"Local refresh failed; see the job log for details."
)
if (
identify_context is not None
and str(identify_context.root) in self.registries
):
self._restore_library_selection(identify_context)
if code == 0:
registry = self.registries[str(identify_context.root)]
linked = self._safe_count(registry, identification="ads_linked")
external = self._safe_count(
registry, identification="identified_without_ads"
)
pending = self._safe_count(
registry, identification="needs_identification"
)
summary = (
f"Identification finished in {identify_context.root.name}: "
f"{linked if linked is not None else 'unknown'} ADS linked, "
f"{external if external is not None else 'unknown'} identified without ADS, "
f"{pending if pending is not None else 'unknown'} need identification. "
"Review the results, then use Rename… if needed."
)
else:
summary = "Identification stopped. See the job log for details."
self.status.showMessage(summary)
self.notice_bar.enqueue(
Notice(
f"identify:{identify_context.root}",
summary,
severity="info" if code == 0 else "warning",
root=identify_context.root,
)
)
self._sync_refresh_actions()
if (
candidate_notice is not None
and code == 0
and refresh_root is not None
and str(refresh_root) in self.registries
and self._refresh_selection_id is not None
):
saved = self.registries[str(refresh_root)].find_location_by_id(
self._refresh_selection_id
)
metadata = saved.get("metadata", {}) if saved is not None else {}
proof = metadata.get(IDENTITY_KEY, {})
if (
saved is not None
and saved.get("status") == "resolved"
and proof.get("sha256") == candidate_notice.candidate_sha256
and proof.get("identity") == candidate_notice.candidate_fingerprint
):
self.notice_bar.dismiss(candidate_notice.key)
if (
not self.combined.enabled
and self._refresh_selection_id is not None
and (refresh_root is None or refresh_root == self.current_root)
):
self._select_file_id(self._refresh_selection_id)
self._refresh_selection_id = None
if (
conflict_context is not None
and conflict_payload is not None
and code >= 0
and not self._close_requested
):
self._review_identity_conflict(
conflict_context, conflict_payload, candidate_notice
)
elif (
candidates
and context is not None
and code == 0
and not self._close_requested
):
root, relative_path, selection_id = context
if str(root) in self.registries:
self._choose_ambiguous_ads_match(
candidates, relative_path, selection_id, root=root
)
if self._close_requested:
self.close()
[docs]
def closeEvent(self, event: Any) -> None:
"""Persist window state and close registry connections."""
self._ads_recheck_timer.stop()
self._ads_recheck_retry_timer.stop()
self._ads_recheck_roots = None
self._ads_recheck_again = False
self._pending_priority_edit = None
if (
self._pending_watch_dialog is not None
and self._pending_watch_dialog[0] == "ads_recheck"
):
root = self._pending_watch_dialog[1]
self._pending_watch_dialog = None
self._ads_recheck_active_root = None
self.watch_controller.resume_writes(root)
if self._ads_token_check is not None and self._ads_token_check.running:
self._close_requested = True
self._ads_token_check.cancel()
if self.job is not None and self.job.running:
self._cancel_job()
event.ignore()
return
if self.job is not None and self.job.running:
self._close_requested = True
self._cancel_job()
event.ignore()
return
if self._ads_library_export_dialog is not None:
self._close_requested = True
self._ads_library_export_dialog.close()
if self._ads_library_export_dialog is not None:
event.ignore()
return
if self._pending_ads_refresh is not None:
pending_root = self._pending_ads_refresh[0]
self._pending_ads_refresh = None
self._pending_watch_dialog = None
self.watch_controller.resume_writes(pending_root)
if not self.watch_controller.stop_all(wait_ms=0):
# A watcher stops after its current PDF. Keep the window responsive
# and close when every watcher has stopped. Network requests have
# timeouts, so the current PDF finishes.
self._close_requested = True
self.status.showMessage(
"Closing after the watcher finishes the current PDF…"
)
QTimer.singleShot(WATCH_STOP_CHECK_MS, self.close)
event.ignore()
return
self._close_requested = True
self.update_checker.cancel()
self.update_downloader.cancel()
if self._about_dialog is not None:
self._about_dialog.close()
if self._update_dialog is not None:
self._update_dialog.close()
self.search_timer.stop()
self._watch_refresh_timer.stop()
self._pending_watch_refresh_roots.clear()
self._watch_refresh_selection_id = None
self._pending_watch_dialog = None
self._manual_metadata_target = None
self._edition_merge_target = None
self._pending_edition_merges.clear()
self._local_refresh_context = None
self._guided_identify_context = None
self._guided_identify_commands.clear()
self._sync_config_roots()
self.config.last_root = str(self.current_root) if self.current_root else None
self.config.last_view = self.current_view.value
self._remember_splitter_sizes()
self.config.window_geometry = bytes(self.saveGeometry().toBase64()).decode()
self.config.window_state = bytes(self.saveState().toBase64()).decode()
self.config.column_widths = {
column.key: self.table.columnWidth(index)
for index, column in enumerate(self.columns)
}
self.config.table_column_order = [
self.columns[self.table.header().logicalIndex(index)].key
for index in range(self.table.header().count())
]
self.config.hidden_table_columns = [
column.key
for index, column in enumerate(self.columns)
if self.table.isColumnHidden(index)
]
self.config.sort_column = self.current_query.order_by
self.config.sort_descending = self.current_query.descending
self.config.secondary_sort_column = self.current_query.secondary_order_by
self.config.tertiary_sort_column = self.current_query.tertiary_order_by
self.config.detail_row_order = self.detail.saved_row_order()
try:
save_config(self.config)
except OSError as exc:
logger.warning("Could not save GUI configuration: %s", exc)
for registry in self.registries.values():
registry.close()
super().closeEvent(event)